fix(scheduler): a firing could be lost between rescheduling and dispatch

`tick` advanced `next_run_at` before dispatching the work, with nothing
recording that the occurrence was owed. A process that died between the two
dropped it silently.

The window is narrower than it first looks — `claim_due` sets `last_run_at`
but does not clear `next_run_at`, so a crash *before* `set_next_run` leaves
the routine due and it re-fires on the next tick. The loss is specifically
between the reschedule and the dispatch. That is tolerable for a message
routine and not tolerable for a scheduled mission, which is why this lands
before mission scheduling does.

`routine_fires` holds one row per (routine, occurrence), claimed before
dispatch and settled after:

- Fresh   — nobody has it; fire.
- Retry   — claimed, never settled: a crash mid-fire. Safe to fire again, as
            no completion was recorded and nothing downstream saw a result.
- Settled — already dispatched; advance the clock and do not run the work.
            This is what keeps a scheduled mission to one container across
            restarts.

A failed dispatch settles terminally rather than staying retryable. Retrying
a persistently failing action every tick is how a broken routine becomes a
denial-of-service against whatever it talks to; the error is kept on the row.

The claim uses `xmax = 0` to distinguish a real insert from a no-op update in
a single statement — `ON CONFLICT DO NOTHING` returns no row at all, so two
schedulers racing one occurrence could both read it as unclaimed.

Also: fan-out capped at 25 per tick with the remainder logged and deferred (a
clock jump or an accidental every-minute cron would otherwise dispatch every
missed occurrence at once — one container each for topology routines), and
`spawn` no longer discards tick errors, so a scheduler that has stopped firing
no longer looks identical to one with nothing to do.

The pre-existing exactly-once test still passes: the claim changes
recoverability, not firing semantics.

Co-Authored-By: Claude Opus 5 <[email protected]>

crates/cm-scheduler/src/lib.rs

@@ -8,6 +8,14 @@ use cm_runtime::Runtime;
 use sqlx::PgPool;
 use time::OffsetDateTime;
 
+/// Most occurrences one tick will dispatch.
+///
+/// A backlog — a clock jump, a long outage, or a cron expression that
+/// accidentally resolves to "every minute" — would otherwise fan out every
+/// missed occurrence at once. For a topology routine that is one container
+/// each. The remainder stays due and is picked up by the following tick.
+const MAX_FIRES_PER_TICK: usize = 25;
+
 pub use cm_runtime::scheduling::next_occurrence;
 
 #[derive(Debug, thiserror::Error)]
@@ -30,12 +38,58 @@ impl Scheduler {
 
     /// Fires every due routine once and reschedules it. Returns how many
     /// fired. Time is a parameter so tests control the clock.
+    ///
+    /// Each occurrence is claimed in `routine_fires` before it is dispatched,
+    /// and settled after. That ordering is what makes a firing survive a
+    /// restart: the clock still advances first (a failing action must not
+    /// stall the schedule), but the claim row remembers that the occurrence
+    /// was owed, so a crash between reschedule and dispatch is retried instead
+    /// of silently skipped — and an occurrence already dispatched is never
+    /// dispatched twice.
     pub async fn tick(&self, now: OffsetDateTime) -> Result<usize, ScheduleError> {
         let due = routines::claim_due(&self.pool, now).await?;
-        for routine in &due {
-            // Reschedule first: a firing failure must not stall the clock. A
-            // one-shot routine (Scheduled mode, a specific date/time) fires once
-            // and never reschedules.
+        // Cap the fan-out. A backlog (clock jump, long outage, a cron that
+        // resolves to "every minute" by accident) would otherwise dispatch
+        // every missed occurrence in one tick — for topology routines that is
+        // one container each.
+        let mut fired = 0usize;
+        for routine in due.iter().take(MAX_FIRES_PER_TICK) {
+            // The occurrence's own timestamp identifies the slot. `claim_due`
+            // does not clear `next_run_at`, so this is still the value that
+            // came due.
+            let slot = routine.next_run_at.unwrap_or(now);
+            match routines::claim_fire(&self.pool, routine.id, slot).await {
+                Ok(routines::FireClaim::Fresh) | Ok(routines::FireClaim::Retry) => {}
+                Ok(routines::FireClaim::Settled) => {
+                    // Already dispatched by a previous tick or replica. Let the
+                    // clock advance below, but do not run the work again.
+                    let one_shot = routine
+                        .action
+                        .get("one_shot")
+                        .and_then(|v| v.as_bool())
+                        .unwrap_or(false);
+                    let next = if one_shot {
+                        None
+                    } else {
+                        next_occurrence(&routine.schedule_cron, now).ok()
+                    };
+                    let _ = routines::set_next_run(&self.pool, routine.id, next).await;
+                    continue;
+                }
+                Err(e) => {
+                    // Could not take the slot. Leaving `next_run_at` untouched
+                    // means the occurrence is still due and the next tick tries
+                    // again — the safe direction.
+                    eprintln!("scheduler: claiming fire for routine {}: {e}", routine.id);
+                    continue;
+                }
+            }
+            fired += 1;
+
+            // Reschedule before dispatching: a firing failure must not stall
+            // the clock. The claim above is what keeps this from losing the
+            // occurrence outright. A one-shot routine (Scheduled mode, a
+            // specific date/time) fires once and never reschedules.
             let one_shot = routine
                 .action
                 .get("one_shot")
@@ -87,6 +141,9 @@ impl Scheduler {
                     };
                     let _ = routine_runs::finish(&self.pool, rid, status, err.as_deref()).await;
                 }
+                let topo_err = res.as_ref().err().cloned();
+                let _ = routines::complete_fire(&self.pool, routine.id, slot, topo_err.as_deref())
+                    .await;
                 continue;
             }
 
@@ -108,15 +165,26 @@ impl Scheduler {
             // Journal the firing for the dashboard routines panel.
             let run_id = routine_runs::start(&self.pool, routine.id).await.ok();
             let res = self.runtime.send_message(session.id, message).await;
+            let send_err = res.as_ref().err().map(|e| format!("{e}"));
             if let Some(rid) = run_id {
                 let (status, err) = match &res {
                     Ok(_) => ("ok", None),
-                    Err(e) => ("error", Some(format!("{e}"))),
+                    Err(_) => ("error", send_err.clone()),
                 };
                 let _ = routine_runs::finish(&self.pool, rid, status, err.as_deref()).await;
             }
+            let _ =
+                routines::complete_fire(&self.pool, routine.id, slot, send_err.as_deref()).await;
         }
-        Ok(due.len())
+        if due.len() > MAX_FIRES_PER_TICK {
+            eprintln!(
+                "scheduler: {} routines were due; fired {MAX_FIRES_PER_TICK} this tick, \
+                 {} deferred to the next one",
+                due.len(),
+                due.len() - MAX_FIRES_PER_TICK,
+            );
+        }
+        Ok(fired)
     }
 
     /// The production loop: ticks on an interval with the real clock.
@@ -125,7 +193,12 @@ impl Scheduler {
             let mut tick = tokio::time::interval(interval);
             loop {
                 tick.tick().await;
-                let _ = self.tick(OffsetDateTime::now_utc()).await;
+                // A persistently failing tick used to be invisible: the result
+                // was discarded, so a scheduler that stopped firing looked
+                // exactly like one with nothing to do.
+                if let Err(e) = self.tick(OffsetDateTime::now_utc()).await {
+                    eprintln!("scheduler: tick failed: {e}");
+                }
             }
         });
     }