From 292e63dfa37dc129c4e6259c9369c6e4cffb416f Mon Sep 17 00:00:00 2001 From: Luan Vu Date: Thu, 9 Apr 2026 23:40:46 +0700 Subject: [PATCH] fix(cron): prevent job flood after restart, interval drift, and SQLite RunJob bug (#796) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## Summary - **Flood after restart**: `recomputeStaleJobs()` now advances past-due jobs (not just NULL) to their next future run time, preventing all missed jobs from firing simultaneously on first tick - **Interval drift**: Anchor-based scheduling computes next run from original scheduled time instead of execution end time, preserving per-job offsets - **O(1) advance**: Replaced O(N) loop with modular arithmetic to prevent CPU starvation after prolonged downtime with short-interval jobs - **SQLite RunJob fix**: Added `next_run_at = NULL` claim + `reloadClaimed` param to match PG store behavior — manual runs were silently skipped - **Manual RunJob consistency**: Nil-anchor guard ensures manual triggers use `now + interval` (not anchor) across all stores - **Zombie at-job fix**: JSON store Start() now disables past-due one-time `at` jobs instead of leaving them enabled with nil NextRunAtMS All fixes applied across JSON, PostgreSQL, and SQLite store implementations. 4 new unit tests covering flood prevention, anchor arithmetic, RunJob scheduling, and at-job disabling. --- internal/cron/service.go | 14 +- internal/cron/service_execution.go | 49 ++-- internal/cron/service_test.go | 227 +++++++++++++++++++ internal/store/pg/cron_scheduler.go | 57 ++++- internal/store/sqlitestore/cron_exec.go | 9 +- internal/store/sqlitestore/cron_scheduler.go | 70 ++++-- 6 files changed, 381 insertions(+), 45 deletions(-) diff --git a/internal/cron/service.go b/internal/cron/service.go index f9cf32d0..c872d125 100644 --- a/internal/cron/service.go +++ b/internal/cron/service.go @@ -58,13 +58,23 @@ func (cs *Service) Start() error { cs.store = Store{Version: 1} } - // Compute next runs for all enabled jobs + // Compute next runs for all enabled jobs: fix NULL and past-due next_run_at. + // Past-due jobs happen when the service was stopped and their scheduled time passed. + // Without this, ALL past-due jobs would fire simultaneously on the first tick. + // NOTE: After prolonged downtime, all past-due "every" jobs with the same interval + // will synchronize (all get next_run_at = now + interval). This is inherent because + // the original schedule anchor is not persisted. After the first execution cycle, + // anchor-based scheduling in executeJobByID preserves spacing going forward. now := nowMS() for i := range cs.store.Jobs { job := &cs.store.Jobs[i] - if job.Enabled && job.State.NextRunAtMS == nil { + if job.Enabled && (job.State.NextRunAtMS == nil || *job.State.NextRunAtMS <= now) { next := cs.computeNextRun(&job.Schedule, now) job.State.NextRunAtMS = next + // Disable past-due one-time "at" jobs instead of leaving them as zombies + if next == nil && job.Schedule.Kind == "at" { + job.Enabled = false + } } } if err := cs.saveUnsafe(); err != nil { diff --git a/internal/cron/service_execution.go b/internal/cron/service_execution.go index 9f3af89e..4e299058 100644 --- a/internal/cron/service_execution.go +++ b/internal/cron/service_execution.go @@ -155,24 +155,31 @@ func (cs *Service) checkJobs() { cs.mu.Lock() now := nowMS() - var dueJobIDs []string + + // Collect due jobs and preserve their original scheduled times. + // The scheduled time is used as anchor for "every" jobs to prevent drift. + type dueJob struct { + id string + scheduledAtMS int64 + } + var dueJobs []dueJob for i := range cs.store.Jobs { job := &cs.store.Jobs[i] if job.Enabled && job.State.NextRunAtMS != nil && *job.State.NextRunAtMS <= now { - dueJobIDs = append(dueJobIDs, job.ID) + dueJobs = append(dueJobs, dueJob{id: job.ID, scheduledAtMS: *job.State.NextRunAtMS}) } } - if len(dueJobIDs) == 0 { + if len(dueJobs) == 0 { cs.mu.Unlock() return } // Clear NextRunAtMS to prevent duplicate execution - dueMap := make(map[string]bool, len(dueJobIDs)) - for _, id := range dueJobIDs { - dueMap[id] = true + dueMap := make(map[string]bool, len(dueJobs)) + for _, dj := range dueJobs { + dueMap[dj.id] = true } for i := range cs.store.Jobs { if dueMap[cs.store.Jobs[i].ID] { @@ -184,18 +191,18 @@ func (cs *Service) checkJobs() { // Execute jobs in parallel — scheduler enforces per-session serialization var wg sync.WaitGroup - for _, jobID := range dueJobIDs { + for _, dj := range dueJobs { wg.Add(1) - go func(id string) { + go func(id string, scheduledAtMS int64) { defer wg.Done() defer safego.Recover(nil, "job_id", id) - cs.executeJobByID(id) - }(jobID) + cs.executeJobByID(id, scheduledAtMS) + }(dj.id, dj.scheduledAtMS) } wg.Wait() } -func (cs *Service) executeJobByID(jobID string) { +func (cs *Service) executeJobByID(jobID string, scheduledAtMS int64) { cs.mu.Lock() var job *Job for i := range cs.store.Jobs { @@ -247,10 +254,22 @@ func (cs *Service) executeJobByID(jobID string) { if cs.store.Jobs[i].DeleteAfterRun { cs.store.Jobs = append(cs.store.Jobs[:i], cs.store.Jobs[i+1:]...) } else { - next := cs.computeNextRun(&cs.store.Jobs[i].Schedule, now) - cs.store.Jobs[i].State.NextRunAtMS = next - if next == nil { - cs.store.Jobs[i].Enabled = false + schedule := &cs.store.Jobs[i].Schedule + // For "every" (interval) jobs, compute next run from the original + // scheduled time (anchor) to prevent drift and synchronization. + if schedule.Kind == "every" && schedule.EveryMS != nil && *schedule.EveryMS > 0 && scheduledAtMS > 0 { + interval := *schedule.EveryMS + // O(1) advance to the next future slot from anchor + elapsed := now - scheduledAtMS + periods := elapsed / interval + next := scheduledAtMS + (periods+1)*interval + cs.store.Jobs[i].State.NextRunAtMS = &next + } else { + next := cs.computeNextRun(schedule, now) + cs.store.Jobs[i].State.NextRunAtMS = next + if next == nil { + cs.store.Jobs[i].Enabled = false + } } } break diff --git a/internal/cron/service_test.go b/internal/cron/service_test.go index d5797503..fb6cc10a 100644 --- a/internal/cron/service_test.go +++ b/internal/cron/service_test.go @@ -352,6 +352,233 @@ func TestService_RunLog_PopulatedByAutoExecution(t *testing.T) { } } +// --- Flood prevention & anchor scheduling --- + +func TestService_Start_AdvancesPastDueJobs(t *testing.T) { + dir := t.TempDir() + storePath := filepath.Join(dir, "cron.json") + + var execCount atomic.Int32 + handler := func(job *Job) (string, error) { + execCount.Add(1) + return "done", nil + } + + // Create service with 3 every-5min jobs, manually set past-due NextRunAtMS + cs1 := NewService(storePath, handler) + interval := int64(300000) // 5 min + now := time.Now().UnixMilli() + for i, offset := range []int64{600000, 1200000, 1800000} { // 10, 20, 30 min ago + name := fmt.Sprintf("past-due-%d", i) + job, err := cs1.AddJob(name, Schedule{Kind: "every", EveryMS: &interval}, "tick", false, "", "", "") + if err != nil { + t.Fatalf("AddJob error: %v", err) + } + // Override NextRunAtMS to past value + for j := range cs1.store.Jobs { + if cs1.store.Jobs[j].ID == job.ID { + past := now - offset + cs1.store.Jobs[j].State.NextRunAtMS = &past + } + } + } + cs1.saveUnsafe() + + // Reload and Start — should advance all jobs to future, not fire them + cs2 := NewService(storePath, handler) + if err := cs2.Start(); err != nil { + t.Fatalf("Start error: %v", err) + } + // Give just enough time for one potential tick, but jobs should be in the future + time.Sleep(1500 * time.Millisecond) + cs2.Stop() + + // Verify no past-due executions happened (jobs were advanced, not fired) + if count := execCount.Load(); count != 0 { + t.Fatalf("expected 0 executions for advanced past-due jobs, got %d", count) + } + + // Verify all jobs have future NextRunAtMS + afterNow := time.Now().UnixMilli() + for _, job := range cs2.ListJobs(true) { + if job.State.NextRunAtMS == nil { + t.Fatalf("job %s should have NextRunAtMS set", job.Name) + } + if *job.State.NextRunAtMS <= afterNow { + t.Fatalf("job %s NextRunAtMS %d should be in the future (> %d)", job.Name, *job.State.NextRunAtMS, afterNow) + } + } +} + +func TestAnchorBasedNextRun_PreservesOffset(t *testing.T) { + // Test the O(1) anchor arithmetic directly — deterministic, no scheduler needed. + // Formula: next = anchor + (elapsed/interval + 1) * interval + + tests := []struct { + name string + anchor int64 // scheduledAtMS + interval int64 // everyMS + now int64 + wantNext int64 + }{ + { + name: "normal_one_period", + anchor: 1000, + interval: 5000, + now: 6500, + // elapsed=5500, periods=5500/5000=1, next=1000+(1+1)*5000=11000 + wantNext: 11000, + }, + { + name: "different_anchor_preserves_offset", + anchor: 2000, + interval: 5000, + now: 6500, + // elapsed=4500, periods=4500/5000=0, next=2000+(0+1)*5000=7000 + wantNext: 7000, + }, + { + name: "exact_boundary", + anchor: 1000, + interval: 5000, + now: 6000, + // elapsed=5000, periods=5000/5000=1, next=1000+(1+1)*5000=11000 + wantNext: 11000, + }, + { + name: "multiple_periods_skipped", + anchor: 1000, + interval: 5000, + now: 22000, + // elapsed=21000, periods=21000/5000=4, next=1000+(4+1)*5000=26000 + wantNext: 26000, + }, + { + name: "small_interval_large_gap", + anchor: 0, + interval: 1000, // 1 second + now: 86400000, // 24 hours later — O(1) handles this without 86400 iterations + // elapsed=86400000, periods=86400000/1000=86400, next=0+(86400+1)*1000=86401000 + wantNext: 86401000, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + elapsed := tt.now - tt.anchor + periods := elapsed / tt.interval + next := tt.anchor + (periods+1)*tt.interval + if next != tt.wantNext { + t.Fatalf("anchor=%d interval=%d now=%d: got next=%d, want %d", + tt.anchor, tt.interval, tt.now, next, tt.wantNext) + } + if next <= tt.now { + t.Fatalf("next %d should be after now %d", next, tt.now) + } + }) + } + + // Verify two jobs with same interval but different anchors maintain offset + anchorA, anchorB := int64(1000), int64(2000) + interval := int64(5000) + now := int64(6500) + + nextA := anchorA + (((now - anchorA) / interval) + 1) * interval + nextB := anchorB + (((now - anchorB) / interval) + 1) * interval + offset := nextA - nextB + if offset != 4000 { // 11000 - 7000 = 4000 (original offset 1000 preserved mod interval) + t.Fatalf("expected 4000ms offset between jobs, got %d", offset) + } +} + +func TestService_RunJob_UsesNowBasedScheduling(t *testing.T) { + dir := t.TempDir() + storePath := filepath.Join(dir, "cron.json") + + handler := func(job *Job) (string, error) { return "ok", nil } + cs := NewService(storePath, handler) + cs.SetRetryConfig(RetryConfig{MaxRetries: 0}) + + interval := int64(60000) // 60s + job, err := cs.AddJob("manual-test", Schedule{Kind: "every", EveryMS: &interval}, "msg", false, "", "", "") + if err != nil { + t.Fatalf("AddJob error: %v", err) + } + + beforeRun := time.Now().UnixMilli() + ran, _, runErr := cs.RunJob(job.ID, true) + afterRun := time.Now().UnixMilli() + + if runErr != nil { + t.Fatalf("RunJob error: %v", runErr) + } + if !ran { + t.Fatal("RunJob should have executed") + } + + found, ok := cs.GetJob(job.ID) + if !ok { + t.Fatal("job should exist after RunJob") + } + if found.State.NextRunAtMS == nil { + t.Fatal("NextRunAtMS should be set after RunJob") + } + + // RunJob uses computeNextRun(now) → now + interval. + // NextRunAtMS should be approximately now + 60s (±1s tolerance for execution time). + expectedMin := beforeRun + interval + expectedMax := afterRun + interval + 1000 + if *found.State.NextRunAtMS < expectedMin || *found.State.NextRunAtMS > expectedMax { + t.Fatalf("NextRunAtMS %d should be between %d and %d (now + interval)", + *found.State.NextRunAtMS, expectedMin, expectedMax) + } +} + +func TestService_Start_DisablesPastDueAtJobs(t *testing.T) { + dir := t.TempDir() + storePath := filepath.Join(dir, "cron.json") + + var execCount atomic.Int32 + handler := func(job *Job) (string, error) { + execCount.Add(1) + return "done", nil + } + + // Add an at-job with time in the past + cs1 := NewService(storePath, handler) + past := time.Now().Add(-time.Hour).UnixMilli() + _, err := cs1.AddJob("past-at", Schedule{Kind: "at", AtMS: &past}, "msg", false, "", "", "") + if err != nil { + t.Fatalf("AddJob error: %v", err) + } + // Override: AddJob rejects past at-jobs, so set it manually + for i := range cs1.store.Jobs { + cs1.store.Jobs[i].State.NextRunAtMS = &past + cs1.store.Jobs[i].Enabled = true + } + cs1.saveUnsafe() + + // Reload and Start + cs2 := NewService(storePath, handler) + if err := cs2.Start(); err != nil { + t.Fatalf("Start error: %v", err) + } + time.Sleep(1500 * time.Millisecond) + cs2.Stop() + + // Verify: past-due at job should be disabled, not executed + if count := execCount.Load(); count != 0 { + t.Fatalf("expected 0 executions for disabled past at-job, got %d", count) + } + jobs := cs2.ListJobs(true) + if len(jobs) != 1 { + t.Fatalf("expected 1 job, got %d", len(jobs)) + } + if jobs[0].Enabled { + t.Fatal("past-due at-job should be disabled") + } +} + // --- helpers --- //go:fix inline diff --git a/internal/store/pg/cron_scheduler.go b/internal/store/pg/cron_scheduler.go index 34239c25..649238b6 100644 --- a/internal/store/pg/cron_scheduler.go +++ b/internal/store/pg/cron_scheduler.go @@ -64,10 +64,11 @@ func (s *PGCronStore) InvalidateCache() { s.mu.Unlock() } -// recomputeStaleJobs fixes enabled jobs that have next_run_at = NULL. -// This happens when the gateway was stopped/crashed while a job was executing, -// or when the previously computed next_run_at was consumed but never recomputed. -// Also resets any jobs stuck in 'running' state from a previous crash. +// recomputeStaleJobs fixes enabled jobs on startup: +// - Resets any jobs stuck in 'running' state from a previous crash. +// - Recomputes next_run_at for jobs where it is NULL (crashed mid-execution). +// - Advances past-due jobs (next_run_at < now) to their next future run time, +// preventing a flood of all missed jobs firing simultaneously after downtime. func (s *PGCronStore) recomputeStaleJobs() { // Reset stale 'running' status — jobs that were mid-execution when the server // crashed will never self-recover, so mark them as interrupted on startup. @@ -78,16 +79,23 @@ func (s *PGCronStore) recomputeStaleJobs() { slog.Info("cron: reset stale running jobs to interrupted", "count", n) } + // Fix jobs with NULL next_run_at OR past-due next_run_at. + // Past-due jobs happen when the server was down and their scheduled time passed. + // Without this, ALL past-due jobs would fire simultaneously on the first tick. + // NOTE: After prolonged downtime, all past-due "every" jobs with the same interval + // will synchronize (all get next_run_at = now + interval). This is inherent because + // the original schedule anchor is not persisted. After the first execution cycle, + // anchor-based scheduling in executeOneJob preserves spacing going forward. + now := time.Now() rows, err := s.db.QueryContext(s.baseCtx, `SELECT id, schedule_kind, cron_expression, run_at, timezone, interval_ms - FROM cron_jobs WHERE enabled = true AND next_run_at IS NULL`) + FROM cron_jobs WHERE enabled = true AND (next_run_at IS NULL OR next_run_at < $1)`, now) if err != nil { slog.Warn("cron: failed to query stale jobs", "error", err) return } defer rows.Close() - now := time.Now() var fixed int for rows.Next() { var id uuid.UUID @@ -128,7 +136,7 @@ func (s *PGCronStore) recomputeStaleJobs() { } if fixed > 0 { - slog.Info("cron: recomputed stale next_run_at on startup", "fixed", fixed) + slog.Info("cron: advanced stale/past-due jobs to next future run", "fixed", fixed) } } @@ -193,6 +201,18 @@ func (s *PGCronStore) checkAndRunDueJobs() { // When false (manual RunJob path), it uses the already-loaded job directly — // skipping the reload avoids the enabled=true filter that would reject disabled jobs. func (s *PGCronStore) executeOneJob(job store.CronJob, handler func(job *store.CronJob) (*store.CronJobResult, error), reloadClaimed bool) { + // Preserve the original scheduled time before reload clears it. + // For "every" jobs, this anchor is used to compute the next run from the + // intended schedule time (not "now"), preventing drift and synchronization + // of interval-based jobs after server restarts. + scheduledAtMS := job.State.NextRunAtMS + + // For manual runs (reloadClaimed=false), don't use anchor — manual triggers + // should reset the schedule to now + interval, not preserve the original offset. + if !reloadClaimed { + scheduledAtMS = nil + } + if reloadClaimed { if id, parseErr := uuid.Parse(job.ID); parseErr == nil { freshJob, ok := s.loadClaimedJob(id) @@ -268,11 +288,28 @@ func (s *PGCronStore) executeOneJob(job store.CronJob, handler func(job *store.C } } else if id, parseErr := uuid.Parse(job.ID); parseErr == nil { schedule := job.Schedule - next := computeNextRun(&schedule, now, s.defaultTZ) var nextRunValue any - if next != nil { - nextRunValue = *next + + // For "every" (interval) jobs, compute next run from the original scheduled + // time (anchor) instead of "now". This prevents: + // 1. Drift: interval is always exact, not interval + execution_time + // 2. Synchronization: after restart, jobs that started at different offsets + // keep their original spacing instead of clustering together + if schedule.Kind == "every" && scheduledAtMS != nil && schedule.EveryMS != nil && *schedule.EveryMS > 0 { + anchor := time.UnixMilli(*scheduledAtMS) + interval := time.Duration(*schedule.EveryMS) * time.Millisecond + // O(1) advance to the next future slot from anchor + elapsed := now.Sub(anchor) + periods := int64(elapsed / interval) + next := anchor.Add(interval * time.Duration(periods+1)) + nextRunValue = next + } else { + next := computeNextRun(&schedule, now, s.defaultTZ) + if next != nil { + nextRunValue = *next + } } + s.db.ExecContext(s.baseCtx, `UPDATE cron_jobs SET last_run_at = $1, last_status = $2, last_error = $3, updated_at = $4, diff --git a/internal/store/sqlitestore/cron_exec.go b/internal/store/sqlitestore/cron_exec.go index 3afa808b..d9ac13ec 100644 --- a/internal/store/sqlitestore/cron_exec.go +++ b/internal/store/sqlitestore/cron_exec.go @@ -28,15 +28,20 @@ func (s *SQLiteCronStore) RunJob(ctx context.Context, jobID string, force bool) return false, "", fmt.Errorf("no job handler configured") } + // Claim the job for manual execution by clearing next_run_at (match PG behavior). + // executeOneJob with reloadClaimed=false skips loadClaimedJob, using the + // already-loaded job directly — allowing manual runs on disabled jobs. if id, parseErr := uuid.Parse(jobID); parseErr == nil { - if _, execErr := s.db.ExecContext(ctx, "UPDATE cron_jobs SET last_status = 'running', updated_at = ? WHERE id = ?", time.Now(), id); execErr != nil { + if _, execErr := s.db.ExecContext(ctx, + "UPDATE cron_jobs SET last_status = 'running', next_run_at = NULL, updated_at = ? WHERE id = ?", + time.Now(), id); execErr != nil { slog.Warn("cron: failed to mark job running", "job", jobID, "error", execErr) } } s.InvalidateCache() s.emitEvent(store.CronEvent{Action: "running", JobID: job.ID, JobName: job.Name, UserID: job.UserID}) - s.executeOneJob(*job, handler) + s.executeOneJob(*job, handler, false) s.mu.Lock() s.cacheLoaded = false diff --git a/internal/store/sqlitestore/cron_scheduler.go b/internal/store/sqlitestore/cron_scheduler.go index 653e1a6f..49539d03 100644 --- a/internal/store/sqlitestore/cron_scheduler.go +++ b/internal/store/sqlitestore/cron_scheduler.go @@ -69,18 +69,26 @@ func (s *SQLiteCronStore) InvalidateCache() { s.mu.Unlock() } -// recomputeStaleJobs fixes enabled jobs with next_run_at = NULL on startup. +// recomputeStaleJobs fixes enabled jobs on startup: +// - Recomputes next_run_at for jobs where it is NULL (crashed mid-execution). +// - Advances past-due jobs (next_run_at < now) to their next future run time, +// preventing a flood of all missed jobs firing simultaneously after downtime. +// +// NOTE: After prolonged downtime, all past-due "every" jobs with the same interval +// will synchronize (all get next_run_at = now + interval). This is inherent because +// the original schedule anchor is not persisted. After the first execution cycle, +// anchor-based scheduling in executeOneJob preserves spacing going forward. func (s *SQLiteCronStore) recomputeStaleJobs() { + now := time.Now() rows, err := s.db.QueryContext(s.baseCtx, `SELECT id, schedule_kind, cron_expression, run_at, timezone, interval_ms - FROM cron_jobs WHERE enabled = 1 AND next_run_at IS NULL`) + FROM cron_jobs WHERE enabled = 1 AND (next_run_at IS NULL OR next_run_at < ?)`, now) if err != nil { slog.Warn("cron: failed to query stale jobs", "error", err) return } defer rows.Close() - now := time.Now() var fixed int for rows.Next() { var id uuid.UUID @@ -124,7 +132,7 @@ func (s *SQLiteCronStore) recomputeStaleJobs() { } if fixed > 0 { - slog.Info("cron: recomputed stale next_run_at on startup", "fixed", fixed) + slog.Info("cron: advanced stale/past-due jobs to next future run", "fixed", fixed) } } @@ -172,7 +180,7 @@ func (s *SQLiteCronStore) checkAndRunDueJobs() { wg.Add(1) go func(job store.CronJob) { defer wg.Done() - s.executeOneJob(job, handler) + s.executeOneJob(job, handler, true) }(job) } wg.Wait() @@ -182,15 +190,30 @@ func (s *SQLiteCronStore) checkAndRunDueJobs() { s.mu.Unlock() } -// executeOneJob runs a single cron job with retry, logs the result, and updates next_run_at. -func (s *SQLiteCronStore) executeOneJob(job store.CronJob, handler func(job *store.CronJob) (*store.CronJobResult, error)) { - if id, parseErr := uuid.Parse(job.ID); parseErr == nil { - freshJob, ok := s.loadClaimedJob(id) - if !ok { - slog.Info("cron job skipped after claim state changed", "id", job.ID) - return +// executeOneJob runs a claimed job. When reloadClaimed is true (scheduler path), +// it re-reads the job from DB to verify claim invariants (enabled + next_run_at IS NULL). +// When false (manual RunJob path), it uses the already-loaded job directly. +func (s *SQLiteCronStore) executeOneJob(job store.CronJob, handler func(job *store.CronJob) (*store.CronJobResult, error), reloadClaimed bool) { + // Preserve the original scheduled time before reload clears it. + // For "every" jobs, this anchor is used to compute the next run from the + // intended schedule time (not "now"), preventing drift and synchronization. + scheduledAtMS := job.State.NextRunAtMS + + // For manual runs (reloadClaimed=false), don't use anchor — manual triggers + // should reset the schedule to now + interval, not preserve the original offset. + if !reloadClaimed { + scheduledAtMS = nil + } + + if reloadClaimed { + if id, parseErr := uuid.Parse(job.ID); parseErr == nil { + freshJob, ok := s.loadClaimedJob(id) + if !ok { + slog.Info("cron job skipped after claim state changed", "id", job.ID) + return + } + job = *freshJob } - job = *freshJob } startTime := time.Now() @@ -253,11 +276,26 @@ func (s *SQLiteCronStore) executeOneJob(job store.CronJob, handler func(job *sto } } else if id, parseErr := uuid.Parse(job.ID); parseErr == nil { schedule := job.Schedule - next := computeNextRun(&schedule, now, s.defaultTZ) var nextRunValue any - if next != nil { - nextRunValue = *next + + // For "every" (interval) jobs, compute next run from the original scheduled + // time (anchor) instead of "now". This prevents drift and synchronization + // of interval-based jobs after server restarts. + if schedule.Kind == "every" && scheduledAtMS != nil && schedule.EveryMS != nil && *schedule.EveryMS > 0 { + anchor := time.UnixMilli(*scheduledAtMS) + interval := time.Duration(*schedule.EveryMS) * time.Millisecond + // O(1) advance to the next future slot from anchor + elapsed := now.Sub(anchor) + periods := int64(elapsed / interval) + next := anchor.Add(interval * time.Duration(periods+1)) + nextRunValue = next + } else { + next := computeNextRun(&schedule, now, s.defaultTZ) + if next != nil { + nextRunValue = *next + } } + s.db.ExecContext(s.baseCtx, `UPDATE cron_jobs SET last_run_at = ?, last_status = ?, last_error = ?, updated_at = ?,