fix(cluster): hold down sequencer-log compaction below armed catch-up

A Calvin scheduler that misses a fan-out (channel full/closed, or it
subscribed after the sequencer already committed epochs for its
vShard) recovers by replaying the sequencer log from an armed
catch-up index. Compaction could previously advance past that index,
permanently losing the only durable copy of a cross-shard txn on that
replica.

Switch catch-up tracking from take-then-clear to peek/confirm
(peek_catch_up_from / clear_catch_up_up_to) so a replay that cannot
complete in one drain tick stays armed instead of being silently
dropped, add min_catch_up_from() to floor the sequencer-group
compactor below the lowest outstanding miss, and arm a spawn-time
catch-up from the earliest retained index when a scheduler first
subscribes to a vShard.

Author: farhan-syah

nodedb-cluster/src/calvin/sequencer/state_machine.rs

@@ -188,6 +188,62 @@ impl SequencerStateMachine {
         map.remove(&vshard)
     }
 
+    /// Arm a catch-up for `vshard` from `index` (min-collapse), so the
+    /// scheduler-side drain replays committed sequencer entries from there.
+    ///
+    /// Called when a scheduler subscribes for a vShard: the sequencer may have
+    /// already committed (and fanned out to a then-absent sender — silently
+    /// skipped) epochs for this vShard before the scheduler existed. A fresh
+    /// node has nothing durably applied to rebuild from, so it would otherwise
+    /// consider itself caught up and never replay those txns. Arming from the
+    /// first available committed index makes the drain replay every committed
+    /// entry for this vShard applied before subscription (idempotent: the
+    /// scheduler's in-flight guard and Reserve/Release no-ops absorb re-apply).
+    pub fn arm_catch_up_from(&self, vshard: u32, index: u64) {
+        self.record_catch_up(vshard, index);
+    }
+
+    /// Read (WITHOUT removing) the catch-up-from Raft index for `vshard`.
+    ///
+    /// The scheduler drain peeks rather than takes so a replay that cannot
+    /// complete this tick (committed index not yet known, transient log-read
+    /// fault) leaves the entry armed for the next tick instead of silently
+    /// dropping it — the loss the old take-then-early-return had.
+    pub fn peek_catch_up_from(&self, vshard: u32) -> Option<u64> {
+        let map = self.catch_up_from.lock().unwrap_or_else(|p| p.into_inner());
+        map.get(&vshard).copied()
+    }
+
+    /// Clear `vshard`'s catch-up entry only if its recorded index is `<= up_to`.
+    ///
+    /// Called after a successful replay of `lo ..= up_to`: the recorded miss is
+    /// now covered, so clear it — unless a concurrent drop has already lowered
+    /// the entry to an index the just-finished replay did not cover (only
+    /// possible for an index `<= up_to` given min-collapse, hence the guard is a
+    /// belt-and-braces no-op in that case). A newer drop recorded at an index
+    /// `> up_to` is preserved for the next drain.
+    pub fn clear_catch_up_up_to(&self, vshard: u32, up_to: u64) {
+        let mut map = self.catch_up_from.lock().unwrap_or_else(|p| p.into_inner());
+        if let Some(&idx) = map.get(&vshard)
+            && idx <= up_to
+        {
+            map.remove(&vshard);
+        }
+    }
+
+    /// The smallest armed catch-up index across ALL vShards, or `None` when no
+    /// catch-up is pending.
+    ///
+    /// The sequencer-group log compactor floors its compaction index at this
+    /// value so a dropped/undelivered fan-out is always replayable from the
+    /// retained log — the hold-down the scheduler-side drain's `LogCompacted`
+    /// arm depends on. Only hosted vShards ever arm a catch-up, so this never
+    /// pins compaction on a vShard this node does not serve.
+    pub fn min_catch_up_from(&self) -> Option<u64> {
+        let map = self.catch_up_from.lock().unwrap_or_else(|p| p.into_inner());
+        map.values().copied().min()
+    }
+
     /// Apply a committed Raft log entry.
     ///
     /// Decodes the `SequencerEntry`, checks epoch monotonicity, fans out to
@@ -721,6 +777,66 @@ mod tests {
         assert_eq!(sm.take_catch_up_from(va), None);
     }
 
+    /// PEEK must not consume: the scheduler drain reads the armed index, and
+    /// only clears it after a confirmed replay. A take-then-early-return (the
+    /// old shape) silently lost the miss when the replay could not complete.
+    #[test]
+    fn peek_catch_up_from_does_not_consume() {
+        let (batch, va, _vb) = make_batch_with_two_vshards();
+        let (tx_a, _rx_a) = mpsc::channel(1);
+        let _ = tx_a.try_send(SchedulerInput::Txn(batch.txns[0].clone()));
+        let mut senders = HashMap::new();
+        senders.insert(va, tx_a);
+        let mut sm = SequencerStateMachine::new(senders, CalvinCompletionRegistry::new_detached());
+
+        sm.apply(9, &encode_entry(&SequencerEntry::EpochBatch { batch }));
+
+        // Repeated peeks keep returning the same armed index.
+        assert_eq!(sm.peek_catch_up_from(va), Some(9));
+        assert_eq!(sm.peek_catch_up_from(va), Some(9));
+    }
+
+    /// Clearing is bounded by the replayed upper bound: a miss covered by the
+    /// replay is cleared, one recorded ABOVE it survives for the next drain.
+    #[test]
+    fn clear_catch_up_up_to_respects_replayed_upper_bound() {
+        let senders = HashMap::new();
+        let sm = SequencerStateMachine::new(senders, CalvinCompletionRegistry::new_detached());
+        let v = 42u32;
+
+        // Armed at 5, replay covered through 10 → cleared.
+        sm.arm_catch_up_from(v, 5);
+        sm.clear_catch_up_up_to(v, 10);
+        assert_eq!(sm.peek_catch_up_from(v), None);
+
+        // Armed at 20, replay only covered through 10 → still armed.
+        sm.arm_catch_up_from(v, 20);
+        sm.clear_catch_up_up_to(v, 10);
+        assert_eq!(sm.peek_catch_up_from(v), Some(20));
+    }
+
+    /// The sequencer-log compaction hold-down floors on the LOWEST armed index
+    /// across all vShards, so no replica's replay range is compacted away.
+    #[test]
+    fn min_catch_up_from_is_lowest_armed_index_across_vshards() {
+        let senders = HashMap::new();
+        let sm = SequencerStateMachine::new(senders, CalvinCompletionRegistry::new_detached());
+        assert_eq!(sm.min_catch_up_from(), None);
+
+        sm.arm_catch_up_from(1, 30);
+        sm.arm_catch_up_from(2, 12);
+        sm.arm_catch_up_from(3, 25);
+        assert_eq!(sm.min_catch_up_from(), Some(12));
+
+        // Draining the lowest lifts the floor to the next outstanding miss.
+        sm.clear_catch_up_up_to(2, 12);
+        assert_eq!(sm.min_catch_up_from(), Some(25));
+
+        sm.clear_catch_up_up_to(1, 30);
+        sm.clear_catch_up_up_to(3, 25);
+        assert_eq!(sm.min_catch_up_from(), None);
+    }
+
     #[test]
     fn catch_up_from_records_dropped_index_on_closed_channel() {
         let (batch, va, vb) = make_batch_with_two_vshards();

nodedb-cluster/src/multi_raft/core.rs

@@ -464,6 +464,18 @@ impl MultiRaft {
         Ok(entries.to_vec())
     }
 
+    /// The lowest committed index still available in `group_id`'s retained log
+    /// (`snapshot_index + 1`), or `None` when the group is absent on this node.
+    ///
+    /// Used to arm a Calvin scheduler catch-up from the earliest replayable
+    /// sequencer index so its drain reads exactly the retained log and never
+    /// faults on a compacted range.
+    pub fn first_available_index(&self, group_id: u64) -> Option<u64> {
+        self.groups
+            .get(&group_id)
+            .map(|n| n.first_available_index())
+    }
+
     /// Auto-compact a group's log if its configured threshold has been
     /// reached, given the DATA-PLANE applied watermark `applied_index`.
     ///

nodedb-raft/src/node/core.rs

@@ -215,6 +215,15 @@ impl<S: LogStorage> RaftNode<S> {
         self.durable_applied
     }
 
+    /// The lowest log index still available in the retained (post-compaction)
+    /// log — `snapshot_index + 1`. A committed-entry read below this yields
+    /// [`RaftError::LogCompacted`]. Used to arm a Calvin scheduler catch-up from
+    /// the earliest replayable index so the drain never faults on a compacted
+    /// range.
+    pub fn first_available_index(&self) -> u64 {
+        self.log.snapshot_index() + 1
+    }
+
     /// Persist `index` as the durable applied floor.
     ///
     /// The caller MUST only pass an index whose state-machine effects are

nodedb/src/control/cluster/calvin/scheduler/driver/core/catch_up.rs

@@ -35,24 +35,33 @@ impl Scheduler {
     /// it; this drain takes them strictly one-at-a-time (SM → release → MultiRaft
     /// → release → SM → release), so the two paths can never form a lock cycle.
     pub(in crate::control::cluster::calvin::scheduler::driver::core) fn drain_catch_up(&mut self) {
-        // 1. SM-lock scope: TAKE the earliest dropped index for this vShard.
-        //    `None` (the common case) means no drop is pending — return O(1).
+        // 1. SM-lock scope: PEEK the earliest armed index for this vShard.
+        //    `None` (the common case) means no catch-up is pending — return O(1).
         //    Otherwise pair it with the committed-index watermark as the replay
-        //    upper bound. Release the SM lock before the MultiRaft read.
+        //    upper bound. We PEEK rather than TAKE: the entry is cleared only
+        //    after a confirmed replay (step 4), so a tick that cannot complete
+        //    the replay (committed index not yet known, transient log-read
+        //    fault) leaves the catch-up armed for the next tick instead of
+        //    silently dropping it. Release the SM lock before the MultiRaft read.
         let (lo, hi) = {
             let sm = self
                 .sequencer_state_machine
                 .lock()
                 .unwrap_or_else(|p| p.into_inner());
-            let Some(lo) = sm.take_catch_up_from(self.vshard_id) else {
+            let Some(lo) = sm.peek_catch_up_from(self.vshard_id) else {
                 return;
             };
             let Some(hi) = sm.current_committed_index() else {
-                // A drop was recorded but nothing is applied yet — nothing to
-                // replay. The take already cleared the entry; the next drop
-                // re-records a fresh index.
+                // Armed but nothing applied yet — leave it armed and retry once
+                // an entry is applied and `hi` is known.
                 return;
             };
+            if lo > hi {
+                // Armed ahead of the committed watermark (e.g. spawn-armed from
+                // the first available index before any entry applied on this
+                // replica). Nothing to replay yet; stay armed.
+                return;
+            }
             (lo, hi)
         };
 
@@ -65,29 +74,31 @@ impl Scheduler {
                 Err(nodedb_cluster::error::ClusterError::Raft(
                     nodedb_raft::RaftError::LogCompacted { .. },
                 )) => {
-                    // The dropped index has been compacted into a snapshot.
-                    // Reachable ONLY via a sequencer-group snapshot-install
-                    // resync, where the installed snapshot already subsumes the
-                    // dropped index — so this replica's state is already correct
-                    // and no replay is owed. Escalate non-silently (error log +
-                    // metric) and RETURN WITHOUT re-arming: the take already
-                    // cleared the entry, and re-inserting it would infinite-loop
-                    // since the log stays compacted. The provably-unreachable fix
-                    // (a sequencer-compaction hold-down keyed on
-                    // `min(catch_up_from)`) is a separately-scoped follow-up.
+                    // The armed index has been compacted below the retained log.
+                    // The sequencer-group compaction hold-down (floored at
+                    // `min_catch_up_from`) is meant to make this unreachable for
+                    // an armed catch-up; if it is nonetheless hit (e.g. a
+                    // snapshot-install resync that already subsumes this index),
+                    // no replay is owed. Escalate non-silently, CLEAR the entry
+                    // to avoid an infinite retry against a permanently-compacted
+                    // index, and return.
                     self.metrics.record_catch_up_log_compacted();
                     tracing::error!(
                         vshard = self.vshard_id,
                         lo,
-                        "calvin catch-up: sequencer log compacted below dropped index; \
+                        "calvin catch-up: sequencer log compacted below armed index; \
                          state is snapshot-covered"
                     );
+                    self.sequencer_state_machine
+                        .lock()
+                        .unwrap_or_else(|p| p.into_inner())
+                        .clear_catch_up_up_to(self.vshard_id, hi);
                     return;
                 }
                 Err(e) => {
-                    // Transient infra fault (e.g. group transiently absent). The
-                    // take cleared the entry; a subsequent drop re-records and a
-                    // later drain retries. Surface it rather than swallow.
+                    // Transient infra fault (e.g. group transiently absent).
+                    // Leave the catch-up ARMED (we peeked, did not take) so a
+                    // later drain retries it. Surface it rather than swallow.
                     tracing::warn!(
                         vshard = self.vshard_id,
                         lo,
@@ -119,6 +130,16 @@ impl Scheduler {
             self.process_scheduler_input(input);
         }
 
+        // Replay of `lo ..= hi` is complete: clear the armed catch-up, but only
+        // up to `hi` — a concurrent drop recorded at an index `> hi` while this
+        // replay ran is preserved for the next drain. This is the CONFIRM step
+        // the peek-not-take at the top defers to; a transient failure above
+        // returned early and left the entry armed.
+        self.sequencer_state_machine
+            .lock()
+            .unwrap_or_else(|p| p.into_inner())
+            .clear_catch_up_up_to(self.vshard_id, hi);
+
         if replayed > 0 {
             self.metrics.record_catch_up_replayed(replayed);
             tracing::info!(

nodedb/src/control/cluster/start_raft/core.rs

@@ -43,6 +43,7 @@ pub fn start_raft(
         loop_build.tracker,
         loop_build.apply_rx,
         loop_build.calvin_read_result_senders,
+        loop_build.sequencer_state_machine,
     );
 
     let ready_rx = finish_observability(

nodedb/src/control/cluster/start_raft/loop_build.rs

@@ -11,8 +11,8 @@ use std::sync::{Arc, Mutex};
 use tokio::sync::mpsc::{self, Sender};
 
 use nodedb_cluster::calvin::{
-    CalvinCompletionRegistry, SequencerConfig, SequencerReceivers, SequencerService, new_inbox,
-    new_reservation_inbox,
+    CalvinCompletionRegistry, SequencerConfig, SequencerReceivers, SequencerService,
+    SequencerStateMachine, new_inbox, new_reservation_inbox,
 };
 
 use crate::control::cluster::calvin::executor::ollp::OllpConfig;
@@ -49,6 +49,9 @@ pub(super) struct LoopBuild {
     pub(super) apply_rx: mpsc::Receiver<ApplyBatch>,
     pub(super) calvin_read_result_senders: Arc<Mutex<BTreeMap<u32, Sender<ReadResultEvent>>>>,
     pub(super) calvin_completion_registry: Arc<CalvinCompletionRegistry>,
+    /// Shared with the compactor wiring so sequencer-group log compaction can
+    /// be held down below any armed scheduler catch-up index.
+    pub(super) sequencer_state_machine: Arc<Mutex<SequencerStateMachine>>,
 }
 
 /// Build the `RaftLoop`, consume `pending_subsystems`, build the sequencer
@@ -184,5 +187,6 @@ pub(super) fn build_raft_loop(
         apply_rx,
         calvin_read_result_senders,
         calvin_completion_registry,
+        sequencer_state_machine,
     })
 }

nodedb/src/control/cluster/start_raft/proposer_wiring.rs

@@ -25,6 +25,7 @@ pub(super) fn wire_proposers(
     tracker: Arc<ProposeTracker>,
     apply_rx: mpsc::Receiver<ApplyBatch>,
     calvin_read_result_senders: Arc<Mutex<BTreeMap<u32, Sender<ReadResultEvent>>>>,
+    sequencer_state_machine: Arc<Mutex<nodedb_cluster::calvin::SequencerStateMachine>>,
 ) {
     // Wire the Raft proposer into SharedState so CP dispatch paths
     // (pgwire, HTTP, array inbound) can route writes through Raft.
@@ -58,14 +59,45 @@ pub(super) fn wire_proposers(
     // `log_compaction_threshold` is `None`.
     // Weak for the same cycle-breaking reason as `raft_proposer` above.
     let raft_loop_for_compact = Arc::downgrade(raft_loop);
+    let sm_for_compact = Arc::clone(&sequencer_state_machine);
     let compactor: Arc<crate::control::wal_replication::RaftCompactor> =
         Arc::new(move |group_id, applied_index| {
             let rl = raft_loop_for_compact
                 .upgrade()
                 .ok_or_else(|| crate::Error::Internal {
                     detail: "raft log compaction: cluster not running".into(),
                 })?;
-            rl.maybe_compact_group(group_id, applied_index)
+
+            // Sequencer-group hold-down. Unlike a data group — whose entries are
+            // replayable from each replica's own durable state — a cross-shard
+            // Calvin txn is re-derived on every replica ONLY from the sequencer
+            // log. A scheduler that missed a fan-out (channel full/closed, or it
+            // had not subscribed yet) recovers by replaying that log from its
+            // armed catch-up index. Compacting past an armed index destroys the
+            // only copy, permanently losing the txn on that replica — which for a
+            // cross-shard graph edge means the edge silently vanishes from that
+            // node's index. Floor the compaction boundary strictly below the
+            // lowest armed catch-up so the replay range always survives.
+            let effective_index = if group_id == nodedb_cluster::calvin::SEQUENCER_GROUP_ID {
+                match sm_for_compact
+                    .lock()
+                    .unwrap_or_else(|p| p.into_inner())
+                    .min_catch_up_from()
+                {
+                    // Keep index `m` itself: compaction discards entries at and
+                    // below its boundary, and the replay range starts AT `m`.
+                    Some(m) => applied_index.min(m.saturating_sub(1)),
+                    None => applied_index,
+                }
+            } else {
+                applied_index
+            };
+            if effective_index == 0 {
+                // Nothing compactable once held down.
+                return Ok(false);
+            }
+
+            rl.maybe_compact_group(group_id, effective_index)
                 .map_err(|e| crate::Error::Internal {
                     detail: format!("raft log compaction: {e}"),
                 })