fix: stack review follow-ups for #120-#127

crates/consensus/primary/src/consensus_bus.rs

@@ -374,7 +374,7 @@ struct ConsensusBusEpochInner {
     /// Count of certificates currently suspended awaiting parents, owned by the certificate
     /// manager. The proposer's backpressure gate reads this, never the mirrored metrics gauge:
     /// control state lives in a component, a metric handle is write-only. Published on each
-    /// suspension, so a drained queue is reflected only at the next suspension.
+    /// suspension and each drain, so the gate releases as soon as the queue empties.
     suspended_cert_count: watch::Sender<usize>,
 
     /// Subscriber sends drain acknowledgment when all in-flight work is complete.

crates/consensus/primary/src/state_sync/cert_manager.rs

@@ -161,14 +161,6 @@ where
                         "certificate suspended - missing parents"
                     );
                     self.pending.insert_pending(cert, missing_parents)?;
-                    let _ =
-                        self.consensus_bus.suspended_cert_count().send(self.pending.num_pending());
-                    // metrics mirror only - decisions read the watch above
-                    self.consensus_bus
-                        .primary_metrics()
-                        .node_metrics
-                        .certificates_currently_suspended
-                        .set(self.pending.num_pending() as i64);
 
                     // Cascade detection: warn when pending queue is growing large
                     let pending_count = self.pending.num_pending();

crates/consensus/primary/src/state_sync/pending_cert_manager.rs

@@ -102,15 +102,23 @@ impl PendingCertificateManager {
             self.missing_for_pending.entry((parent_round, parent)).or_default().insert(digest);
         }
 
-        let _ = self.consensus_bus.suspended_cert_count().send(self.pending.len());
+        self.publish_count();
+
+        Ok(())
+    }
+
+    /// Publishes the pending count to the proposer's backpressure watch and the metrics mirror.
+    ///
+    /// `send_replace`, not `send`: nobody holds a receiver (the proposer borrows the sender), and
+    /// a `send` with no receivers discards the value instead of storing it.
+    fn publish_count(&self) {
+        self.consensus_bus.suspended_cert_count().send_replace(self.pending.len());
         // metrics mirror only - decisions read the watch above
         self.consensus_bus
             .primary_metrics()
             .node_metrics
             .certificates_currently_suspended
             .set(self.pending.len() as i64);
-
-        Ok(())
     }
 
     /// When a certificate is accepted, returns all of its children that are now ready to be
@@ -162,6 +170,8 @@ impl PendingCertificateManager {
             }
         }
 
+        // the drain must publish too, or the proposer brake stays stuck at the last suspension
+        self.publish_count();
         Ok(ready_certificates)
     }
 

crates/consensus/primary/src/tests/cert_manager_tests.rs

@@ -14,6 +14,8 @@ struct TestTypes<DB = MemDatabase> {
     manager: CertificateManager<DB>,
     /// The committee fixture.
     fixture: CommitteeFixture<DB>,
+    /// The bus the manager publishes on.
+    bus: ConsensusBus,
 }
 
 fn create_test_types() -> TestTypes<MemDatabase> {
@@ -26,9 +28,9 @@ fn create_test_types() -> TestTypes<MemDatabase> {
     let gc_round = AtomicRound::new(0);
     let highest_processed_round = AtomicRound::new(0);
 
-    let manager = CertificateManager::new(config, cb, gc_round, highest_processed_round);
+    let manager = CertificateManager::new(config, cb.clone(), gc_round, highest_processed_round);
 
-    TestTypes { manager, fixture }
+    TestTypes { manager, fixture, bus: cb }
 }
 
 #[tokio::test]
@@ -85,3 +87,41 @@ async fn test_accept_pending_certs() -> eyre::Result<()> {
     assert_eq!(expected_pending_len, manager.pending.num_pending());
     Ok(())
 }
+
+/// The proposer's backpressure brake reads this watch, so a count left stale-high after the
+/// drain would throttle proposing forever.
+#[tokio::test]
+async fn suspended_cert_count_follows_the_drain() -> eyre::Result<()> {
+    let TestTypes { mut manager, fixture, bus } = create_test_types();
+    let committee = fixture.committee();
+    let num_authorities = fixture.num_authorities();
+
+    let genesis =
+        Certificate::genesis(&committee).iter().map(|x| x.digest()).collect::<BTreeSet<_>>();
+    let keys: Vec<_> = fixture.authorities().map(|a| (a.id(), a.keypair().copy())).collect();
+    let (certificates, _) =
+        make_optimal_signed_certificates(1..=3, &genesis, &committee, keys.as_slice());
+    let mut first_round: Vec<_> = certificates
+        .into_iter()
+        .map(|mut c| {
+            c.set_signature_verification_state(SignatureVerificationState::VerifiedDirectly(
+                c.aggregated_signature().expect("signature valid"),
+            ));
+            c
+        })
+        .collect();
+    let later_rounds = first_round.split_off(num_authorities);
+    let suspended = later_rounds.len();
+
+    let shutdown = Notifier::new();
+    let shutdown_rx = shutdown.subscribe();
+    let res = manager.process_verified_certificates(later_rounds, &shutdown_rx).await;
+    assert_matches!(res, Err(CertManagerError::Pending(_)));
+    assert_eq!(*bus.suspended_cert_count().borrow(), suspended, "suspension publishes the count");
+
+    // the first round unlocks everything above it
+    manager.process_verified_certificates(first_round, &shutdown_rx).await?;
+    assert_eq!(manager.pending.num_pending(), 0, "every pending cert drained");
+    assert_eq!(*bus.suspended_cert_count().borrow(), 0, "the drain publishes the count");
+    Ok(())
+}

crates/consensus/worker/src/batch-builder/src/error.rs

@@ -35,7 +35,7 @@ pub enum BatchBuilderError {
     /// Error building batch because this transaction would case the batch to exceed max size (in
     /// bytes).
     #[error(
-        "The transaction was not included becuase it would exceed the max batch size. Tx size: {0} bytes - max size: {1} bytes."
+        "The transaction was not included because it would exceed the max batch size. Tx size: {0} bytes - max size: {1} bytes."
     )]
     MaxBatchSize(usize, usize),
     /// An operation that requires canonical state did not have it.

crates/consensus/worker/src/batch-builder/src/lib.rs

@@ -319,7 +319,7 @@ impl BatchBuilder {
         };
 
         let current_seq = awaiting.current_seq();
-        let start_time = std::time::Instant::now();
+        let start_time = awaiting.state.started;
 
         let outcome = match res.map_err(BatchBuilderError::from).and_then(|r| r) {
             Ok(out) => out,

crates/consensus/worker/src/batch-builder/src/pipeline.rs

@@ -9,7 +9,7 @@ use crate::{
     batch::SelectedForSeal, error::BatchBuilderResult, BOUNDARY_QUIESCE_WINDOW_SECS, MAX_SEAL_AHEAD,
 };
 use rayls_execution_evm::in_flight::SealMarks;
-use std::fmt;
+use std::{fmt, time::Instant};
 use tokio::sync::oneshot;
 
 /// Phase with no candidate transactions pending a seal.
@@ -35,13 +35,16 @@ pub struct AwaitingQuorum {
     /// Whether a candidate event arrived mid-build, so the seal must re-accumulate rather than
     /// return to clean and lose the wake.
     pub(crate) event_arrived_while_waiting: bool,
+    /// When the build task was spawned, so the resolution log reports the real quorum latency.
+    pub(crate) started: Instant,
 }
 
 impl fmt::Debug for AwaitingQuorum {
     fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
         f.debug_struct("AwaitingQuorum")
             .field("rx", &"<oneshot::Receiver>")
             .field("event_arrived_while_waiting", &self.event_arrived_while_waiting)
+            .field("started", &self.started)
             .finish()
     }
 }
@@ -205,7 +208,11 @@ impl BatchPipeline<Accumulating> {
         rx: oneshot::Receiver<BatchBuilderResult<TaskOutcome>>,
     ) -> BatchPipeline<AwaitingQuorum> {
         BatchPipeline {
-            state: AwaitingQuorum { rx, event_arrived_while_waiting: false },
+            state: AwaitingQuorum {
+                rx,
+                event_arrived_while_waiting: false,
+                started: Instant::now(),
+            },
             data: self.data,
         }
     }
@@ -236,7 +243,11 @@ impl BatchPipeline<BacklogDraining> {
         rx: oneshot::Receiver<BatchBuilderResult<TaskOutcome>>,
     ) -> BatchPipeline<AwaitingQuorum> {
         BatchPipeline {
-            state: AwaitingQuorum { rx, event_arrived_while_waiting: false },
+            state: AwaitingQuorum {
+                rx,
+                event_arrived_while_waiting: false,
+                started: Instant::now(),
+            },
             data: self.data,
         }
     }

crates/consensus/worker/src/network/message.rs

@@ -147,6 +147,21 @@ mod tests {
         assert!(matches!(decoded, WorkerResponse::Error(WorkerRPCError(s)) if s == "boom"));
     }
 
+    /// Pins `Vec<Bytes>` to the `Vec<Vec<u8>>` bytes an un-upgraded peer emits and expects.
+    #[test]
+    fn worker_gossip_txn_payload_is_bcs_identical_to_vec_vec_u8() {
+        let vecs: Vec<Vec<u8>> = vec![vec![1, 2, 3], vec![], vec![4; 300]];
+        let bytes: Vec<Bytes> = vecs.iter().cloned().map(Bytes::from).collect();
+        assert_eq!(encode(&bytes), encode(&vecs));
+
+        let gossip = WorkerGossip::Txn(bytes);
+        let mut old_wire = vec![1u8]; // Txn is bcs variant index 1
+        old_wire.extend(encode(&vecs));
+        assert_eq!(encode(&gossip), old_wire);
+        let decoded: WorkerGossip = try_decode(&old_wire).unwrap();
+        assert_eq!(decoded, gossip);
+    }
+
     #[test]
     fn worker_request_submit_txns_is_appended_last() {
         assert_eq!(encode(&WorkerRequest::SubmitTxns { transactions: vec![] })[0], 3);

crates/execution/evm/src/in_flight/marks.rs

@@ -71,6 +71,7 @@ pub(crate) enum Armed {
 
 /// A capability handle scoping mark writes to the sealing role, returned by
 /// `InFlightTracker::arm_sealing` so a caller cannot mark hashes without first arming.
+#[must_use = "dropping the handle discards the sealing capability the arm just minted"]
 #[derive(Debug)]
 pub struct SealMarks {
     tracker: InFlightTracker,
@@ -100,6 +101,7 @@ pub struct ForwardProbe {
 
 /// A capability handle scoping mark writes to the forwarding role, returned by
 /// `InFlightTracker::arm_forwarding` so a caller cannot mark hashes without first arming.
+#[must_use = "dropping the handle discards the forwarding capability the arm just minted"]
 #[derive(Debug, Clone)]
 pub struct ForwardMarks {
     tracker: InFlightTracker,

crates/execution/evm/src/in_flight/mod.rs

@@ -331,6 +331,16 @@ impl InFlightTracker {
 
     fn consume_stash(guard: &mut Inner, role: MarkRole) -> usize {
         let Some(backup) = guard.pending_restore.take() else { return 0 };
+        if backup.version != MARK_BACKUP_VERSION {
+            info!(
+                target: "rayls::txpool",
+                discarded = backup.marks.len(),
+                saved_version = backup.version,
+                current_version = MARK_BACKUP_VERSION,
+                "discarded restored in-flight marks from another schema version"
+            );
+            return 0;
+        }
         if backup.role != role {
             info!(
                 target: "rayls::txpool",

crates/execution/evm/src/in_flight/tests.rs

@@ -609,3 +609,17 @@ fn clear_with_nothing_armed_keeps_the_stash() {
         "an actor-less clear must not eat the next actor's stash"
     );
 }
+
+/// bcs is positional, so a backup from another schema version can decode into wrong marks; it is
+/// rejected whole rather than restored best-effort.
+#[test]
+fn arm_rejects_a_backup_from_another_schema_version() {
+    let tracker = InFlightTracker::with_fresh_metrics();
+    tracker.stash_restore(MarkBackup {
+        version: MARK_BACKUP_VERSION + 1,
+        role: MarkRole::Sealing,
+        marks: vec![SavedMark { hash: hash(1), kind: SavedMarkKind::Sent { attempts: 0 } }],
+    });
+    let _seal = tracker.arm_sealing(DuePolicy::ttl(Duration::from_secs(60)));
+    assert!(!tracker.is_in_flight(&hash(1)), "a foreign-version backup restores nothing");
+}

crates/execution/evm/src/txn_pool.rs

@@ -178,6 +178,8 @@ impl WorkerTxPool {
             blockchain_provider.canonical_state_stream(),
             task_spawner.clone(),
             MaintainPoolConfig {
+                // bounds the queued (non-executable) sub-pool only; a pending tx is never
+                // lifetime-evicted, so this caps nonce-gapped stranding, not seal latency
                 max_tx_lifetime: Duration::from_mins(5),
                 no_local_exemptions: true,
                 ..Default::default()

crates/execution/evm/src/txn_pool/backup.rs

@@ -433,7 +433,7 @@ mod tests {
             !loading_pool.mark_backup_path().exists(),
             "load deletes the backup for at-most-once replay"
         );
-        loader.arm_forwarding(policy);
+        let _fwd = loader.arm_forwarding(policy);
         assert!(loader.is_in_flight(&hash), "the reloaded mark is live once forwarding re-arms");
     }
 }

crates/infrastructure/storage/src/stores/batch_ordering_store.rs

@@ -4,7 +4,7 @@ use crate::{
 };
 use rayls_infrastructure_types::{
     batch_ordering::{AuthoritySeqState, BatchOrderingState, StoredBatchOrderingState},
-    decode, try_decode, B256Map, Batch, Database, DbTx, B256,
+    try_decode, B256Map, Batch, Database, DbTx, B256,
 };
 use std::{collections::BTreeMap, sync::Arc};
 use tracing::warn;
@@ -77,7 +77,7 @@ impl<DB: Database> BatchOrderingStore for DB {
 }
 
 /// Rebuilds each authority's parked map by pairing every reference with its reloaded body,
-/// dropping any entry whose body is absent from `Batches`.
+/// dropping any entry whose body is absent from `Batches` or no longer decodes.
 ///
 /// 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.
@@ -89,11 +89,18 @@ fn reconstruct_parked(
     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);
+            match bodies.get(&reference.batch_digest).map(|bytes| try_decode::<Batch>(bytes)) {
+                Some(Ok(batch)) => {
                     parked.insert(seq, reference.into_prepared(Arc::new(batch)));
                 }
+                Some(Err(e)) => warn!(
+                    target: "engine",
+                    ?addr,
+                    seq,
+                    batch_digest = ?reference.batch_digest,
+                    %e,
+                    "dropping parked batch on restart: corrupt Batches row"
+                ),
                 None => warn!(
                     target: "engine",
                     ?addr,
@@ -108,3 +115,46 @@ fn reconstruct_parked(
     }
     BatchOrderingState { epoch: stored.epoch, authorities }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use rayls_infrastructure_types::{
+        batch_ordering::{ParkedRef, StoredAuthoritySeqState},
+        Address,
+    };
+
+    /// A parked body that no longer decodes is dropped like a missing one: boot-time recovery
+    /// must not abort the node on a corrupt row when the ordering can refill the seq gap.
+    #[test]
+    fn reconstruct_drops_a_parked_batch_whose_body_is_corrupt() {
+        let digest = B256::repeat_byte(1);
+        let reference = ParkedRef {
+            batch_digest: digest,
+            beneficiary: Address::ZERO,
+            output_digest: B256::ZERO,
+            output_nonce: 0,
+            timestamp: 0,
+            epoch: 0,
+            worker_id: 0,
+            batch_index: 0,
+            drained: false,
+            gas_limit: 0,
+        };
+        let stored = StoredBatchOrderingState {
+            epoch: 0,
+            authorities: BTreeMap::from([(
+                Address::ZERO,
+                StoredAuthoritySeqState {
+                    last_executed_seq: None,
+                    parked: BTreeMap::from([(1, reference)]),
+                },
+            )]),
+        };
+        let bodies = B256Map::from_iter([(digest, vec![0xff, 0xff, 0xff])]);
+
+        let state = reconstruct_parked(stored, &bodies);
+
+        assert!(state.authorities[&Address::ZERO].parked.is_empty(), "the corrupt body is dropped");
+    }
+}

crates/middleware/orchestrator/src/engine/node_inner.rs

@@ -3,7 +3,6 @@
 
 use super::txn_forwarder::TxnForwarder;
 use crate::types::ExecutionError;
-use eyre::OptionExt;
 use jsonrpsee::http_client::HttpClient;
 use rayls_batch_builder::{BatchBuilder, BatchBuilderConfig, OwnWatermarkReceiver};
 use rayls_batch_validator::BatchValidator;
@@ -166,7 +165,7 @@ impl ExecutionNodeInner {
         let transaction_pool = self
             .workers
             .get(worker_id as usize)
-            .ok_or_eyre("worker components missing for {worker_id}")?
+            .ok_or_else(|| eyre::eyre!("worker components missing for worker {worker_id}"))?
             .pool();
 
         let own_executed_sequence = self
@@ -221,7 +220,7 @@ impl ExecutionNodeInner {
         let transaction_pool = self
             .workers
             .get(worker_id as usize)
-            .ok_or_eyre("worker components missing for {worker_id}")?
+            .ok_or_else(|| eyre::eyre!("worker components missing for worker {worker_id}"))?
             .pool();
 
         let reth_env = self.reth_env.clone();