From a4c93a51226d4c4a155334339fa169beb4603e4c Mon Sep 17 00:00:00 2001 From: mixeme Date: Fri, 19 Jun 2026 08:22:35 +0300 Subject: [PATCH] T3.4: Convert scheduler to drive app.Service; inject Clock MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The scheduler no longer shares a *[]domain.Job with the GUI. It is now a thin timing loop with an injected Clock that calls a tick callback; the application service is the sole writer of job and runtime state. - scheduler: add Clock interface + RealClock (clock.go); strip all job logic from scheduler.go (NewScheduler(clock, tick)); rewrite tests to cover the loop with a fake clock. - app.Service: add RunDue(now) (pause + one-run-per-tick policy, records back through the service) and Start(Clock)/Stop() owning a cancelable run context; prime each job's first next-run at construction. Capture the run context under the lock for executeRun. - gui: talk only to app.Service (no shared state) — Open() the service, keep a refreshed snapshot, route every mutation through the service, and react to changes via a single Subscribe listener. - Tests: add RunDue (due/not-due/paused) and Start-drives-RunDue cases. Verified with CGO + MSYS2 UCRT64: go vet ./... clean, go test -race ./... green (GUI included), full module builds. Co-Authored-By: Claude Opus 4.8 --- docs/REFACTORING.md | 2 +- src/app/operations.go | 43 +++++- src/app/operations_test.go | 102 ++++++++++++ src/app/service.go | 76 +++++++-- src/gui/app.go | 236 ++++++++++++---------------- src/scheduler/clock.go | 55 +++++++ src/scheduler/scheduler.go | 265 ++++---------------------------- src/scheduler/scheduler_test.go | 122 ++++++++------- 8 files changed, 459 insertions(+), 442 deletions(-) create mode 100644 src/scheduler/clock.go diff --git a/docs/REFACTORING.md b/docs/REFACTORING.md index 74cb86b..d9b56f2 100644 --- a/docs/REFACTORING.md +++ b/docs/REFACTORING.md @@ -263,7 +263,7 @@ Track progress here. Mark tasks complete as they land and pass review. - [x] T3.1 — Create `src/app/service.go`; owns state behind mutex - [x] T3.2 — Add `src/app/events.go`; Event types + Observer - [x] T3.3 — Add state-mutating operations to service -- [ ] T3.4 — Convert `scheduler` to use service; inject Clock +- [x] T3.4 — Convert `scheduler` to use service; inject Clock - [ ] T3.5 — Move display helpers to `src/app/format.go` - [ ] T3.6 — Add `src/app` unit tests (no Fyne) diff --git a/src/app/operations.go b/src/app/operations.go index ad3da03..cb9ba68 100644 --- a/src/app/operations.go +++ b/src/app/operations.go @@ -1,6 +1,7 @@ package app import ( + "context" "errors" "fmt" "strings" @@ -200,6 +201,40 @@ func (s *Service) RunNow(id int) error { return err } +// RunDue is the scheduler's per-tick entry point: it starts whatever is due at +// the given time. It is a no-op while globally paused. At most one job is started +// per call so scheduled shell commands in this single process do not overlap; a +// job already running is skipped. Run results are recorded back through the +// Service, so the Service stays the sole writer of job and runtime state. The +// time is supplied by the scheduler's clock, which lets tests drive +// due-evaluation deterministically. +func (s *Service) RunDue(now time.Time) { + s.mu.Lock() + var startedID int + if !s.paused { + for index := range s.jobs { + job := &s.jobs[index] + runtime := s.runtimeForLocked(job) + if !job.Enabled || runtime.NextDue.IsZero() || now.Before(runtime.NextDue) { + continue + } + if runtime.LastState == "Running" { + continue + } + // Async save errors cannot be returned to a caller here; surfacing them + // is deferred to T5.1 with the rest of the swallowed saves. + _ = s.startRunLocked(job, runtime, "Schedule") + startedID = job.ID + break + } + } + s.mu.Unlock() + + if startedID != 0 { + s.emit(JobChanged{JobID: startedID}) + } +} + // UpdateSettings validates and persists a new application configuration. The // loaded jobs are re-saved because the jobs directory may have changed, and log // cleanup runs so a tightened retention policy takes effect immediately. @@ -239,14 +274,16 @@ func (s *Service) startRunLocked(job *domain.Job, runtime *domain.JobRuntime, tr runtime.Output = runningOutput(jobCopy, trigger, time.Now()) runtime.NextDue = time.Time{} err := s.store.SaveJobs(s.jobs) - go s.executeRun(jobCopy, trigger) + // Capture ctx under the lock so a concurrent Start/Stop cannot swap it out + // from under the goroutine after we release mu. + go s.executeRun(s.ctx, jobCopy, trigger) return err } // executeRun runs the job off the lock, then records the result back through the // Service under the lock and announces it. It runs on its own goroutine. -func (s *Service) executeRun(jobCopy domain.Job, trigger string) { - record := s.runJob(s.ctx, &jobCopy, trigger, s.store.Paths.LogsDir) +func (s *Service) executeRun(ctx context.Context, jobCopy domain.Job, trigger string) { + record := s.runJob(ctx, &jobCopy, trigger, s.store.Paths.LogsDir) s.mu.Lock() if current := s.findByIDLocked(jobCopy.ID); current != nil { diff --git a/src/app/operations_test.go b/src/app/operations_test.go index be466e9..62fac74 100644 --- a/src/app/operations_test.go +++ b/src/app/operations_test.go @@ -3,6 +3,7 @@ package app import ( "context" "path/filepath" + "sync/atomic" "testing" "time" @@ -250,6 +251,107 @@ func TestRunNowRefusedWhilePaused(t *testing.T) { } } +func TestRunDueStartsDueJob(t *testing.T) { + svc := newTempService(t, []domain.Job{{ID: 1, Name: "A", Schedule: "@every 1m", Command: "echo", Enabled: true}}) + + done := make(chan domain.RunRecord, 1) + svc.runJob = func(_ context.Context, job *domain.Job, trigger string, _ string) domain.RunRecord { + if trigger != "Schedule" { + t.Errorf("trigger = %q, want Schedule", trigger) + } + return domain.RunRecord{Time: "2026-06-19 12:00:00", JobID: job.ID, JobName: job.Name, State: "Success", Output: "ok"} + } + svc.Subscribe(ObserverFunc(func(e Event) { + if rr, ok := e.(RunRecorded); ok && rr.Record.JobID == 1 && rr.Record.State == "Success" { + select { + case done <- rr.Record: + default: + } + } + })) + + // The job's next-due was primed ~1m ahead at construction; tick well past it. + svc.RunDue(time.Now().Add(2 * time.Minute)) + + select { + case <-done: + case <-time.After(2 * time.Second): + t.Fatal("RunDue did not start the due job") + } + if rt := svc.Runtime(1); rt.LastState != "Success" || rt.Output != "ok" { + t.Errorf("runtime after scheduled run = %+v", rt) + } +} + +func TestRunDueSkipsJobNotYetDue(t *testing.T) { + svc := newTempService(t, []domain.Job{{ID: 1, Name: "A", Schedule: "@every 1m", Command: "echo", Enabled: true}}) + var ran int32 + svc.runJob = func(context.Context, *domain.Job, string, string) domain.RunRecord { + atomic.AddInt32(&ran, 1) + return domain.RunRecord{} + } + + // Next-due is ~1m out, so nothing is due "now". + svc.RunDue(time.Now()) + time.Sleep(50 * time.Millisecond) + if atomic.LoadInt32(&ran) != 0 { + t.Error("RunDue ran a job before it was due") + } +} + +func TestRunDueDoesNothingWhilePaused(t *testing.T) { + svc := newTempService(t, []domain.Job{{ID: 1, Name: "A", Schedule: "@every 1m", Command: "echo", Enabled: true}}) + var ran int32 + svc.runJob = func(context.Context, *domain.Job, string, string) domain.RunRecord { + atomic.AddInt32(&ran, 1) + return domain.RunRecord{} + } + if err := svc.SetGlobalPause(true); err != nil { + t.Fatalf("SetGlobalPause: %v", err) + } + + svc.RunDue(time.Now().Add(2 * time.Minute)) + time.Sleep(50 * time.Millisecond) + if atomic.LoadInt32(&ran) != 0 { + t.Error("RunDue ran a job while globally paused") + } +} + +// appFakeClock is a scheduler.Clock whose tick and "now" the test controls, used +// to verify Start wires the loop to RunDue without the wall clock. +type appFakeClock struct { + ticks chan time.Time + now time.Time +} + +func (c *appFakeClock) Now() time.Time { return c.now } +func (c *appFakeClock) Ticks() <-chan time.Time { return c.ticks } +func (c *appFakeClock) Stop() {} + +func TestStartDrivesRunDueOnTick(t *testing.T) { + svc := newTempService(t, []domain.Job{{ID: 1, Name: "A", Schedule: "@every 1m", Command: "echo", Enabled: true}}) + + done := make(chan struct{}, 1) + svc.runJob = func(context.Context, *domain.Job, string, string) domain.RunRecord { + select { + case done <- struct{}{}: + default: + } + return domain.RunRecord{State: "Success"} + } + + clock := &appFakeClock{ticks: make(chan time.Time, 1), now: time.Now().Add(2 * time.Minute)} + svc.Start(clock) + defer svc.Stop() + + clock.ticks <- clock.now + select { + case <-done: + case <-time.After(2 * time.Second): + t.Fatal("Start did not drive a run from a clock tick") + } +} + func TestUpdateSettingsPersistsAndValidates(t *testing.T) { svc := newTempService(t, nil) diff --git a/src/app/service.go b/src/app/service.go index e803e4d..6ba2797 100644 --- a/src/app/service.go +++ b/src/app/service.go @@ -3,9 +3,11 @@ package app import ( "context" "sync" + "time" "gitea.mixdep.ru/mix/gosentry/src/domain" "gitea.mixdep.ru/mix/gosentry/src/runner" + "gitea.mixdep.ru/mix/gosentry/src/scheduler" "gitea.mixdep.ru/mix/gosentry/src/storage" ) @@ -16,11 +18,15 @@ import ( // race on a shared *[]Job. // // State ownership and the locking contract were established in T3.1; the -// event/observer machinery in T3.2. T3.3 adds the state-mutating intents +// event/observer machinery in T3.2. T3.3 added the state-mutating intents // (CreateJob, UpdateJob, DeleteJob, SetEnabled, RunNow, SetGlobalPause, -// UpdateSettings) in operations.go: the Service is now the sole writer of job -// and runtime state, persisting through the store and announcing changes via -// events. +// UpdateSettings) in operations.go: the Service is the sole writer of job and +// runtime state, persisting through the store and announcing changes via events. +// +// T3.4 makes the Service drive scheduling too. It owns the timing loop through a +// scheduler.Scheduler that calls RunDue on every tick; the scheduler holds no +// job state and never touches the slice directly. The old shared *[]domain.Job +// between GUI and scheduler is gone — both go through the Service. // // Locking contract: mu is a plain, non-reentrant mutex. Exported methods take // it; unexported helpers ending in "Locked" assume the caller already holds it. @@ -35,18 +41,23 @@ type Service struct { // schedules caches a parsed Schedule per job ID so timing math does not // re-parse the schedule string on every use. paused is the global pause flag. - // Both are guarded by mu. (The scheduler still keeps its own copy until T3.4 - // converts it to drive the Service instead of sharing state.) + // Both are guarded by mu. schedules map[int]domain.Schedule paused bool // runJob is the run seam. It defaults to runner.RunJob and is overridden in - // tests with a fake so the run-now path can be exercised without spawning real - // processes. ctx is the lifecycle context passed to runs; T3.4 wires a - // cancelable Start/Stop, for now it is context.Background(). + // tests with a fake so the run paths can be exercised without spawning real + // processes. ctx is the lifecycle context passed to runs; Start replaces it + // with a cancelable context so Stop can abort in-flight runs, and until Start + // it is context.Background(). runJob func(ctx context.Context, job *domain.Job, trigger string, logsDir string) domain.RunRecord ctx context.Context + // sched is the timing loop installed by Start; cancel tears down ctx on Stop. + // Both are guarded by mu. + sched *scheduler.Scheduler + cancel context.CancelFunc + // observers and their guard live in events.go. dispatchMu is separate from mu // so that emitting an event never requires (or is held under) the state lock: // the Service must release mu before dispatching, per the locking contract. @@ -67,12 +78,51 @@ func NewService(store *storage.Store, jobs []domain.Job) *Service { runJob: runner.RunJob, ctx: context.Background(), } + // Parse every schedule once, then compute each job's first next-run so the + // Service is ready to schedule the moment it exists — mirroring the old + // scheduler's reset-on-construction. No lock is needed: construction is + // single-threaded, before Start launches the timing loop. + now := time.Now() for index := range s.jobs { - s.parseScheduleLocked(&s.jobs[index]) + job := &s.jobs[index] + s.parseScheduleLocked(job) + s.refreshNextRunFromLocked(job, s.runtimes[job.ID], now) } return s } +// Start begins scheduling. It installs a cancelable run context and a timing +// loop driven by the given clock; every tick calls RunDue. Pass +// scheduler.NewRealClock() in production. Start is expected once, during setup, +// before any concurrent use. +func (s *Service) Start(clock scheduler.Clock) { + s.mu.Lock() + ctx, cancel := context.WithCancel(context.Background()) + s.ctx = ctx + s.cancel = cancel + s.sched = scheduler.NewScheduler(clock, s.RunDue) + sched := s.sched + s.mu.Unlock() + + sched.Start() +} + +// Stop halts scheduling and cancels the run context so in-flight runs see a +// canceled context. It is safe to call when Start was never called. +func (s *Service) Stop() { + s.mu.Lock() + sched := s.sched + cancel := s.cancel + s.mu.Unlock() + + if sched != nil { + sched.Stop() + } + if cancel != nil { + cancel() + } +} + // Open loads the store and constructs a Service from it in one step. It is the // convenience entry point for the application; tests inject a pre-built store // via NewService instead. @@ -105,9 +155,9 @@ func (s *Service) Jobs() []domain.Job { // Runtime returns the transient runtime state for a job ID, or nil if no job // with that ID is loaded. The returned pointer is the live runtime; reads of it -// are only safe while no concurrent mutation is in flight, which holds during -// the current single-threaded transition and is tightened as the scheduler -// moves behind the Service in T3.4. +// are only safe while no concurrent mutation is in flight. The scheduler now +// drives the Service rather than sharing state, so the remaining concurrent +// reader is the UI listener, which T4.1 marshals onto the main thread. func (s *Service) Runtime(id int) *domain.JobRuntime { s.mu.Lock() defer s.mu.Unlock() diff --git a/src/gui/app.go b/src/gui/app.go index 53caf71..4c0a131 100644 --- a/src/gui/app.go +++ b/src/gui/app.go @@ -17,9 +17,7 @@ import ( "gitea.mixdep.ru/mix/gosentry/src/domain" "gitea.mixdep.ru/mix/gosentry/src/platform/autostart" "gitea.mixdep.ru/mix/gosentry/src/platform/desktop" - "gitea.mixdep.ru/mix/gosentry/src/runner" "gitea.mixdep.ru/mix/gosentry/src/scheduler" - "gitea.mixdep.ru/mix/gosentry/src/storage" "fyne.io/fyne/v2" fyneapp "fyne.io/fyne/v2/app" @@ -165,18 +163,33 @@ func serveSingleInstance(listener net.Listener, w fyne.Window) { } func newMainView(w fyne.Window) (fyne.CanvasObject, func(time.Duration, bool)) { - store, jobs, err := storage.OpenStore() + svc, err := app.Open() if err != nil { return container.NewPadded(widget.NewLabel("Failed to load GoSentry configuration: " + err.Error())), func(time.Duration, bool) {} } + store := svc.Store() if iconPath, err := desktop.InstallDesktopIntegration(appID, store.Paths.ExecutablePath, assets.IconBytes()); err == nil { store.Paths.DesktopIcon = iconPath } - // Transient execution state lives in a runtime map keyed by job ID, separate - // from the durable jobs slice. The scheduler shares the same map so background - // runs and GUI edits observe one in-memory copy of each job's status. - runtimes := domain.NewRuntimes(jobs) + // app.Service is the single owner of job and runtime state. The GUI keeps a + // read snapshot of the durable jobs plus a map of the live runtime pointers, + // both refreshed from the Service after every change. The Service — not the + // GUI — mutates state and drives the scheduler, so there is no shared *[]Job. + jobs := svc.Jobs() + runtimes := make(map[int]*domain.JobRuntime, len(jobs)) + syncFromService := func() { + jobs = svc.Jobs() + for id := range runtimes { + delete(runtimes, id) + } + for _, current := range jobs { + if runtime := svc.Runtime(current.ID); runtime != nil { + runtimes[current.ID] = runtime + } + } + } + syncFromService() runtimeFor := func(index int) *domain.JobRuntime { if index < 0 || index >= len(jobs) { return &domain.JobRuntime{} @@ -188,10 +201,6 @@ func newMainView(w fyne.Window) (fyne.CanvasObject, func(time.Duration, bool)) { } events := collectActivity(jobs, runtimes) - // The GUI keeps the loaded jobs slice in memory and persists changes after - // each edit/run. This keeps the first version responsive and easy to reason - // about; a database would be unnecessary overhead for one YAML file. - nextJobID := nextID(jobs) selected := 0 selectedFolder := allFolders schedulerPaused := false @@ -275,15 +284,15 @@ func newMainView(w fyne.Window) (fyne.CanvasObject, func(time.Duration, bool)) { selectedLogs = append(selectedLogs[:0], runtime.Logs...) } refresh := func() { - // Several callbacks mutate jobs, filters, and event history. A single - // refresh closure keeps the different widgets synchronized after each - // mutation without introducing a heavier state-management layer. + // Several callbacks change jobs, filters, and event history. A single + // refresh closure re-reads the Service snapshot and keeps the different + // widgets synchronized after each change, without a heavier state layer. + syncFromService() filteredJobs = filteredJobIndexes(jobs, selectedFolder) updateDetails(selected) jobLogs.Refresh() history.Refresh() } - var sched *scheduler.Scheduler list := widget.NewList( func() int { return len(filteredJobs) }, @@ -338,25 +347,23 @@ func newMainView(w fyne.Window) (fyne.CanvasObject, func(time.Duration, bool)) { addButton := widget.NewButtonWithIcon("New job", theme.ContentAddIcon(), func() { showJobDialog(w, "New job", job{Schedule: "@every 1m", Command: "echo GoSentry job ran", Enabled: true}, func(saved job) { - saved.ID = nextJobID - nextJobID++ - jobs = append(jobs, saved) - runtime := domain.NewRuntime(saved) - runtimes[saved.ID] = runtime - selected = len(jobs) - 1 - created := newEvent(saved.ID, saved.Name, "Created", "Job was added") - // UI events are kept in memory for the current session. They explain - // user actions in History, while command output remains in log files. - runtime.Logs = append([]event{created}, runtime.Logs...) - events = append(events, created) - _ = store.SaveJobs(jobs) + // The Service assigns the ID, stores the job, records the "Created" + // activity, and emits events. The observer appends those to History; we + // only refresh the snapshot and move the selection to the new job. + created, err := svc.CreateJob(saved) + if err != nil { + dialog.ShowError(err, w) + return + } + syncFromService() folderSelect.Options = folderOptions(jobs) folderSelect.Refresh() - targetFolder := filterValue(saved.Folder) + targetFolder := filterValue(created.Folder) if selectedFolder != allFolders && selectedFolder != targetFolder { selectedFolder = targetFolder folderSelect.SetSelected(targetFolder) } + selected = indexOfID(jobs, created.ID) filteredJobs = filteredJobIndexes(jobs, selectedFolder) list.Refresh() list.Select(displayIndex(filteredJobs, selected)) @@ -368,31 +375,15 @@ func newMainView(w fyne.Window) (fyne.CanvasObject, func(time.Duration, bool)) { return } showJobDialog(w, "Edit job", jobs[selected], func(saved job) { + // The job keeps its ID, so the Service preserves the runtime (keyed by + // ID), reflects any enabled/disabled change, recomputes the next run, and + // emits the "Updated" activity the observer records. saved.ID = jobs[selected].ID - jobs[selected] = saved - // Runtime state (activity, output, status) is keyed by job ID and the ID - // is unchanged, so it survives the edit automatically. Reflect a possible - // enabled/disabled change into the status; the scheduler recomputes the - // next-run string below. - runtime := runtimes[saved.ID] - if runtime != nil { - if saved.Enabled { - if runtime.LastState == "" || runtime.LastState == "Paused" { - runtime.LastState = "Ready" - } - } else { - runtime.LastState = "Paused" - } + if err := svc.UpdateJob(saved); err != nil { + dialog.ShowError(err, w) + return } - updated := newEvent(saved.ID, saved.Name, "Updated", "Job settings changed") - if runtime != nil { - runtime.Logs = append([]event{updated}, runtime.Logs...) - } - events = append(events, updated) - if sched != nil { - sched.RefreshSchedule(selected) - } - _ = store.SaveJobs(jobs) + syncFromService() folderSelect.Options = folderOptions(jobs) folderSelect.Refresh() list.Refresh() @@ -409,7 +400,9 @@ func newMainView(w fyne.Window) (fyne.CanvasObject, func(time.Duration, bool)) { dialog.ShowInformation("Scheduler paused", "Global pause is active. Resume the scheduler before running jobs.", w) return } - if !sched.RunNow(selected) { + // RunNow refuses an already-running job (it returns an error); the GUI has + // always ignored that case silently, so the run simply does not start. + if err := svc.RunNow(jobs[selected].ID); err != nil { return } list.Refresh() @@ -417,36 +410,23 @@ func newMainView(w fyne.Window) (fyne.CanvasObject, func(time.Duration, bool)) { }) stopAllButton := widget.NewButtonWithIcon("Pause all", theme.MediaStopIcon(), nil) stopAllButton.OnTapped = func() { + // SetGlobalPause flips the Service's pause flag, updates every job's + // next-run text, and emits the activity record the observer logs. Mirror the + // new state into the local flag and the controls; revert it if the save fails. schedulerPaused = !schedulerPaused + if err := svc.SetGlobalPause(schedulerPaused); err != nil { + schedulerPaused = !schedulerPaused + dialog.ShowError(err, w) + return + } if schedulerPaused { schedulerState.SetText("Scheduler paused") stopAllButton.SetText("Resume all") stopAllButton.SetIcon(theme.MediaPlayIcon()) - for index := range jobs { - if runtime := runtimes[jobs[index].ID]; runtime != nil && jobs[index].Enabled { - runtime.NextRun = "Scheduler paused" - } - } - if sched != nil { - sched.SetPaused(true) - } - events = append(events, newEvent(0, "Scheduler", "Paused", "All job execution paused")) } else { schedulerState.SetText("Scheduler running") stopAllButton.SetText("Pause all") stopAllButton.SetIcon(theme.MediaStopIcon()) - for index := range jobs { - runtime := runtimes[jobs[index].ID] - if runtime != nil && jobs[index].Enabled && runtime.NextRun == "Scheduler paused" { - // The scheduler will calculate the exact next run when it is - // resumed; this interim text prevents a stale paused timestamp. - runtime.NextRun = "Waiting for scheduler" - } - } - if sched != nil { - sched.SetPaused(false) - } - events = append(events, newEvent(0, "Scheduler", "Resumed", "All job execution resumed")) } list.Refresh() refresh() @@ -455,29 +435,14 @@ func newMainView(w fyne.Window) (fyne.CanvasObject, func(time.Duration, bool)) { if selected < 0 || selected >= len(jobs) { return } - current := &jobs[selected] - current.Enabled = !current.Enabled - runtime := runtimeFor(selected) - if current.Enabled { - runtime.LastState = "Ready" - runtime.NextRun = "Waiting for scheduler" - resumed := newEvent(current.ID, current.Name, "Resumed", "Job was enabled") - runtime.Logs = append([]event{resumed}, runtime.Logs...) - events = append(events, resumed) - if sched != nil { - sched.RefreshSchedule(selected) - } - } else { - runtime.LastState = "Paused" - runtime.NextRun = "Paused" - paused := newEvent(current.ID, current.Name, "Paused", "Job was disabled") - runtime.Logs = append([]event{paused}, runtime.Logs...) - events = append(events, paused) - if sched != nil { - sched.RefreshSchedule(selected) - } + // SetEnabled toggles the job, updates its runtime/next-run, and records the + // "Resumed"/"Paused" activity the observer logs. + current := jobs[selected] + if err := svc.SetEnabled(current.ID, !current.Enabled); err != nil { + dialog.ShowError(err, w) + return } - _ = store.SaveJobs(jobs) + syncFromService() list.Refresh() refresh() }) @@ -492,8 +457,14 @@ func newMainView(w fyne.Window) (fyne.CanvasObject, func(time.Duration, bool)) { if !confirm { return } - jobs = append(jobs[:selected], jobs[selected+1:]...) - delete(runtimes, deleted.ID) + // The Service removes the job and its runtime, persists, and records the + // "Deleted" activity the observer logs; the GUI re-reads the snapshot and + // fixes up the folder filter and selection. + if err := svc.DeleteJob(deleted.ID); err != nil { + dialog.ShowError(err, w) + return + } + syncFromService() folderSelect.Options = folderOptions(jobs) folderSelect.Refresh() filteredJobs = filteredJobIndexes(jobs, selectedFolder) @@ -507,8 +478,6 @@ func newMainView(w fyne.Window) (fyne.CanvasObject, func(time.Duration, bool)) { } else { selected = filteredJobs[0] } - events = append(events, newEvent(deleted.ID, deleted.Name, "Deleted", "Job was removed")) - _ = store.SaveJobs(jobs) list.Refresh() if selected >= 0 { list.Select(displayIndex(filteredJobs, selected)) @@ -542,20 +511,26 @@ func newMainView(w fyne.Window) (fyne.CanvasObject, func(time.Duration, bool)) { jobLogs, ) - sched = scheduler.NewScheduler(store, &jobs, runtimes, func(record domain.RunRecord) { - // Scheduled runs happen on the scheduler goroutine. The callback updates - // the shared in-memory event list so History reflects background activity. - events = append(events, record) + // The Service announces every change through events. This single listener is + // where the GUI reacts: it appends run/activity records to History and redraws. + // Scheduled and manual completions fire it from the run goroutine; UI actions + // fire it synchronously. Marshaling these widget updates onto the main thread + // (fyne.Do) is wired in T4.1 — for now this matches the prior direct refresh. + svc.Subscribe(app.ObserverFunc(func(ev app.Event) { + if recorded, ok := ev.(app.RunRecorded); ok { + events = append(events, recorded.Record) + } refresh() - }) - sched.Start() + list.Refresh() + })) + svc.Start(scheduler.NewRealClock()) fixedSidebar := container.New(minWidthLayout{width: minJobsSidebarWidth}, sidebar) jobsView := container.NewBorder(nil, nil, fixedSidebar, nil, container.NewPadded(details)) tabs := container.NewAppTabs( container.NewTabItemWithIcon("Jobs", theme.ListIcon(), jobsView), container.NewTabItemWithIcon("History", theme.HistoryIcon(), history), - container.NewTabItemWithIcon("Settings", theme.SettingsIcon(), settingsView(w, store, &jobs)), + container.NewTabItemWithIcon("Settings", theme.SettingsIcon(), settingsView(w, svc)), ) tabs.SetTabLocation(container.TabLocationTop) @@ -644,14 +619,13 @@ func collectActivity(jobs []job, runtimes map[int]*domain.JobRuntime) []event { return events } -func nextID(jobs []job) int { - next := 1 - for _, current := range jobs { - if current.ID >= next { - next = current.ID + 1 +func indexOfID(jobs []job, id int) int { + for index, current := range jobs { + if current.ID == id { + return index } } - return next + return 0 } func detailRow(label string, value fyne.CanvasObject) fyne.CanvasObject { @@ -938,7 +912,8 @@ func logFileName(path string) string { return path } -func settingsView(w fyne.Window, store *storage.Store, jobs *[]job) fyne.CanvasObject { +func settingsView(w fyne.Window, svc *app.Service) fyne.CanvasObject { + store := svc.Store() startOnLogin := widget.NewCheck("Start on login", nil) startOnLogin.SetChecked(store.Config.StartOnLogin) autostartStatus := widget.NewLabel("") @@ -989,7 +964,6 @@ func settingsView(w fyne.Window, store *storage.Store, jobs *[]job) fyne.CanvasO settingsStatus.SetText("Max log age days must be a positive number") return } - store.Config.LogsDir = strings.TrimSpace(logsDir.Text) if strings.TrimSpace(jobsDir.Text) == "" { settingsStatus.SetText("Jobs directory is required") return @@ -998,35 +972,29 @@ func settingsView(w fyne.Window, store *storage.Store, jobs *[]job) fyne.CanvasO settingsStatus.SetText("Logs directory is required") return } - store.Config.JobsDir = strings.TrimSpace(jobsDir.Text) - store.Config.MaxLogFiles = files - store.Config.MaxLogAgeDays = days - store.Config.StartOnLogin = startOnLogin.Checked - store.Config.KeepRunningInTray = minimizeToTray.Checked - store.Config.NotifyOnFailure = notifications.Checked - if err := store.SaveConfig(); err != nil { + // Build the new config from the form and hand it to the Service, which + // validates it, persists config and jobs to the (possibly new) directory, + // and runs log cleanup so tightened retention limits take effect at once. + config := store.Config + config.JobsDir = strings.TrimSpace(jobsDir.Text) + config.LogsDir = strings.TrimSpace(logsDir.Text) + config.MaxLogFiles = files + config.MaxLogAgeDays = days + config.StartOnLogin = startOnLogin.Checked + config.KeepRunningInTray = minimizeToTray.Checked + config.NotifyOnFailure = notifications.Checked + if err := svc.UpdateSettings(config); err != nil { settingsStatus.SetText("Save failed: " + err.Error()) return } + // Autostart is platform integration the Service leaves to the caller (until + // T5.2 introduces an injectable autostart.Manager), so apply it here. if err := autostart.SetAutostart(store.Config.StartOnLogin, store.Paths.ExecutablePath, store.Paths.DesktopIcon); err != nil { refreshAutostartStatus() settingsStatus.SetText("Saved, autostart failed: " + err.Error()) return } refreshAutostartStatus() - // When the jobs directory changes, save the currently loaded jobs to the - // newly resolved path immediately. That makes the setting visible on disk - // without requiring a restart or a separate migration command. - if err := store.SaveJobs(*jobs); err != nil { - settingsStatus.SetText("Jobs save failed: " + err.Error()) - return - } - // Cleanup runs on settings save so a user who tightens retention limits - // sees the new policy take effect right away. - if err := runner.CleanupLogs(store.Paths.LogsDir, store.Config.MaxLogFiles, store.Config.MaxLogAgeDays); err != nil { - settingsStatus.SetText("Saved, cleanup failed: " + err.Error()) - return - } settingsStatus.SetText("Saved") }) diff --git a/src/scheduler/clock.go b/src/scheduler/clock.go new file mode 100644 index 0000000..b12b29e --- /dev/null +++ b/src/scheduler/clock.go @@ -0,0 +1,55 @@ +package scheduler + +import "time" + +// Clock supplies the scheduler with the current time and a stream of ticks. +// Hiding both behind an interface lets tests drive the loop deterministically — +// firing ticks and controlling "now" — instead of waiting on the wall clock. +// Production uses RealClock. +type Clock interface { + // Now returns the current time. It is the value passed to the tick callback + // on each tick, so a fake can make due-evaluation deterministic. + Now() time.Time + // Ticks returns a channel that delivers a value on every scheduler tick. The + // scheduler reads it for the lifetime of the loop. + Ticks() <-chan time.Time + // Stop releases the resources backing Ticks. The scheduler calls it once when + // the loop exits. + Stop() +} + +// RealClock is the production Clock: wall-clock time and a one-second ticker. +// +// A one-second cadence is accurate enough for cron-style desktop automation — +// five-field cron expressions have minute precision, while @every values may be +// shorter for testing and lightweight local tasks — and it keeps a single timer +// instead of one per job. +type RealClock struct { + ticker *time.Ticker +} + +// NewRealClock returns a real clock. The underlying ticker is created lazily on +// the first Ticks call so a clock that is never started leaks nothing. +func NewRealClock() *RealClock { + return &RealClock{} +} + +// Now returns the wall-clock time. +func (c *RealClock) Now() time.Time { + return time.Now() +} + +// Ticks starts (once) and returns the one-second ticker channel. +func (c *RealClock) Ticks() <-chan time.Time { + if c.ticker == nil { + c.ticker = time.NewTicker(time.Second) + } + return c.ticker.C +} + +// Stop halts the ticker if it was ever started. +func (c *RealClock) Stop() { + if c.ticker != nil { + c.ticker.Stop() + } +} diff --git a/src/scheduler/scheduler.go b/src/scheduler/scheduler.go index ec24ba5..b9f52db 100644 --- a/src/scheduler/scheduler.go +++ b/src/scheduler/scheduler.go @@ -2,266 +2,55 @@ package scheduler import ( "context" - "fmt" - "strings" - "sync" "time" - - "gitea.mixdep.ru/mix/gosentry/src/domain" - "gitea.mixdep.ru/mix/gosentry/src/runner" - "gitea.mixdep.ru/mix/gosentry/src/storage" ) -// Scheduler owns the timing loop for jobs that are currently loaded in the GUI. -// It receives a pointer to the jobs slice because the GUI edits the same slice; -// this keeps the early architecture simple while storage and scheduling are -// still in one desktop process. +// Scheduler is a thin timing loop. It owns no job or runtime state: on every +// clock tick it calls the injected tick function with the current time, and that +// function — the application service's RunDue — decides what, if anything, to +// run. Keeping all state and mutation in the service makes the service the sole +// writer (resolving the old shared-*[]Job data race) and reduces the scheduler +// to a loop that is trivially testable with a fake Clock. type Scheduler struct { - store *storage.Store - jobs *[]domain.Job - runtimes map[int]*domain.JobRuntime - onChange func(domain.RunRecord) + clock Clock + tick func(now time.Time) - mu sync.Mutex - ctx context.Context - cancel context.CancelFunc - paused bool - schedules map[int]domain.Schedule // parsed once per job on load/edit + ctx context.Context + cancel context.CancelFunc } -// NewScheduler shares the durable jobs slice and the transient runtime map with -// the GUI. Both still point at the same in-memory state for now; Phase 3 moves -// ownership behind an application service. -func NewScheduler(store *storage.Store, jobs *[]domain.Job, runtimes map[int]*domain.JobRuntime, onChange func(domain.RunRecord)) *Scheduler { +// NewScheduler builds a scheduler that calls tick on every Clock tick. The clock +// is injected so tests can drive the loop without the wall clock. +func NewScheduler(clock Clock, tick func(now time.Time)) *Scheduler { ctx, cancel := context.WithCancel(context.Background()) - s := &Scheduler{ - store: store, - jobs: jobs, - runtimes: runtimes, - onChange: onChange, - ctx: ctx, - cancel: cancel, - schedules: make(map[int]domain.Schedule), + return &Scheduler{ + clock: clock, + tick: tick, + ctx: ctx, + cancel: cancel, } - s.resetNextRuns(time.Now()) - return s -} - -// runtimeFor returns the runtime state for a job, lazily creating it if the map -// has no entry yet. This keeps the scheduler robust if a job is added to the -// shared slice without a matching runtime. -func (s *Scheduler) runtimeFor(job *domain.Job) *domain.JobRuntime { - runtime, ok := s.runtimes[job.ID] - if !ok || runtime == nil { - runtime = domain.NewRuntime(*job) - s.runtimes[job.ID] = runtime - } - return runtime } +// Start launches the loop on its own goroutine and returns immediately. func (s *Scheduler) Start() { - // A one-second ticker is accurate enough for cron-style desktop automation - // and avoids the complexity of maintaining one timer per job. Five-field cron - // expressions have minute precision, while @every values may be shorter for - // testing and lightweight local tasks. - ticker := time.NewTicker(time.Second) go func() { - defer ticker.Stop() + ticks := s.clock.Ticks() + defer s.clock.Stop() for { select { case <-s.ctx.Done(): return - case now := <-ticker.C: - s.tick(now) + case <-ticks: + // Pass the clock's notion of "now" rather than the tick value so a + // fake clock can control due-evaluation precisely. + s.tick(s.clock.Now()) } } }() } +// Stop ends the loop. A tick already in progress finishes; no further ticks are +// delivered. func (s *Scheduler) Stop() { s.cancel() } - -func (s *Scheduler) SetPaused(paused bool) { - s.mu.Lock() - defer s.mu.Unlock() - - s.paused = paused - now := time.Now() - // Pause state is reflected into each job's display string so the list view is - // understandable even before the next scheduler tick. - for index := range *s.jobs { - job := &(*s.jobs)[index] - runtime := s.runtimeFor(job) - if !job.Enabled { - runtime.NextRun = "Paused" - continue - } - if paused { - runtime.NextRun = "Scheduler paused" - continue - } - s.prepareNextRun(job, runtime, now) - } - _ = s.store.SaveJobs(*s.jobs) -} - -func (s *Scheduler) RunNow(index int) bool { - s.mu.Lock() - defer s.mu.Unlock() - - if index < 0 || index >= len(*s.jobs) { - return false - } - // Manual runs share the same runner and log writer as scheduled runs. The - // Trigger field is the only difference, which keeps History comparable and - // prevents "Run now" from becoming a separate behavior path. - return s.startRunLocked(index, "Manual") -} - -func (s *Scheduler) RefreshSchedule(index int) { - s.mu.Lock() - defer s.mu.Unlock() - - if index < 0 || index >= len(*s.jobs) { - return - } - job := &(*s.jobs)[index] - runtime := s.runtimeFor(job) - s.parseJobSchedule(job) // re-parse in case the schedule string changed - if !job.Enabled { - runtime.NextRun = "Paused" - return - } - if s.paused { - runtime.NextRun = "Scheduler paused" - return - } - s.prepareNextRun(job, runtime, time.Now()) -} - -func (s *Scheduler) tick(now time.Time) { - var changed bool - - s.mu.Lock() - if !s.paused { - for index := range *s.jobs { - job := &(*s.jobs)[index] - runtime := s.runtimeFor(job) - if !job.Enabled || runtime.NextDue.IsZero() || now.Before(runtime.NextDue) { - continue - } - // Run only one due job per tick for now. That avoids overlapping shell - // commands in the GUI process and keeps the first version predictable; - // a future worker pool can add concurrency once cancellation and status - // reporting are more explicit. - changed = s.startRunLocked(index, "Schedule") - break - } - } - s.mu.Unlock() - _ = changed -} - -func (s *Scheduler) startRunLocked(index int, trigger string) bool { - job := &(*s.jobs)[index] - runtime := s.runtimeFor(job) - if runtime.LastState == "Running" { - return false - } - - jobCopy := *job - runtime.LastState = "Running" - runtime.NextRun = "Running" - runtime.Output = runningOutput(jobCopy, trigger, time.Now()) - runtime.NextDue = time.Time{} - _ = s.store.SaveJobs(*s.jobs) - - go func() { - record := runner.RunJob(s.ctx, &jobCopy, trigger, s.store.Paths.LogsDir) - - s.mu.Lock() - if current := s.findJobByIDLocked(jobCopy.ID); current != nil { - currentRuntime := s.runtimeFor(current) - currentRuntime.LastRun = record.Time - currentRuntime.LastState = record.State - currentRuntime.Output = record.Output - currentRuntime.Logs = append([]domain.RunRecord{record}, currentRuntime.Logs...) - if len(currentRuntime.Logs) > 50 { - currentRuntime.Logs = currentRuntime.Logs[:50] - } - s.prepareNextRun(current, currentRuntime, time.Now()) - _ = runner.CleanupLogs(s.store.Paths.LogsDir, s.store.Config.MaxLogFiles, s.store.Config.MaxLogAgeDays) - _ = s.store.SaveJobs(*s.jobs) - } - s.mu.Unlock() - - if s.onChange != nil { - s.onChange(record) - } - }() - return true -} - -func (s *Scheduler) findJobByIDLocked(id int) *domain.Job { - for index := range *s.jobs { - if (*s.jobs)[index].ID == id { - return &(*s.jobs)[index] - } - } - return nil -} - -func runningOutput(job domain.Job, trigger string, started time.Time) string { - var builder strings.Builder - builder.WriteString("status:\n") - builder.WriteString("Running since " + started.Format("2006-01-02 15:04:05") + "\n\n") - builder.WriteString("trigger:\n") - builder.WriteString(trigger + "\n\n") - builder.WriteString("command:\n") - builder.WriteString(job.Command + "\n\n") - builder.WriteString("arguments:\n") - builder.WriteString(runner.LogArguments(job.Arguments)) - builder.WriteString("\n\nsuccess_exit_codes:\n") - builder.WriteString(runner.SuccessExitCodesText(job)) - builder.WriteString("\n\nstart_only:\n") - builder.WriteString(fmt.Sprintf("%t", job.StartOnly)) - return builder.String() -} - -func (s *Scheduler) resetNextRuns(now time.Time) { - for index := range *s.jobs { - job := &(*s.jobs)[index] - runtime := s.runtimeFor(job) - s.parseJobSchedule(job) // parse once on load - if !job.Enabled { - runtime.NextRun = "Paused" - continue - } - s.prepareNextRun(job, runtime, now) - } - _ = s.store.SaveJobs(*s.jobs) -} - -// parseJobSchedule caches a parsed domain.Schedule for the job. Invalid -// schedule strings are silently dropped from the cache so prepareNextRun can -// distinguish them from valid ones. -func (s *Scheduler) parseJobSchedule(job *domain.Job) { - sched, err := domain.Parse(job.Schedule) - if err != nil { - delete(s.schedules, job.ID) - return - } - s.schedules[job.ID] = sched -} - -func (s *Scheduler) prepareNextRun(job *domain.Job, runtime *domain.JobRuntime, from time.Time) { - sched, ok := s.schedules[job.ID] - if !ok { - runtime.NextRun = "Invalid schedule" - runtime.NextDue = time.Time{} - return - } - runtime.NextDue = sched.Next(from) - runtime.NextRun = runtime.NextDue.Format("2006-01-02 15:04:05") -} diff --git a/src/scheduler/scheduler_test.go b/src/scheduler/scheduler_test.go index dea7ab3..56978e1 100644 --- a/src/scheduler/scheduler_test.go +++ b/src/scheduler/scheduler_test.go @@ -1,69 +1,85 @@ package scheduler import ( - "strings" + "sync" "testing" "time" - - "gitea.mixdep.ru/mix/gosentry/src/domain" ) -func TestPrepareNextRunSetsDisplayString(t *testing.T) { - jobs := []domain.Job{{Schedule: "*/5 * * * *", Enabled: true}} - s := &Scheduler{jobs: &jobs, runtimes: domain.NewRuntimes(jobs), schedules: make(map[int]domain.Schedule)} - s.parseJobSchedule(&jobs[0]) - runtime := s.runtimeFor(&jobs[0]) - from := time.Date(2026, 6, 14, 12, 3, 0, 0, time.UTC) +// fakeClock is a Clock whose ticks and "now" are driven by the test instead of +// the wall clock, so the scheduler loop can be exercised deterministically. +type fakeClock struct { + ticks chan time.Time - s.prepareNextRun(&jobs[0], runtime, from) - - want := "2026-06-14 12:05:00" - if runtime.NextRun != want { - t.Errorf("NextRun: got %q, want %q", runtime.NextRun, want) - } - wantDue := time.Date(2026, 6, 14, 12, 5, 0, 0, time.UTC) - if !runtime.NextDue.Equal(wantDue) { - t.Errorf("NextDue: got %v, want %v", runtime.NextDue, wantDue) - } + mu sync.Mutex + now time.Time + stopped bool } -func TestPrepareNextRunSetsInvalidScheduleLabel(t *testing.T) { - jobs := []domain.Job{{Schedule: "not-a-cron", Enabled: true}} - s := &Scheduler{jobs: &jobs, runtimes: domain.NewRuntimes(jobs), schedules: make(map[int]domain.Schedule)} - // parseJobSchedule will drop the invalid spec, so schedules map stays empty. - s.parseJobSchedule(&jobs[0]) - runtime := s.runtimeFor(&jobs[0]) - - s.prepareNextRun(&jobs[0], runtime, time.Now()) - - if runtime.NextRun != "Invalid schedule" { - t.Errorf("NextRun: got %q, want 'Invalid schedule'", runtime.NextRun) - } - if !runtime.NextDue.IsZero() { - t.Errorf("NextDue should be zero for invalid schedule, got %v", runtime.NextDue) - } +func newFakeClock(now time.Time) *fakeClock { + return &fakeClock{ticks: make(chan time.Time, 1), now: now} } -func TestRunningOutputIncludesInvocation(t *testing.T) { - started := time.Date(2026, 6, 17, 23, 40, 0, 0, time.Local) - job := domain.Job{ - Name: "Backup", - Command: `C:\Program Files\FreeFileSync\FreeFileSync.exe`, - Arguments: `D:\Local\Jobs\Auto.ffs_batch`, - SuccessExitCodes: "0,1", - } +func (c *fakeClock) Now() time.Time { + c.mu.Lock() + defer c.mu.Unlock() + return c.now +} - output := runningOutput(job, "Manual", started) - for _, want := range []string{ - "Running since 2026-06-17 23:40:00", - "Manual", - job.Command, - job.Arguments, - "0,1", - "start_only", - } { - if !strings.Contains(output, want) { - t.Fatalf("expected running output to contain %q, got:\n%s", want, output) +func (c *fakeClock) Ticks() <-chan time.Time { return c.ticks } + +func (c *fakeClock) Stop() { + c.mu.Lock() + c.stopped = true + c.mu.Unlock() +} + +func (c *fakeClock) isStopped() bool { + c.mu.Lock() + defer c.mu.Unlock() + return c.stopped +} + +// fire advances the clock to t and delivers one tick. +func (c *fakeClock) fire(t time.Time) { + c.mu.Lock() + c.now = t + c.mu.Unlock() + c.ticks <- t +} + +func TestSchedulerCallsTickWithClockNow(t *testing.T) { + clock := newFakeClock(time.Unix(0, 0)) + got := make(chan time.Time, 1) + s := NewScheduler(clock, func(now time.Time) { got <- now }) + s.Start() + defer s.Stop() + + want := time.Date(2026, 6, 19, 12, 0, 0, 0, time.UTC) + clock.fire(want) + + select { + case now := <-got: + if !now.Equal(want) { + t.Errorf("tick now = %v, want %v", now, want) } + case <-time.After(time.Second): + t.Fatal("scheduler did not call tick after a clock tick") + } +} + +func TestSchedulerStopReleasesClock(t *testing.T) { + clock := newFakeClock(time.Now()) + s := NewScheduler(clock, func(time.Time) {}) + s.Start() + s.Stop() + + // After Stop the loop exits and releases the clock via the deferred Stop. + deadline := time.Now().Add(time.Second) + for !clock.isStopped() { + if time.Now().After(deadline) { + t.Fatal("clock was not stopped after scheduler Stop") + } + time.Sleep(time.Millisecond) } }