fix(cron): preserve newer claims when overlapping runs complete

internal/cli/cron_claim_test.go

@@ -2,6 +2,10 @@ package cli
 
 import (
 	"bytes"
+	"io"
+	"os"
+	"path/filepath"
+	"strings"
 	"sync"
 	"testing"
 	"time"
@@ -39,3 +43,115 @@ func TestFireJobClaimPreventsDoubleFire(t *testing.T) {
 		t.Fatalf("NextRunAt must advance past now, got %s", d.NextRunAt)
 	}
 }
+
+func TestFireJobOverlappingAdjacentSlots(t *testing.T) {
+	for _, olderFirst := range []bool{false, true} {
+		name := "newer completes first"
+		if olderFirst {
+			name = "older completes first"
+		}
+		t.Run(name, func(t *testing.T) {
+			root := t.TempDir()
+			stores := []*cron.Store{
+				cron.NewStore(cron.StoreOptions{RootDir: root}),
+				cron.NewStore(cron.StoreOptions{RootDir: root}),
+			}
+			now := time.Date(2026, 6, 9, 9, 0, 0, 0, time.UTC)
+			job, err := stores[0].Add(cron.Job{Expr: "* * * * *", Prompt: "fire me", NextRunAt: now, FireCount: 7})
+			if err != nil {
+				t.Fatal(err)
+			}
+			var release, done [2]chan struct{}
+			for i := range stores {
+				release[i], done[i] = make(chan struct{}), make(chan struct{})
+				// Always release blocked runners, including on assertion failures.
+				defer func() {
+					select {
+					case <-release[i]:
+					default:
+						close(release[i])
+					}
+					<-done[i]
+				}()
+				started := make(chan struct{})
+				go func() {
+					defer close(done[i])
+					fireJob(stores[i], func() time.Time { return now.Add(time.Duration(i) * time.Minute) }, job, io.Discard, io.Discard,
+						func(_ []string, _, _ io.Writer) int {
+							close(started)
+							<-release[i]
+							return i // Include a failed execution in the accounting.
+						})
+				}()
+				select {
+				case <-started:
+				case <-done[i]:
+					t.Fatal("runner did not claim its slot")
+				}
+			}
+			order := []int{1, 0}
+			if olderFirst {
+				order = []int{0, 1}
+			}
+			for n, i := range order {
+				close(release[i])
+				<-done[i]
+				current, err := stores[0].Get(job.ID)
+				if err != nil {
+					t.Fatal(err)
+				}
+				if current.FireCount != 8+n || !current.NextRunAt.Equal(now.Add(2*time.Minute)) {
+					t.Fatalf("after completion %d: FireCount=%d NextRunAt=%s; want count %d and next %s", n+1, current.FireCount, current.NextRunAt, 8+n, now.Add(2*time.Minute))
+				}
+				claimed, err := claimFire(stores[1], now.Add(time.Minute), job.ID)
+				if err != nil || claimed {
+					t.Fatalf("already claimed slot became available: claimed=%v err=%v", claimed, err)
+				}
+			}
+			runs, err := stores[0].Runs(job.ID)
+			if err != nil || len(runs) != 2 {
+				t.Fatalf("runs=%v err=%v", runs, err)
+			}
+			for n, i := range order {
+				if !runs[n].At.Equal(now.Add(time.Duration(i)*time.Minute)) || runs[n].ExitCode != i {
+					t.Fatalf("unexpected run record: %+v", runs[n])
+				}
+			}
+		})
+	}
+}
+
+func TestFireJobCompletionReadErrorDoesNotOverwriteState(t *testing.T) {
+	root := t.TempDir()
+	store := cron.NewStore(cron.StoreOptions{RootDir: root})
+	now := time.Date(2026, 6, 9, 9, 0, 0, 0, time.UTC)
+	job, err := store.Add(cron.Job{Expr: "* * * * *", Prompt: "fire me", NextRunAt: now})
+	if err != nil {
+		t.Fatal(err)
+	}
+	metadata := filepath.Join(root, job.ID, "metadata.json")
+	var stdout, stderr bytes.Buffer
+	fireJob(store, func() time.Time { return now }, job, &stdout, &stderr,
+		func(_ []string, _, _ io.Writer) int {
+			// A corrupt read must not cause completion to replace unknown newer state
+			// with the stale pre-execution job. This works without permission tricks.
+			if err := os.WriteFile(metadata, []byte("unreadable job"), 0o600); err != nil {
+				t.Fatal(err)
+			}
+			return 0
+		})
+	data, err := os.ReadFile(metadata)
+	if err != nil || string(data) != "unreadable job" {
+		t.Fatalf("completion overwrote unreadable state: %q, err=%v", data, err)
+	}
+	if !strings.Contains(stderr.String(), "failed to persist job state") {
+		t.Fatalf("missing persistence warning: %s", stderr.String())
+	}
+	if want := "fired " + job.ID + " -> exit 0 (next: unknown)\n"; stdout.String() != want {
+		t.Fatalf("completion summary = %q; want %q", stdout.String(), want)
+	}
+	runs, err := store.Runs(job.ID)
+	if err != nil || len(runs) != 1 || runs[0].ExitCode != 0 {
+		t.Fatalf("execution should still be recorded: runs=%v err=%v", runs, err)
+	}
+}

internal/cli/cron_run.go

@@ -129,17 +129,15 @@ func reconcileOverdue(store *cron.Store, now func() time.Time, ids []string, std
 	}
 }
 
-// fireJob runs one job via the exec runner, records the outcome, advances the
-// schedule, and persists. The foreground loop is single-goroutine, so the
-// previous fire has already returned before the next tick — no overlap.
+// fireJob claims one job, runs it via the exec runner, and records the outcome.
+// Separate scheduler processes can have adjacent slots executing concurrently.
 func fireJob(store *cron.Store, now func() time.Time, job cron.Job, stdout io.Writer, stderr io.Writer, exec execRunner) {
 	fired := now()
 	// Atomically claim this fire before running it so concurrent schedulers (or
 	// --once overlapping the forever loop) can't both execute the same due job: the
 	// winner advances NextRunAt under the per-job lock, and a loser sees the
-	// advanced time and skips. The post-exec advance recomputes the same
-	// NextRunAt (sched.Next(fired) is deterministic), so the claim serializes the
-	// fire decision without changing any persisted schedule state (M9).
+	// advanced time and skips. Only the claim writes schedule state; completion
+	// must not overwrite a later slot's claim with this run's stale job copy.
 	claimed, claimErr := claimFire(store, fired, job.ID)
 	if claimErr != nil {
 		if errors.Is(claimErr, cron.ErrJobNotFound) {
@@ -178,44 +176,25 @@ func fireJob(store *cron.Store, now func() time.Time, job cron.Job, stdout io.Wr
 		rec.Error = cronTruncate(detail, 500)
 	}
 
-	job.FireCount++
-	// Advance the schedule. If the expression can no longer produce a future run
-	// (became invalid, or is an impossible spec whose Next is zero), pause the job
-	// so it cannot re-fire on every tick.
+	// Explain schedules that claimFire paused because they cannot advance.
 	if sched, perr := cron.Parse(job.Expr); perr != nil {
-		job.Status = cron.StatusPaused
 		if rec.Error == "" {
 			rec.Error = "invalid schedule; job paused: " + perr.Error()
 		}
 	} else if nxt := sched.Next(fired.Truncate(time.Minute)); nxt.IsZero() {
-		job.Status = cron.StatusPaused
 		if rec.Error == "" {
 			rec.Error = "schedule no longer fires; job paused"
 		}
-	} else {
-		// Minute-aligned input keeps this advance identical to claimFire's and lets
-		// the DST fall-back collapse guard engage (AUDIT-M4).
-		job.NextRunAt = nxt
 	}
-	// Re-read and persist the advanced state ATOMICALLY under the per-job lock. The
-	// job may have been paused or removed while it executed, and this in-memory copy
-	// is stale from tick start. Mutate closes the read-modify-write window a single
-	// scheduler alone could only narrow (Store.Mutate took a cross-process lock): a
-	// concurrent scheduler or an external pause/remove landing here can no longer be
-	// clobbered.
+	// Count this completion against current state under the cross-process lock.
+	// The claim already advanced or paused the schedule. Leave that state alone:
+	// another scheduler may have claimed a later slot or a user may have edited it.
 	persisted, err := store.Mutate(job.ID, func(current cron.Job, readErr error) (cron.Job, error) {
 		if readErr != nil {
-			// A transient read failure (IO/permission) is NOT removal — persist the
-			// computed next state anyway so the schedule advances and the job does not
-			// re-fire next tick.
-			fmt.Fprintf(stderr, "warning: could not re-read job %s before persist: %v\n", job.ID, readErr)
-			return job, nil
-		}
-		current.FireCount = job.FireCount
-		current.NextRunAt = job.NextRunAt
-		if current.Status != cron.StatusPaused {
-			current.Status = job.Status
+			// Never replace unknown newer state with the pre-execution snapshot.
+			return current, readErr
 		}
+		current.FireCount++
 		return current, nil
 	})
 	if errors.Is(err, cron.ErrJobNotFound) {
@@ -229,19 +208,21 @@ func fireJob(store *cron.Store, now func() time.Time, job cron.Job, stdout io.Wr
 	if aerr := store.AppendRun(job.ID, rec); aerr != nil {
 		fmt.Fprintf(stderr, "warning: failed to record run for %s: %v\n", job.ID, aerr)
 	}
+	next := "unknown"
 	if err != nil {
 		fmt.Fprintf(stderr, "warning: failed to persist job state for %s: %v\n", job.ID, err)
-		persisted = job
+	} else {
+		next = formatCronTime(persisted.NextRunAt)
 	}
-	fmt.Fprintf(stdout, "fired %s -> exit %d (next: %s)\n", job.ID, code, formatCronTime(persisted.NextRunAt))
+	fmt.Fprintf(stdout, "fired %s -> exit %d (next: %s)\n", job.ID, code, next)
 }
 
 // claimFire atomically claims a due job's fire under the per-job lock and reports
 // whether THIS caller should run it. It returns false only when the job was paused
 // externally or another scheduler already advanced NextRunAt past `fired` (so a
 // valid recurring job never double-fires). A valid due job has its NextRunAt
-// advanced here to claim the slot; an invalid or exhausted schedule is left for the
-// caller to fire-once and pause (preserving existing single-scheduler behavior).
+// advanced here to claim the slot; an invalid or exhausted schedule is paused
+// here and the caller still fires once (preserving single-scheduler behavior).
 func claimFire(store *cron.Store, fired time.Time, id string) (bool, error) {
 	claimed := true
 	_, err := store.Mutate(id, func(current cron.Job, readErr error) (cron.Job, error) {
@@ -268,8 +249,8 @@ func claimFire(store *cron.Store, fired time.Time, id string) (bool, error) {
 		}
 		// Unparseable or unadvanceable schedule (an impossible spec whose Next is
 		// zero): pause the job inside this same locked claim so a concurrent scheduler
-		// sees a non-active job and does NOT also fire it. The winner still fires once
-		// and fireJob's post-exec keeps it paused. Without this, both callers leave
+		// sees a non-active job and does NOT also fire it. The winner still fires once.
+		// Without this, both callers leave
 		// NextRunAt unchanged, both see the job due, and both fire the same slot
 		// (AUDIT-M5).
 		current.Status = cron.StatusPaused

internal/cron/store.go

@@ -228,8 +228,7 @@
 // Remove). It re-reads the CURRENT on-disk job and passes it to mutate along with
 // any read error; mutate returns the job to persist. A removed job aborts with
 // ErrJobNotFound (no recreate). A transient read error (not removal) is surfaced
-// via readErr so the caller can still persist a best-effort state — the fire path
-// advances the schedule regardless, to avoid a re-fire.
+// via readErr; the callback can return it to abort without overwriting state.
 func (s *Store) Mutate(id string, mutate func(current Job, readErr error) (Job, error)) (Job, error) {
 	if !validID(id) {
 		return Job{}, fmt.Errorf("invalid cron job id %q", id)
@@ -340,7 +339,7 @@
 	return f.Close()
 }
 
 func (s *Store) Runs(id string) ([]RunRecord, error) {
 	if !validID(id) {
 		return nil, fmt.Errorf("invalid cron job id %q", id)
 	}