From 263717874cc546559f4100cce71dd2b79e87bf2a Mon Sep 17 00:00:00 2001 From: Mikhail Yenuchenko Date: Thu, 6 Aug 2026 17:06:55 +0300 Subject: [PATCH] fix: remove no-op SaveJobs calls, fix PendingRuns lifecycle and cap Implements items 4-5 of the whole-project review's suggested order (docs/PROJECT_REVIEW_PLAN.md): - Drop the three SaveJobs calls in the run lifecycle (startRunLocked, executeRun, SetGlobalPause): none of them change a durable Job field, everything they touch lives on JobRuntime, which is never persisted. Retire TestStartRunLockedRollbackOnSaveFailure with the rollback it guarded, since a run can no longer fail to start this way. - Clear PendingRuns (the "queue" overlap policy's backlog) when a job is disabled or the scheduler is globally paused, so resuming or re-enabling a job no longer replays a deferred run left over from before the pause/disable. Cap it at maxPendingRuns (10) so a job whose runs take longer than its own interval stops accumulating an unbounded backlog. Surface the queued count in the details pane via DisplayStats. Co-Authored-By: Claude Sonnet 5 --- docs/CHANGELOG.md | 7 ++++ docs/STANDARDS.md | 9 +++++ docs/TESTS.md | 5 ++- src/app/format.go | 21 ++++++++-- src/app/format_test.go | 11 ++++++ src/app/operations.go | 14 +++++-- src/app/operations_test.go | 17 ++++++++ src/app/run.go | 65 ++++++++++++------------------- src/app/run_test.go | 80 ++++++++++++++++++++++---------------- 9 files changed, 147 insertions(+), 82 deletions(-) diff --git a/docs/CHANGELOG.md b/docs/CHANGELOG.md index 0098655..967adf1 100644 --- a/docs/CHANGELOG.md +++ b/docs/CHANGELOG.md @@ -32,6 +32,13 @@ the app icon (experimental).** loss mid-write can no longer leave a truncated or empty file. `Service.Stop()` is now called when the app quits, which also makes the run context cancellation reach in-flight runs on shutdown. +- Fixed the "queue" overlap policy's backlog (`PendingRuns`): it no longer + survives a global pause or a job being disabled, so resuming or re-enabling a + job can no longer replay a deferred run left over from before the pause/ + disable. It is also capped at 10 queued occurrences, so a job whose runs take + longer than its own interval no longer accumulates an unbounded backlog that + then runs back-to-back indefinitely. The job details pane now shows the + queued-run count (", N queued") whenever it is non-zero. **Jobs:** diff --git a/docs/STANDARDS.md b/docs/STANDARDS.md index 96c628b..179efc0 100644 --- a/docs/STANDARDS.md +++ b/docs/STANDARDS.md @@ -79,6 +79,15 @@ change to their shape has to stay compatible on its own. mid-session (see [ROADMAP.md](ROADMAP.md)). - **`--start-in-tray` defers to config.** A stale autostart shortcut that still passes the flag does not hide the window when `KeepRunningInTray` is off. +- **`JobRuntime.PendingRuns` (the "queue" overlap policy's backlog) is capped at + `maxPendingRuns` (10) and cleared on pause or disable.** A job whose runs take + longer than its interval stops accumulating backlog once the cap is hit — + further overlaps are dropped like the "skip" policy until the backlog drains + below the cap. `SetGlobalPause(true)` and `SetEnabled(id, false)` both zero + the counter, so resuming or re-enabling a job never replays a deferred run for + an occurrence that fired before the pause/disable. The details pane appends + ", N queued" to the statistics line via `DisplayStats` whenever the count is + non-zero. ## Out of scope diff --git a/docs/TESTS.md b/docs/TESTS.md index f23842c..236f476 100644 --- a/docs/TESTS.md +++ b/docs/TESTS.md @@ -136,6 +136,7 @@ Tests all mutating operations on the Service, scheduler integration, and setting | `TestDeleteJobNotFound` | Verifies that `DeleteJob` returns an error for an unknown job ID. | | `TestSetEnabledNotFound` | Verifies that `SetEnabled` returns an error for an unknown job ID. | | `TestSetEnabledToggles` | Verifies that `SetEnabled` flips the enabled flag and persists the change. | +| `TestSetEnabledClearsPendingRuns` | Verifies that disabling a job zeroes a `PendingRuns` backlog it was carrying, so re-enabling it later does not replay a stale deferred run. | #### Global pause / run-now / run-due @@ -187,11 +188,11 @@ and scheduler edge cases using injected `runJob` and `primeDue`. | `TestRunDueSkipDropsOverlap` | Global skip: no second concurrent run, `PendingRuns` stays 0. | | `TestRunDueQueueRerunsAfterFinish` | Queue: one deferred run after an in-flight finish; also covers an empty per-job policy inheriting the global default. | | `TestRunDueQueueDrainsMultipleOverlaps` | Queue: multiple missed ticks drain as separate runs. | +| `TestRunDueQueueCapsPendingRuns` | Regression: `PendingRuns` stops growing at `maxPendingRuns` instead of accumulating without bound for a job that never keeps up with its schedule. | | `TestRunDuePerJobQueueOverridesGlobalSkip` | Per-job `queue` beats global `skip`. | | `TestRunDuePerJobSkipOverridesGlobalQueue` | Per-job `skip` beats global `queue`. | | `TestRunNowSequentialGuard` | Manual run refused while another job runs in sequential mode. | -| `TestStartRunLockedRollbackOnSaveFailure` | Regression: run does not start when `SaveJobs` fails. | -| `TestRunDueQueueDrainSkippedWhenPaused` | Queued overlaps are not drained while the scheduler is paused. | +| `TestRunDueQueueDrainSkippedWhenPaused` | Queued overlaps are not drained while the scheduler is paused, and pausing clears the backlog rather than leaving it to fire a stale deferred run on resume. | | `TestEffectiveTimeout` | Verifies the three-state resolution: `nil` inherits the global default, a positive value overrides it, and an explicit `0` means no timeout without inheriting. | --- diff --git a/src/app/format.go b/src/app/format.go index 739be56..6c09cf7 100644 --- a/src/app/format.go +++ b/src/app/format.go @@ -82,14 +82,29 @@ func DisplayInvocation(job domain.Job) string { return job.Command + " " + strings.ReplaceAll(strings.TrimSpace(job.Arguments), "\n", " ") } -// DisplayStats returns a one-line execution-time summary for a job runtime. -// Returns "No runs recorded" when no runs have been counted yet. +// DisplayStats returns a one-line execution-time summary for a job runtime, +// with the queued-run depth appended whenever the "queue" overlap policy has +// deferred runs waiting (see maxPendingRuns). Returns "No runs recorded" when +// no runs have been counted yet, still showing the queue depth if one exists. func DisplayStats(rt *domain.JobRuntime) string { if rt == nil || rt.RunCount == 0 { + if rt != nil && rt.PendingRuns > 0 { + return "No runs recorded" + pendingRunsSuffix(rt.PendingRuns) + } return "No runs recorded" } return fmt.Sprintf("%d runs, %d failed, last %d ms, avg %d ms, max %d ms", - rt.RunCount, rt.FailCount, rt.LastDurationMS, rt.AvgDurationMS, rt.MaxDurationMS) + rt.RunCount, rt.FailCount, rt.LastDurationMS, rt.AvgDurationMS, rt.MaxDurationMS) + + pendingRunsSuffix(rt.PendingRuns) +} + +// pendingRunsSuffix formats the queued-run depth for DisplayStats, empty when +// nothing is queued. +func pendingRunsSuffix(pending int) string { + if pending <= 0 { + return "" + } + return fmt.Sprintf(", %d queued", pending) } // DisplayOverlapPolicy formats a job's effective overlap policy for the details diff --git a/src/app/format_test.go b/src/app/format_test.go index 613bd6f..5005be0 100644 --- a/src/app/format_test.go +++ b/src/app/format_test.go @@ -126,6 +126,17 @@ func TestDisplayStats(t *testing.T) { if got := DisplayStats(rtNoFail); got != wantNoFail { t.Errorf("DisplayStats no-fail = %q, want %q", got, wantNoFail) } + + // A "queue" overlap backlog is appended to whichever form applies, so it stays + // visible even before the first run has completed. + if got, want := DisplayStats(&domain.JobRuntime{PendingRuns: 2}), "No runs recorded, 2 queued"; got != want { + t.Errorf("DisplayStats pending, no runs = %q, want %q", got, want) + } + rtPending := &domain.JobRuntime{RunCount: 5, FailCount: 2, LastDurationMS: 450, AvgDurationMS: 380, MaxDurationMS: 520, PendingRuns: 3} + wantPending := "5 runs, 2 failed, last 450 ms, avg 380 ms, max 520 ms, 3 queued" + if got := DisplayStats(rtPending); got != wantPending { + t.Errorf("DisplayStats pending = %q, want %q", got, wantPending) + } } func TestEventLine(t *testing.T) { diff --git a/src/app/operations.go b/src/app/operations.go index 5137075..af10c8c 100644 --- a/src/app/operations.go +++ b/src/app/operations.go @@ -148,6 +148,10 @@ func (s *Service) SetEnabled(id int, enabled bool) error { runtime.LastState = "Paused" runtime.NextRun = "Paused" runtime.NextDue = time.Time{} + // A disabled job's own occurrences stop firing, so a "queue" backlog it was + // carrying no longer corresponds to anything: clear it rather than replaying + // stale deferred runs if the job is re-enabled later. + runtime.PendingRuns = 0 record = uiRecord(id, job.Name, "Paused", "Job was disabled") } prependLog(runtime, record) @@ -175,12 +179,16 @@ func (s *Service) SetGlobalPause(paused bool) error { for index := range s.jobs { job := &s.jobs[index] runtime := s.runtimeForLocked(job) + if paused { + // A "queue" backlog counts occurrences missed *while paused is off*; once + // paused, none of those correspond to anything the user would expect + // replayed on resume, so drop it rather than letting a stale counter fire + // a deferred run for an occurrence from before the pause. + runtime.PendingRuns = 0 + } s.refreshNextRunFromLocked(job, runtime, now) } err := s.store.SaveConfig() - if err == nil { - err = s.store.SaveJobs(s.jobs) - } s.mu.Unlock() if err != nil { diff --git a/src/app/operations_test.go b/src/app/operations_test.go index 07b0862..05900fa 100644 --- a/src/app/operations_test.go +++ b/src/app/operations_test.go @@ -217,6 +217,23 @@ func TestSetEnabledToggles(t *testing.T) { } } +// TestSetEnabledClearsPendingRuns verifies that disabling a job drops any +// "queue" overlap backlog it was carrying, so re-enabling it later does not +// replay a deferred run for an occurrence that fired before the disable. +func TestSetEnabledClearsPendingRuns(t *testing.T) { + svc := newTempService(t, []domain.Job{{ID: 1, Name: "A", Schedule: "@every 1m", Command: "echo", Enabled: true}}) + svc.mu.Lock() + svc.runtimes[1].PendingRuns = 2 + svc.mu.Unlock() + + if err := svc.SetEnabled(1, false); err != nil { + t.Fatalf("SetEnabled false: %v", err) + } + if rt := svc.Runtime(1); rt.PendingRuns != 0 { + t.Errorf("PendingRuns after disable = %d, want 0", rt.PendingRuns) + } +} + func TestSetGlobalPauseUpdatesRuntimesAndEmits(t *testing.T) { svc := newTempService(t, []domain.Job{ {ID: 1, Name: "On", Schedule: "@every 1m", Command: "echo", Enabled: true}, diff --git a/src/app/run.go b/src/app/run.go index b00817c..40c8352 100644 --- a/src/app/run.go +++ b/src/app/run.go @@ -11,6 +11,13 @@ import ( "gitea.mixdep.ru/mix/gosentry/src/runner" ) +// maxPendingRuns bounds how many missed occurrences the "queue" overlap policy +// will defer for one job. Without a ceiling a job whose runs take longer than +// its interval would queue one more occurrence on every tick forever, so once +// the cap is reached further overlaps are dropped exactly as the "skip" policy +// would drop them, until the backlog drains below the cap again. +const maxPendingRuns = 10 + // RunNow starts a manual run of a job. Global pause stops only the scheduler's // automatic runs (see RunDue), so a manual "Run now" is allowed even while // paused — it is the user's explicit, one-off action. It will not start a job @@ -36,14 +43,12 @@ func (s *Service) RunNow(id int) error { s.mu.Unlock() return errors.New("another job is already running (sequential mode)") } - err := s.startRunLocked(job, runtime, "Manual", time.Now()) + s.startRunLocked(job, runtime, "Manual", time.Now()) s.mu.Unlock() - if err == nil { - // Reflect the "Running" transition; the run's completion emits again later. - s.emit(JobChanged{JobID: id}) - } - return err + // Reflect the "Running" transition; the run's completion emits again later. + s.emit(JobChanged{JobID: id}) + return nil } // RunDue is the scheduler's per-tick entry point: it starts whatever is due at @@ -63,7 +68,6 @@ func (s *Service) RunNow(id int) error { func (s *Service) RunDue(now time.Time) { s.mu.Lock() var started []int - var startErr error if !s.paused { sequential := s.store.Config.ExecutionMode == domain.ExecutionModeSequential running := s.anyRunningLocked() @@ -77,7 +81,7 @@ func (s *Service) RunDue(now time.Time) { // The job came due again while its own run is still in flight. // Apply the effective overlap policy and step past this // occurrence. - if s.effectiveOverlapPolicy(job) == domain.OverlapPolicyQueue { + if s.effectiveOverlapPolicy(job) == domain.OverlapPolicyQueue && runtime.PendingRuns < maxPendingRuns { runtime.PendingRuns++ } s.advanceNextDueLocked(job, runtime, now) @@ -88,19 +92,13 @@ func (s *Service) RunDue(now time.Time) { // tick once the in-flight run has finished. continue } - if err := s.startRunLocked(job, runtime, "Schedule", now); err != nil { - startErr = err - continue - } + s.startRunLocked(job, runtime, "Schedule", now) started = append(started, job.ID) running = true } } s.mu.Unlock() - if startErr != nil { - s.emit(ErrorOccurred{Err: fmt.Errorf("save jobs before scheduled run: %w", startErr)}) - } for _, id := range started { s.emit(JobChanged{JobID: id}) } @@ -116,29 +114,19 @@ type runEnv struct { } // 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. -// now is the reference time for next-due advancement and the running placeholder. -func (s *Service) startRunLocked(job *domain.Job, runtime *domain.JobRuntime, trigger string, now time.Time) error { +// scheduled occurrence, and launches the run on a background goroutine. Neither +// step touches a durable field — both live on JobRuntime, which is never +// persisted — so there is nothing to save here. 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. now is the reference time for +// next-due advancement and the running placeholder. +func (s *Service) startRunLocked(job *domain.Job, runtime *domain.JobRuntime, trigger string, now time.Time) { jobCopy := *job - prevState := runtime.LastState - prevNextRun := runtime.NextRun - prevOutput := runtime.Output - prevNextDue := runtime.NextDue - runtime.LastState = "Running" runtime.NextRun = "Running" runtime.Output = runningOutput(jobCopy, trigger, now) s.advanceNextDueLocked(job, runtime, now) - if err := s.store.SaveJobs(s.jobs); err != nil { - runtime.LastState = prevState - runtime.NextRun = prevNextRun - runtime.Output = prevOutput - runtime.NextDue = prevNextDue - return err - } env := runEnv{ logsDir: s.store.Paths.LogsDir, maxFiles: s.store.Config.MaxLogFiles, @@ -148,7 +136,6 @@ func (s *Service) startRunLocked(job *domain.Job, runtime *domain.JobRuntime, tr // 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, env) - return nil } // executeRun runs the job off the lock, then records the result back through the @@ -160,7 +147,7 @@ func (s *Service) executeRun(ctx context.Context, jobCopy domain.Job, trigger st record, logErr := s.runJob(ctx, &jobCopy, trigger, env.logsDir, env.timeout) s.mu.Lock() - var cleanupErr, saveErr error + var cleanupErr error var rerunStarted bool if current := s.findByIDLocked(jobCopy.ID); current != nil { runtime := s.runtimeForLocked(current) @@ -174,11 +161,10 @@ func (s *Service) executeRun(ctx context.Context, jobCopy domain.Job, trigger st runtime.PendingRuns-- // A scheduled occurrence fired while this run was active under the // "queue" policy; start one deferred run now. - saveErr = s.startRunLocked(current, runtime, "Schedule", time.Now()) - rerunStarted = saveErr == nil + s.startRunLocked(current, runtime, "Schedule", time.Now()) + rerunStarted = true } else { s.refreshNextRunLocked(current, runtime) - saveErr = s.store.SaveJobs(s.jobs) } cleanupErr = runner.CleanupLogs(env.logsDir, env.maxFiles, env.maxAge) } @@ -190,9 +176,6 @@ func (s *Service) executeRun(ctx context.Context, jobCopy domain.Job, trigger st if cleanupErr != nil { s.emit(ErrorOccurred{Err: fmt.Errorf("log cleanup after run %q: %w", jobCopy.Name, cleanupErr)}) } - if saveErr != nil { - s.emit(ErrorOccurred{Err: fmt.Errorf("save jobs after run %q: %w", jobCopy.Name, saveErr)}) - } s.emit(RunRecorded{Record: record}) if !rerunStarted { s.emit(JobChanged{JobID: jobCopy.ID}) diff --git a/src/app/run_test.go b/src/app/run_test.go index a0c064b..098206e 100644 --- a/src/app/run_test.go +++ b/src/app/run_test.go @@ -2,7 +2,6 @@ package app import ( "context" - "os" "sync/atomic" "testing" "time" @@ -359,6 +358,42 @@ func TestRunDueQueueDrainsMultipleOverlaps(t *testing.T) { } } +// TestRunDueQueueCapsPendingRuns verifies that a job whose runs never keep up +// with its schedule stops accumulating PendingRuns at maxPendingRuns instead of +// growing without bound. +func TestRunDueQueueCapsPendingRuns(t *testing.T) { + svc := newQueueService(t, domain.ExecutionModeParallel, domain.OverlapPolicyQueue, []domain.Job{ + {ID: 1, Name: "A", Schedule: "@every 1h", Command: "echo", Enabled: true}, + }) + + release := make(chan struct{}) + svc.runJob = func(_ context.Context, job *domain.Job, _ string, _ string, _ time.Duration) (domain.RunRecord, error) { + <-release + return domain.RunRecord{Time: "t", JobID: job.ID, JobName: job.Name, State: "Success"}, nil + } + done := completions(svc) + t.Cleanup(func() { + close(release) + waitRecord(t, done) + }) + + primeDue(t, svc, 1) + svc.RunDue(time.Now()) + + // Far more due ticks than the cap while the first run stays in flight. + for range maxPendingRuns + 5 { + primeDue(t, svc, 1) + svc.RunDue(time.Now()) + } + + svc.mu.Lock() + pending := svc.runtimes[1].PendingRuns + svc.mu.Unlock() + if pending != maxPendingRuns { + t.Fatalf("PendingRuns = %d, want capped at %d", pending, maxPendingRuns) + } +} + // TestRunDuePerJobQueueOverridesGlobalSkip verifies that a job carrying its own // "queue" policy queues a re-run even though the global default is "skip": the // effective policy is resolved per job, so the job-level value wins. @@ -493,37 +528,9 @@ func TestRunNowSequentialGuard(t *testing.T) { waitRecord(t, done) } -// TestStartRunLockedRollbackOnSaveFailure is a regression test for CODE_REVIEW -// finding #2: a run must not start when persisting the Running state fails. -func TestStartRunLockedRollbackOnSaveFailure(t *testing.T) { - svc := newTempService(t, []domain.Job{{ID: 1, Name: "A", Schedule: "@every 1h", Command: "echo", Enabled: true}}) - if err := svc.store.SaveJobs(svc.jobs); err != nil { - t.Fatalf("seed jobs.json: %v", err) - } - if err := os.Chmod(svc.store.Paths.JobsPath, 0o444); err != nil { - t.Fatalf("chmod jobs.json: %v", err) - } - t.Cleanup(func() { _ = os.Chmod(svc.store.Paths.JobsPath, 0o644) }) - - var started int32 - svc.runJob = func(_ context.Context, job *domain.Job, _ string, _ string, _ time.Duration) (domain.RunRecord, error) { - atomic.AddInt32(&started, 1) - return domain.RunRecord{Time: "t", JobID: job.ID, JobName: job.Name, State: "OK"}, nil - } - - if err := svc.RunNow(1); err == nil { - t.Fatal("expected RunNow to fail when jobs.json is not writable") - } - if atomic.LoadInt32(&started) != 0 { - t.Error("run goroutine must not start when SaveJobs fails") - } - if rt := svc.Runtime(1); rt == nil || rt.LastState == "Running" { - t.Errorf("runtime should roll back from Running, got %+v", rt) - } -} - // TestRunDueQueueDrainSkippedWhenPaused verifies that queued overlap runs are not -// drained while the scheduler is globally paused. +// drained while the scheduler is globally paused, and that pausing clears the +// backlog rather than leaving it to fire a stale deferred run on resume. func TestRunDueQueueDrainSkippedWhenPaused(t *testing.T) { svc := newQueueService(t, domain.ExecutionModeParallel, domain.OverlapPolicyQueue, []domain.Job{ {ID: 1, Name: "A", Schedule: "@every 1h", Command: "echo", Enabled: true}, @@ -561,6 +568,13 @@ func TestRunDueQueueDrainSkippedWhenPaused(t *testing.T) { t.Fatalf("SetGlobalPause: %v", err) } + svc.mu.Lock() + pending = svc.runtimes[1].PendingRuns + svc.mu.Unlock() + if pending != 0 { + t.Errorf("pausing must clear a queued backlog, PendingRuns = %d, want 0", pending) + } + close(release) waitRecord(t, done) expectNoEntry(t, entered) @@ -568,8 +582,8 @@ func TestRunDueQueueDrainSkippedWhenPaused(t *testing.T) { svc.mu.Lock() pending = svc.runtimes[1].PendingRuns svc.mu.Unlock() - if pending != 1 { - t.Errorf("paused scheduler must not drain queue, PendingRuns = %d, want 1", pending) + if pending != 0 { + t.Errorf("paused scheduler must not drain queue, PendingRuns = %d, want 0", pending) } if got := atomic.LoadInt32(&calls); got != 1 { t.Errorf("runner called %d time(s), want 1", got)