Phase 11 of PROJECT_REVIEW_PLAN.md: the themed cleanup pass over every low-severity finding still open (2.2-2.3, 3.4-3.6, 4.3-4.7, 6.4-6.7, 7.1-7.3, 8.2-8.3, 9.1-9.4, and the under-documented decisions in §10/§11). Behavioral fixes: - Reassign duplicate job IDs in a hand-edited jobs.json instead of letting two jobs share one runtime, schedule entry, and SeedStats bucket. - Disambiguate run-log file names that collide within the same second. - Compute AvgDurationMS as DurationSumMS/TimedRunCount instead of an incremental integer mean, so it always matches the seeded-from-logs average instead of drifting from truncation error. - Clean absolute paths in ResolveConfiguredPath so two spellings of the same jobs file do not trigger a spurious adoption. - Report InstallDesktopIcon failures through ErrorOccurred instead of discarding them silently. - Move settingsView's blocking AutostartStatus (PowerShell on Windows) off the UI thread. - Give notify-timing.tsv its own extension so CleanupLogs no longer manages it as a run log. - Replace the settingsView Save handler's second copy of validateConfig's rules with a bare parse, letting the Service's own error surface. Cleanups: - Delete collectActivity, the dead yaml tags on RunRecord, and the logArguments/LogArguments alias. - Fold the two systemTrayRegistered/mainWindowHidden globals into one trayState instance Run owns and threads through Settings and the single-instance reveal path. - Fix stale comments/docs: the frozen window-size restore claim, a reference to a renamed recordRun, README's "Pause all" and notification wording, the PowerShell quoting note for TESTS.md's coverage command, and scripts/test.bat's UTF-8 checkmarks under a non-UTF-8 code page. - Document the single-instance fallback's consequence and the unauthenticated instance-channel port in STANDARDS.md; record the config-shim retirement plan in ROADMAP.md. 3.5, 7.3, and 9.4 turned out to already be fixed by earlier phases; no change needed for those three. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
631 lines
21 KiB
Go
631 lines
21 KiB
Go
package app
|
|
|
|
import (
|
|
"context"
|
|
"sync/atomic"
|
|
"testing"
|
|
"time"
|
|
|
|
"gitea.mixdep.ru/mix/gosentry/src/domain"
|
|
)
|
|
|
|
// newQueueService builds a temp-backed Service with a chosen execution mode and
|
|
// overlap policy. Config is set before any run starts, so mutating it directly is
|
|
// safe (construction is single-threaded).
|
|
func newQueueService(t *testing.T, mode domain.ExecutionMode, policy domain.OverlapPolicy, jobs []domain.Job) *Service {
|
|
t.Helper()
|
|
svc := newTempService(t, jobs)
|
|
svc.store.Config.ExecutionMode = mode
|
|
svc.store.Config.OverlapPolicy = policy
|
|
return svc
|
|
}
|
|
|
|
// primeDue forces the given jobs to be due by backdating their NextDue. Tests tick
|
|
// RunDue at the real wall clock; a long "@every 1h" schedule then advances a
|
|
// started job's NextDue an hour out, so it does not spuriously re-fire on a later
|
|
// tick unless a test re-primes it.
|
|
func primeDue(t *testing.T, svc *Service, ids ...int) {
|
|
t.Helper()
|
|
svc.mu.Lock()
|
|
defer svc.mu.Unlock()
|
|
for _, id := range ids {
|
|
if rt := svc.runtimes[id]; rt != nil {
|
|
rt.NextDue = time.Now().Add(-time.Second)
|
|
}
|
|
}
|
|
}
|
|
|
|
// completions subscribes a recorder that forwards every RunRecorded onto a
|
|
// channel, so tests can wait for runs to finish (and drain in-flight runs before
|
|
// the temp dir is cleaned up).
|
|
func completions(svc *Service) <-chan domain.RunRecord {
|
|
ch := make(chan domain.RunRecord, 16)
|
|
svc.Subscribe(ObserverFunc(func(e Event) {
|
|
if rr, ok := e.(RunRecorded); ok {
|
|
ch <- rr.Record
|
|
}
|
|
}))
|
|
return ch
|
|
}
|
|
|
|
func waitRecord(t *testing.T, ch <-chan domain.RunRecord) domain.RunRecord {
|
|
t.Helper()
|
|
select {
|
|
case r := <-ch:
|
|
return r
|
|
case <-time.After(2 * time.Second):
|
|
t.Fatal("timed out waiting for a run record")
|
|
return domain.RunRecord{}
|
|
}
|
|
}
|
|
|
|
func expectNoEntry(t *testing.T, entered <-chan int) {
|
|
t.Helper()
|
|
select {
|
|
case id := <-entered:
|
|
t.Fatalf("a job (%d) started unexpectedly", id)
|
|
case <-time.After(200 * time.Millisecond):
|
|
}
|
|
}
|
|
|
|
// TestUpdateStats verifies that aggregate statistics are folded correctly after
|
|
// a sequence of fake runs with varying durations and states.
|
|
func TestUpdateStats(t *testing.T) {
|
|
rt := &domain.JobRuntime{}
|
|
|
|
// First run: success, 200 ms.
|
|
updateStats(rt, domain.RunRecord{State: "OK", DurationMS: 200})
|
|
if rt.RunCount != 1 || rt.FailCount != 0 {
|
|
t.Fatalf("after run 1: RunCount=%d FailCount=%d, want 1/0", rt.RunCount, rt.FailCount)
|
|
}
|
|
if rt.LastDurationMS != 200 || rt.MaxDurationMS != 200 || rt.AvgDurationMS != 200 {
|
|
t.Errorf("after run 1: last=%d max=%d avg=%d, want 200/200/200",
|
|
rt.LastDurationMS, rt.MaxDurationMS, rt.AvgDurationMS)
|
|
}
|
|
|
|
// Second run: failure, 400 ms.
|
|
updateStats(rt, domain.RunRecord{State: "Failed", DurationMS: 400})
|
|
if rt.RunCount != 2 || rt.FailCount != 1 {
|
|
t.Fatalf("after run 2: RunCount=%d FailCount=%d, want 2/1", rt.RunCount, rt.FailCount)
|
|
}
|
|
if rt.LastDurationMS != 400 || rt.MaxDurationMS != 400 {
|
|
t.Errorf("after run 2: last=%d max=%d, want 400/400", rt.LastDurationMS, rt.MaxDurationMS)
|
|
}
|
|
if rt.AvgDurationMS != 300 {
|
|
t.Errorf("after run 2: avg=%d, want 300", rt.AvgDurationMS)
|
|
}
|
|
|
|
// Third run: success, 100 ms — avg should be (200+400+100)/3 = 233.
|
|
updateStats(rt, domain.RunRecord{State: "OK", DurationMS: 100})
|
|
if rt.LastDurationMS != 100 || rt.MaxDurationMS != 400 {
|
|
t.Errorf("after run 3: last=%d max=%d, want 100/400", rt.LastDurationMS, rt.MaxDurationMS)
|
|
}
|
|
if rt.AvgDurationMS != 233 {
|
|
t.Errorf("after run 3: avg=%d, want 233", rt.AvgDurationMS)
|
|
}
|
|
// AvgDurationMS must always be exactly DurationSumMS/TimedRunCount — a stored
|
|
// sum divided once, not an incremental mean that truncates on every step and
|
|
// compounds error over a long-running job.
|
|
if rt.DurationSumMS != 700 {
|
|
t.Errorf("DurationSumMS = %d, want 700", rt.DurationSumMS)
|
|
}
|
|
if want := rt.DurationSumMS / int64(rt.TimedRunCount); rt.AvgDurationMS != want {
|
|
t.Errorf("AvgDurationMS = %d, want DurationSumMS/TimedRunCount = %d", rt.AvgDurationMS, want)
|
|
}
|
|
}
|
|
|
|
func TestUpdateStatsSkipsZeroDuration(t *testing.T) {
|
|
rt := &domain.JobRuntime{}
|
|
updateStats(rt, domain.RunRecord{State: "OK", DurationMS: 200})
|
|
updateStats(rt, domain.RunRecord{State: "OK", DurationMS: 0})
|
|
if rt.RunCount != 2 {
|
|
t.Fatalf("RunCount = %d, want 2", rt.RunCount)
|
|
}
|
|
if rt.TimedRunCount != 1 {
|
|
t.Fatalf("TimedRunCount = %d, want 1", rt.TimedRunCount)
|
|
}
|
|
if rt.AvgDurationMS != 200 {
|
|
t.Errorf("AvgDurationMS = %d, want 200 (zero-duration run excluded)", rt.AvgDurationMS)
|
|
}
|
|
if rt.LastDurationMS != 200 {
|
|
t.Errorf("LastDurationMS = %d, want 200", rt.LastDurationMS)
|
|
}
|
|
}
|
|
|
|
// TestRunDueParallelStartsAllDueJobs verifies that in parallel mode every due job
|
|
// starts at once: both runs are in flight (blocked in the runner) before either
|
|
// is released.
|
|
func TestRunDueParallelStartsAllDueJobs(t *testing.T) {
|
|
svc := newQueueService(t, domain.ExecutionModeParallel, domain.OverlapPolicySkip, []domain.Job{
|
|
{ID: 1, Name: "A", Schedule: "@every 1h", Command: "echo", Enabled: true},
|
|
{ID: 2, Name: "B", Schedule: "@every 1h", Command: "echo", Enabled: true},
|
|
})
|
|
|
|
entered := make(chan int, 2)
|
|
release := make(chan struct{})
|
|
svc.runJob = func(_ context.Context, job *domain.Job, _ string, _ string, _ time.Duration) (domain.RunRecord, error) {
|
|
entered <- job.ID
|
|
<-release
|
|
return domain.RunRecord{Time: "t", JobID: job.ID, JobName: job.Name, State: "Success"}, nil
|
|
}
|
|
done := completions(svc)
|
|
|
|
primeDue(t, svc, 1, 2)
|
|
svc.RunDue(time.Now())
|
|
|
|
// Both runs must reach the runner (and block) before any is released, which is
|
|
// only possible if RunDue started them concurrently.
|
|
got := map[int]bool{}
|
|
for i := 0; i < 2; i++ {
|
|
select {
|
|
case id := <-entered:
|
|
got[id] = true
|
|
case <-time.After(2 * time.Second):
|
|
t.Fatalf("only %d job(s) started in parallel, want 2", len(got))
|
|
}
|
|
}
|
|
if !got[1] || !got[2] {
|
|
t.Fatalf("started jobs = %v, want both 1 and 2", got)
|
|
}
|
|
|
|
close(release)
|
|
waitRecord(t, done)
|
|
waitRecord(t, done)
|
|
}
|
|
|
|
// TestRunDueSequentialSerializes verifies that in sequential mode only one due job
|
|
// runs at a time: the second due job waits until the first finishes and a later
|
|
// tick picks it up.
|
|
func TestRunDueSequentialSerializes(t *testing.T) {
|
|
svc := newQueueService(t, domain.ExecutionModeSequential, domain.OverlapPolicySkip, []domain.Job{
|
|
{ID: 1, Name: "A", Schedule: "@every 1h", Command: "echo", Enabled: true},
|
|
{ID: 2, Name: "B", Schedule: "@every 1h", Command: "echo", Enabled: true},
|
|
})
|
|
|
|
entered := make(chan int, 2)
|
|
release := make(chan struct{})
|
|
svc.runJob = func(_ context.Context, job *domain.Job, _ string, _ string, _ time.Duration) (domain.RunRecord, error) {
|
|
entered <- job.ID
|
|
<-release
|
|
return domain.RunRecord{Time: "t", JobID: job.ID, JobName: job.Name, State: "Success"}, nil
|
|
}
|
|
done := completions(svc)
|
|
|
|
primeDue(t, svc, 1, 2)
|
|
svc.RunDue(time.Now())
|
|
|
|
// Exactly one job — the first in order — starts; the second is held back.
|
|
if id := <-entered; id != 1 {
|
|
t.Fatalf("first started job = %d, want 1", id)
|
|
}
|
|
expectNoEntry(t, entered)
|
|
|
|
// Let the first run finish, then tick again: now the second job runs.
|
|
close(release)
|
|
waitRecord(t, done)
|
|
svc.RunDue(time.Now())
|
|
if id := <-entered; id != 2 {
|
|
t.Fatalf("second started job = %d, want 2", id)
|
|
}
|
|
waitRecord(t, done)
|
|
}
|
|
|
|
// TestRunDueSkipDropsOverlap verifies that under the "skip" overlap policy a job
|
|
// coming due again while its own run is in flight does not queue or start a second
|
|
// run.
|
|
func TestRunDueSkipDropsOverlap(t *testing.T) {
|
|
svc := newQueueService(t, domain.ExecutionModeParallel, domain.OverlapPolicySkip, []domain.Job{
|
|
{ID: 1, Name: "A", Schedule: "@every 1h", Command: "echo", Enabled: true},
|
|
})
|
|
|
|
entered := make(chan int, 2)
|
|
release := make(chan struct{})
|
|
var calls int32
|
|
svc.runJob = func(_ context.Context, job *domain.Job, _ string, _ string, _ time.Duration) (domain.RunRecord, error) {
|
|
atomic.AddInt32(&calls, 1)
|
|
entered <- job.ID
|
|
<-release
|
|
return domain.RunRecord{Time: "t", JobID: job.ID, JobName: job.Name, State: "Success"}, nil
|
|
}
|
|
done := completions(svc)
|
|
|
|
primeDue(t, svc, 1)
|
|
svc.RunDue(time.Now())
|
|
if id := <-entered; id != 1 {
|
|
t.Fatalf("started job = %d, want 1", id)
|
|
}
|
|
|
|
// The job is now in flight; make it due again and tick. Skip must drop it.
|
|
primeDue(t, svc, 1)
|
|
svc.RunDue(time.Now())
|
|
expectNoEntry(t, entered)
|
|
|
|
svc.mu.Lock()
|
|
pending := svc.runtimes[1].PendingRuns
|
|
svc.mu.Unlock()
|
|
if pending != 0 {
|
|
t.Error("skip policy must not queue deferred runs")
|
|
}
|
|
|
|
close(release)
|
|
waitRecord(t, done)
|
|
// No re-run is queued, so the runner is invoked exactly once.
|
|
expectNoEntry(t, entered)
|
|
if got := atomic.LoadInt32(&calls); got != 1 {
|
|
t.Errorf("runner called %d time(s), want 1", got)
|
|
}
|
|
}
|
|
|
|
// TestRunDueQueueRerunsAfterFinish verifies that under the "queue" overlap policy a
|
|
// job coming due again while running increments PendingRuns and re-runs after the
|
|
// in-flight run finishes.
|
|
func TestRunDueQueueRerunsAfterFinish(t *testing.T) {
|
|
svc := newQueueService(t, domain.ExecutionModeParallel, domain.OverlapPolicyQueue, []domain.Job{
|
|
{ID: 1, Name: "A", Schedule: "@every 1h", Command: "echo", Enabled: true},
|
|
})
|
|
// Empty per-job OverlapPolicy inherits the global queue policy.
|
|
if svc.jobs[0].OverlapPolicy != "" {
|
|
t.Fatalf("test setup: job OverlapPolicy = %q, want empty (inherit)", svc.jobs[0].OverlapPolicy)
|
|
}
|
|
|
|
entered := make(chan int, 2)
|
|
release := make(chan struct{})
|
|
var calls int32
|
|
svc.runJob = func(_ context.Context, job *domain.Job, _ string, _ string, _ time.Duration) (domain.RunRecord, error) {
|
|
atomic.AddInt32(&calls, 1)
|
|
entered <- job.ID
|
|
<-release
|
|
return domain.RunRecord{Time: "t", JobID: job.ID, JobName: job.Name, State: "Success"}, nil
|
|
}
|
|
done := completions(svc)
|
|
|
|
primeDue(t, svc, 1)
|
|
svc.RunDue(time.Now())
|
|
if id := <-entered; id != 1 {
|
|
t.Fatalf("started job = %d, want 1", id)
|
|
}
|
|
|
|
// Re-due the running job and tick: queue must increment PendingRuns without
|
|
// starting a second concurrent run.
|
|
primeDue(t, svc, 1)
|
|
svc.RunDue(time.Now())
|
|
expectNoEntry(t, entered)
|
|
|
|
svc.mu.Lock()
|
|
pending := svc.runtimes[1].PendingRuns
|
|
svc.mu.Unlock()
|
|
if pending != 1 {
|
|
t.Fatalf("queue policy must queue one deferred run, PendingRuns = %d", pending)
|
|
}
|
|
|
|
// Releasing the first run lets executeRun start the deferred run automatically.
|
|
close(release)
|
|
waitRecord(t, done)
|
|
if id := <-entered; id != 1 {
|
|
t.Fatalf("re-run job = %d, want 1", id)
|
|
}
|
|
waitRecord(t, done)
|
|
|
|
if got := atomic.LoadInt32(&calls); got != 2 {
|
|
t.Errorf("runner called %d time(s), want 2 (original + queued re-run)", got)
|
|
}
|
|
svc.mu.Lock()
|
|
pending = svc.runtimes[1].PendingRuns
|
|
svc.mu.Unlock()
|
|
if pending != 0 {
|
|
t.Errorf("PendingRuns must be cleared after the re-run starts, got %d", pending)
|
|
}
|
|
}
|
|
|
|
// TestRunDueQueueDrainsMultipleOverlaps verifies that each missed occurrence
|
|
// under the queue policy eventually runs after the in-flight run finishes.
|
|
func TestRunDueQueueDrainsMultipleOverlaps(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{})
|
|
var calls int32
|
|
svc.runJob = func(_ context.Context, job *domain.Job, _ string, _ string, _ time.Duration) (domain.RunRecord, error) {
|
|
if atomic.LoadInt32(&calls) == 0 {
|
|
<-release
|
|
}
|
|
atomic.AddInt32(&calls, 1)
|
|
return domain.RunRecord{Time: "t", JobID: job.ID, JobName: job.Name, State: "Success"}, nil
|
|
}
|
|
done := completions(svc)
|
|
|
|
primeDue(t, svc, 1)
|
|
svc.RunDue(time.Now())
|
|
|
|
// Three extra due ticks while the first run is still in flight.
|
|
for range 3 {
|
|
primeDue(t, svc, 1)
|
|
svc.RunDue(time.Now())
|
|
}
|
|
|
|
svc.mu.Lock()
|
|
pending := svc.runtimes[1].PendingRuns
|
|
svc.mu.Unlock()
|
|
if pending != 3 {
|
|
t.Fatalf("PendingRuns = %d, want 3 queued occurrences", pending)
|
|
}
|
|
|
|
close(release)
|
|
for range 4 {
|
|
waitRecord(t, done)
|
|
}
|
|
|
|
if got := atomic.LoadInt32(&calls); got != 4 {
|
|
t.Errorf("runner called %d time(s), want 4 (original + 3 queued)", got)
|
|
}
|
|
svc.mu.Lock()
|
|
pending = svc.runtimes[1].PendingRuns
|
|
svc.mu.Unlock()
|
|
if pending != 0 {
|
|
t.Errorf("PendingRuns = %d after drain, want 0", pending)
|
|
}
|
|
}
|
|
|
|
// 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.
|
|
func TestRunDuePerJobQueueOverridesGlobalSkip(t *testing.T) {
|
|
svc := newQueueService(t, domain.ExecutionModeParallel, domain.OverlapPolicySkip, []domain.Job{
|
|
{ID: 1, Name: "A", Schedule: "@every 1h", Command: "echo", Enabled: true, OverlapPolicy: string(domain.OverlapPolicyQueue)},
|
|
})
|
|
|
|
entered := make(chan int, 2)
|
|
release := make(chan struct{})
|
|
var calls int32
|
|
svc.runJob = func(_ context.Context, job *domain.Job, _ string, _ string, _ time.Duration) (domain.RunRecord, error) {
|
|
atomic.AddInt32(&calls, 1)
|
|
entered <- job.ID
|
|
<-release
|
|
return domain.RunRecord{Time: "t", JobID: job.ID, JobName: job.Name, State: "Success"}, nil
|
|
}
|
|
done := completions(svc)
|
|
|
|
primeDue(t, svc, 1)
|
|
svc.RunDue(time.Now())
|
|
if id := <-entered; id != 1 {
|
|
t.Fatalf("started job = %d, want 1", id)
|
|
}
|
|
|
|
// Re-due the running job and tick. Despite the global "skip", the job's own
|
|
// "queue" policy must mark it Pending.
|
|
primeDue(t, svc, 1)
|
|
svc.RunDue(time.Now())
|
|
expectNoEntry(t, entered)
|
|
|
|
svc.mu.Lock()
|
|
pending := svc.runtimes[1].PendingRuns
|
|
svc.mu.Unlock()
|
|
if pending != 1 {
|
|
t.Fatalf("per-job queue policy must queue a deferred run, PendingRuns = %d", pending)
|
|
}
|
|
|
|
// Releasing the first run lets executeRun start the deferred re-run.
|
|
close(release)
|
|
waitRecord(t, done)
|
|
if id := <-entered; id != 1 {
|
|
t.Fatalf("re-run job = %d, want 1", id)
|
|
}
|
|
waitRecord(t, done)
|
|
if got := atomic.LoadInt32(&calls); got != 2 {
|
|
t.Errorf("runner called %d time(s), want 2 (original + queued re-run)", got)
|
|
}
|
|
}
|
|
|
|
// TestRunDuePerJobSkipOverridesGlobalQueue verifies the reverse override: a job
|
|
// carrying its own "skip" policy drops an overlapping run even though the global
|
|
// default is "queue".
|
|
func TestRunDuePerJobSkipOverridesGlobalQueue(t *testing.T) {
|
|
svc := newQueueService(t, domain.ExecutionModeParallel, domain.OverlapPolicyQueue, []domain.Job{
|
|
{ID: 1, Name: "A", Schedule: "@every 1h", Command: "echo", Enabled: true, OverlapPolicy: string(domain.OverlapPolicySkip)},
|
|
})
|
|
|
|
entered := make(chan int, 2)
|
|
release := make(chan struct{})
|
|
var calls int32
|
|
svc.runJob = func(_ context.Context, job *domain.Job, _ string, _ string, _ time.Duration) (domain.RunRecord, error) {
|
|
atomic.AddInt32(&calls, 1)
|
|
entered <- job.ID
|
|
<-release
|
|
return domain.RunRecord{Time: "t", JobID: job.ID, JobName: job.Name, State: "Success"}, nil
|
|
}
|
|
done := completions(svc)
|
|
|
|
primeDue(t, svc, 1)
|
|
svc.RunDue(time.Now())
|
|
if id := <-entered; id != 1 {
|
|
t.Fatalf("started job = %d, want 1", id)
|
|
}
|
|
|
|
// Re-due the running job and tick. Despite the global "queue", the job's own
|
|
// "skip" policy must drop it without marking Pending.
|
|
primeDue(t, svc, 1)
|
|
svc.RunDue(time.Now())
|
|
expectNoEntry(t, entered)
|
|
|
|
svc.mu.Lock()
|
|
pending := svc.runtimes[1].PendingRuns
|
|
svc.mu.Unlock()
|
|
if pending != 0 {
|
|
t.Errorf("per-job skip policy must not queue deferred runs, PendingRuns = %d", pending)
|
|
}
|
|
|
|
close(release)
|
|
waitRecord(t, done)
|
|
expectNoEntry(t, entered)
|
|
if got := atomic.LoadInt32(&calls); got != 1 {
|
|
t.Errorf("runner called %d time(s), want 1", got)
|
|
}
|
|
}
|
|
|
|
// TestRunNowSequentialGuard verifies the sequential-mode guard in RunNow: a manual
|
|
// run is refused while another job is running, and allowed once nothing is.
|
|
func TestRunNowSequentialGuard(t *testing.T) {
|
|
svc := newQueueService(t, domain.ExecutionModeSequential, domain.OverlapPolicySkip, []domain.Job{
|
|
{ID: 1, Name: "A", Schedule: "@every 1h", Command: "echo", Enabled: true},
|
|
{ID: 2, Name: "B", Schedule: "@every 1h", Command: "echo", Enabled: true},
|
|
})
|
|
|
|
entered := make(chan int, 2)
|
|
release := make(chan struct{})
|
|
svc.runJob = func(_ context.Context, job *domain.Job, _ string, _ string, _ time.Duration) (domain.RunRecord, error) {
|
|
entered <- job.ID
|
|
<-release
|
|
return domain.RunRecord{Time: "t", JobID: job.ID, JobName: job.Name, State: "Success"}, nil
|
|
}
|
|
done := completions(svc)
|
|
|
|
if err := svc.RunNow(1); err != nil {
|
|
t.Fatalf("first RunNow: %v", err)
|
|
}
|
|
<-entered // job 1 is in flight
|
|
|
|
if err := svc.RunNow(2); err == nil {
|
|
t.Error("expected RunNow to be refused while another job runs (sequential mode)")
|
|
}
|
|
|
|
// Once job 1 finishes, a manual run of job 2 is allowed.
|
|
close(release)
|
|
waitRecord(t, done)
|
|
if err := svc.RunNow(2); err != nil {
|
|
t.Fatalf("RunNow after first finished: %v", err)
|
|
}
|
|
if id := <-entered; id != 2 {
|
|
t.Fatalf("second manual run job = %d, want 2", id)
|
|
}
|
|
waitRecord(t, done)
|
|
}
|
|
|
|
// TestRunDueQueueDrainSkippedWhenPaused verifies that queued overlap runs are not
|
|
// 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},
|
|
})
|
|
|
|
entered := make(chan int, 2)
|
|
release := make(chan struct{})
|
|
var calls int32
|
|
svc.runJob = func(_ context.Context, job *domain.Job, _ string, _ string, _ time.Duration) (domain.RunRecord, error) {
|
|
atomic.AddInt32(&calls, 1)
|
|
entered <- job.ID
|
|
<-release
|
|
return domain.RunRecord{Time: "t", JobID: job.ID, JobName: job.Name, State: "OK"}, nil
|
|
}
|
|
done := completions(svc)
|
|
|
|
primeDue(t, svc, 1)
|
|
svc.RunDue(time.Now())
|
|
if id := <-entered; id != 1 {
|
|
t.Fatalf("started job = %d, want 1", id)
|
|
}
|
|
|
|
primeDue(t, svc, 1)
|
|
svc.RunDue(time.Now())
|
|
expectNoEntry(t, entered)
|
|
|
|
svc.mu.Lock()
|
|
pending := svc.runtimes[1].PendingRuns
|
|
svc.mu.Unlock()
|
|
if pending != 1 {
|
|
t.Fatalf("expected one queued overlap, PendingRuns = %d", pending)
|
|
}
|
|
|
|
if err := svc.SetGlobalPause(true); err != nil {
|
|
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)
|
|
|
|
svc.mu.Lock()
|
|
pending = svc.runtimes[1].PendingRuns
|
|
svc.mu.Unlock()
|
|
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)
|
|
}
|
|
}
|
|
|
|
// TestEffectiveTimeout verifies the three-state resolution: an unset (nil)
|
|
// Job.TimeoutSeconds falls back to the global default, a positive value
|
|
// overrides it, and an explicit 0 means "no timeout" without inheriting.
|
|
func TestEffectiveTimeout(t *testing.T) {
|
|
svc := newTempService(t, nil)
|
|
svc.store.Config.DefaultTimeoutSeconds = 30
|
|
|
|
inherit := &domain.Job{TimeoutSeconds: nil}
|
|
if got, want := svc.effectiveTimeout(inherit), 30*time.Second; got != want {
|
|
t.Errorf("inherited timeout = %s, want %s", got, want)
|
|
}
|
|
|
|
own := &domain.Job{TimeoutSeconds: domain.TimeoutSecondsPtr(5)}
|
|
if got, want := svc.effectiveTimeout(own), 5*time.Second; got != want {
|
|
t.Errorf("per-job timeout = %s, want %s", got, want)
|
|
}
|
|
|
|
// An explicit per-job 0 must beat a positive global default rather than be
|
|
// mistaken for "unset".
|
|
none := &domain.Job{TimeoutSeconds: domain.TimeoutSecondsPtr(0)}
|
|
if got, want := svc.effectiveTimeout(none), time.Duration(0); got != want {
|
|
t.Errorf("explicit per-job zero timeout = %s, want %s (no timeout)", got, want)
|
|
}
|
|
|
|
svc.store.Config.DefaultTimeoutSeconds = 0
|
|
if got, want := svc.effectiveTimeout(inherit), time.Duration(0); got != want {
|
|
t.Errorf("inherited timeout with no global default = %s, want %s (no timeout)", got, want)
|
|
}
|
|
}
|