fix: harden batch ordering, proposer requeue, and boot recovery

crates/consensus/primary/src/consensus/state.rs

@@ -549,18 +549,10 @@ impl<DB: Database> Consensus<DB> {
             Err(e) => return Err(e),
         };
         if self.active {
-            // We extract a list of headers from this specific validator that
-            // have been agreed upon, and signal this back to the narwhal sub-system
-            // to be used to re-send batches that have not made it to a commit.
+            // Committed certificates are reported back to the primary so the proposer can drop
+            // its own committed headers and re-propose the batches of those that missed commit.
             let mut committed_certificates = Vec::new();
 
-            // Each cert is tagged with whether its subdag reaches the epoch boundary: the
-            // subscriber drops those post-boundary outputs, so the proposer must keep
-            // (not clean) their batches for rescue. Computed here where both the subdag
-            // commit_timestamp and the epoch_boundary are known; carried per-cert to
-            // the proposer (no shared transition flag).
-            let epoch_boundary = self.consensus_config.epoch_boundary();
-
             // Output the sequence in the right order.
             let csd_len = committed_sub_dags.len();
             for (i, committed_sub_dag) in committed_sub_dags.into_iter().enumerate() {
@@ -581,10 +573,8 @@ impl<DB: Database> Consensus<DB> {
 
                 tracing::debug!(target: "rayls::consensus_state", "Commit in Sequence {:?}", committed_sub_dag.leader.nonce());
 
-                let dropped = committed_sub_dag.reaches_epoch_boundary(epoch_boundary);
-
                 for certificate in &committed_sub_dag.certificates {
-                    committed_certificates.push((certificate.clone(), dropped));
+                    committed_certificates.push(certificate.clone());
                 }
 
                 // NOTE: The size of the sub-dag can be arbitrarily large (depending on the network
@@ -601,7 +591,7 @@ impl<DB: Database> Consensus<DB> {
                 // expected by primary.
                 let leader_commit_round = committed_certificates
                     .iter()
-                    .map(|(c, _)| c.round())
+                    .map(|c| c.round())
                     .max()
                     .expect("committed_certificates isn't empty");
 

crates/consensus/primary/src/consensus_bus.rs

@@ -341,9 +341,7 @@ struct ConsensusBusEpochInner {
     /// only if it already sent us its whole history.
     new_certificates: MeteredMpscChannel<Certificate>,
     /// Outputs the sequence of ordered certificates to the primary (for cleanup and feedback).
-    /// Each cert's `bool` is `true` when its committed subdag reaches the epoch boundary, so its
-    /// output is dropped by the subscriber cut and its batches must not be cleaned up.
-    committed_certificates: MeteredMpscChannel<(Round, Vec<(Certificate, bool)>)>,
+    committed_certificates: MeteredMpscChannel<(Round, Vec<Certificate>)>,
 
     /// Sends missing certificates to the `CertificateFetcher`.
     /// Receives certificates with missing parents from the `Synchronizer`.
@@ -359,9 +357,8 @@ struct ConsensusBusEpochInner {
     /// Updates when headers were committed by consensus.
     ///
     /// NOTE: this does not mean the header was executed yet.
-    /// Each round's `bool` is `true` when its commit reaches the epoch boundary (output dropped by
-    /// the subscriber cut), so the proposer must skip `NodeBatchesCache` cleanup for that header.
-    committed_own_headers: MeteredMpscChannel<(Round, Vec<(Round, bool)>)>,
+    /// Carries the rounds of this authority's own headers included in the commit.
+    committed_own_headers: MeteredMpscChannel<(Round, Vec<Round>)>,
 
     /// Outputs the sequence of ordered certificates to the application layer.
     sequence: MeteredMpscChannel<CommittedSubDag>,
@@ -507,7 +504,7 @@ impl ConsensusBus {
 
     /// Outputs the sequence of ordered certificates to the primary (for cleanup and feedback).
     /// Can only be subscribed to once.
-    pub fn committed_certificates(&self) -> &impl RaylsSender<(Round, Vec<(Certificate, bool)>)> {
+    pub fn committed_certificates(&self) -> &impl RaylsSender<(Round, Vec<Certificate>)> {
         &self.inner_epoch.committed_certificates
     }
 
@@ -566,7 +563,7 @@ impl ConsensusBus {
     ///
     /// NOTE: this does not mean the header was executed yet.
     /// Can only be subscribed to once.
-    pub fn committed_own_headers(&self) -> &impl RaylsSender<(Round, Vec<(Round, bool)>)> {
+    pub fn committed_own_headers(&self) -> &impl RaylsSender<(Round, Vec<Round>)> {
         &self.inner_epoch.committed_own_headers
     }
 
@@ -606,23 +603,15 @@ impl ConsensusBus {
         rx
     }
 
-    /// Track the most recently executed blocks (a bounded window, newest at the tip).
+    /// The most recently executed blocks (a bounded window, newest at the tip).
     ///
-    /// Safe to read for block *numbers* and *hashes* - those are monotonic. But the tip's nonce
-    /// (`epoch << 32 | round`) is NOT monotonic - neither half: draining a parked (out-of-order
-    /// seq) batch executes a block belonging to an OLDER output that still lands as the newest
-    /// height, so the tip's round (and, for a batch carried over from a previous epoch, its epoch)
-    /// can regress far below the true execution frontier.
-    ///
-    /// Example: execution has genuinely reached round 498. A batch for an earlier seq, mapping to
-    /// round 200, was parked; the gap then fills and it is drained and executed now. That fresh
-    /// block gets the next (highest) block number and becomes the tip, but its nonce encodes round
-    /// 200. A caller reading the round off the tip sees 200, not 498. The proposer throttle did
-    /// exactly this: with consensus at round 500 it computed lag `500 - 200 = 300 > threshold` and
-    /// throttled forever, wedging proposals - when the real lag was `500 - 498 = 2`.
-    ///
-    /// For the frontier epoch/round read the monotonic [`Self::executed_anchor`] instead, or scan
-    /// this window for the max-nonce block.
+    /// Block numbers and hashes in the window are monotonic; the tip's nonce
+    /// (`epoch << 32 | round`) is not, in either half. Draining a parked (out-of-order seq) batch
+    /// executes a block belonging to an older output that still lands as the newest height, so a
+    /// round or epoch read off the tip can regress far below the true execution frontier, and a
+    /// lag computed from it (a throttle, a catch-up gate) wedges on a false gap. For the
+    /// frontier epoch/round read the monotonic [`Self::executed_anchor`] instead, or scan this
+    /// window for the max-nonce block.
     pub fn recently_executed_blocks(&self) -> &watch::Sender<RecentlyExecutedBlocks> {
         &self.inner_app.tx_recently_executed_blocks
     }
@@ -641,13 +630,14 @@ impl ConsensusBus {
 
     /// True when the node-scoped engine has executed everything it admitted (queue empty, nothing
     /// in flight). A mode transition waits on this so the engine's admitted backlog finishes before
-    /// the next epoch's `get_missing_consensus` snapshot — otherwise a concurrently-finishing
+    /// the next epoch's `get_missing_consensus` snapshot - otherwise a concurrently-finishing
     /// output is dropped as stale (the demote→rejoin flap race).
     pub fn engine_idle(&self) -> &watch::Sender<bool> {
         &self.inner_app.tx_engine_idle
     }
 
-    /// Signal that execution replay of missed consensus outputs is complete.
+    /// Whether execution replay of missed consensus outputs is complete.
+    ///
     /// Set by the subscriber after replaying, read by the proposer before creating headers.
     pub fn execution_replay_complete(&self) -> &watch::Sender<bool> {
         &self.inner_app.tx_execution_replay_complete

crates/consensus/primary/src/proposer/recovery.rs

@@ -1,9 +1,14 @@
 use crate::proposer::{types::ProposerDigest, Proposer, DIGEST_QUEUE_WARN_THRESHOLD};
-use rayls_infrastructure_storage::tables::NodeBatchesCache;
-use rayls_infrastructure_types::{Database, DbTxMut, Round};
+use rayls_infrastructure_types::{batch_tracker::RequeueReason, Database, Header, Round};
 use std::collections::VecDeque;
 use tracing::{debug, warn};
 
+/// Rounds below a foreign commit round before an uncommitted own header is re-proposed.
+///
+/// Ordinary commit lag (certificates still collecting votes) stays inside the grace window, so
+/// only a header that has clearly missed its commit is retransmitted.
+pub(super) const FALLBACK_REQUEUE_GRACE_ROUNDS: Round = 4;
+
 impl<DB: Database> Proposer<DB> {
     /// Rayls: Push a batch digest; never drops.
     ///
@@ -22,114 +27,112 @@ impl<DB: Database> Proposer<DB> {
         self.digests.push_back(digest);
     }
 
-    /// Rayls: Evict proposed headers beyond gc_depth from current round.
+    /// Re-queues the payload digests of `headers` ahead of the pending queue and reports them to
+    /// the batch tracker under `reason`; returns how many digests were re-queued.
+    fn requeue_front<'a>(
+        &mut self,
+        headers: impl IntoIterator<Item = &'a Header>,
+        reason: RequeueReason,
+    ) -> usize {
+        let mut requeued: VecDeque<ProposerDigest> = headers
+            .into_iter()
+            .flat_map(|header| header.payload().into_iter())
+            .map(|(digest, worker_id)| ProposerDigest { digest: *digest, worker_id: *worker_id })
+            .collect();
+        let count = requeued.len();
+        self.consensus_bus
+            .batch_tracker()
+            .digests_requeued_in_proposer(requeued.iter().map(|d| d.digest), reason);
+        requeued.append(&mut self.digests);
+        self.digests = requeued;
+        count
+    }
+
+    /// Rayls: Drop proposed headers more than `gc_depth` below the committed round, re-queueing
+    /// their digests (upstream never evicts proposed headers; the requeue keeps its never-drop
+    /// rule on this fork-added path).
+    ///
+    /// The horizon keys on the committed round, not `self.round`: commits trail the proposal
+    /// frontier, so a header in the lag window is still committable. Commit rounds are monotone,
+    /// so every future leader round L >= C and an evicted round R <= C - gc_depth is already
+    /// excluded by `order_dag`; re-proposing its digests cannot double-commit the old header.
+    /// Re-queueing is mandatory: the digests are quorum'd and seq-consumed, so dropping them gaps
+    /// the original seq permanently on peers. Queue growth is bounded by the seal-ahead gate.
     pub(super) fn evict_old_proposed_headers(&mut self) {
-        if self.round <= self.gc_depth {
+        let committed = *self.consensus_bus.committed_round_updates().borrow();
+        let Some(gc_round) = committed.checked_sub(self.gc_depth) else {
+            return;
+        };
+        if gc_round == 0 {
             return;
         }
-        let gc_round = self.round - self.gc_depth;
-        let old_count = self.proposed_headers.len();
 
-        // Remove all headers from rounds at or before gc_round
-        self.proposed_headers.retain(|&round, _| round > gc_round);
-
-        let evicted = old_count - self.proposed_headers.len();
-        if evicted > 0 {
-            debug!(
-                target: "primary::proposer",
-                evicted,
-                gc_round,
-                current_round = self.round,
-                remaining = self.proposed_headers.len(),
-                "Evicted old proposed headers"
-            );
+        // rounds <= gc_round stay behind in `self.proposed_headers`, the rest are retained
+        let retained = self.proposed_headers.split_off(&gc_round.saturating_add(1));
+        let evicted_headers = std::mem::replace(&mut self.proposed_headers, retained);
+        if evicted_headers.is_empty() {
+            return;
         }
+
+        let requeued_digests = self.requeue_front(evicted_headers.values(), RequeueReason::GcEvict);
+
+        debug!(
+            target: "primary::proposer",
+            evicted = evicted_headers.len(),
+            requeued_digests,
+            gc_round,
+            current_round = self.round,
+            remaining = self.proposed_headers.len(),
+            "Evicted old proposed headers, requeued their digests"
+        );
     }
 
-    /// Process notifications that Proposer's own headers have been committed in the DAG for a
-    /// particular round.
-    ///
-    /// Committed headers are removed from the collection of `self.proposed_headers`. Headers
-    /// that are skipped with no hope of being committed (proposed in a previous round) are also
-    /// removed after adding the expired header's proposed block digests and system messages to
-    /// the beginning of the queue.
+    /// Processes a commit notification for the proposer's own headers.
     ///
-    /// This method ensures batches that were previously proposed but weren't committed are
-    /// added back to the queue so their transactions are included in the next proposal.
+    /// Committed rounds leave `self.proposed_headers`; headers old enough that they can no longer
+    /// commit are removed too, with their payload digests re-queued at the front so the batches
+    /// land in the next proposal.
     pub(super) fn process_committed_headers(
         &mut self,
         commit_round: Round,
-        committed_headers: Vec<(Round, bool)>,
+        committed_headers: Vec<Round>,
     ) {
-        // Each `(round, dropped)`: skip NodeBatchesCache cleanup for a header whose subdag reaches
-        // the epoch boundary — the subscriber drops its output, so orphan_batches must still find
-        // its batches to rescue them. `dropped` is computed per-commit in the committer (where the
-        // subdag commit_timestamp and epoch_boundary are known) and carried on the channel — no
-        // shared transition flag, no TOCTOU.
-
-        // drain every committed round (not just the lowest-matching one) so later
-        // rounds cannot be re-queued by the retransmit loop below
-        for (round, dropped) in committed_headers.iter().copied() {
-            let Some(header) = self.proposed_headers.remove(&round) else { continue };
-            if dropped {
-                continue;
-            }
-            let _ = self.proposer_store.with_write_txn(|txn| {
-                for (batch_hash, _) in header.payload() {
-                    let _ = txn.remove::<NodeBatchesCache>(batch_hash);
-                }
-                Ok(())
-            });
+        // drain every committed round (not just the lowest) so the retransmit split below cannot
+        // re-queue a round that already committed
+        for round in committed_headers.iter().copied() {
+            self.proposed_headers.remove(&round);
         }
 
         // Fall back to the commit round when none of our own headers committed: otherwise a
         // validator whose proposals keep getting rejected strands its quorum'd digests (consumed
-        // seqs) in proposed_headers until GC, leaving a permanent per-authority seq gap on peers.
-        let highest_committed =
-            committed_headers.iter().map(|(r, _)| *r).max().unwrap_or(commit_round);
-        let Some(&lowest_uncommitted) = self.proposed_headers.keys().next() else { return };
-        if lowest_uncommitted >= highest_committed {
+        // seqs) in proposed_headers until the GC-horizon requeue, parking its batches on peers for
+        // gc_depth rounds. Unlike the own-commit key this is NOT airtight: a requeued header below
+        // a foreign commit round may still commit later (this authority's last_committed did not
+        // advance), so the original AND the re-proposal can both land.
+        // `ExecutedBatchRegistry::try_register` absorbs that duplicate at execution admission, so
+        // the registry dedup is load-bearing here: do not weaken one without the other. The
+        // horizon sits `FALLBACK_REQUEUE_GRACE_ROUNDS` below the commit round so ordinary commit
+        // lag never triggers it.
+        let highest_committed = committed_headers
+            .iter()
+            .copied()
+            .max()
+            .unwrap_or_else(|| commit_round.saturating_sub(FALLBACK_REQUEUE_GRACE_ROUNDS));
+        // Split at the horizon: rounds below it are retransmitted; rounds at or above are fresh
+        // (certificates still collecting votes) and the normal commit path handles them, so
+        // re-queueing those would only create duplicates.
+        let fresh = self.proposed_headers.split_off(&highest_committed);
+        let retransmitted = std::mem::replace(&mut self.proposed_headers, fresh);
+        if retransmitted.is_empty() {
             return;
         }
 
-        // re-insert batches for any proposed header from a round below the current commit
-        //
-        // ensure batches are FIFO to re-send them
-        //
-        // payloads: oldest -> newest
-        let mut digests_to_resend = VecDeque::new();
-        // Oldest to newest rounds.
-        let mut retransmit_rounds = Vec::new();
-
-        // loop through proposed headers in order by round
-        for (header_round, header) in &mut self.proposed_headers {
-            let mut digests = header
-                .payload()
-                .into_iter()
-                .map(|(k, v)| ProposerDigest { digest: *k, worker_id: *v })
-                .collect();
-
-            // add payloads and system messages from oldest to newest
-            digests_to_resend.append(&mut digests);
-            retransmit_rounds.push(*header_round);
-        }
-
-        // process rounds that need to be retransmitted
-        if retransmit_rounds.is_empty() {
-            return;
-        }
-
-        let num_digests_to_resend = digests_to_resend.len();
-
-        // prepend missing batches from previous round and update `self`
-        digests_to_resend.append(&mut self.digests);
-        self.digests = digests_to_resend;
-
-        // remove the old headers that failed
-        // the proposed blocks are included in the next header
-        for round in &retransmit_rounds {
-            self.proposed_headers.remove(round);
-        }
+        // Re-queued, never dropped: these digests are quorum'd and seq-consumed, so dropping one
+        // gaps that seq permanently on peers. Pinned by
+        // `digests_survive_gc_advance_and_later_round_commit`.
+        let retransmit_rounds: Vec<Round> = retransmitted.keys().copied().collect();
+        let num_digests_to_resend =
+            self.requeue_front(retransmitted.values(), RequeueReason::CommitLag);
 
         warn!(
             target: "primary::proposer",

crates/consensus/primary/src/state_handler.rs

@@ -37,28 +37,15 @@ impl StateHandler {
         );
     }
 
-    async fn handle_sequenced(
-        &mut self,
-        commit_round: Round,
-        certificates: Vec<(Certificate, bool)>,
-    ) {
-        // Now we are going to signal which of our own batches have been committed, carrying each
-        // one's boundary-drop flag so the proposer cleans pre-boundary headers but keeps dropped
-        // ones for rescue.
-        let own_rounds_committed: Vec<(Round, bool)> = certificates
+    async fn handle_sequenced(&mut self, commit_round: Round, certificates: Vec<Certificate>) {
+        // report which of this authority's own headers the commit covered
+        let own_rounds_committed: Vec<Round> = certificates
             .iter()
-            .filter_map(|(cert, dropped)| {
-                if cert.header().author() == &self.authority_id {
-                    Some((cert.header().round(), *dropped))
-                } else {
-                    None
-                }
-            })
+            .filter(|cert| cert.header().author() == &self.authority_id)
+            .map(|cert| cert.header().round())
             .collect();
         debug!(target: "primary::state_handler", "Own committed rounds {:?} at round {:?}", own_rounds_committed, commit_round);
 
-        // If a reporting channel is available send the committed own
-        // headers to it.
         if let Err(e) = self
             .consensus_bus
             .committed_own_headers()