From b3e699792ab90b876eee7f0c9efaf49b7e6bdd11 Mon Sep 17 00:00:00 2001 From: mixeme Date: Tue, 23 Jun 2026 07:54:12 +0300 Subject: [PATCH] P3.3: Apply execution mode + overlap policy in dispatch Rework RunDue/startRunLocked/executeRun so the ExecutionMode and OverlapPolicy config knobs take effect: - startRunLocked advances NextDue to the next occurrence instead of zeroing it, keeping the schedule marching during an in-flight run. - RunDue scans all due jobs: parallel starts every due, non-running job; sequential defers a due job while any other job runs. When a job comes due again mid-run, skip drops it and queue marks it Pending; either way NextDue is advanced past the fired occurrence. - executeRun re-runs a Pending job once the current run finishes. - RunNow gains a sequential-mode guard refusing a manual run while another job is running. - Add anyRunningLocked and advanceNextDueLocked helpers. Co-Authored-By: Claude Opus 4.8 --- docs/PRE-RELEASE-TASKS.md | 2 +- src/app/run.go | 113 ++++++++++++++++++++++++++++++-------- 2 files changed, 92 insertions(+), 23 deletions(-) diff --git a/docs/PRE-RELEASE-TASKS.md b/docs/PRE-RELEASE-TASKS.md index 76462a9..ebb5a5a 100644 --- a/docs/PRE-RELEASE-TASKS.md +++ b/docs/PRE-RELEASE-TASKS.md @@ -96,7 +96,7 @@ These land together because both edit `domain/job.go` and `storage/store.go`. ### Phase 3 — Task-queue model + settings - [x] P3.1 — Config/runtime fields + defaults - [x] P3.2 — Split dispatch into `app/run.go` -- [ ] P3.3 — Rework `RunDue`/`executeRun` for mode + overlap policy +- [x] P3.3 — Rework `RunDue`/`executeRun` for mode + overlap policy - [ ] P3.4 — Settings Queue selects - [ ] P3.5 — Queue tests diff --git a/src/app/run.go b/src/app/run.go index 9fb425e..54514dc 100644 --- a/src/app/run.go +++ b/src/app/run.go @@ -13,10 +13,12 @@ import ( // RunNow starts a manual run of a job. It refuses to run while globally paused — // the pause is an emergency stop for all execution — and will not start a job -// that is already running. The run itself happens on a background goroutine that -// records the result through the Service, so RunNow returns as soon as the run -// is started. The error reports why a run could not be started (or a failure to -// persist the "Running" status), not the run's own outcome. +// that is already running. In sequential execution mode it also refuses while +// any other job is running, so a manual run never breaks the one-at-a-time +// guarantee. The run itself happens on a background goroutine that records the +// result through the Service, so RunNow returns as soon as the run is started. +// The error reports why a run could not be started (or a failure to persist the +// "Running" status), not the run's own outcome. func (s *Service) RunNow(id int) error { s.mu.Lock() if s.paused { @@ -33,6 +35,10 @@ func (s *Service) RunNow(id int) error { s.mu.Unlock() return fmt.Errorf("job %d is already running", id) } + if s.store.Config.ExecutionMode == domain.ExecutionModeSequential && s.anyRunningLocked() { + s.mu.Unlock() + return errors.New("another job is already running (sequential mode)") + } err := s.startRunLocked(job, runtime, "Manual") s.mu.Unlock() @@ -42,17 +48,27 @@ func (s *Service) RunNow(id int) error { } // 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. +// the given time. It is a no-op while globally paused. 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. +// +// Dispatch obeys two configured knobs. The execution mode decides whether +// distinct due jobs run together (parallel) or one at a time (sequential): in +// sequential mode a due job is left for a later tick while any other job is +// running. The overlap policy decides what happens when a job comes due again +// while its own previous run is still in flight: "skip" drops the new run, +// "queue" marks it Pending so executeRun re-runs it the moment the current run +// finishes. Either way NextDue is advanced past the fired occurrence so the same +// moment is not re-evaluated on every tick. func (s *Service) RunDue(now time.Time) { s.mu.Lock() - var startedID int + var started []int var startErr error if !s.paused { + sequential := s.store.Config.ExecutionMode == domain.ExecutionModeSequential + queue := s.store.Config.OverlapPolicy == domain.OverlapPolicyQueue + running := s.anyRunningLocked() for index := range s.jobs { job := &s.jobs[index] runtime := s.runtimeForLocked(job) @@ -60,11 +76,24 @@ func (s *Service) RunDue(now time.Time) { continue } if runtime.LastState == "Running" { + // The job came due again while its own run is still in flight. + // Apply the overlap policy and step past this occurrence. + if queue { + runtime.Pending = true + } + s.advanceNextDueLocked(job, runtime, now) continue } - startErr = s.startRunLocked(job, runtime, "Schedule") - startedID = job.ID - break + if sequential && running { + // One-at-a-time: leave this job due and pick it up on a later + // tick once the in-flight run has finished. + continue + } + if err := s.startRunLocked(job, runtime, "Schedule"); err != nil { + startErr = err + } + started = append(started, job.ID) + running = true } } s.mu.Unlock() @@ -72,19 +101,22 @@ func (s *Service) RunDue(now time.Time) { if startErr != nil { s.emit(ErrorOccurred{Err: fmt.Errorf("save jobs before scheduled run: %w", startErr)}) } - if startedID != 0 { - s.emit(JobChanged{JobID: startedID}) + for _, id := range started { + s.emit(JobChanged{JobID: id}) } } -// startRunLocked transitions a job to "Running", persists that, and launches the -// run on a background goroutine. The caller must hold mu. +// startRunLocked transitions a job to "Running", advances its NextDue to the next +// scheduled occurrence, persists that, and launches the run on a background +// goroutine. Advancing (rather than zeroing) NextDue keeps the schedule marching +// while the run is in flight, which is what lets RunDue notice a fresh occurrence +// firing during a long run and apply the overlap policy. The caller must hold mu. func (s *Service) startRunLocked(job *domain.Job, runtime *domain.JobRuntime, trigger string) error { jobCopy := *job runtime.LastState = "Running" runtime.NextRun = "Running" runtime.Output = runningOutput(jobCopy, trigger, time.Now()) - runtime.NextDue = time.Time{} + s.advanceNextDueLocked(job, runtime, time.Now()) err := s.store.SaveJobs(s.jobs) // Capture ctx under the lock so a concurrent Start/Stop cannot swap it out // from under the goroutine after we release mu. @@ -93,7 +125,10 @@ func (s *Service) startRunLocked(job *domain.Job, runtime *domain.JobRuntime, tr } // 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. +// Service under the lock and announces it. If the job was marked Pending while +// running (the "queue" overlap policy), and it is still enabled and the scheduler +// is not paused, the deferred run is started immediately. It runs on its own +// goroutine. func (s *Service) executeRun(ctx context.Context, jobCopy domain.Job, trigger string) { record := s.runJob(ctx, &jobCopy, trigger, s.store.Paths.LogsDir) @@ -105,9 +140,17 @@ func (s *Service) executeRun(ctx context.Context, jobCopy domain.Job, trigger st runtime.LastState = record.State runtime.Output = record.Output prependLog(runtime, record) - s.refreshNextRunLocked(current, runtime) + rerun := runtime.Pending && current.Enabled && !s.paused + runtime.Pending = false + if rerun { + // A scheduled occurrence fired while this run was active under the + // "queue" policy; start that deferred run now. + saveErr = s.startRunLocked(current, runtime, "Schedule") + } else { + s.refreshNextRunLocked(current, runtime) + saveErr = s.store.SaveJobs(s.jobs) + } cleanupErr = runner.CleanupLogs(s.store.Paths.LogsDir, s.store.Config.MaxLogFiles, s.store.Config.MaxLogAgeDays) - saveErr = s.store.SaveJobs(s.jobs) } s.mu.Unlock() @@ -121,6 +164,32 @@ func (s *Service) executeRun(ctx context.Context, jobCopy domain.Job, trigger st s.emit(JobChanged{JobID: jobCopy.ID}) } +// anyRunningLocked reports whether any loaded job is currently in the "Running" +// state. It backs the sequential-mode guards in RunNow and RunDue. The caller +// must hold mu. +func (s *Service) anyRunningLocked() bool { + for index := range s.jobs { + runtime, ok := s.runtimes[s.jobs[index].ID] + if ok && runtime != nil && runtime.LastState == "Running" { + return true + } + } + return false +} + +// advanceNextDueLocked moves a job's NextDue to the next scheduled time after +// from, leaving the NextRun display string untouched so callers can keep it +// showing "Running" during a run. A missing schedule cache (an unparseable +// schedule) zeroes NextDue. The caller must hold mu. +func (s *Service) advanceNextDueLocked(job *domain.Job, runtime *domain.JobRuntime, from time.Time) { + sched, ok := s.schedules[job.ID] + if !ok { + runtime.NextDue = time.Time{} + return + } + runtime.NextDue = sched.Next(from) +} + // runningOutput is the placeholder output shown while a job is running, before // the real command output replaces it. func runningOutput(job domain.Job, trigger string, started time.Time) string {