diff --git a/crates/consensus/primary/src/consensus/state.rs b/crates/consensus/primary/src/consensus/state.rs
index e2364b27..97129e0b 100644
--- a/crates/consensus/primary/src/consensus/state.rs
+++ b/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");
 
diff --git a/crates/consensus/primary/src/consensus_bus.rs b/crates/consensus/primary/src/consensus_bus.rs
index c8d4a5d6..8c9c4946 100644
--- a/crates/consensus/primary/src/consensus_bus.rs
+++ b/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
diff --git a/crates/consensus/primary/src/proposer/recovery.rs b/crates/consensus/primary/src/proposer/recovery.rs
index 1d412641..7a772d05 100644
--- a/crates/consensus/primary/src/proposer/recovery.rs
+++ b/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",
diff --git a/crates/consensus/primary/src/state_handler.rs b/crates/consensus/primary/src/state_handler.rs
index d38d6de8..7e5e5189 100644
--- a/crates/consensus/primary/src/state_handler.rs
+++ b/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()
diff --git a/crates/consensus/primary/src/tests/proposer_tests.rs b/crates/consensus/primary/src/tests/proposer_tests.rs
index 3881b666..8cad52f0 100644
--- a/crates/consensus/primary/src/tests/proposer_tests.rs
+++ b/crates/consensus/primary/src/tests/proposer_tests.rs
@@ -233,29 +233,24 @@ async fn test_retransmit_headers_on_gap() {
         vec![4u32, 6u32, 8u32, 10u32],
         vec![4u32, 6u32, 8u32, 10u32],
         vec![6u32, 8u32, 10u32],
+        // committed={1,4,7}: rounds at or above the horizon (8, 10) stay proposed - their
+        // certificates are still collecting votes and the normal commit path handles them
+        vec![8u32, 10u32],
+        // committed={5,6,7,8}: 10 sits at the horizon's fresh side and stays proposed
+        vec![10u32],
         vec![],
         vec![],
-        vec![],
-        vec![],
-        // 4 and 8 removed as committed; 6 and 10 re-queued by the retransmit loop
-        vec![],
+        // skip-then-commit: 4 and 8 removed as committed; 10 is fresh (>= horizon 8) and stays
+        vec![10u32],
     ];
     let expected_digests_cases = [
         vec![FixedBytes::<32>::with_last_byte(100)],
         vec![FixedBytes::<32>::with_last_byte(100)],
         vec![FixedBytes::<32>::with_last_byte(100)],
-        vec![
-            FixedBytes::<32>::with_last_byte(6),
-            FixedBytes::<32>::with_last_byte(8),
-            FixedBytes::<32>::with_last_byte(10),
-            FixedBytes::<32>::with_last_byte(100),
-        ],
-        // committed={5,6,7,8}: 6 and 8 are removed as committed, only 4 and 10 retransmitted
-        vec![
-            FixedBytes::<32>::with_last_byte(4),
-            FixedBytes::<32>::with_last_byte(10),
-            FixedBytes::<32>::with_last_byte(100),
-        ],
+        vec![FixedBytes::<32>::with_last_byte(6), FixedBytes::<32>::with_last_byte(100)],
+        // committed={5,6,7,8}: 6 and 8 are removed as committed, only the straggler 4 (below
+        // the horizon) retransmits; 10 stays proposed
+        vec![FixedBytes::<32>::with_last_byte(4), FixedBytes::<32>::with_last_byte(100)],
         // committed={10}: 10 removed, 4/6/8 retransmitted
         vec![
             FixedBytes::<32>::with_last_byte(4),
@@ -271,13 +266,9 @@ async fn test_retransmit_headers_on_gap() {
             FixedBytes::<32>::with_last_byte(10),
             FixedBytes::<32>::with_last_byte(100),
         ],
-        // skip-then-commit: 4 and 8 removed as committed (NOT re-queued); only the
-        // truly-uncommitted rounds 6 and 10 are re-queued by the retransmit loop.
-        vec![
-            FixedBytes::<32>::with_last_byte(6),
-            FixedBytes::<32>::with_last_byte(10),
-            FixedBytes::<32>::with_last_byte(100),
-        ],
+        // skip-then-commit: 4 and 8 removed as committed (NOT re-queued); the superseded
+        // round 6 requeues, the fresh round 10 stays proposed.
+        vec![FixedBytes::<32>::with_last_byte(6), FixedBytes::<32>::with_last_byte(100)],
     ];
 
     for i in 0..proposed_headers_cases.len() {
@@ -316,8 +307,7 @@ async fn test_retransmit_headers_on_gap() {
             worker_id: 1u16,
         });
 
-        proposer
-            .process_committed_headers(1, commited_headers.iter().map(|r| (*r, false)).collect());
+        proposer.process_committed_headers(1, commited_headers.to_vec());
 
         let updated_digests =
             proposer.digests.iter().map(|digest| digest.digest).collect::<Vec<_>>();
@@ -329,3 +319,241 @@ async fn test_retransmit_headers_on_gap() {
         assert_eq!(updated_digests, expected_digests.clone());
     }
 }
+
+/// A header carrying exactly one digest, as the proposer would have built it.
+fn header_with(author: AuthorityIdentifier, round: Round, digest: B256) -> Header {
+    let mut payload = IndexMap::new();
+    payload.insert(digest, 0u16);
+    Header::new(author, round, 1, payload, BTreeSet::new(), BlockNumHash::default())
+}
+
+/// The digests currently queued in the proposer, front to back.
+fn queued_digests<DB: Database>(proposer: &Proposer<DB>) -> Vec<B256> {
+    proposer.digests.iter().map(|d| d.digest).collect()
+}
+
+/// Builds a proposer holding one proposed-but-uncommitted header at `round` carrying `digest`,
+/// with one unrelated digest already queued.
+fn proposer_with_pending_header(
+    fixture: &CommitteeFixture<MemDatabase>,
+    round: Round,
+    digest: FixedBytes<32>,
+) -> Proposer<MemDatabase> {
+    let committee = fixture.committee();
+    let primary = fixture.authorities().next().unwrap();
+    let mut proposer = Proposer::new(
+        primary.consensus_config(),
+        primary.consensus_config().authority_id().expect("authority"),
+        ConsensusBus::new(),
+        LeaderSchedule::new(committee.clone(), LeaderSwapTable::default()),
+        TaskManager::default().get_spawner(),
+    );
+
+    proposer.proposed_headers.insert(round, header_with(primary.id(), round, digest));
+    proposer
+        .digests
+        .push_back(ProposerDigest { digest: FixedBytes::<32>::with_last_byte(100), worker_id: 1 });
+    proposer
+}
+
+/// The proposer queue never drops digests: dropping a quorum'd digest would gap the
+/// per-authority seq stream and park every later batch from this authority. Growth past the
+/// warn threshold only logs.
+#[tokio::test]
+async fn push_digest_never_drops() {
+    let fixture = CommitteeFixture::builder(MemDatabase::default).build();
+    let primary = fixture.authorities().next().unwrap();
+    let cb = ConsensusBus::new();
+    let task_manager = TaskManager::default();
+    let mut proposer = Proposer::new(
+        primary.consensus_config(),
+        primary.consensus_config().authority_id().expect("authority"),
+        cb.clone(),
+        LeaderSchedule::new(fixture.committee(), LeaderSwapTable::default()),
+        task_manager.get_spawner(),
+    );
+
+    // fill one past the warn threshold: every digest is retained
+    for _ in 0..=DIGEST_QUEUE_WARN_THRESHOLD {
+        proposer.push_digest(ProposerDigest { digest: B256::random(), worker_id: 0 });
+    }
+    assert_eq!(
+        proposer.digests.len(),
+        DIGEST_QUEUE_WARN_THRESHOLD + 1,
+        "growth past the warn threshold is retained, not evicted"
+    );
+}
+
+/// The fallback horizon (a commit carrying none of our own headers) must forgive ordinary
+/// commit lag: a header a round or two behind the commit is routinely still committable, so
+/// reproposing it manufactures a duplicate commit and parks its successor seqs. Fails against
+/// the ungraced fallback, which requeues at one round of lag.
+#[tokio::test]
+async fn fallback_requeue_forgives_ordinary_commit_lag() {
+    let fixture = CommitteeFixture::builder(MemDatabase::default).build();
+    let header_digest = FixedBytes::<32>::with_last_byte(1);
+    let mut proposer = proposer_with_pending_header(&fixture, 1, header_digest);
+
+    // a foreign commit one round above the header: within the grace, nothing moves
+    proposer.process_committed_headers(2, vec![]);
+
+    assert_eq!(proposer.proposed_headers.keys().copied().collect::<Vec<_>>(), vec![1]);
+    let digests: Vec<_> = proposer.digests.iter().map(|d| d.digest).collect();
+    assert_eq!(digests, vec![FixedBytes::<32>::with_last_byte(100)]);
+}
+
+/// Beyond the grace the fallback must still rescue: a header this far behind foreign commits
+/// is stranded (its votes were rejected), and only the requeue returns its quorum'd,
+/// seq-consumed digests to circulation before the GC horizon.
+#[tokio::test]
+async fn fallback_requeue_still_rescues_beyond_grace() {
+    let fixture = CommitteeFixture::builder(MemDatabase::default).build();
+    let header_digest = FixedBytes::<32>::with_last_byte(1);
+    let mut proposer = proposer_with_pending_header(&fixture, 1, header_digest);
+
+    // a foreign commit past the grace: the header is stranded and its digests requeue in front
+    let beyond_grace = 1 + super::recovery::FALLBACK_REQUEUE_GRACE_ROUNDS + 1;
+    proposer.process_committed_headers(beyond_grace, vec![]);
+
+    assert!(proposer.proposed_headers.is_empty());
+    let digests: Vec<_> = proposer.digests.iter().map(|d| d.digest).collect();
+    assert_eq!(digests, vec![header_digest, FixedBytes::<32>::with_last_byte(100)]);
+}
+
+/// The fallback requeue (no own header committed, keyed on the leader's commit round) stops at
+/// that round: headers above it are fresh, their certificates are still collecting votes, and
+/// the normal commit path handles them. Requeueing them manufactures a duplicate commit
+/// (original + re-proposal) for zero rescue value; the execution registry absorbs it, but it is a
+/// backstop, not a reason to create duplicates.
+#[tokio::test]
+async fn fallback_requeue_stops_at_the_commit_round() {
+    let fixture = CommitteeFixture::builder(MemDatabase::default).build();
+    let primary = fixture.authorities().next().unwrap();
+    let task_manager = TaskManager::default();
+    let mut proposer = Proposer::new(
+        primary.consensus_config(),
+        primary.consensus_config().authority_id().expect("authority"),
+        ConsensusBus::new(),
+        LeaderSchedule::new(fixture.committee(), LeaderSwapTable::default()),
+        task_manager.get_spawner(),
+    );
+
+    let straggler = B256::random();
+    let fresh_low = B256::random();
+    let fresh_high = B256::random();
+    proposer.proposed_headers.insert(91, header_with(primary.id(), 91, straggler));
+    proposer.proposed_headers.insert(103, header_with(primary.id(), 103, fresh_low));
+    proposer.proposed_headers.insert(105, header_with(primary.id(), 105, fresh_high));
+
+    // a foreign-only commit at round 99: none of our headers are in it
+    proposer.process_committed_headers(99, vec![]);
+
+    assert_eq!(
+        queued_digests(&proposer),
+        vec![straggler],
+        "only the straggler below the commit round is requeued"
+    );
+    assert_eq!(
+        proposer.proposed_headers.keys().copied().collect::<Vec<_>>(),
+        vec![103, 105],
+        "headers at or above the commit round stay proposed - their certs are in flight"
+    );
+}
+
+/// A quorum'd digest handed to the proposer is never silently discarded: on every path that
+/// could lose one it is either in a proposed header or in the queue. GC eviction requeues
+/// just like a superseded commit.
+///
+/// Load-bearing beyond the proposer: a dropped digest gaps the per-authority seq stream and
+/// parks every later batch from this authority, and the in-flight TTL does not heal it - the
+/// sweep re-seals those txs under a NEW seq, leaving the original gap.
+#[tokio::test]
+async fn digests_survive_gc_advance_and_later_round_commit() {
+    let fixture = CommitteeFixture::builder(MemDatabase::default).build();
+    let primary = fixture.authorities().next().unwrap();
+    let task_manager = TaskManager::default();
+    let bus = ConsensusBus::new();
+    let mut proposer = Proposer::new(
+        primary.consensus_config(),
+        primary.consensus_config().authority_id().expect("authority"),
+        bus.clone(),
+        LeaderSchedule::new(fixture.committee(), LeaderSwapTable::default()),
+        task_manager.get_spawner(),
+    );
+    // set explicitly so the test does not ride on fixture parameter defaults
+    proposer.gc_depth = 10;
+    proposer.round = 100;
+    // the eviction horizon keys on the COMMITTED round: rounds <= 85 are evictable
+    bus.committed_round_updates().send_replace(95);
+
+    // a quorum'd digest that is queued but not yet proposed
+    let queued = B256::random();
+    proposer.push_digest(ProposerDigest { digest: queued, worker_id: 0 });
+
+    // two proposed-but-uncommitted headers, both inside the retained window
+    let early = B256::random();
+    let late = B256::random();
+    proposer.proposed_headers.insert(91, header_with(primary.id(), 91, early));
+    proposer.proposed_headers.insert(99, header_with(primary.id(), 99, late));
+
+    // GC runs after every successful proposal: it must not evict a header still inside the
+    // window, and must never touch the queue of un-proposed digests.
+    proposer.evict_old_proposed_headers();
+    assert_eq!(
+        proposer.proposed_headers.keys().copied().collect::<Vec<_>>(),
+        vec![91, 99],
+        "GC must retain headers above the gc horizon - their digests are still recoverable"
+    );
+    assert_eq!(
+        queued_digests(&proposer),
+        vec![queued],
+        "GC must not touch the un-proposed digest queue"
+    );
+
+    // the LATER round commits while the earlier one never does: round 91's digest must come
+    // back to the queue, FIFO-ahead of the still-queued one, rather than being dropped with
+    // its header.
+    proposer.process_committed_headers(99, vec![99]);
+    assert!(proposer.proposed_headers.is_empty(), "committed and superseded rounds are drained");
+    assert_eq!(
+        queued_digests(&proposer),
+        vec![early, queued],
+        "the uncommitted round's digest is requeued (oldest first), never discarded"
+    );
+
+    // and a later GC sweep, with the round jumped far past every digest's origin round, still
+    // leaves both queued.
+    proposer.round = 500;
+    proposer.evict_old_proposed_headers();
+    assert_eq!(
+        queued_digests(&proposer),
+        vec![early, queued],
+        "a round advance past the digests' rounds must not discard them"
+    );
+
+    // Commits trail the frontier: proposer at 500, committed at 460, so the eviction horizon
+    // is 450 - NOT 490. Rounds 400/401 sit below it (the leaderless-partition case), 455 sits
+    // in the commit-lag window (450, 490]: still committable by a future leader, so evicting
+    // it would double-commit its batches once it lands. 495 is above every horizon.
+    bus.committed_round_updates().send_replace(460);
+    let old_low = B256::random();
+    let old_high = B256::random();
+    let lag_window = B256::random();
+    let survivor = B256::random();
+    proposer.proposed_headers.insert(400, header_with(primary.id(), 400, old_low));
+    proposer.proposed_headers.insert(401, header_with(primary.id(), 401, old_high));
+    proposer.proposed_headers.insert(455, header_with(primary.id(), 455, lag_window));
+    proposer.proposed_headers.insert(495, header_with(primary.id(), 495, survivor));
+
+    proposer.evict_old_proposed_headers();
+    assert_eq!(
+        proposer.proposed_headers.keys().copied().collect::<Vec<_>>(),
+        vec![455, 495],
+        "eviction keys on the committed round: the commit-lag window stays committable"
+    );
+    assert_eq!(
+        queued_digests(&proposer),
+        vec![old_low, old_high, early, queued],
+        "gc-evicted headers requeue their digests oldest-round-first, ahead of the queue"
+    );
+}
diff --git a/crates/consensus/state-sync/src/lib.rs b/crates/consensus/state-sync/src/lib.rs
index a4cad764..a1ae9587 100644
--- a/crates/consensus/state-sync/src/lib.rs
+++ b/crates/consensus/state-sync/src/lib.rs
@@ -240,7 +240,6 @@ pub fn save_consensus<DB: Database>(
             let _ = txn.remove::<ConsensusBlockNumbersByDigest>(stale_digest);
         }
 
-        // Do not clean NodeBatchesCache table here. It must be clean in `process_committed_headers` in the proposer`
         Ok(())
     })?;
 
@@ -250,22 +249,13 @@ pub fn save_consensus<DB: Database>(
 /// The canonical consensus-chain tip, used to seed the live subscriber's header *numbering* so a
 /// number the network already consumed is never reused.
 ///
-/// Returns the highest `ConsensusBlocks` header on the canonical chain. The raw table tip is not
-/// always canonical: a drain race at an epoch boundary can leave prior-epoch outputs saved above
-/// the certified checkpoint (a "post-boundary leak"). Seeding numbering from such a leak offsets
-/// every new-epoch header by the leak count and forks the chain via a divergent `ConsensusHeader`
-/// digest (which feeds the block's `mix_hash` / `parent_beacon_block_root`). When the tip is a
-/// prior-epoch header above the certified checkpoint, the certified (committee-signed) checkpoint
-/// is returned instead, so the new epoch's first number is the one every validator agrees on.
-///
-/// A current-epoch tip is returned directly, so numbering still sits >= the execution anchor when
-/// execution lags the commit (a crash/static-file heal must not regress numbering and reuse a
-/// number). IMPORTANT: use this ONLY for numbering. Do NOT anchor re-execution/replay on it - that
-/// must stay on the SSOT `executed_anchor`, or committed-but-unexecuted outputs get skipped on
-/// restart and the chain forks. See the note in `epoch_manager/core.rs`.
-///
-/// Returns `None` only when the consensus DB is empty (fresh boot); execution is at 0 then too,
-/// since `save_consensus` persists each header before it executes.
+/// The raw `ConsensusBlocks` tip is not always canonical: a drain race at an epoch boundary can
+/// leave prior-epoch outputs saved above the certified checkpoint, and numbering from them offsets
+/// every new-epoch header and forks the chain through a divergent `ConsensusHeader` digest. Such
+/// a tip yields the committee-signed checkpoint instead; a current-epoch tip is returned as is, so
+/// numbering stays >= the execution anchor when execution lags the commit. Use this for numbering
+/// only: replay must anchor on `executed_anchor`, or committed-but-unexecuted outputs are skipped
+/// on restart. Returns `None` only when the consensus DB is empty.
 pub fn consensus_chain_tip<DB: Database>(config: &ConsensusConfig<DB>) -> Option<ConsensusHeader> {
     let db = config.node_storage();
     let (_, tip) = db.last_record::<ConsensusBlocks>()?;
@@ -441,21 +431,13 @@ fn collect_replayable_headers<DB: Database>(
     result
 }
 
-/// Collect and return any consensus headers that were not executed before last shutdown.
-/// This will be consensus that was reached but had not executed before a shutdown.
-///
-/// ## Interaction with checkpoint-based crash recovery
-///
-/// The manager's `recover_partial_transition()` runs BEFORE this function is called.
-/// That method checks for incomplete epoch transitions (via a DB checkpoint) and either
-/// completes the remaining phases or clears a stale checkpoint. By the time this
-/// function executes, the checkpoint system guarantees that `recently_executed_blocks` accurately
-/// reflects the last executed block.
+/// Collects the consensus headers that were reached but not executed before the last shutdown.
 ///
-/// If a checkpoint still exists when this function runs, it indicates an unexpected
-/// ordering issue -- recovery should have handled it already. We log a warning but
-/// proceed, as the worst case is a harmless replay: the `ExecutorEngine` drops
-/// duplicate/out-of-order outputs via `last_seen_output_number` (processor `<=` check).
+/// `recover_partial_transition()` runs first and completes or clears any epoch-transition
+/// checkpoint, so `recently_executed_blocks` reflects the last executed block by the time this
+/// runs. A checkpoint still present here is an ordering bug; it is logged and the walk proceeds,
+/// since the worst case is a harmless replay that the `ExecutorEngine` drops via
+/// `last_seen_output_number`.
 pub async fn get_missing_consensus<DB: Database>(
     config: &ConsensusConfig<DB>,
     consensus_bus: &ConsensusBus,
diff --git a/crates/consensus/worker/src/batch-builder/tests/it/build_batches.rs b/crates/consensus/worker/src/batch-builder/tests/it/build_batches.rs
index fcf9ef64..58400d7d 100644
--- a/crates/consensus/worker/src/batch-builder/tests/it/build_batches.rs
+++ b/crates/consensus/worker/src/batch-builder/tests/it/build_batches.rs
@@ -199,6 +199,16 @@ async fn test_make_batch_el_to_cl() {
         .expect("batch in store");
     assert_eq!(batch_from_store.beneficiary, address);
 
+    // The seal writes Batches only: nothing reads NodeBatchesCache any more, so a write there is
+    // a regression.
+    assert!(
+        store
+            .get::<rayls_infrastructure_storage::tables::NodeBatchesCache>(&expected_batch.digest())
+            .expect("cache read")
+            .is_none(),
+        "seal must not write the dead NodeBatchesCache table"
+    );
+
     // Sealed transactions are marked in flight, not evicted, so they stay pending (RPC-visible)
     // until execution drains them; every pending tx here reached quorum, so none is re-sealable.
     // (test_make_batch_no_ack_txs_in_pool_still covers the no-quorum case.)
diff --git a/crates/consensus/worker/src/batch-validator/src/validator.rs b/crates/consensus/worker/src/batch-validator/src/validator.rs
index 78350402..93105768 100644
--- a/crates/consensus/worker/src/batch-validator/src/validator.rs
+++ b/crates/consensus/worker/src/batch-validator/src/validator.rs
@@ -2,7 +2,7 @@
 
 use rayls_execution_evm::{
     bytes_to_txn, chainspec::RaylsHardforks, recover_signed_transaction, reth_env::RethEnv,
-    EthPooledTransaction, FixedBytes, WorkerTxPool,
+    EthPooledTransaction, FixedBytes, PoolErrorKind, WorkerTxPool,
 };
 use rayls_infrastructure_types::{
     gas_accumulator::BaseFeeContainer, max_batch_size, BatchValidation, BatchValidationError,
@@ -14,7 +14,7 @@ use rayon::iter::{IntoParallelRefIterator as _, ParallelIterator as _};
 use dashmap::DashMap;
 use rustc_hash::FxHasher;
 use std::hash::Hasher;
-use tracing::{trace, warn};
+use tracing::{debug, trace, warn};
 
 /// Type convenience for implementing block validation errors.
 type BatchValidationResult<T> = Result<T, BatchValidationError>;
@@ -154,9 +154,18 @@ impl BatchValidation for BatchValidator {
             let tx_pool = tx_pool.clone();
             self.reth_env.get_task_spawner().spawn_task("submit-tx-batch", async move {
                 for tx in parsed_txns.into_iter().flatten() {
-                    let res = tx_pool.add_raw_transaction_external(tx).await;
-                    if let Err(e) = res {
-                        warn!(target: "worker::validator", "failed to submit gossipped txn: {e}");
+                    match tx_pool.add_raw_transaction_external(tx).await {
+                        Ok(_) => {}
+                        // A hash this pool already holds is ordinary gossip dedup: multiple
+                        // observers forward overlapping pools, and a forwarder re-sends what
+                        // it cannot confirm landed. Under load this is the common outcome, so
+                        // logging it at warn buries the submissions that did fail.
+                        Err(e) if matches!(e.kind, PoolErrorKind::AlreadyImported) => {
+                            debug!(target: "worker::validator", "gossipped txn already in pool: {e}");
+                        }
+                        Err(e) => {
+                            warn!(target: "worker::validator", "failed to submit gossipped txn: {e}");
+                        }
                     }
                 }
             });
@@ -278,7 +287,7 @@ impl BatchValidator {
         let tip = self.reth_env.canonical_tip();
         let next_block = tip.number + 1;
         if chain_spec.is_eip1559_active_at_block(next_block) {
-            // per-block EIP-1559 active — skip exact match
+            // per-block EIP-1559 active - skip exact match
             return Ok(());
         }
         let expected_base_fee = self.base_fee.base_fee();
diff --git a/crates/consensus/worker/src/worker.rs b/crates/consensus/worker/src/worker.rs
index 49b330b1..d088ab8f 100644
--- a/crates/consensus/worker/src/worker.rs
+++ b/crates/consensus/worker/src/worker.rs
@@ -14,9 +14,7 @@ use rayls_infrastructure_config::ConsensusConfig;
 use rayls_infrastructure_network_types::{
     local::LocalNetwork, WorkerOwnBatchMessage, WorkerToPrimaryClient,
 };
-use rayls_infrastructure_storage::tables::{
-    BatchSeqCounter, Batches, ConsensusBlocks, NodeBatchesCache,
-};
+use rayls_infrastructure_storage::tables::{BatchSeqCounter, Batches, ConsensusBlocks};
 use rayls_infrastructure_types::{
     batch_tracker::{BatchTracker, SealFailureReason},
     error::BlockSealError,
@@ -361,12 +359,6 @@ impl<DB: Database, QW: QuorumWaiterTrait> Worker<DB, QW> {
         );
 
         let (batch, digest) = sealed_batch.split();
-        if let Err(e) = self.store.insert::<NodeBatchesCache>(&digest, &batch) {
-            // Cache the batch early, avoid race conditions.
-            // Note the cache should be cleared every epoch after processing.
-            error!(target: "worker::batch_provider", "Store failed (batch cache) with error: {:?}", e);
-            return Err(BlockSealError::FatalDBFailure);
-        }
         if let Some(tracker) = &self.batch_tracker {
             tracker.batch_sealed(digest, batch.transactions.len(), &sender_nonce_ranges);
         }
diff --git a/crates/execution/evm/src/lib.rs b/crates/execution/evm/src/lib.rs
index 9000e137..57256b2a 100644
--- a/crates/execution/evm/src/lib.rs
+++ b/crates/execution/evm/src/lib.rs
@@ -23,7 +23,7 @@ pub use reth_provider::{AccountReader, CanonStateNotificationStream, ExecutionOu
 pub use reth_rpc_eth_types::EthApiError;
 pub use reth_tracing::FileWorkerGuard;
 pub use reth_transaction_pool::{
-    error::{InvalidPoolTransactionError, PoolError, PoolTransactionError},
+    error::{InvalidPoolTransactionError, PoolError, PoolErrorKind, PoolTransactionError},
     identifier::SenderIdentifiers,
     BestTransactions, EthPooledTransaction, PoolTransaction, TransactionPool as TransactionPoolT,
 };
diff --git a/crates/execution/evm/src/reth_env/config.rs b/crates/execution/evm/src/reth_env/config.rs
index b3d711e0..d14745be 100644
--- a/crates/execution/evm/src/reth_env/config.rs
+++ b/crates/execution/evm/src/reth_env/config.rs
@@ -91,16 +91,15 @@ pub struct RethConfig(pub(crate) NodeConfig<RethChainSpec>);
 
 const DEFAULT_UNUSED_ADDR: IpAddr = IpAddr::V4(Ipv4Addr::UNSPECIFIED);
 
-/// All the rpc modules we allow.
-/// Disallow admin
-pub(super) const ALL_MODULES: [RethRpcModule; 6] = [
+/// RPC modules the node exposes; `admin` is deliberately excluded.
+pub(super) const ALL_MODULES: [RethRpcModule; 7] = [
     RethRpcModule::Eth,
     RethRpcModule::Net,
     RethRpcModule::Web3,
     RethRpcModule::Debug,
     RethRpcModule::Trace,
     RethRpcModule::Rpc,
-    // RethRpcModule::Txpool,
+    RethRpcModule::Txpool,
 ];
 
 impl RethConfig {
diff --git a/crates/infrastructure/storage/src/lib.rs b/crates/infrastructure/storage/src/lib.rs
index 31f72b52..8580f65b 100644
--- a/crates/infrastructure/storage/src/lib.rs
+++ b/crates/infrastructure/storage/src/lib.rs
@@ -92,9 +92,10 @@ pub mod tables {
     #[cfg(feature = "cold-storage")]
     use crate::cold::ColdLocation;
     use rayls_infrastructure_types::{
-        batch_ordering::BatchOrderingState as TypeBatchOrderingState, AuthorityIdentifier, Batch,
-        BlockHash, Certificate, CertificateDigest, ConsensusHeader, Epoch, EpochCertificate,
-        EpochRecord, EpochTransitionCheckpoint, Header, Round, VoteInfo, WorkerId, B256,
+        batch_ordering::StoredBatchOrderingState as TypeStoredBatchOrderingState,
+        AuthorityIdentifier, Batch, BlockHash, Certificate, CertificateDigest, ConsensusHeader,
+        Epoch, EpochCertificate, EpochRecord, EpochTransitionCheckpoint, Header, Round, VoteInfo,
+        WorkerId, B256,
     };
 
     tables!(
@@ -113,7 +114,9 @@ pub mod tables {
         ConsensusBlockNumbersByDigest;crate::CONSENSUS_BLOCK_NUMBER_BY_DIGEST_CF;<BlockHash, u64>,
         // This is a cache to store verified but unprocessed consensus headers, remove once processed.
         ConsensusBlocksCache;crate::CONSENSUS_BLOCK_CACHE_CF;<u64, ConsensusHeader>,
-        // This is a cache to store this nodes batches before consensus, remove once in a ConsensusHeader.
+        // No longer written: graceful-shutdown txpool persistence replaced the seal-time batch
+        // cache. Retained so existing databases keep a stable column-family set; still cleared
+        // on foreign-DB sanitization to purge rows written by older binaries.
         NodeBatchesCache;crate::NODE_BATCHES_CACHE_CF;<BlockHash, Batch>,
         // These tables are for the epoch chain not the normal consensus.
         EpochRecords;crate::EPOCH_RECORDS_CF;<Epoch, EpochRecord>,
@@ -131,7 +134,7 @@ pub mod tables {
         // Node identity: stores this validator's AuthorityIdentifier for foreign DB detection.
         NodeIdentity;crate::NODE_IDENTITY_CF;<u8, AuthorityIdentifier>,
         // Batch ordering state for the current epoch.
-        BatchOrderingState;crate::BATCH_ORDERING_STATE_CF;<u8, TypeBatchOrderingState>
+        BatchOrderingState;crate::BATCH_ORDERING_STATE_CF;<u8, TypeStoredBatchOrderingState>
     );
 
     // Cold-tier tables, compiled only with the `cold-storage` feature.
diff --git a/crates/infrastructure/storage/src/stores/batch_ordering_store.rs b/crates/infrastructure/storage/src/stores/batch_ordering_store.rs
index 9dedd828..ce6e67dd 100644
--- a/crates/infrastructure/storage/src/stores/batch_ordering_store.rs
+++ b/crates/infrastructure/storage/src/stores/batch_ordering_store.rs
@@ -1,24 +1,110 @@
-use crate::{tables::BatchOrderingState as TableBatchOrderingState, StoreResult};
-use rayls_infrastructure_types::{batch_ordering::BatchOrderingState, Database};
+use crate::{
+    tables::{BatchOrderingState as TableBatchOrderingState, Batches},
+    StoreResult,
+};
+use rayls_infrastructure_types::{
+    batch_ordering::{AuthoritySeqState, BatchOrderingState, StoredBatchOrderingState},
+    decode, try_decode, B256Map, Batch, Database, DbTx, B256,
+};
+use std::{collections::BTreeMap, sync::Arc};
+use tracing::warn;
 
-/// The ordering state key - always 0 since we store one epoch's worth at a time.
+/// Key of the single ordering-state row; only the current epoch's state is kept.
 pub const ORDERING_KEY: u8 = 0;
 
-/// Trait for persisting batch ordering state.
+/// Persistence of the per-epoch batch ordering state.
 pub trait BatchOrderingStore {
-    /// Write the entire batch ordering state for the current epoch.
+    /// Writes the entire batch ordering state for the current epoch.
     fn write_batch_ordering_state(&self, ordering: &BatchOrderingState) -> StoreResult<()>;
 
-    /// Read the batch ordering state.
+    /// Reads the batch ordering state, reloading parked batch bodies from the `Batches` table.
     fn read_batch_ordering_state(&self) -> StoreResult<Option<BatchOrderingState>>;
 }
 
 impl<DB: Database> BatchOrderingStore for DB {
     fn write_batch_ordering_state(&self, ordering: &BatchOrderingState) -> StoreResult<()> {
-        self.insert::<TableBatchOrderingState>(&ORDERING_KEY, ordering)
+        // Persist parked batches by digest, not by value: a committed batch's `Batches` row
+        // survives the reboot, so the transaction bytes would only be duplicated in the blob.
+        let stored = StoredBatchOrderingState::from(ordering);
+        self.insert::<TableBatchOrderingState>(&ORDERING_KEY, &stored)
     }
 
     fn read_batch_ordering_state(&self) -> StoreResult<Option<BatchOrderingState>> {
-        self.get::<TableBatchOrderingState>(&ORDERING_KEY)
+        // Every read txn here does raw reads only; decoding happens after it closes. Holding a
+        // read txn across the deserialization of a parking storm's worth of transaction bytes
+        // would trip the MDBX read-txn timeout mid-recovery.
+        let Some(raw) = self.with_read_txn(|txn| {
+            Ok(txn
+                .raw_get::<TableBatchOrderingState>(&ORDERING_KEY)?
+                .map(|bytes| bytes.into_owned()))
+        })?
+        else {
+            return Ok(None);
+        };
+
+        // A `ParkedRef` is fixed-size, so the compact decode succeeds only on the current format: a
+        // legacy blob with any parked entry leaves trailing bytes and fails it, while an all-empty
+        // blob is byte-identical and decodes either way harmlessly.
+        let Ok(stored) = try_decode::<StoredBatchOrderingState>(&raw) else {
+            // Backwards compatibility: an older binary wrote the parked batches by value, so the
+            // blob carries the bodies. Decode it directly; the next persist rewrites the compact
+            // format.
+            let legacy = try_decode(&raw).map_err(|e| {
+                eyre::eyre!("undecodable batch ordering state (neither format): {e}")
+            })?;
+            return Ok(Some(legacy));
+        };
+
+        // Collect the parked bodies in one more short txn (raw reads only), then decode and
+        // assemble once it has closed.
+        let digests: Vec<B256> = stored
+            .authorities
+            .values()
+            .flat_map(|auth| auth.parked.values().map(|reference| reference.batch_digest))
+            .collect();
+        let bodies = self.with_read_txn(|txn| {
+            let mut bodies = B256Map::default();
+            for digest in &digests {
+                if let Some(bytes) = txn.raw_get::<Batches>(digest)? {
+                    bodies.insert(*digest, bytes.into_owned());
+                }
+            }
+            Ok(bodies)
+        })?;
+
+        Ok(Some(reconstruct_parked(stored, &bodies)))
+    }
+}
+
+/// Rebuilds each authority's parked map by pairing every reference with its reloaded body,
+/// dropping any entry whose body is absent from `Batches`.
+///
+/// A dropped entry leaves a seq gap the ordering waits to refill; a committed parked batch's row
+/// always survives the reboot, so the drop is defensive.
+fn reconstruct_parked(
+    stored: StoredBatchOrderingState,
+    bodies: &B256Map<Vec<u8>>,
+) -> BatchOrderingState {
+    let mut authorities = BTreeMap::new();
+    for (addr, auth) in stored.authorities {
+        let mut parked = BTreeMap::new();
+        for (seq, reference) in auth.parked {
+            match bodies.get(&reference.batch_digest) {
+                Some(bytes) => {
+                    let batch: Batch = decode(bytes);
+                    parked.insert(seq, reference.into_prepared(Arc::new(batch)));
+                }
+                None => warn!(
+                    target: "engine",
+                    ?addr,
+                    seq,
+                    batch_digest = ?reference.batch_digest,
+                    "dropping parked batch on restart: no Batches row for its digest"
+                ),
+            }
+        }
+        authorities
+            .insert(addr, AuthoritySeqState { last_executed_seq: auth.last_executed_seq, parked });
     }
+    BatchOrderingState { epoch: stored.epoch, authorities }
 }
diff --git a/crates/infrastructure/types/src/batch_tracker.rs b/crates/infrastructure/types/src/batch_tracker.rs
index ebcd9076..579e7833 100644
--- a/crates/infrastructure/types/src/batch_tracker.rs
+++ b/crates/infrastructure/types/src/batch_tracker.rs
@@ -45,6 +45,11 @@ pub enum BatchStage {
     /// Parked batch discarded at the epoch boundary instead of force executed; its txs stay
     /// pooled.
     DiscardedAtBoundary = 1 << 13,
+    /// Proposer requeued the digest for re-proposal (GC eviction or a missed commit window).
+    RequeuedInProposer = 1 << 14,
+    /// Parked batch force-drained by the epoch-boundary reset (pre OutputSeqNormalization only;
+    /// the fork discards instead).
+    ForceDrained = 1 << 15,
 }
 
 impl fmt::Display for BatchStage {
@@ -64,6 +69,8 @@ impl fmt::Display for BatchStage {
             Self::Parked => "Parked",
             Self::SealFailed => "SealFailed",
             Self::DiscardedAtBoundary => "DiscardedAtBoundary",
+            Self::RequeuedInProposer => "RequeuedInProposer",
+            Self::ForceDrained => "ForceDrained",
         })
     }
 }
@@ -79,6 +86,24 @@ pub enum SealFailureReason {
     ReportUnacknowledged,
 }
 
+/// Why the proposer requeued a header's digests for re-proposal.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum RequeueReason {
+    /// GC evicted the proposed header past the committed-round horizon.
+    GcEvict,
+    /// The header missed its commit window while a later round committed.
+    CommitLag,
+}
+
+impl fmt::Display for RequeueReason {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        f.write_str(match self {
+            Self::GcEvict => "gc_evict",
+            Self::CommitLag => "commit_lag",
+        })
+    }
+}
+
 impl fmt::Display for SealFailureReason {
     fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
         f.write_str(match self {
@@ -159,7 +184,7 @@ impl BatchEntry {
     }
 
     fn stages_str(&self) -> String {
-        const ALL: [BatchStage; 14] = [
+        const ALL: [BatchStage; 16] = [
             BatchStage::Sealed,
             BatchStage::QuorumReached,
             BatchStage::ReportedToPrimary,
@@ -174,6 +199,8 @@ impl BatchEntry {
             BatchStage::Parked,
             BatchStage::SealFailed,
             BatchStage::DiscardedAtBoundary,
+            BatchStage::RequeuedInProposer,
+            BatchStage::ForceDrained,
         ];
         ALL.iter().filter(|s| self.has(**s)).map(|s| s.to_string()).collect::<Vec<_>>().join(",")
     }
@@ -191,7 +218,6 @@ struct OutputEntry {
 pub struct BatchTracker {
     batches: DashMap<crate::BlockHash, BatchEntry>,
     outputs: DashMap<u64, OutputEntry>,
-    // counters
     total_tracked: AtomicU64,
     total_dropped_proposer: AtomicU64,
     total_txs_dropped: AtomicU64,
@@ -216,7 +242,7 @@ impl Default for BatchTracker {
 }
 
 impl BatchTracker {
-    /// Create a new tracker.
+    /// Creates an empty tracker.
     pub fn new() -> Self {
         Self {
             batches: DashMap::new(),
@@ -282,6 +308,18 @@ impl BatchTracker {
         trace!(target: "batch_tracker", ?digest, "batch_reported_to_primary");
     }
 
+    /// Proposer requeued a header's digests for re-proposal.
+    pub fn digests_requeued_in_proposer(
+        &self,
+        digests: impl IntoIterator<Item = crate::BlockHash>,
+        reason: RequeueReason,
+    ) {
+        for digest in digests {
+            self.batches.entry(digest).or_default().mark(BatchStage::RequeuedInProposer);
+            trace!(target: "batch_tracker", ?digest, %reason, "digest_requeued_in_proposer");
+        }
+    }
+
     /// Proposer received a digest.
     pub fn digest_queued_in_proposer(&self, digest: crate::BlockHash) {
         let mut entry = self.batches.entry(digest).or_default();
@@ -377,6 +415,15 @@ impl BatchTracker {
         trace!(target: "batch_tracker", ?digest, "batch_parked");
     }
 
+    /// Parked batch force-drained by the epoch-boundary reset: its predecessor seq never executed
+    /// in the batch's created epoch, so it executes out of order at the boundary. Pre
+    /// OutputSeqNormalization only; the fork discards instead.
+    pub fn batch_force_drained(&self, digest: crate::BlockHash, seq: u64) {
+        let mut entry = self.batches.entry(digest).or_default();
+        entry.mark(BatchStage::ForceDrained);
+        warn!(target: "batch_tracker", ?digest, seq, "batch_force_drained");
+    }
+
     /// Parked batch discarded whole at the epoch boundary instead of force executed out of
     /// order; its txs stay pooled and are re-sealed in the new epoch.
     pub fn batch_discarded_at_boundary(&self, digest: crate::BlockHash, seq: u64) {
@@ -459,11 +506,10 @@ impl BatchTracker {
         }
     }
 
-    /// Check for batches stuck at intermediate stages and log gaps.
+    /// Logs batches stuck at an intermediate stage and evicts stale entries.
     ///
-    /// Call periodically (e.g. every 30s or on each new block notification).
-    /// Combines stuck-batch detection and cleanup in a single `retain` pass
-    /// to avoid multiple O(n) iterations over the DashMap.
+    /// Call periodically. Detection and cleanup share one `retain` pass so the map is walked
+    /// once.
     pub fn check_gaps(&self) {
         let now = Instant::now();
         let stale_threshold = std::time::Duration::from_secs(60);
@@ -499,7 +545,6 @@ impl BatchTracker {
             true // keep
         });
 
-        // Always log periodic summary at info level so it's visible
         let total_tracked = self.total_tracked.load(Ordering::Relaxed);
         let total_dropped = self.total_dropped_proposer.load(Ordering::Relaxed);
         let total_txs_dropped = self.total_txs_dropped.load(Ordering::Relaxed);
@@ -533,7 +578,7 @@ impl BatchTracker {
         self.check_output_gaps();
     }
 
-    /// Detect gaps in output numbers.
+    /// Warns on gaps between tracked output numbers.
     fn check_output_gaps(&self) {
         let mut numbers: Vec<u64> = self.outputs.iter().map(|e| *e.key()).collect();
         if numbers.len() < 2 {
@@ -554,7 +599,7 @@ impl BatchTracker {
         }
     }
 
-    /// Log a summary of current tracking state.
+    /// Logs a summary of the current tracking state.
     pub fn summary(&self) {
         trace!(
             target: "batch_tracker",
diff --git a/crates/infrastructure/types/src/lib.rs b/crates/infrastructure/types/src/lib.rs
index 4376faa4..3698ee2c 100644
--- a/crates/infrastructure/types/src/lib.rs
+++ b/crates/infrastructure/types/src/lib.rs
@@ -58,8 +58,9 @@ pub use alloy::{
     genesis::{Genesis, GenesisAccount},
     hex::{self, FromHex},
     primitives::{
-        address, hex_literal, keccak256, map::B256Set, Address, BlockHash, BlockNumber, Bloom,
-        Bytes, Sealable, TxHash, TxKind, B256, U160, U256,
+        address, hex_literal, keccak256,
+        map::{B256Map, B256Set},
+        Address, BlockHash, BlockNumber, Bloom, Bytes, Sealable, TxHash, TxKind, B256, U160, U256,
     },
     rpc::types::{AccessList, Withdrawals},
     signers::Signature as EthSignature,
diff --git a/crates/infrastructure/types/src/processor/batch_ordering.rs b/crates/infrastructure/types/src/processor/batch_ordering.rs
index 4d6108dc..7999eabf 100644
--- a/crates/infrastructure/types/src/processor/batch_ordering.rs
+++ b/crates/infrastructure/types/src/processor/batch_ordering.rs
@@ -1,8 +1,8 @@
 //! Per-authority batch sequence ordering with parking for out-of-order batches.
 
-use std::collections::{BTreeMap, HashMap};
+use std::{collections::BTreeMap, sync::Arc};
 
-use crate::{Address, Epoch, PreparedBatch};
+use crate::{Address, Batch, Epoch, PreparedBatch, WorkerId, B256};
 use serde::{Deserialize, Serialize};
 
 /// Maximum number of parked batches per authority before forced out-of-order execution.
@@ -22,7 +22,7 @@ pub enum AcceptResult {
 /// Per-authority ordering state.
 #[derive(Clone, Debug, Default, Serialize, Deserialize)]
 pub struct AuthoritySeqState {
-    /// The highest seq we've executed for this authority. None means first batch not yet seen.
+    /// The highest seq executed for this authority; `None` until the first batch is seen.
     pub last_executed_seq: Option<u64>,
     /// Batches waiting for their predecessor, keyed by seq.
     pub parked: BTreeMap<u64, PreparedBatch>,
@@ -34,5 +34,169 @@ pub struct BatchOrderingState {
     /// The epoch these ordering states belong to.
     pub epoch: Epoch,
     /// Per-authority ordering state, keyed by ECDSA address.
-    pub authorities: HashMap<Address, AuthoritySeqState>,
+    pub authorities: BTreeMap<Address, AuthoritySeqState>,
+}
+
+/// A parked batch persisted by reference.
+///
+/// Carries every [`PreparedBatch`] field except the batch body, which is reloaded from the
+/// `Batches` table on restart - a committed batch's row outlives the reboot, so persisting the
+/// transaction bytes in the ordering blob only duplicates them. Dropping them keeps `persist`
+/// cheap even in the degraded regime that parks the per-authority limit on every output.
+#[derive(Clone, Debug, Serialize, Deserialize)]
+pub struct ParkedRef {
+    /// Digest addressing the batch body in the `Batches` table.
+    pub batch_digest: B256,
+    /// ECDSA address of the authority.
+    pub beneficiary: Address,
+    /// The ConsensusHeader digest.
+    pub output_digest: B256,
+    /// The output nonce (epoch << 32 | round).
+    pub output_nonce: u64,
+    /// Commit timestamp from the output.
+    pub timestamp: u64,
+    /// The epoch from the output (for gas limit calc).
+    pub epoch: Epoch,
+    /// Worker ID from the batch.
+    pub worker_id: WorkerId,
+    /// Original batch index in the subdag.
+    pub batch_index: usize,
+    /// True when this batch was drained from the parking area.
+    pub drained: bool,
+    /// Block gas limit.
+    pub gas_limit: u64,
+}
+
+impl From<&PreparedBatch> for ParkedRef {
+    fn from(prepared: &PreparedBatch) -> Self {
+        Self {
+            batch_digest: prepared.batch_digest,
+            beneficiary: prepared.beneficiary,
+            output_digest: prepared.output_digest,
+            output_nonce: prepared.output_nonce,
+            timestamp: prepared.timestamp,
+            epoch: prepared.epoch,
+            worker_id: prepared.worker_id,
+            batch_index: prepared.batch_index,
+            drained: prepared.drained,
+            gas_limit: prepared.gas_limit,
+        }
+    }
+}
+
+impl ParkedRef {
+    /// Pairs the reference with its reloaded batch body to rebuild the full [`PreparedBatch`].
+    pub fn into_prepared(self, batch: Arc<Batch>) -> PreparedBatch {
+        PreparedBatch {
+            batch,
+            batch_digest: self.batch_digest,
+            beneficiary: self.beneficiary,
+            output_digest: self.output_digest,
+            output_nonce: self.output_nonce,
+            timestamp: self.timestamp,
+            epoch: self.epoch,
+            worker_id: self.worker_id,
+            batch_index: self.batch_index,
+            drained: self.drained,
+            gas_limit: self.gas_limit,
+        }
+    }
+}
+
+/// Per-authority ordering state as persisted: parked batches held by reference.
+#[derive(Clone, Debug, Default, Serialize, Deserialize)]
+pub struct StoredAuthoritySeqState {
+    /// The highest seq executed for this authority; `None` until the first batch is seen.
+    pub last_executed_seq: Option<u64>,
+    /// Parked batches by seq, held by digest reference.
+    pub parked: BTreeMap<u64, ParkedRef>,
+}
+
+/// Batch ordering state as persisted: the on-disk form of [`BatchOrderingState`] that stores
+/// parked batches by digest instead of by value.
+#[derive(Clone, Debug, Default, Serialize, Deserialize)]
+pub struct StoredBatchOrderingState {
+    /// The epoch these ordering states belong to.
+    pub epoch: Epoch,
+    /// Per-authority ordering state, keyed by ECDSA address.
+    pub authorities: BTreeMap<Address, StoredAuthoritySeqState>,
+}
+
+impl From<&BatchOrderingState> for StoredBatchOrderingState {
+    fn from(state: &BatchOrderingState) -> Self {
+        Self {
+            epoch: state.epoch,
+            authorities: state
+                .authorities
+                .iter()
+                .map(|(addr, auth)| {
+                    (
+                        *addr,
+                        StoredAuthoritySeqState {
+                            last_executed_seq: auth.last_executed_seq,
+                            parked: auth
+                                .parked
+                                .iter()
+                                .map(|(seq, prepared)| (*seq, ParkedRef::from(prepared)))
+                                .collect(),
+                        },
+                    )
+                })
+                .collect(),
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::{encode, try_decode, Batch, ExecHeader};
+
+    fn legacy_state_with_parked() -> BatchOrderingState {
+        let batch = Batch::new_for_test(vec![vec![0u8; 64]], ExecHeader::default(), 0, 0, 7);
+        let digest = batch.digest();
+        let prepared = PreparedBatch {
+            batch: Arc::new(batch),
+            batch_digest: digest,
+            beneficiary: Address::from([9u8; 20]),
+            output_digest: B256::ZERO,
+            output_nonce: 0,
+            timestamp: 0,
+            epoch: 3,
+            worker_id: 0,
+            batch_index: 0,
+            drained: false,
+            gas_limit: 30_000_000,
+        };
+        let mut state = BatchOrderingState { epoch: 3, ..Default::default() };
+        state.authorities.insert(
+            prepared.beneficiary,
+            AuthoritySeqState { last_executed_seq: Some(6), parked: [(7, prepared)].into() },
+        );
+        state
+    }
+
+    /// The restart read tells the two on-disk formats apart by decode success: a legacy (by-value)
+    /// blob with any parked entry must fail the compact decode and pass the legacy one, and an
+    /// empty blob must decode either way. Otherwise the read would misinterpret one format as the
+    /// other instead of falling back.
+    #[test]
+    fn stored_and_legacy_ordering_blobs_are_distinguishable_by_decode() {
+        let legacy_bytes = encode(&legacy_state_with_parked());
+        assert!(
+            try_decode::<StoredBatchOrderingState>(&legacy_bytes).is_err(),
+            "a legacy blob with parked entries must not decode as the compact format"
+        );
+        assert!(
+            try_decode::<BatchOrderingState>(&legacy_bytes).is_ok(),
+            "the legacy fallback must decode the legacy blob"
+        );
+
+        let empty_bytes = encode(&BatchOrderingState { epoch: 5, ..Default::default() });
+        assert_eq!(
+            try_decode::<StoredBatchOrderingState>(&empty_bytes).expect("empty decodes").epoch,
+            5,
+            "an all-empty blob is byte-identical across formats"
+        );
+    }
 }
diff --git a/crates/middleware/orchestrator/src/engine/node_inner.rs b/crates/middleware/orchestrator/src/engine/node_inner.rs
index 38132d3a..afb18288 100644
--- a/crates/middleware/orchestrator/src/engine/node_inner.rs
+++ b/crates/middleware/orchestrator/src/engine/node_inner.rs
@@ -147,6 +147,14 @@ impl ExecutionNodeInner {
         initial_batch_seq: u64,
         epoch_boundary: u64,
     ) -> eyre::Result<()> {
+        // The node-scoped in-flight tracker is shared by every worker pool: a second worker's
+        // boundary clear or reconcile would wipe the sibling's marks. Fence the assumption until
+        // multi-worker mark scoping exists.
+        eyre::ensure!(
+            self.workers.len() == 1,
+            "in-flight dedup assumes a single worker (found {})",
+            self.workers.len()
+        );
         // check for worker components and initialize if they're missing
         let transaction_pool = self
             .workers
diff --git a/crates/middleware/orchestrator/src/epoch_manager/core.rs b/crates/middleware/orchestrator/src/epoch_manager/core.rs
index a540ae66..5c304706 100644
--- a/crates/middleware/orchestrator/src/epoch_manager/core.rs
+++ b/crates/middleware/orchestrator/src/epoch_manager/core.rs
@@ -673,7 +673,7 @@ where
             self.collect_epoch_votes(&primary, epoch_rec, &epoch_task_manager).await;
         }
 
-        // biased: shutdown > boundary > mode_transition > task crash
+        // biased: node shutdown > consensus shutdown > boundary > mode_transition > task crash
         // snapshot before select: join() fires shutdown as side-effect
         let was_externally_shutdown = epoch_shutdown_rx.noticed();
 
@@ -682,6 +682,12 @@ where
 
             _ = node_ended => RunningOutcome::NodeShutdown,
 
+            // An external consensus shutdown arriving during the select resolves here rather than
+            // through the join arm below, where a critical task exiting Ok in response to it would
+            // be misclassified as a crash and kill the node. `was_externally_shutdown` only
+            // samples the state before the select, so it cannot catch a notify that lands during.
+            _ = epoch_shutdown_rx => RunningOutcome::NodeShutdown,
+
             res = self.detect_epoch_boundary(epoch_boundary, to_engine, consensus_output) => {
                 match res {
                     Ok((target_hash, boundary_output)) => {
diff --git a/crates/middleware/orchestrator/src/epoch_manager/state.rs b/crates/middleware/orchestrator/src/epoch_manager/state.rs
index 696504a2..eef20f11 100644
--- a/crates/middleware/orchestrator/src/epoch_manager/state.rs
+++ b/crates/middleware/orchestrator/src/epoch_manager/state.rs
@@ -70,7 +70,7 @@ where
                 self.prev_epoch_record.as_ref(),
                 epoch,
             );
-            // Neither memory nor disk has it — e.g. a restart before vote quorum
+            // Neither memory nor disk has it - e.g. a restart before vote quorum
             // persisted this node's copy. Peers closed the previous epoch and hold
             // its certified record, so fetch it directly instead of failing and
             // waiting for the async collector to backfill (which may not win the
@@ -149,7 +149,7 @@ where
         // and the durable canonical tip. Read the reth canonical head directly rather than the
         // in-memory recently_executed_blocks (which is fed asynchronously by the engine-update
         // task): the tip is the epoch-closing block the engine finalized before this runs,
-        // so parent_state is deterministic and race-free — and identical to the value the
+        // so parent_state is deterministic and race-free - and identical to the value the
         // pre-anchor code committed.
         let parent_state = engine.get_reth_env().await.canonical_tip().num_hash();
 
@@ -248,7 +248,8 @@ where
             txn.clear_table::<LastProposed>()?;
             txn.clear_table::<Votes>()?;
             txn.clear_table::<Payload>()?;
-            // node-specific long-lived tables
+            // node-specific long-lived tables (NodeBatchesCache is no longer written; cleared to
+            // purge rows an older binary may have left in an imported DB)
             txn.clear_table::<NodeBatchesCache>()?;
             txn.clear_table::<EpochTransitionCheckpoints>()?;
             txn.clear_table::<BatchSeqCounter>()?;
diff --git a/crates/middleware/processor/src/batch/ordering.rs b/crates/middleware/processor/src/batch/ordering.rs
index 466d358a..731ca62f 100644
--- a/crates/middleware/processor/src/batch/ordering.rs
+++ b/crates/middleware/processor/src/batch/ordering.rs
@@ -2,7 +2,7 @@
 
 use std::sync::Arc;
 
-use std::collections::{BTreeMap, HashMap};
+use std::{cmp::Ordering, collections::BTreeMap};
 
 use parking_lot::Mutex;
 use rayls_infrastructure_storage::{
@@ -13,7 +13,7 @@ use rayls_infrastructure_types::{
     batch_ordering::{AuthoritySeqState, BatchOrderingState, MAX_PARKED_PER_AUTHORITY},
     batch_tracker::BatchTracker,
     executed_batch_registry::ExecutedBatchRegistry,
-    AcceptResult, Address, Database, DbTx, Epoch, PreparedBatch,
+    leader_epoch_and_batch_digests, AcceptResult, Address, Database, DbTx, Epoch, PreparedBatch,
 };
 use tracing::{debug, info, warn};
 
@@ -68,8 +68,8 @@ impl<DB: Database> BatchOrdering<DB> {
 
         if let Some(last_seq) = auth.last_executed_seq {
             if batch_seq <= last_seq {
-                // stale reproposal: park would land under an already-executed seq
-                // where drain_consecutive never looks; defer to dedup registry instead
+                // stale re-proposal: parking would land under an already-executed seq where
+                // drain_consecutive never looks; the dedup registry absorbs it instead
                 info!(
                     target: "engine",
                     ?authority,
@@ -90,7 +90,6 @@ impl<DB: Database> BatchOrdering<DB> {
                         batch_digest = ?prepared.batch_digest,
                         "parking limit reached, executing batch out of order"
                     );
-                    // fall through - accept as in-order (forced)
                     auth.last_executed_seq = Some(batch_seq);
                     // Drop the parked entries the forced jump abandoned: drain_consecutive only
                     // looks at last_executed_seq + 1, so they can never drain, and leaving them
@@ -113,20 +112,18 @@ impl<DB: Database> BatchOrdering<DB> {
                     batch_digest = ?prepared.batch_digest,
                     "parking out-of-order batch"
                 );
-                // Park without registering in the dedup guard - the batch
-                // has not been executed yet. Registration happens when the
-                // batch is eventually drained via drain_consecutive.
+                // Not registered in the dedup guard yet: the batch has not executed. Registration
+                // happens when drain_consecutive releases it.
                 auth.parked.insert(batch_seq, prepared);
                 return AcceptResult::Parked;
             }
         }
 
-        // first batch from this authority OR in-order - accept
         auth.last_executed_seq = Some(batch_seq);
         AcceptResult::InOrder(prepared)
     }
 
-    /// Build ordering state for `current_epoch`, preferring persisted state over a history reseed.
+    /// Builds the ordering for `current_epoch`, preferring persisted state over a history reseed.
     ///
     /// Persisted state one epoch behind is kept, not reseeded: a restart between the boundary block
     /// and the next epoch's first output leaves the closing epoch's parked batches undrained (so
@@ -154,10 +151,11 @@ impl<DB: Database> BatchOrdering<DB> {
         }
         Self::new(store, state)
     }
-    /// Drain consecutive parked batches starting from the next expected seq.
+
+    /// Drains consecutive parked batches starting from the next expected seq into `collected`.
     ///
-    /// When `check_dedup` is true (V2), each drained batch is registered in the
-    /// dedup guard. When false (V1), batches were pre-registered at park time.
+    /// With `check_dedup` each drained batch is registered in the dedup guard and skipped if
+    /// already executed; without it the batches were registered at park time.
     pub fn drain_consecutive(
         &self,
         authority: Address,
@@ -207,18 +205,32 @@ impl<DB: Database> BatchOrdering<DB> {
         }
     }
 
-    /// On epoch change, drain ALL parked batches sorted deterministically by `(beneficiary, seq)`.
+    /// Drains every parked batch on an epoch change, sorted by `(beneficiary, seq)` so all
+    /// validators execute them in the same order.
     ///
-    /// Returns an empty vec if the epoch has not changed.
+    /// Returns an empty vec when the epoch has not advanced.
     pub fn drain_epoch(&self, new_epoch: Epoch) -> Vec<PreparedBatch> {
         let mut state = self.inner.batch_ordering_state.lock();
-        if state.epoch == new_epoch {
-            return Vec::new();
+        match new_epoch.cmp(&state.epoch) {
+            Ordering::Equal => return Vec::new(),
+            // Never rewind: the epoch is a monotonic position. A crash after finalize but before
+            // the next output can replay an already-passed epoch's output; draining and moving the
+            // epoch backward here would double-drain the parked set and desync the frame.
+            Ordering::Less => {
+                warn!(
+                    target: "engine",
+                    current_epoch = state.epoch,
+                    replayed_epoch = new_epoch,
+                    "ignoring a batch-ordering drain for an already-passed epoch"
+                );
+                return Vec::new();
+            }
+            Ordering::Greater => {}
         }
 
         let total_parked: usize = state.authorities.values().map(|a| a.parked.len()).sum();
         let mut drained: Vec<PreparedBatch> = Vec::with_capacity(total_parked);
-        for (authority, auth_state) in state.authorities.drain() {
+        for (authority, auth_state) in std::mem::take(&mut state.authorities) {
             for (seq, parked) in auth_state.parked {
                 warn!(
                     target: "engine",
@@ -232,7 +244,6 @@ impl<DB: Database> BatchOrdering<DB> {
             }
         }
 
-        // sort by (beneficiary, seq) for deterministic execution order
         drained
             .sort_by(|a, b| a.beneficiary.cmp(&b.beneficiary).then(a.batch.seq.cmp(&b.batch.seq)));
 
@@ -258,27 +269,27 @@ impl<DB: Database> BatchOrdering<DB> {
 
     /// Write the current ordering state to the store.
     pub fn persist(&self) {
-        let state = self.inner.batch_ordering_state.lock();
+        // Snapshot under the lock, write outside it: holding the state lock across the DB write
+        // stalls every try_accept/drain for the write's duration (the persist-starvation class).
+        let state = self.inner.batch_ordering_state.lock().clone();
         self.inner
             .batch_ordering_store
             .write_batch_ordering_state(&state)
             .expect("DB write failed");
     }
 
-    /// Snapshot the highest seq executed (or recovered) for `authority`.
-    ///
-    /// `None` means no batch has been observed for this authority in the
-    /// current epoch.
+    /// Returns the highest seq executed (or recovered) for `authority`, or `None` when no batch
+    /// has been observed for it in the current epoch.
     pub fn last_executed_seq(&self, authority: Address) -> Option<u64> {
         self.inner.batch_ordering_state.lock().authorities.get(&authority)?.last_executed_seq
     }
 
-    /// Snapshot the number of authorities tracked in the current epoch.
+    /// Returns the number of authorities tracked in the current epoch.
     pub fn tracked_authorities(&self) -> usize {
         self.inner.batch_ordering_state.lock().authorities.len()
     }
 
-    /// Snapshot the number of parked batches for `authority`.
+    /// Returns the number of parked batches for `authority`.
     pub fn parked_count(&self, authority: Address) -> usize {
         self.inner
             .batch_ordering_state
@@ -290,34 +301,42 @@ impl<DB: Database> BatchOrdering<DB> {
     }
 }
 
-/// Walk `ConsensusBlocks` in reverse, accumulating per-beneficiary max
-/// `batch.seq` for `current_epoch`. Stops at the first block of an earlier
-/// epoch (BatchOrdering state is per-epoch). Returns an empty map if the
-/// store contains no blocks for `current_epoch`.
+/// Walks `ConsensusBlocks` in reverse, accumulating the per-beneficiary max `batch.seq` for
+/// `current_epoch`; stops at the first block of an earlier epoch. Returns an empty map when the
+/// store holds no blocks for `current_epoch`.
 fn recover_authorities_from_history<DB: Database>(
     store: &DB,
     current_epoch: Epoch,
-) -> HashMap<Address, AuthoritySeqState> {
+) -> BTreeMap<Address, AuthoritySeqState> {
     let by_addr = store
         .with_read_txn(|txn| {
-            let mut by_addr: HashMap<Address, u64> = HashMap::new();
-            for (_block_num, consensus_block) in txn.reverse_iter::<ConsensusBlocks>() {
-                if consensus_block.sub_dag.leader_epoch() < current_epoch {
+            // The walk covers a whole epoch of ConsensusBlocks on a cold boot; under the default
+            // long-read cap a slow disk aborts the txn mid-walk. Opt out like the catch-up
+            // accumulator does: this runs once at boot before any writer contends.
+            txn.disable_long_read_safety();
+            let mut by_addr: BTreeMap<Address, u64> = BTreeMap::new();
+            for (_key, value) in txn.reverse_raw_iter::<ConsensusBlocks>() {
+                let (leader_epoch, batch_digests) = leader_epoch_and_batch_digests(&value)?;
+                if leader_epoch < current_epoch {
                     break;
                 }
-                for cert in &consensus_block.sub_dag.certificates {
-                    for (batch_digest, _wid) in cert.header.payload() {
-                        let Ok(Some(batch)) = txn.get::<Batches>(batch_digest) else { continue };
-                        by_addr
-                            .entry(batch.beneficiary)
-                            .and_modify(|s| *s = (*s).max(batch.seq))
-                            .or_insert(batch.seq);
-                    }
+                for batch_digest in batch_digests {
+                    let Ok(Some(batch)) = txn.get::<Batches>(&batch_digest) else { continue };
+                    // batch.beneficiary must equal the live path's key,
+                    // authority_execution_address(cert.origin()). If they diverge, a restart
+                    // seeds watermarks under addresses the live path never looks up, silently
+                    // disabling gap detection.
+                    by_addr
+                        .entry(batch.beneficiary)
+                        .and_modify(|s| *s = (*s).max(batch.seq))
+                        .or_insert(batch.seq);
                 }
             }
             Ok(by_addr)
         })
-        .unwrap_or_default();
+        // Fail closed: with the cap lifted a failure here is a real DB error, and booting with
+        // empty watermarks silently disables gap detection for the epoch.
+        .expect("BatchOrdering: consensus-history read failed on boot");
 
     debug!(
         target: "engine",
@@ -478,6 +497,11 @@ mod tests {
         let mut persisted = BatchOrderingState { epoch: 87, ..Default::default() };
         let parked: BTreeMap<u64, PreparedBatch> =
             (1436..=1440u64).map(|seq| (seq, make_parked(auth, seq))).collect();
+        // Production precondition: every committed batch's body row exists in the store the
+        // ordering blob lives in; the by-digest persistence reloads bodies from those rows.
+        for prepared in parked.values() {
+            store.insert::<Batches>(&prepared.batch_digest, &prepared.batch).expect("seed body");
+        }
         persisted
             .authorities
             .insert(auth, AuthoritySeqState { last_executed_seq: Some(1434), parked });
@@ -569,4 +593,35 @@ mod tests {
             "pre-fork the stranded entries stay parked for the boundary force-drain"
         );
     }
+
+    #[test]
+    fn drain_epoch_never_rewinds_to_an_already_passed_epoch() {
+        // A crash after finalize but before the next output can replay an older epoch's output.
+        // drain_epoch for that stale epoch must be a no-op: it must not drain the current parked
+        // set nor move the epoch backward (the epoch is a monotonic position).
+        let store = MemDatabase::default();
+        let ord = BatchOrdering::new_with_empty_state(store);
+        let auth = Address::from([1u8; 20]);
+
+        // advance to epoch 88 with one parked batch belonging to it
+        assert_eq!(ord.drain_epoch(88).len(), 0);
+        assert!(matches!(
+            ord.try_accept(auth, 1, make_parked(auth, 1), false),
+            AcceptResult::InOrder(_)
+        ));
+        assert!(matches!(
+            ord.try_accept(auth, 3, make_parked(auth, 3), false),
+            AcceptResult::Parked
+        ));
+
+        // replay of an already-passed epoch: no drain, no rewind
+        let drained = ord.drain_epoch(87);
+        assert!(drained.is_empty(), "a stale-epoch drain must not drain the current parked set");
+        assert_eq!(ord.parked_count(auth), 1, "the current epoch's parked batch must survive");
+        assert_eq!(
+            ord.inner.batch_ordering_state.lock().epoch,
+            88,
+            "the epoch must not move backward"
+        );
+    }
 }
diff --git a/crates/middleware/processor/src/execution/orchestrator.rs b/crates/middleware/processor/src/execution/orchestrator.rs
index 917f2795..ddc0ef65 100644
--- a/crates/middleware/processor/src/execution/orchestrator.rs
+++ b/crates/middleware/processor/src/execution/orchestrator.rs
@@ -145,7 +145,9 @@ impl<DB: Database> Processor<DB> {
             );
         }
 
-        debug_assert_eq!(
+        // Release assert: a mismatch would otherwise surface as a messageless index abort in the
+        // raw digest indexing below.
+        assert_eq!(
             batches.len(),
             output.batch_digests.len(),
             "uneven number of sealed blocks from batches and batch digests"
@@ -334,7 +336,9 @@ impl<DB: Database> Processor<DB> {
                 }
             }
             // V1: no dedup check (batches were registered at park time)
-
+            if let Some(tracker) = &self.batch_tracker {
+                tracker.batch_force_drained(parked.batch_digest, parked.batch.seq);
+            }
             let (header, _) =
                 self.execute_prepared_batch(&parked, canonical_header, None, &mut executed_blocks)?;
             canonical_header = header;
diff --git a/crates/middleware/processor/tests/it/batch_ordering_restart.rs b/crates/middleware/processor/tests/it/batch_ordering_restart.rs
index e3d2f108..cd7ecbd1 100644
--- a/crates/middleware/processor/tests/it/batch_ordering_restart.rs
+++ b/crates/middleware/processor/tests/it/batch_ordering_restart.rs
@@ -229,6 +229,15 @@ impl BatchOrderingHarness {
             let mut batch = Batch::new_for_test(vec![], ExecHeader::default(), 0, 0, seq);
             batch.beneficiary = beneficiary;
             batch.base_fee_per_gas = MIN_PROTOCOL_BASE_FEE;
+            // Seed the batch body row. In production the engine's store is the consensus DB, where
+            // every committed batch is stored before execution and the by-digest parked
+            // persistence reloads bodies on restart; the harness's separate ordering DB models
+            // that here.
+            self.ordering_store
+                .as_ref()
+                .expect("ordering_store open")
+                .insert::<rayls_infrastructure_storage::tables::Batches>(&batch.digest(), &batch)
+                .expect("seed batch body row");
             digests.push_back(batch.digest());
             certified.push(CertifiedBatch { address: beneficiary, batches: vec![batch] });
         }