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 <noreply@anthropic.com>
This commit is contained in:
@@ -32,6 +32,13 @@ the app icon (experimental).**
|
|||||||
loss mid-write can no longer leave a truncated or empty file. `Service.Stop()`
|
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
|
is now called when the app quits, which also makes the run context
|
||||||
cancellation reach in-flight runs on shutdown.
|
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:**
|
**Jobs:**
|
||||||
|
|
||||||
|
|||||||
@@ -79,6 +79,15 @@ change to their shape has to stay compatible on its own.
|
|||||||
mid-session (see [ROADMAP.md](ROADMAP.md)).
|
mid-session (see [ROADMAP.md](ROADMAP.md)).
|
||||||
- **`--start-in-tray` defers to config.** A stale autostart shortcut that still
|
- **`--start-in-tray` defers to config.** A stale autostart shortcut that still
|
||||||
passes the flag does not hide the window when `KeepRunningInTray` is off.
|
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
|
## Out of scope
|
||||||
|
|
||||||
|
|||||||
+3
-2
@@ -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. |
|
| `TestDeleteJobNotFound` | Verifies that `DeleteJob` returns an error for an unknown job ID. |
|
||||||
| `TestSetEnabledNotFound` | Verifies that `SetEnabled` 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. |
|
| `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
|
#### 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. |
|
| `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. |
|
| `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. |
|
| `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`. |
|
| `TestRunDuePerJobQueueOverridesGlobalSkip` | Per-job `queue` beats global `skip`. |
|
||||||
| `TestRunDuePerJobSkipOverridesGlobalQueue` | Per-job `skip` beats global `queue`. |
|
| `TestRunDuePerJobSkipOverridesGlobalQueue` | Per-job `skip` beats global `queue`. |
|
||||||
| `TestRunNowSequentialGuard` | Manual run refused while another job runs in sequential mode. |
|
| `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, and pausing clears the backlog rather than leaving it to fire a stale deferred run on resume. |
|
||||||
| `TestRunDueQueueDrainSkippedWhenPaused` | Queued overlaps are not drained while the scheduler is paused. |
|
|
||||||
| `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. |
|
| `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. |
|
||||||
|
|
||||||
---
|
---
|
||||||
|
|||||||
+18
-3
@@ -82,14 +82,29 @@ func DisplayInvocation(job domain.Job) string {
|
|||||||
return job.Command + " " + strings.ReplaceAll(strings.TrimSpace(job.Arguments), "\n", " ")
|
return job.Command + " " + strings.ReplaceAll(strings.TrimSpace(job.Arguments), "\n", " ")
|
||||||
}
|
}
|
||||||
|
|
||||||
// DisplayStats returns a one-line execution-time summary for a job runtime.
|
// DisplayStats returns a one-line execution-time summary for a job runtime,
|
||||||
// Returns "No runs recorded" when no runs have been counted yet.
|
// 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 {
|
func DisplayStats(rt *domain.JobRuntime) string {
|
||||||
if rt == nil || rt.RunCount == 0 {
|
if rt == nil || rt.RunCount == 0 {
|
||||||
|
if rt != nil && rt.PendingRuns > 0 {
|
||||||
|
return "No runs recorded" + pendingRunsSuffix(rt.PendingRuns)
|
||||||
|
}
|
||||||
return "No runs recorded"
|
return "No runs recorded"
|
||||||
}
|
}
|
||||||
return fmt.Sprintf("%d runs, %d failed, last %d ms, avg %d ms, max %d ms",
|
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
|
// DisplayOverlapPolicy formats a job's effective overlap policy for the details
|
||||||
|
|||||||
@@ -126,6 +126,17 @@ func TestDisplayStats(t *testing.T) {
|
|||||||
if got := DisplayStats(rtNoFail); got != wantNoFail {
|
if got := DisplayStats(rtNoFail); got != wantNoFail {
|
||||||
t.Errorf("DisplayStats no-fail = %q, want %q", 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) {
|
func TestEventLine(t *testing.T) {
|
||||||
|
|||||||
+11
-3
@@ -148,6 +148,10 @@ func (s *Service) SetEnabled(id int, enabled bool) error {
|
|||||||
runtime.LastState = "Paused"
|
runtime.LastState = "Paused"
|
||||||
runtime.NextRun = "Paused"
|
runtime.NextRun = "Paused"
|
||||||
runtime.NextDue = time.Time{}
|
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")
|
record = uiRecord(id, job.Name, "Paused", "Job was disabled")
|
||||||
}
|
}
|
||||||
prependLog(runtime, record)
|
prependLog(runtime, record)
|
||||||
@@ -175,12 +179,16 @@ func (s *Service) SetGlobalPause(paused bool) error {
|
|||||||
for index := range s.jobs {
|
for index := range s.jobs {
|
||||||
job := &s.jobs[index]
|
job := &s.jobs[index]
|
||||||
runtime := s.runtimeForLocked(job)
|
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)
|
s.refreshNextRunFromLocked(job, runtime, now)
|
||||||
}
|
}
|
||||||
err := s.store.SaveConfig()
|
err := s.store.SaveConfig()
|
||||||
if err == nil {
|
|
||||||
err = s.store.SaveJobs(s.jobs)
|
|
||||||
}
|
|
||||||
s.mu.Unlock()
|
s.mu.Unlock()
|
||||||
|
|
||||||
if err != nil {
|
if err != nil {
|
||||||
|
|||||||
@@ -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) {
|
func TestSetGlobalPauseUpdatesRuntimesAndEmits(t *testing.T) {
|
||||||
svc := newTempService(t, []domain.Job{
|
svc := newTempService(t, []domain.Job{
|
||||||
{ID: 1, Name: "On", Schedule: "@every 1m", Command: "echo", Enabled: true},
|
{ID: 1, Name: "On", Schedule: "@every 1m", Command: "echo", Enabled: true},
|
||||||
|
|||||||
+24
-41
@@ -11,6 +11,13 @@ import (
|
|||||||
"gitea.mixdep.ru/mix/gosentry/src/runner"
|
"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
|
// 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
|
// 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
|
// 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()
|
s.mu.Unlock()
|
||||||
return errors.New("another job is already running (sequential mode)")
|
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()
|
s.mu.Unlock()
|
||||||
|
|
||||||
if err == nil {
|
// Reflect the "Running" transition; the run's completion emits again later.
|
||||||
// Reflect the "Running" transition; the run's completion emits again later.
|
s.emit(JobChanged{JobID: id})
|
||||||
s.emit(JobChanged{JobID: id})
|
return nil
|
||||||
}
|
|
||||||
return err
|
|
||||||
}
|
}
|
||||||
|
|
||||||
// RunDue is the scheduler's per-tick entry point: it starts whatever is due at
|
// 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) {
|
func (s *Service) RunDue(now time.Time) {
|
||||||
s.mu.Lock()
|
s.mu.Lock()
|
||||||
var started []int
|
var started []int
|
||||||
var startErr error
|
|
||||||
if !s.paused {
|
if !s.paused {
|
||||||
sequential := s.store.Config.ExecutionMode == domain.ExecutionModeSequential
|
sequential := s.store.Config.ExecutionMode == domain.ExecutionModeSequential
|
||||||
running := s.anyRunningLocked()
|
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.
|
// The job came due again while its own run is still in flight.
|
||||||
// Apply the effective overlap policy and step past this
|
// Apply the effective overlap policy and step past this
|
||||||
// occurrence.
|
// occurrence.
|
||||||
if s.effectiveOverlapPolicy(job) == domain.OverlapPolicyQueue {
|
if s.effectiveOverlapPolicy(job) == domain.OverlapPolicyQueue && runtime.PendingRuns < maxPendingRuns {
|
||||||
runtime.PendingRuns++
|
runtime.PendingRuns++
|
||||||
}
|
}
|
||||||
s.advanceNextDueLocked(job, runtime, now)
|
s.advanceNextDueLocked(job, runtime, now)
|
||||||
@@ -88,19 +92,13 @@ func (s *Service) RunDue(now time.Time) {
|
|||||||
// tick once the in-flight run has finished.
|
// tick once the in-flight run has finished.
|
||||||
continue
|
continue
|
||||||
}
|
}
|
||||||
if err := s.startRunLocked(job, runtime, "Schedule", now); err != nil {
|
s.startRunLocked(job, runtime, "Schedule", now)
|
||||||
startErr = err
|
|
||||||
continue
|
|
||||||
}
|
|
||||||
started = append(started, job.ID)
|
started = append(started, job.ID)
|
||||||
running = true
|
running = true
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
s.mu.Unlock()
|
s.mu.Unlock()
|
||||||
|
|
||||||
if startErr != nil {
|
|
||||||
s.emit(ErrorOccurred{Err: fmt.Errorf("save jobs before scheduled run: %w", startErr)})
|
|
||||||
}
|
|
||||||
for _, id := range started {
|
for _, id := range started {
|
||||||
s.emit(JobChanged{JobID: id})
|
s.emit(JobChanged{JobID: id})
|
||||||
}
|
}
|
||||||
@@ -116,29 +114,19 @@ type runEnv struct {
|
|||||||
}
|
}
|
||||||
|
|
||||||
// startRunLocked transitions a job to "Running", advances its NextDue to the next
|
// startRunLocked transitions a job to "Running", advances its NextDue to the next
|
||||||
// scheduled occurrence, persists that, and launches the run on a background
|
// scheduled occurrence, and launches the run on a background goroutine. Neither
|
||||||
// goroutine. Advancing (rather than zeroing) NextDue keeps the schedule marching
|
// step touches a durable field — both live on JobRuntime, which is never
|
||||||
// while the run is in flight, which is what lets RunDue notice a fresh occurrence
|
// persisted — so there is nothing to save here. Advancing (rather than zeroing)
|
||||||
// firing during a long run and apply the overlap policy. The caller must hold mu.
|
// NextDue keeps the schedule marching while the run is in flight, which is what
|
||||||
// now is the reference time for next-due advancement and the running placeholder.
|
// lets RunDue notice a fresh occurrence firing during a long run and apply the
|
||||||
func (s *Service) startRunLocked(job *domain.Job, runtime *domain.JobRuntime, trigger string, now time.Time) error {
|
// 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
|
jobCopy := *job
|
||||||
prevState := runtime.LastState
|
|
||||||
prevNextRun := runtime.NextRun
|
|
||||||
prevOutput := runtime.Output
|
|
||||||
prevNextDue := runtime.NextDue
|
|
||||||
|
|
||||||
runtime.LastState = "Running"
|
runtime.LastState = "Running"
|
||||||
runtime.NextRun = "Running"
|
runtime.NextRun = "Running"
|
||||||
runtime.Output = runningOutput(jobCopy, trigger, now)
|
runtime.Output = runningOutput(jobCopy, trigger, now)
|
||||||
s.advanceNextDueLocked(job, runtime, 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{
|
env := runEnv{
|
||||||
logsDir: s.store.Paths.LogsDir,
|
logsDir: s.store.Paths.LogsDir,
|
||||||
maxFiles: s.store.Config.MaxLogFiles,
|
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
|
// Capture ctx under the lock so a concurrent Start/Stop cannot swap it out
|
||||||
// from under the goroutine after we release mu.
|
// from under the goroutine after we release mu.
|
||||||
go s.executeRun(s.ctx, jobCopy, trigger, env)
|
go s.executeRun(s.ctx, jobCopy, trigger, env)
|
||||||
return nil
|
|
||||||
}
|
}
|
||||||
|
|
||||||
// executeRun runs the job off the lock, then records the result back through the
|
// 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)
|
record, logErr := s.runJob(ctx, &jobCopy, trigger, env.logsDir, env.timeout)
|
||||||
|
|
||||||
s.mu.Lock()
|
s.mu.Lock()
|
||||||
var cleanupErr, saveErr error
|
var cleanupErr error
|
||||||
var rerunStarted bool
|
var rerunStarted bool
|
||||||
if current := s.findByIDLocked(jobCopy.ID); current != nil {
|
if current := s.findByIDLocked(jobCopy.ID); current != nil {
|
||||||
runtime := s.runtimeForLocked(current)
|
runtime := s.runtimeForLocked(current)
|
||||||
@@ -174,11 +161,10 @@ func (s *Service) executeRun(ctx context.Context, jobCopy domain.Job, trigger st
|
|||||||
runtime.PendingRuns--
|
runtime.PendingRuns--
|
||||||
// A scheduled occurrence fired while this run was active under the
|
// A scheduled occurrence fired while this run was active under the
|
||||||
// "queue" policy; start one deferred run now.
|
// "queue" policy; start one deferred run now.
|
||||||
saveErr = s.startRunLocked(current, runtime, "Schedule", time.Now())
|
s.startRunLocked(current, runtime, "Schedule", time.Now())
|
||||||
rerunStarted = saveErr == nil
|
rerunStarted = true
|
||||||
} else {
|
} else {
|
||||||
s.refreshNextRunLocked(current, runtime)
|
s.refreshNextRunLocked(current, runtime)
|
||||||
saveErr = s.store.SaveJobs(s.jobs)
|
|
||||||
}
|
}
|
||||||
cleanupErr = runner.CleanupLogs(env.logsDir, env.maxFiles, env.maxAge)
|
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 {
|
if cleanupErr != nil {
|
||||||
s.emit(ErrorOccurred{Err: fmt.Errorf("log cleanup after run %q: %w", jobCopy.Name, cleanupErr)})
|
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})
|
s.emit(RunRecorded{Record: record})
|
||||||
if !rerunStarted {
|
if !rerunStarted {
|
||||||
s.emit(JobChanged{JobID: jobCopy.ID})
|
s.emit(JobChanged{JobID: jobCopy.ID})
|
||||||
|
|||||||
+47
-33
@@ -2,7 +2,6 @@ package app
|
|||||||
|
|
||||||
import (
|
import (
|
||||||
"context"
|
"context"
|
||||||
"os"
|
|
||||||
"sync/atomic"
|
"sync/atomic"
|
||||||
"testing"
|
"testing"
|
||||||
"time"
|
"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
|
// TestRunDuePerJobQueueOverridesGlobalSkip verifies that a job carrying its own
|
||||||
// "queue" policy queues a re-run even though the global default is "skip": the
|
// "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.
|
// effective policy is resolved per job, so the job-level value wins.
|
||||||
@@ -493,37 +528,9 @@ func TestRunNowSequentialGuard(t *testing.T) {
|
|||||||
waitRecord(t, done)
|
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
|
// 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) {
|
func TestRunDueQueueDrainSkippedWhenPaused(t *testing.T) {
|
||||||
svc := newQueueService(t, domain.ExecutionModeParallel, domain.OverlapPolicyQueue, []domain.Job{
|
svc := newQueueService(t, domain.ExecutionModeParallel, domain.OverlapPolicyQueue, []domain.Job{
|
||||||
{ID: 1, Name: "A", Schedule: "@every 1h", Command: "echo", Enabled: true},
|
{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)
|
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)
|
close(release)
|
||||||
waitRecord(t, done)
|
waitRecord(t, done)
|
||||||
expectNoEntry(t, entered)
|
expectNoEntry(t, entered)
|
||||||
@@ -568,8 +582,8 @@ func TestRunDueQueueDrainSkippedWhenPaused(t *testing.T) {
|
|||||||
svc.mu.Lock()
|
svc.mu.Lock()
|
||||||
pending = svc.runtimes[1].PendingRuns
|
pending = svc.runtimes[1].PendingRuns
|
||||||
svc.mu.Unlock()
|
svc.mu.Unlock()
|
||||||
if pending != 1 {
|
if pending != 0 {
|
||||||
t.Errorf("paused scheduler must not drain queue, PendingRuns = %d, want 1", pending)
|
t.Errorf("paused scheduler must not drain queue, PendingRuns = %d, want 0", pending)
|
||||||
}
|
}
|
||||||
if got := atomic.LoadInt32(&calls); got != 1 {
|
if got := atomic.LoadInt32(&calls); got != 1 {
|
||||||
t.Errorf("runner called %d time(s), want 1", got)
|
t.Errorf("runner called %d time(s), want 1", got)
|
||||||
|
|||||||
Reference in New Issue
Block a user