fix(cron): prevent job flood after restart, interval drift, and SQLite RunJob bug (#796)

## 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

@@ -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 {

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

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

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,

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

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 = ?,