From e6ebb58283d39477d0d1bf224b13b215dff77dd8 Mon Sep 17 00:00:00 2001 From: Mario Ignatov Date: Wed, 26 Aug 2026 16:02:01 +0300 Subject: [PATCH 1/7] fix: store the suspended-cert count so the proposer brake engages and releases - the watch has no receiver (the proposer borrows the sender), so `send` failed and discarded every value: the backpressure brake read 0 forever - `send_replace` stores unconditionally; publish from the pending manager on both suspension and drain so the count falls back to 0 once parents resolve - drop the duplicate publish in the cert manager; `insert_pending` already owns it - red test: suspend, drain, assert the watch reads 0 - addresses the PR #127 (perf/hot-path-optimizations) review blocker: proposer livelock on backpressure drain; the fix also covers the dead-watch case the review missed --- crates/consensus/primary/src/consensus_bus.rs | 2 +- .../primary/src/state_sync/cert_manager.rs | 8 ---- .../src/state_sync/pending_cert_manager.rs | 16 +++++-- .../primary/src/tests/cert_manager_tests.rs | 44 ++++++++++++++++++- 4 files changed, 56 insertions(+), 14 deletions(-) diff --git a/crates/consensus/primary/src/consensus_bus.rs b/crates/consensus/primary/src/consensus_bus.rs index d82bc649..b088ef78 100644 --- a/crates/consensus/primary/src/consensus_bus.rs +++ b/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, /// Subscriber sends drain acknowledgment when all in-flight work is complete. diff --git a/crates/consensus/primary/src/state_sync/cert_manager.rs b/crates/consensus/primary/src/state_sync/cert_manager.rs index 7eb4041b..430dfaa1 100644 --- a/crates/consensus/primary/src/state_sync/cert_manager.rs +++ b/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(); diff --git a/crates/consensus/primary/src/state_sync/pending_cert_manager.rs b/crates/consensus/primary/src/state_sync/pending_cert_manager.rs index 5abf8c50..da848fe9 100644 --- a/crates/consensus/primary/src/state_sync/pending_cert_manager.rs +++ b/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) } diff --git a/crates/consensus/primary/src/tests/cert_manager_tests.rs b/crates/consensus/primary/src/tests/cert_manager_tests.rs index 14056524..af1a6983 100644 --- a/crates/consensus/primary/src/tests/cert_manager_tests.rs +++ b/crates/consensus/primary/src/tests/cert_manager_tests.rs @@ -14,6 +14,8 @@ struct TestTypes { manager: CertificateManager, /// The committee fixture. fixture: CommitteeFixture, + /// The bus the manager publishes on. + bus: ConsensusBus, } fn create_test_types() -> TestTypes { @@ -26,9 +28,9 @@ fn create_test_types() -> TestTypes { 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::>(); + 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(()) +} From 020378b05660b2a69c50b1c77197e5cc43ee61ce Mon Sep 17 00:00:00 2001 From: Mario Ignatov Date: Wed, 26 Aug 2026 16:02:01 +0300 Subject: [PATCH 2/7] fix: reject an in-flight mark backup from another schema version - `MARK_BACKUP_VERSION` was written but never read on restore; bcs is positional, so a reshaped backup could decode into wrong marks instead of being refused - `#[must_use]` on `SealMarks`/`ForwardMarks`: dropping the handle silently discards the capability the arm minted; one test was doing exactly that - addresses the PR #120 (feature/txpool-in-flight-tracker-core) review: unchecked `MarkBackup::version` and the missing `#[must_use]` on the arm handles --- crates/execution/evm/src/in_flight/marks.rs | 2 ++ crates/execution/evm/src/in_flight/mod.rs | 10 ++++++++++ crates/execution/evm/src/in_flight/tests.rs | 14 ++++++++++++++ crates/execution/evm/src/txn_pool/backup.rs | 2 +- 4 files changed, 27 insertions(+), 1 deletion(-) diff --git a/crates/execution/evm/src/in_flight/marks.rs b/crates/execution/evm/src/in_flight/marks.rs index 1e32d47d..4b01ea49 100644 --- a/crates/execution/evm/src/in_flight/marks.rs +++ b/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, diff --git a/crates/execution/evm/src/in_flight/mod.rs b/crates/execution/evm/src/in_flight/mod.rs index dc094ade..cea20b9e 100644 --- a/crates/execution/evm/src/in_flight/mod.rs +++ b/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", diff --git a/crates/execution/evm/src/in_flight/tests.rs b/crates/execution/evm/src/in_flight/tests.rs index 66b7357d..addef7a7 100644 --- a/crates/execution/evm/src/in_flight/tests.rs +++ b/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"); +} diff --git a/crates/execution/evm/src/txn_pool/backup.rs b/crates/execution/evm/src/txn_pool/backup.rs index 6ad51b2a..ef74e881 100644 --- a/crates/execution/evm/src/txn_pool/backup.rs +++ b/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"); } } From df15ecaad0d02978d9f2b6577cfd4f5b1e17d35c Mon Sep 17 00:00:00 2001 From: Mario Ignatov Date: Wed, 26 Aug 2026 16:02:01 +0300 Subject: [PATCH 3/7] fix: drop a parked batch whose body no longer decodes instead of aborting - boot recovery already drops a parked ref with no `Batches` row; a corrupt row used the infallible decode and aborted the node on the same defensive path - addresses the PR #123 (feature/txpool-ordering-and-proposer-hardening) review: infallible `decode` in `reconstruct_parked` --- .../src/stores/batch_ordering_store.rs | 60 +++++++++++++++++-- 1 file changed, 55 insertions(+), 5 deletions(-) diff --git a/crates/infrastructure/storage/src/stores/batch_ordering_store.rs b/crates/infrastructure/storage/src/stores/batch_ordering_store.rs index ce6e67dd..57bc5a81 100644 --- a/crates/infrastructure/storage/src/stores/batch_ordering_store.rs +++ b/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 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::(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"); + } +} From 143f84f01aa048304ded8b01435ad220a9de7e99 Mon Sep 17 00:00:00 2001 From: Mario Ignatov Date: Wed, 26 Aug 2026 16:02:01 +0300 Subject: [PATCH 4/7] fix: interpolate the worker id in the missing-components error - `ok_or_eyre` takes a `Display` value, so the braces were emitted literally - addresses the PR #125 (feature/txpool-in-flight-tracker-forwarding-affinity) review: `ok_or_eyre` literal at both `node_inner.rs` sites --- crates/middleware/orchestrator/src/engine/node_inner.rs | 5 ++--- 1 file changed, 2 insertions(+), 3 deletions(-) diff --git a/crates/middleware/orchestrator/src/engine/node_inner.rs b/crates/middleware/orchestrator/src/engine/node_inner.rs index ca3b0ca8..a14f4014 100644 --- a/crates/middleware/orchestrator/src/engine/node_inner.rs +++ b/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(); From 8f505365d720ace9dd401b922ffbe3659b725cae Mon Sep 17 00:00:00 2001 From: Mario Ignatov Date: Wed, 26 Aug 2026 16:02:01 +0300 Subject: [PATCH 5/7] fix: measure batch quorum latency from the build spawn - `elapsed_ms` was stamped after the quorum result had already arrived, so it logged the resolution overhead, never the wait; `AwaitingQuorum` now carries the spawn instant - fix "becuase" in the max-batch-size error text - addresses the PR #124 (feature/txpool-in-flight-tracker-builder-pipeline) review: `elapsed_ms` always ~0, plus the typo nit --- .../worker/src/batch-builder/src/error.rs | 2 +- .../worker/src/batch-builder/src/lib.rs | 2 +- .../worker/src/batch-builder/src/pipeline.rs | 17 ++++++++++++++--- 3 files changed, 16 insertions(+), 5 deletions(-) diff --git a/crates/consensus/worker/src/batch-builder/src/error.rs b/crates/consensus/worker/src/batch-builder/src/error.rs index 337431f2..f965539c 100644 --- a/crates/consensus/worker/src/batch-builder/src/error.rs +++ b/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. diff --git a/crates/consensus/worker/src/batch-builder/src/lib.rs b/crates/consensus/worker/src/batch-builder/src/lib.rs index 9f3d95b7..47e2823d 100644 --- a/crates/consensus/worker/src/batch-builder/src/lib.rs +++ b/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, diff --git a/crates/consensus/worker/src/batch-builder/src/pipeline.rs b/crates/consensus/worker/src/batch-builder/src/pipeline.rs index 29846885..ca546a9e 100644 --- a/crates/consensus/worker/src/batch-builder/src/pipeline.rs +++ b/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,6 +35,8 @@ 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 { @@ -42,6 +44,7 @@ impl fmt::Debug for AwaitingQuorum { f.debug_struct("AwaitingQuorum") .field("rx", &"") .field("event_arrived_while_waiting", &self.event_arrived_while_waiting) + .field("started", &self.started) .finish() } } @@ -205,7 +208,11 @@ impl BatchPipeline { rx: oneshot::Receiver>, ) -> BatchPipeline { 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 { rx: oneshot::Receiver>, ) -> BatchPipeline { BatchPipeline { - state: AwaitingQuorum { rx, event_arrived_while_waiting: false }, + state: AwaitingQuorum { + rx, + event_arrived_while_waiting: false, + started: Instant::now(), + }, data: self.data, } } From 9e6bddd419df78a6f6c2608419d65181487c55d6 Mon Sep 17 00:00:00 2001 From: Mario Ignatov Date: Wed, 26 Aug 2026 16:02:02 +0300 Subject: [PATCH 6/7] test: pin the txn gossip payload to the pre-Bytes wire encoding - the `Vec` identity with `Vec>` was asserted only in a comment; the pin encodes both and cross-decodes the old bytes into the new variant - addresses the PR #125 (feature/txpool-in-flight-tracker-forwarding-affinity) review: the `Vec` wire identity was comment-only --- crates/consensus/worker/src/network/message.rs | 15 +++++++++++++++ 1 file changed, 15 insertions(+) diff --git a/crates/consensus/worker/src/network/message.rs b/crates/consensus/worker/src/network/message.rs index 314ab825..9f637158 100644 --- a/crates/consensus/worker/src/network/message.rs +++ b/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` to the `Vec>` 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![vec![1, 2, 3], vec![], vec![4; 300]]; + let bytes: Vec = 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); From 02563e4d398daad2124c969695c0e24320571683 Mon Sep 17 00:00:00 2001 From: Mario Ignatov Date: Wed, 26 Aug 2026 16:02:02 +0300 Subject: [PATCH 7/7] fix: log a panic in the pre-drain txpool backup task - the post-drain twin logs; the pre-drain snapshot swallowed the `JoinError`, leaving no trace when the node entered the drain with no backup - note at `max_tx_lifetime` that reth applies it to the queued sub-pool only - addresses the PR #121 (feature/txpool-reth-pool-maintenance) review: swallowed `JoinError`; the `max_tx_lifetime` note answers the same review's eviction concern, which misread the knob --- crates/execution/evm/src/txn_pool.rs | 2 ++ crates/middleware/orchestrator/src/epoch_manager/core.rs | 5 ++++- 2 files changed, 6 insertions(+), 1 deletion(-) diff --git a/crates/execution/evm/src/txn_pool.rs b/crates/execution/evm/src/txn_pool.rs index 0cd07062..e72cf0d3 100644 --- a/crates/execution/evm/src/txn_pool.rs +++ b/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() diff --git a/crates/middleware/orchestrator/src/epoch_manager/core.rs b/crates/middleware/orchestrator/src/epoch_manager/core.rs index 90a2f71f..622714fd 100644 --- a/crates/middleware/orchestrator/src/epoch_manager/core.rs +++ b/crates/middleware/orchestrator/src/epoch_manager/core.rs @@ -423,13 +423,16 @@ where // first reconcile. Best-effort: a panic here must not skip the drain. { let pools = engine.get_all_worker_transaction_pools().await; - let _ = tokio::task::spawn_blocking(move || { + let saved = tokio::task::spawn_blocking(move || { for pool in &pools { pool.save_backup(); pool.save_mark_backup(); } }) .await; + if let Err(e) = saved { + error!(target: "engine", %e, "pre-drain txpool backup task panicked"); + } } match engine_done_rx.await {