T3.4: Convert scheduler to drive app.Service; inject Clock

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 <noreply@anthropic.com>
2026-06-19 08:22:35 +03:00
parent d8ab9acf7e
commit a4c93a5122
8 changed files with 459 additions and 442 deletions
@@ -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 {
@@ -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)

@@ -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()