diff --git a/crates/batch-builder/README.md b/crates/batch-builder/README.md index 0101a939e..eeea98ab3 100644 --- a/crates/batch-builder/README.md +++ b/crates/batch-builder/README.md @@ -97,6 +97,13 @@ This is a non-fatal error. Although this is inefficient, it is considered an acceptable limitation of the protocol at this time. Future iterations are planned to address this inefficiency. +One local mitigation is in place (issue #1329): when this node validates a peer's batch, the batch's transaction hashes are deferred by this node's builder for `PEER_BATCH_DEFER_TTL` (10 seconds, the default batch vote timeout), bounded by `PEER_BATCH_SEEN_MAX_TXS` remembered hashes. +The builder skips a deferred hash and, with it, that sender's later nonces: those nonces wait exactly as long as the deferred earlier nonce does, which they could not execute ahead of anyway. +A transaction a client sent to every validator is therefore not packed by every worker at once. +The builder only skips the transaction for this build, and it seals nothing at all when every pending transaction is deferred (`BuildOutcome::NothingToSeal`), because an empty batch is rejected by peers and penalized as fatal. +Each arming costs a transaction at most one TTL: the deferral expires on its own, the entry stays immune to re-arming until it is forgotten at twice the TTL, and execution still tolerates duplicates. +A flood of peer batches cannot evict a live entry; once the window is full further hashes are simply not remembered, so a flood can only switch the deferral off, never re-arm it. + ### Safety of Early Pool Updates #### Quorum failure does not corrupt pool state diff --git a/crates/batch-builder/src/batch.rs b/crates/batch-builder/src/batch.rs index 8488713ad..552d942fb 100644 --- a/crates/batch-builder/src/batch.rs +++ b/crates/batch-builder/src/batch.rs @@ -31,6 +31,11 @@ pub struct BatchBuilderOutput { /// minus the just-mined cost, so parked insufficient-funds transactions are not spuriously /// promoted (see `build_batch`). pub(crate) changed_accounts: Vec, + /// The number of transactions skipped because a validated peer batch already carries them. + /// + /// Reported as a metric so an operator can see how much duplicate work the deferral window + /// avoids (issue #1329). + pub(crate) peer_deferred: usize, } /// Construct an TN batch using the best transactions from the pool. @@ -71,18 +76,36 @@ pub fn build_batch( let mut unsupported_transactions = Vec::new(); let mut sender_nonces: HashMap = HashMap::new(); let mut sender_costs: HashMap = HashMap::new(); + let mut peer_deferred: usize = 0; // begin loop through sorted "best" transactions in pending pool // and execute them to build the block while let Some(pool_tx) = best_txs.next() { + // a validated peer batch may already carry this transaction: that peer is proposing it + // right now, so packing a copy here only spends batch space, bandwidth and a vote round + // before execution skips the copy for free (issue #1329) + let deferred_by_peer = pool.is_peer_deferred(pool_tx.hash()); + // ensure block has capacity (in gas) for this transaction - if total_possible_gas + pool_tx.gas_limit() > gas_limit { - // the tx could exceed max gas limit for the block - // marking as invalid within the context of the `BestTransactions` pulled in this - // current iteration all dependents for this transaction are now considered invalid - // before continuing loop - best_txs.exceeds_gas_limit(&pool_tx, gas_limit); - debug!(target: "worker::batch_builder", ?pool_tx, "marking tx invalid due to gas constraint"); + let exceeds_gas_limit = total_possible_gas + pool_tx.gas_limit() > gas_limit; + + // either guard skips the transaction: + // - the tx could exceed max gas limit for the block + // - the tx is already in flight inside a peer's batch + // + // marking as invalid within the context of the `BestTransactions` pulled in this + // current iteration all dependents for this transaction are now considered invalid + // before continuing loop. For the deferral that is deliberate: a later nonce from the + // same sender would land nonce-gapped and only be skipped at execution. + if deferred_by_peer || exceeds_gas_limit { + if deferred_by_peer { + best_txs.peer_deferred(&pool_tx); + peer_deferred = peer_deferred.saturating_add(1); + debug!(target: "worker::batch_builder", ?pool_tx, "deferring tx already packed by a validated peer batch"); + } else { + best_txs.exceeds_gas_limit(&pool_tx, gas_limit); + debug!(target: "worker::batch_builder", ?pool_tx, "marking tx invalid due to gas constraint"); + } continue; } @@ -195,7 +218,7 @@ pub fn build_batch( .collect(); // return output - BatchBuilderOutput { batch, mined_transactions, changed_accounts } + BatchBuilderOutput { batch, mined_transactions, changed_accounts, peer_deferred } } #[cfg(test)] @@ -207,6 +230,64 @@ mod tests { use tn_reth::{test_utils::TransactionFactory, RethChainSpec}; use tn_types::{test_genesis, BatchBuilderArgs, Bytes, B256, MIN_PROTOCOL_BASE_FEE, U256}; + /// A transaction a validated peer batch already carries must not be packed again here: the + /// duplicate costs batch space, bandwidth and a vote round, and execution skips it for free + /// (issue #1329). The sender's later nonces must be skipped in the same build too, because a + /// nonce-gapped copy would only be caught at execution. + #[test] + fn peer_batched_transactions_are_deferred_with_their_successors() { + let chain: Arc = Arc::new(test_genesis().into()); + let mut factory_a = TransactionFactory::new(); + let mut factory_b = TransactionFactory::new_random(); + + // sender A submits nonces 0 and 1, sender B submits nonce 0 + let a_nonce_0 = factory_a.create_eip1559_encoded( + chain.clone(), + None, + 100, + None, + U256::from(1), + Bytes::new(), + ); + let a_nonce_1 = factory_a.create_eip1559_encoded( + chain.clone(), + None, + 100, + None, + U256::from(1), + Bytes::new(), + ); + let b_nonce_0 = factory_b.create_eip1559_encoded( + chain.clone(), + None, + 100, + None, + U256::from(1), + Bytes::new(), + ); + + let hash_a0 = *tn_reth::recover_raw_transaction(&a_nonce_0).expect("tx a0").hash(); + let hash_a1 = *tn_reth::recover_raw_transaction(&a_nonce_1).expect("tx a1").hash(); + let hash_b0 = *tn_reth::recover_raw_transaction(&b_nonce_0).expect("tx b0").hash(); + + let pool = TestPool::new(&[a_nonce_0.clone(), a_nonce_1, b_nonce_0.clone()]); + + // a peer's validated batch carries A's first transaction + pool.record_peer_batch(&[hash_a0]); + + let args = BatchBuilderArgs { pool, beneficiary: Address::ZERO, epoch: 0 }; + let BatchBuilderOutput { batch, mined_transactions, peer_deferred, .. } = + build_batch(args, 0, MIN_PROTOCOL_BASE_FEE); + + // only B's transaction is packed + assert_eq!(batch.transactions, vec![b_nonce_0]); + assert_eq!(mined_transactions, vec![hash_b0]); + // neither A's deferred transaction nor its nonce-gapped successor is mined + assert!(!mined_transactions.contains(&hash_a0)); + assert!(!mined_transactions.contains(&hash_a1)); + assert_eq!(peer_deferred, 1); + } + /// The optimistic `changed_accounts` update must carry the sender's real balance, not an /// inflated `U256::MAX`. An inflated balance lets the pool promote a sender's parked /// insufficient-funds transactions into the next batch, which is a griefing/amplification diff --git a/crates/batch-builder/src/lib.rs b/crates/batch-builder/src/lib.rs index eb2789bb3..a7ef2b851 100644 --- a/crates/batch-builder/src/lib.rs +++ b/crates/batch-builder/src/lib.rs @@ -64,6 +64,15 @@ enum BuildOutcome { /// Any other non-fatal seal failure (quorum, timeout, reporting). The pool keeps its /// transactions and the loop retries on the next delay tick, as before. Failed, + /// The build produced no transactions, so nothing was sent to the worker (issue #1329). + /// + /// The pending pool is not empty here: every transaction in it was deferred because a + /// validated peer batch already carries it. Sealing that build would broadcast a batch with + /// no transactions, which peers reject as + /// [`BatchValidationError::EmptyBatch`](tn_types::error::BatchValidationError::EmptyBatch) + /// and score as a fatal penalty. The run loop treats this as a quiet tick: no pool update, + /// no error log, and no refusal backoff, because forward admission was never exercised. + NothingToSeal, } /// Ceiling for the refusal backoff (issue #1145). @@ -210,74 +219,97 @@ impl BatchBuilder { let (ack, rx) = oneshot::channel(); // this is safe to call without a semaphore bc it's held as a single `Option` - let BatchBuilderOutput { batch, mined_transactions, changed_accounts } = build_batch(build_args, worker_id, base_fee); - let batch = batch.seal_slow(); - span.record("batch", batch.digest().to_string()); - - // forward to worker and wait for ack that quorum was reached - if let Err(e) = to_worker.send((batch, ack)).await { - error!(target: "worker::batch_builder", ?e, "failed to send next batch to worker"); - // try to return error if worker channel closed - let _ = result.send(Err(BatchBuilderError::WorkerChannelClosed)); - return Err(e.into()); - } + let BatchBuilderOutput { batch, mined_transactions, changed_accounts, peer_deferred } = build_batch(build_args, worker_id, base_fee); + + // report the transactions this build left to an in-flight peer batch (issue #1329) + metrics + .peer_deferred_txs_total + .increment(u64::try_from(peer_deferred).unwrap_or(u64::MAX)); + // A build in which every pending transaction was deferred yields a batch with no + // transactions. Peers reject an empty batch (`BatchValidationError::EmptyBatch`) and + // score the sender a fatal penalty, so report no work and send nothing to the worker + // (issue #1329). The deferral metric above is still recorded: this build is exactly + // the case an operator watches that metric for. + if batch.transactions.is_empty() { + debug!( + target: "worker::batch_builder", + peer_deferred, + "every pending transaction is deferred by a peer batch; sealing nothing" + ); + result.send(Ok(BuildOutcome::NothingToSeal)).err().into_iter().for_each(|e| { + error!(target: "worker::batch_builder", ?e, "failed to send no-work outcome to block builder task"); + }); + Ok(()) + } else { + let batch = batch.seal_slow(); + span.record("batch", batch.digest().to_string()); + + // forward to worker and wait for ack that quorum was reached + if let Err(e) = to_worker.send((batch, ack)).await { + error!(target: "worker::batch_builder", ?e, "failed to send next batch to worker"); + // try to return error if worker channel closed + let _ = result.send(Err(BatchBuilderError::WorkerChannelClosed)); + return Err(e.into()); + } - // wait for worker to ack quorum reached then update pool with mined transactions - match rx.await { - Ok(res) => { - // measures build + broadcast + quorum, regardless of outcome - metrics.seal_duration_seconds.record(seal_start.elapsed()); - match res { - Ok(_) => { - debug!(target: "worker::batch-builder", ?res, "received ack"); - metrics.batches_sealed_total.increment(1); - // signal to Self that this task is complete - if let Err(e) = result.send(Ok(BuildOutcome::Mined(MinedBatchResult { mined_transactions, changed_accounts }))) { - error!(target: "worker::batch_builder", ?e, "failed to send block builder result to block builder task"); - } - } - Err(error) => { - metrics.record_seal_failure(worker_id, &error); - let converted = match error { - BlockSealError::FatalDBFailure => { - // fatal - return error - Err(BatchBuilderError::FatalDBFailure) - } - // The observer refusal is expected steady state whenever no - // committee endpoint is reachable, so the per-attempt line - // stays at debug; the run loop owns the state-change-gated - // logging and the retry backoff (issue #1145). - BlockSealError::NotValidator => { - debug!(target: "worker::batch_builder", "batch seal refused: no forward admitted the batch"); - Ok(BuildOutcome::Refused) + // wait for worker to ack quorum reached then update pool with mined txs + match rx.await { + Ok(res) => { + // measures build + broadcast + quorum, regardless of outcome + metrics.seal_duration_seconds.record(seal_start.elapsed()); + match res { + Ok(_) => { + debug!(target: "worker::batch-builder", ?res, "received ack"); + metrics.batches_sealed_total.increment(1); + // signal to Self that this task is complete + if let Err(e) = result.send(Ok(BuildOutcome::Mined(MinedBatchResult { mined_transactions, changed_accounts }))) { + error!(target: "worker::batch_builder", ?e, "failed to send block builder result to block builder task"); } - BlockSealError::QuorumRejected - | BlockSealError::AntiQuorum - | BlockSealError::Timeout - | BlockSealError::FailedToReport - | BlockSealError::FailedQuorum => { - error!(target: "worker::batch_builder", ?error, "error while sealing batch"); - // potentially non-fatal error - // - // NOTE: this will apply no changes to transaction pool - Ok(BuildOutcome::Failed) + } + Err(error) => { + metrics.record_seal_failure(worker_id, &error); + let converted = match error { + BlockSealError::FatalDBFailure => { + // fatal - return error + Err(BatchBuilderError::FatalDBFailure) + } + // The observer refusal is expected steady state whenever + // no committee endpoint is reachable, so the per-attempt + // line stays at debug; the run loop owns the + // state-change-gated logging and the retry backoff + // (issue #1145). + BlockSealError::NotValidator => { + debug!(target: "worker::batch_builder", "batch seal refused: no forward admitted the batch"); + Ok(BuildOutcome::Refused) + } + BlockSealError::QuorumRejected + | BlockSealError::AntiQuorum + | BlockSealError::Timeout + | BlockSealError::FailedToReport + | BlockSealError::FailedQuorum => { + error!(target: "worker::batch_builder", ?error, "error while sealing batch"); + // potentially non-fatal error + // + // NOTE: this applies no changes to transaction pool + Ok(BuildOutcome::Failed) + } + }; + + if let Err(e) = result.send(converted) { + error!(target: "worker::batch_builder", ?e, "failed to send block builder result to block builder task"); } - }; - - if let Err(e) = result.send(converted) { - error!(target: "worker::batch_builder", ?e, "failed to send block builder result to block builder task"); } } } - } - Err(e) => { - error!(target: "worker::batch_builder", ?e, "quorum waiter failed ack failed"); - if let Err(e) = result.send(Err(e.into())) { - error!(target: "worker::batch_builder", ?e, "failed to send block builder result to block builder task"); + Err(e) => { + error!(target: "worker::batch_builder", ?e, "quorum waiter failed ack failed"); + if let Err(e) = result.send(Err(e.into())) { + error!(target: "worker::batch_builder", ?e, "failed to send block builder result to block builder task"); + } } } + Ok(()) } - Ok(()) }.instrument(span_clone)); // return oneshot channel for receiving completion status @@ -446,6 +478,16 @@ impl BatchBuilder { mined_transactions: vec![], changed_accounts: vec![], }, + // nothing was sent to the worker, so there is no seal to judge: take the + // empty-mined path below (interval reset, one deferred build) without a + // pool update, an error log, or backoff bookkeeping (issue #1329) + BuildOutcome::NothingToSeal => { + debug!( + target: "worker::batch_builder", + "build sealed nothing: every pending transaction is deferred by a peer batch" + ); + MinedBatchResult { mined_transactions: vec![], changed_accounts: vec![] } + } }; // NOTE: a mined batch that pruned nothing applies no pool update; it also @@ -509,7 +551,7 @@ mod tests { payload::BuildArguments, recover_raw_transaction, test_utils::{create_committee_from_state, TransactionFactory}, - ForwardTargetPolicy, RethChainSpec, WorkerRpcForwarder, + ForwardTargetPolicy, RethChainSpec, TxPool as _, WorkerRpcForwarder, }; use tn_storage::{open_db, tables::NodeBatchesCache}; use tn_types::{ @@ -971,6 +1013,79 @@ mod tests { TestTools { tx_factory, execution_components, task_manager } } + /// A pool whose every pending transaction is already carried by a validated peer batch must + /// seal nothing (issue #1329). + /// + /// This is the steady state the deferral window creates on a validator whose whole pool sits + /// in a peer's in-flight batch: the build gate sees a non-empty pending pool, the build + /// defers every transaction, and the resulting batch has no transactions. Sending it would + /// earn a fatal penalty from every peer (`BatchValidationError::EmptyBatch`), so the task + /// reports [`BuildOutcome::NothingToSeal`] and sends nothing to the worker. + /// + /// Deterministic: the build task is awaited to completion, so a batch sent to the worker + /// would already sit in the channel when `try_recv` runs. + #[tokio::test] + async fn test_all_deferred_pool_seals_nothing() { + let tmp_dir = TempDir::new().unwrap(); + let TestTools { mut tx_factory, execution_components, task_manager } = + get_test_tools(tmp_dir.path()); + let TestExecutionComponents { reth_env, txpool, chain, .. } = execution_components; + let address = Address::from(U160::from(33)); + let (to_worker, mut from_batch_builder) = tokio::sync::mpsc::channel(2); + + let batch_builder = BatchBuilder::new( + &reth_env, + txpool.clone(), + to_worker, + address, + Duration::from_millis(1), + task_manager.get_spawner(), + 0, + MIN_PROTOCOL_BASE_FEE, + 0, + ) + .expect("batch builder"); + + let gas_price = reth_env.get_gas_price().unwrap(); + let value = U256::from(10).checked_pow(U256::from(18)).expect("1e18 doesn't overflow U256"); + let transaction = tx_factory.create_eip1559( + chain.clone(), + None, + gas_price, + Some(Address::ZERO), + value, // 1 TEL + Bytes::new(), + ); + + let hash = tx_factory.submit_tx_to_pool(transaction.clone(), txpool.clone()).await; + assert_eq!(&hash, transaction.hash()); + assert_eq!(txpool.pool_size().pending, 1, "the build gate sees a pending transaction"); + + // a peer batch this node validated already carries the pool's only pending transaction + txpool.record_peer_batch(&[hash]); + assert!(txpool.is_peer_deferred(&hash)); + + // await the build task itself: no sleeps, no polling of the run loop. The timeout is a + // failure detector, not a wait: a build that seals reaches the ack wait and never + // resolves, so it must surface as a failed assertion rather than a hung test. + let outcome = timeout(Duration::from_secs(5), batch_builder.spawn_execution_task()) + .await + .expect("the build task finishes without waiting on a seal ack") + .expect("build task reports its outcome") + .expect("an all-deferred build is not a fatal error"); + + assert_matches!(outcome, BuildOutcome::NothingToSeal); + assert!( + from_batch_builder.try_recv().is_err(), + "an empty batch must never reach the worker" + ); + assert_eq!( + txpool.pool_size().pending, + 1, + "the transaction stays pending for a later build" + ); + } + /// Test all possible errors from the worker while trying to reach quorum from peers. /// /// Non-fatal errors return empty vecs of mined transactions. diff --git a/crates/batch-builder/src/metrics.rs b/crates/batch-builder/src/metrics.rs index d0434d8d0..143f51243 100644 --- a/crates/batch-builder/src/metrics.rs +++ b/crates/batch-builder/src/metrics.rs @@ -22,6 +22,8 @@ pub(crate) struct BatchBuilderMetrics { pub(crate) batches_sealed_total: Counter, /// Time from spawning a batch build until the worker's quorum ack resolves. pub(crate) seal_duration_seconds: Histogram, + /// Total transactions skipped because a validated peer batch already carries them (#1329). + pub(crate) peer_deferred_txs_total: Counter, } impl BatchBuilderMetrics { diff --git a/crates/batch-builder/src/test_utils.rs b/crates/batch-builder/src/test_utils.rs index 1accdf2e7..df1f2d1f3 100644 --- a/crates/batch-builder/src/test_utils.rs +++ b/crates/batch-builder/src/test_utils.rs @@ -4,10 +4,11 @@ use crate::{build_batch, BatchBuilderOutput}; use std::{ collections::{BTreeMap, HashSet, VecDeque}, sync::Arc, + time::Duration, }; use tn_reth::{ - new_pool_txn, BestTransactions, InvalidPoolTransactionError, PoolTxn, PoolTxnId, - SenderIdentifiers, TxPool, + new_pool_txn, BestTransactions, InvalidPoolTransactionError, PeerBatchTxs, PoolTxn, PoolTxnId, + SenderId, SenderIdentifiers, TxPool, }; use tn_types::{Address, Batch, BatchBuilderArgs, Recovered, TransactionTrait as _, TxHash, U256}; @@ -25,14 +26,33 @@ pub fn execute_test_batch(test_batch: &mut Batch) { // Don't reset base_fee_per_gas, some tests need that value to remain. } +/// The deferral TTL every [`TestPool`] uses. +/// +/// Long enough that a build started after [`TxPool::record_peer_batch`] always observes the +/// deferral, so builder tests never race the clock. +const TEST_PEER_BATCH_TTL: Duration = Duration::from_secs(3600); + /// A test pool that ensures every transaction is in the pending pool -#[derive(Default, Clone, Debug)] +#[derive(Clone, Debug)] pub(crate) struct TestPool { transactions: Vec>, by_id: BTreeMap>, /// Per-sender balances returned by [`TxPool::get_account_balance`]. A sender that is absent /// here reports [`U256::MAX`], preserving the behavior of tests that do not exercise balance. balances: BTreeMap, + /// Transactions seen inside a validated peer batch, deferred by the builder (issue #1329). + peer_batch_txs: PeerBatchTxs, +} + +impl Default for TestPool { + fn default() -> Self { + Self { + transactions: Vec::new(), + by_id: BTreeMap::new(), + balances: BTreeMap::new(), + peer_batch_txs: PeerBatchTxs::new(TEST_PEER_BATCH_TTL), + } + } } impl TxPool for TestPool { @@ -52,6 +72,12 @@ impl TxPool for TestPool { fn get_account_balance(&self, address: Address) -> U256 { self.balances.get(&address).copied().unwrap_or(U256::MAX) } + fn record_peer_batch(&self, hashes: &[TxHash]) { + self.peer_batch_txs.record(hashes) + } + fn is_peer_deferred(&self, hash: &TxHash) -> bool { + self.peer_batch_txs.is_deferred(hash) + } } impl TestPool { @@ -93,7 +119,7 @@ impl TestPool { valid_tx }) .collect(); - Self { transactions, by_id: by_id.into_iter().collect(), balances: BTreeMap::new() } + Self { transactions, by_id: by_id.into_iter().collect(), ..Default::default() } } fn best_transactions_int(&self) -> Box>> { @@ -137,7 +163,12 @@ struct BestTestTransactions { /// then can be moved from the `all` set to the `independent` set. independent: VecDeque>, /// There might be the case where a yielded transactions is invalid, this will track it. - invalid: HashSet, + /// + /// Senders, not hashes, mirroring reth's `BestTransactions`: marking a transaction invalid + /// must also skip its descendants (the sender's later nonces), which are already unlocked by + /// the time the caller marks it. Tracking hashes alone would still yield the successor and + /// pack a nonce-gapped batch. + invalid: HashSet, /// Flag to control whether to skip blob transactions (EIP4844). skip_blobs: bool, } @@ -145,7 +176,7 @@ struct BestTestTransactions { impl BestTestTransactions { /// Mark the transaction and it's descendants as invalid. fn mark_invalid(&mut self, tx: &Arc) { - self.invalid.insert(*tx.hash()); + self.invalid.insert(tx.sender_id()); } } @@ -176,8 +207,8 @@ impl Iterator for BestTestTransactions { let best = self.independent.pop_front()?.clone(); let hash = best.transaction.transaction().hash(); - // skip transactions that were marked as invalid - if self.invalid.contains(hash) { + // skip transactions whose sender was marked invalid (this transaction or an ancestor) + if self.invalid.contains(&best.sender_id()) { tracing::debug!( target: "test-txpool", "[{:?}] skipping invalid transaction", diff --git a/crates/batch-builder/tests/it/build_batches.rs b/crates/batch-builder/tests/it/build_batches.rs index a15137896..e8058eaf3 100644 --- a/crates/batch-builder/tests/it/build_batches.rs +++ b/crates/batch-builder/tests/it/build_batches.rs @@ -12,7 +12,7 @@ use tn_engine::execute_consensus_output; use tn_network_types::{local::LocalNetwork, MockWorkerToPrimary}; use tn_reth::{ payload::BuildArguments, recover_raw_transaction, test_utils::TransactionFactory, - RethChainSpec, RethEnv, + RethChainSpec, RethEnv, TxPool as _, }; use tn_storage::{open_db, tables::NodeBatchesCache}; use tn_test_utils::wait_until; @@ -564,3 +564,142 @@ async fn test_canonical_notification_updates_pool() -> eyre::Result<()> { Ok(()) } + +/// A transaction a validated peer batch already carries must not be packed into this node's own +/// batch. The client can submit one signed transaction to every committee validator, so without +/// the deferral every worker packs a copy, every copy passes peer validation and takes a vote +/// round, and only the first executed copy pays (issue #1329). +#[tokio::test] +async fn test_peer_batched_tx_is_not_repacked() -> eyre::Result<()> { + let tmp_dir = TempDir::new().expect("temp dir"); + let task_manager = TaskManager::default(); + + // + //=== Consensus Layer + // + + let network_client = LocalNetwork::new_with_empty_id(); + let db_path = tmp_dir.path().join("c-db"); + let _ = std::fs::create_dir_all(&db_path); + let store = open_db(db_path); + + // Mock the primary client to always succeed. + let mock_server = MockWorkerToPrimary(); + network_client + .set_worker_to_primary_local_handler(Arc::new(mock_server)) + .expect("register mock primary handler"); + + let qw = TestMakeBlockQuorumWaiter::new_test(); + let mut batch_provider = Worker::new( + 0, + Some(qw.clone()), + network_client, + store.clone(), + Duration::from_secs(5), + WorkerNetworkHandle::new_for_test(task_manager.get_spawner()), + Arc::new(NoopTxnForwarder), + Vec::new(), + ); + batch_provider.spawn_batch_builder("test builder", &task_manager); + + // + //=== Execution Layer + // + + // adiri genesis funds the default factory; fund a second, independent sender so the two + // transactions below never share a nonce sequence + let genesis = test_genesis(); + let mut factory_b = TransactionFactory::new_random(); + let genesis = genesis.extend_accounts([( + factory_b.address(), + GenesisAccount::default().with_balance(U256::MAX), + )]); + let chain: Arc = Arc::new(genesis.into()); + + let reth_env = + RethEnv::new_for_temp_chain(chain.clone(), tmp_dir.path(), &task_manager, None).unwrap(); + let txpool = reth_env.init_txn_pool(BaseFeeContainer::default()).unwrap(); + let address = Address::from(U160::from(333)); + + let batch_builder = BatchBuilder::new( + &reth_env, + txpool.clone(), + batch_provider.batches_tx(), + address, + Duration::from_secs(1), + task_manager.get_spawner(), + 0, + MIN_PROTOCOL_BASE_FEE, + 0, + ) + .expect("batch builder"); + + let gas_price = reth_env.get_gas_price().unwrap(); + let value = U256::from(10).checked_pow(U256::from(18)).expect("1e18 doesn't overflow U256"); + let mut factory_a = TransactionFactory::new(); + + // two transactions from two senders, both admitted to this node's pool + let tx1 = factory_a.create_eip1559( + chain.clone(), + None, + gas_price, + Some(Address::ZERO), + value, // 1 TEL + Bytes::new(), + ); + let tx2 = factory_b.create_eip1559( + chain.clone(), + None, + gas_price, + Some(Address::ZERO), + value, // 1 TEL + Bytes::new(), + ); + + let added_result = factory_a.submit_tx_to_pool(tx1.clone(), txpool.clone()).await; + assert_matches!(added_result, hash if &hash == tx1.hash()); + let added_result = factory_b.submit_tx_to_pool(tx2.clone(), txpool.clone()).await; + assert_matches!(added_result, hash if &hash == tx2.hash()); + assert_eq!(txpool.pool_size().pending, 2); + + // a peer proposes a valid batch that carries tx1: the same transaction the client also sent + // to this node + let peer_batch = Batch { + transactions: vec![tx1.encoded_2718()], + epoch: 0, + beneficiary: Address::ZERO, + base_fee_per_gas: MIN_PROTOCOL_BASE_FEE, + worker_id: 0, + received_at: None, + } + .seal_slow(); + + let batch_validator = + BatchValidator::new(reth_env.clone(), Some(txpool.clone()), 0, MIN_PROTOCOL_BASE_FEE, 0); + assert!(batch_validator.validate_batch(peer_batch).is_ok()); + + // validating the peer batch deferred exactly its own transactions + assert!(txpool.is_peer_deferred(tx1.hash())); + assert!(!txpool.is_peer_deferred(tx2.hash())); + + // + //=== Test batch flow + // + + let _batch_builder = tokio::spawn(batch_builder.run()); + + // wait for this node's batch to be stored + wait_until(Duration::from_secs(5), "batch stored", || async { + Ok(store.iter::().next().is_some()) + }) + .await?; + + // the peer's transaction is in no batch this node produced; its own transaction is + let stored: Vec> = store + .iter::() + .flat_map(|(_, batch)| batch.transactions().to_vec()) + .collect(); + assert_eq!(stored, vec![tx2.encoded_2718()]); + + Ok(()) +} diff --git a/crates/batch-validator/src/validator.rs b/crates/batch-validator/src/validator.rs index 0e3144fb2..0cf1c59e8 100644 --- a/crates/batch-validator/src/validator.rs +++ b/crates/batch-validator/src/validator.rs @@ -1,11 +1,13 @@ //! Block validator use rayon::iter::{IntoParallelRefIterator as _, ParallelIterator as _}; -use tn_reth::{recover_raw_transaction, recover_signed_transaction, RethEnv, WorkerTxPool}; +use tn_reth::{ + recover_raw_transaction, recover_signed_transaction, RethEnv, TxPool as _, WorkerTxPool, +}; use tn_types::{ batch_allowlisted_tx_type, max_batch_gas, max_batch_size, BatchValidation, BatchValidationError, BlockHash, Epoch, SealedBatch, TransactionSigned, TransactionTrait as _, - Typed2718 as _, WorkerId, + TxHash, Typed2718 as _, WorkerId, }; /// Type convenience for implementing block validation errors. @@ -37,6 +39,10 @@ impl BatchValidation for BatchValidator { /// Validate a peer's batch. /// /// Workers do not execute full batches. This method validates the required information. + /// + /// On success (and only on success) the batch's transaction hashes are recorded in the + /// worker pool's deferral window, so this node's batch builder skips them while the peer + /// batch is in flight (issue #1329). A node without a pool (an observer) records nothing. fn validate_batch(&self, sealed_batch: SealedBatch) -> BatchValidationResult<()> { // ensure digest matches batch let (batch, digest) = sealed_batch.split(); @@ -79,6 +85,16 @@ impl BatchValidation for BatchValidator { // validate base fee- all batches for a worker and epoch have the same base fee. self.validate_basefee(batch.base_fee_per_gas)?; + + // the batch is valid: remember its transactions so this node's own builder does not pack + // a copy of something a peer is already proposing (issue #1329). Recording happens only + // after every check passes, so an invalid batch never defers anything, and the deferral + // expires on its own (see `PEER_BATCH_DEFER_TTL`) if the peer batch is abandoned. + if let Some(pool) = &self.tx_pool { + let hashes: Vec = decoded_txs.iter().map(|tx| *tx.hash()).collect(); + pool.record_peer_batch(&hashes); + } + Ok(()) } diff --git a/crates/tn-reth/README.md b/crates/tn-reth/README.md index 94d7dbbce..9afcddbbd 100644 --- a/crates/tn-reth/README.md +++ b/crates/tn-reth/README.md @@ -30,6 +30,13 @@ from the code at the cited paths; when the code and this file disagree, the code and continues (`src/env/execution.rs`). Transactions whose signer cannot be recovered are dropped deterministically instead of halting the network (issue #933) and counted by the alertable `tn_reth_unrecoverable_txs_dropped_total` metric (`src/metrics.rs`). +- Peer batch deferral (`src/peer_batch.rs`, issue #1329): a batch this node validates records its + transaction hashes in the pool's `PeerBatchTxs` window, and the batch builder skips a remembered + hash for `PEER_BATCH_DEFER_TTL` (10 seconds), bounded by `PEER_BATCH_SEEN_MAX_TXS` hashes. A + full window drops further hashes instead of evicting live entries, which are forgotten only at + twice the TTL, so a flood of peer batches can switch a deferral off but never re-arm one. + Duplicates across workers stay tolerated at execution; this only stops this node from packing a + copy of what a peer is already proposing. ## Header field mapping diff --git a/crates/tn-reth/src/lib.rs b/crates/tn-reth/src/lib.rs index 06b972f72..74985558e 100644 --- a/crates/tn-reth/src/lib.rs +++ b/crates/tn-reth/src/lib.rs @@ -112,13 +112,15 @@ pub use reth_rpc_eth_types::EthApiError; pub use reth_tracing::{FileWorkerGuard, Layers}; pub use reth_transaction_pool::{ error::{InvalidPoolTransactionError, PoolError, PoolTransactionError}, - identifier::SenderIdentifiers, + identifier::{SenderId, SenderIdentifiers}, BestTransactions, EthPooledTransaction, TransactionPool as TransactionPoolT, }; mod cli; pub mod dirs; pub mod payload; +pub mod peer_batch; +pub use peer_batch::{PeerBatchTxs, PEER_BATCH_DEFER_TTL, PEER_BATCH_SEEN_MAX_TXS}; pub mod traits; pub mod txn_pool; pub use txn_pool::*; diff --git a/crates/tn-reth/src/peer_batch.rs b/crates/tn-reth/src/peer_batch.rs new file mode 100644 index 000000000..bc466eb90 --- /dev/null +++ b/crates/tn-reth/src/peer_batch.rs @@ -0,0 +1,344 @@ +//! Short-lived memory of transactions already packed by a validated peer batch. +//! +//! A client can submit one signed transaction to every committee validator's RPC. Each worker +//! packs it into its own batch, every batch passes peer validation, and only the first executed +//! copy pays: the later copies are skipped for free at execution. The duplicates still consume +//! batch space, bandwidth, and vote rounds on every worker lane. +//! +//! This window is the local, protocol-neutral half of the fix. When a node validates a peer's +//! batch it remembers that batch's transaction hashes for a bounded time, and its own batch +//! builder defers those hashes (and, through `mark_invalid`, the sender's later nonces, which +//! could not execute ahead of the deferred nonce anyway, so they wait exactly as long as it +//! does) instead of packing a copy. No peer is penalized and nothing leaves the pool: after the +//! peer batch executes, the canonical update drops the transaction from the pool anyway, so the +//! memory only matters while the peer batch is in flight or lost. A build in which every pending +//! transaction is deferred seals nothing at all (`BuildOutcome::NothingToSeal` in +//! `tn-batch-builder`), because an empty batch is a message peers reject and penalize as fatal. + +use std::{ + collections::{HashMap, VecDeque}, + sync::{Arc, Mutex, PoisonError}, + time::{Duration, Instant}, +}; +use tn_types::TxHash; + +/// How long a transaction stays deferred after a peer batch that carries it validates. +/// +/// This equals the default `batch_vote_timeout` (`crates/config/src/node.rs`). A peer batch that +/// has not gathered its quorum by then has been abandoned by its producer, and one that has is on +/// its way into a header. +/// +/// Each arming defers a transaction for at most one TTL. A fresh arming is only possible after +/// the entry is forgotten at twice the TTL (see [`PeerBatchWindow::forget_expired`]), and during +/// the immune half this node's builder packs the transaction if it is still pending. A byzantine +/// peer that keeps reporting a batch it never certifies therefore adds at most one TTL of delay +/// per two TTLs and cannot censor. +/// +/// An honest producer that misses its vote quorum rebuilds the same transactions and re-reports +/// the same digest; that repeat report lands inside the immune window and is a no-op. +pub const PEER_BATCH_DEFER_TTL: Duration = Duration::from_secs(10); + +/// Hard cap on the number of remembered transaction hashes. +/// +/// Entries are never evicted early. While the window holds this many live entries a further hash +/// is not remembered at all, and the builder treats that transaction exactly as it did before +/// this window existed (it packs it). Hashes become recordable again only as entries age out at +/// twice the TTL. +/// +/// Evicting to make room would be a censorship lever. Batch validation checks neither balance nor +/// nonce, so a byzantine validator can push this many junk hashes through structurally valid +/// batches, evict a target's immune entry, and re-arm the target on every cycle: unbounded +/// deferral of one sender on every honest builder. With the drop-when-full policy a flood of peer +/// batches can only switch the deferral off, never re-arm an entry. +/// +/// Memory bound: about 100 bytes per entry (a 32-byte hash plus an `Instant` in both the map and +/// the order queue, plus hash-map overhead), so under 7 MB at the cap. +/// +/// Honest load: 65,536 hashes is about 46 full batches of 21,000-gas transfers (`max_batch_gas` +/// is 30M), or twenty seconds of peer batches at 3,276 transactions per second, which is longer +/// than the two TTLs any entry survives. +pub const PEER_BATCH_SEEN_MAX_TXS: usize = 65_536; + +/// A cheap-clone handle to the shared deferral window. +/// +/// Cloning shares the same window: the batch validator records into the clone held by the pool, +/// and the batch builder reads it through the pool it builds from. +#[derive(Clone, Debug)] +pub struct PeerBatchTxs { + /// The shared window. A `std::sync::Mutex` (not an async lock) because every critical + /// section is a handful of map operations and the builder is a synchronous function. + inner: Arc>, +} + +/// The window's state. +/// +/// `order` mirrors `seen` in insertion order, which equals time order because an existing entry +/// is never refreshed or re-timestamped and the clock is always sampled under the lock. +#[derive(Debug)] +struct PeerBatchWindow { + /// How long an entry defers its transaction. + ttl: Duration, + /// Upper bound on `seen.len()`. + cap: usize, + /// The remembered hashes with the instant each was first recorded. + seen: HashMap, + /// The same entries in insertion (time) order, oldest at the front. + order: VecDeque<(Instant, TxHash)>, +} + +impl Default for PeerBatchTxs { + fn default() -> Self { + Self::new(PEER_BATCH_DEFER_TTL) + } +} + +impl PeerBatchTxs { + /// Create an empty window that defers a recorded hash for `ttl`. + pub fn new(ttl: Duration) -> Self { + Self::with_cap(ttl, PEER_BATCH_SEEN_MAX_TXS) + } + + /// Create an empty window with an explicit capacity. + /// + /// Private: tests use a small capacity to exercise the full-window policy without recording + /// 65,536 hashes, and every other caller goes through [`Self::new`]. + fn with_cap(ttl: Duration, cap: usize) -> Self { + Self { + inner: Arc::new(Mutex::new(PeerBatchWindow { + ttl, + cap, + seen: HashMap::new(), + order: VecDeque::new(), + })), + } + } + + /// Remember `hashes` as packed by a validated peer batch, as of now. + /// + /// The clock is sampled after the lock is taken, not before. Batch validation runs on many + /// network handler tasks at once, and sampling first would let a recorder that lost the race + /// push a decreasing `Instant` into `order`, which the `take_while` prune in + /// [`PeerBatchWindow::forget_expired`] reads as sorted. + pub fn record(&self, hashes: &[TxHash]) { + let mut window = self.inner.lock().unwrap_or_else(PoisonError::into_inner); + let now = Instant::now(); + window.record(hashes, now); + } + + /// Remember `hashes` as packed by a validated peer batch, as of `now`. + /// + /// A hash already in the window keeps its original timestamp, so a peer that re-reports the + /// same batch cannot extend the deferral, and an entry whose deferral has expired stays + /// immune to re-arming until it is forgotten at twice the TTL. Each arming therefore costs + /// the transaction at most one TTL, and buying another one costs the reporter a full TTL in + /// which this node's builder packs the transaction if it is still pending. + #[cfg(test)] + fn record_at(&self, hashes: &[TxHash], now: Instant) { + let mut window = self.inner.lock().unwrap_or_else(PoisonError::into_inner); + window.record(hashes, now); + } + + /// Return true if `hash` was recorded by a validated peer batch within the TTL, as of now. + pub fn is_deferred(&self, hash: &TxHash) -> bool { + let window = self.inner.lock().unwrap_or_else(PoisonError::into_inner); + let now = Instant::now(); + window.is_deferred(hash, now) + } + + /// Return true if `hash` was recorded by a validated peer batch within the TTL, as of `now`. + #[cfg(test)] + fn is_deferred_at(&self, hash: &TxHash, now: Instant) -> bool { + let window = self.inner.lock().unwrap_or_else(PoisonError::into_inner); + window.is_deferred(hash, now) + } + + /// The number of remembered hashes, including expired entries not yet forgotten. + pub fn len(&self) -> usize { + self.inner.lock().unwrap_or_else(PoisonError::into_inner).seen.len() + } + + /// Return true if nothing is remembered. + pub fn is_empty(&self) -> bool { + self.len() == 0 + } +} + +impl PeerBatchWindow { + /// Forget what has aged out, then remember every hash that still fits. + fn record(&mut self, hashes: &[TxHash], now: Instant) { + self.forget_expired(now); + hashes.iter().for_each(|hash| self.insert(*hash, now)); + } + + /// Return true if `hash` is remembered and its deferral has not elapsed at `now`. + fn is_deferred(&self, hash: &TxHash, now: Instant) -> bool { + let ttl = self.ttl; + self.seen.get(hash).is_some_and(|seen_at| now.duration_since(*seen_at) < ttl) + } + + /// Drop entries older than twice the TTL. + /// + /// Twice the TTL, not the TTL itself, is what makes an expired entry immune to re-arming for + /// one more TTL: a peer that keeps reporting the same batch cannot chain deferrals. + fn forget_expired(&mut self, now: Instant) { + let forget_after = self.ttl.saturating_mul(2); + let stale = self + .order + .iter() + .take_while(|(seen_at, _)| now.duration_since(*seen_at) >= forget_after) + .count(); + let seen = &mut self.seen; + self.order.drain(..stale).for_each(|(_, hash)| { + seen.remove(&hash); + }); + } + + /// Record `hash` at `now` unless it is already remembered or the window is full. + /// + /// A full window drops the hash instead of evicting a live entry, which is what stops a + /// flood from re-arming a target's deferral (see [`PEER_BATCH_SEEN_MAX_TXS`]). + fn insert(&mut self, hash: TxHash, now: Instant) { + let fresh = self.seen.len() < self.cap && !self.seen.contains_key(&hash); + fresh.then_some(hash).into_iter().for_each(|hash| { + self.seen.insert(hash, now); + self.order.push_back((now, hash)); + }); + } +} + +#[cfg(test)] +mod tests { + use super::*; + + /// Build a distinct hash from a small number. + fn hash(n: u8) -> TxHash { + TxHash::with_last_byte(n) + } + + #[test] + fn fresh_hash_is_deferred_and_unknown_hash_is_not() { + let t0 = Instant::now(); + let window = PeerBatchTxs::new(PEER_BATCH_DEFER_TTL); + window.record_at(&[hash(1)], t0); + + assert!(window.is_deferred_at(&hash(1), t0)); + assert!(!window.is_deferred_at(&hash(2), t0)); + assert_eq!(window.len(), 1); + assert!(!window.is_empty()); + } + + #[test] + fn deferral_expires_at_exactly_the_ttl() { + let t0 = Instant::now(); + let ttl = PEER_BATCH_DEFER_TTL; + let window = PeerBatchTxs::new(ttl); + window.record_at(&[hash(1)], t0); + + // still deferred one millisecond before the TTL elapses + assert!(window.is_deferred_at(&hash(1), t0 + ttl - Duration::from_millis(1))); + // the TTL is exclusive: at exactly the TTL the transaction is buildable again + assert!(!window.is_deferred_at(&hash(1), t0 + ttl)); + } + + #[test] + fn re_record_within_the_ttl_does_not_refresh() { + let t0 = Instant::now(); + let ttl = PEER_BATCH_DEFER_TTL; + let window = PeerBatchTxs::new(ttl); + window.record_at(&[hash(1)], t0); + // a peer re-reports the same batch halfway through the deferral + window.record_at(&[hash(1)], t0 + ttl / 2); + + // the deferral still ends one TTL after the FIRST report + assert!(!window.is_deferred_at(&hash(1), t0 + ttl)); + assert_eq!(window.len(), 1); + } + + #[test] + fn re_record_after_expiry_is_immune_until_forgotten() { + let t0 = Instant::now(); + let ttl = PEER_BATCH_DEFER_TTL; + let window = PeerBatchTxs::new(ttl); + window.record_at(&[hash(1)], t0); + + // expired but not yet forgotten: a fresh report cannot re-arm the deferral + let re_report = t0 + ttl + Duration::from_millis(1); + window.record_at(&[hash(1)], re_report); + assert!(!window.is_deferred_at(&hash(1), re_report)); + assert!(!window.is_deferred_at(&hash(1), t0 + ttl * 2 - Duration::from_millis(1))); + } + + #[test] + fn re_record_after_twice_the_ttl_defers_again() { + let t0 = Instant::now(); + let ttl = PEER_BATCH_DEFER_TTL; + let window = PeerBatchTxs::new(ttl); + window.record_at(&[hash(1)], t0); + + // the entry is forgotten at twice the TTL, so a later peer batch defers it once more + let re_report = t0 + ttl * 2; + window.record_at(&[hash(1)], re_report); + assert!(window.is_deferred_at(&hash(1), re_report)); + assert_eq!(window.len(), 1); + } + + #[test] + fn full_window_drops_new_hashes_until_entries_age_out() { + let t0 = Instant::now(); + let ttl = PEER_BATCH_DEFER_TTL; + let window = PeerBatchTxs::with_cap(ttl, 2); + // fill the window to its capacity + window.record_at(&[hash(1), hash(2)], t0); + + // a hash recorded while the window is full is not remembered at all + let later = t0 + Duration::from_secs(1); + window.record_at(&[hash(3)], later); + assert!(!window.is_deferred_at(&hash(3), later), "the new hash is dropped, not remembered"); + assert_eq!(window.len(), 2, "the window never exceeds its capacity"); + assert!(window.is_deferred_at(&hash(1), later), "no live entry is evicted"); + + // once the filled entries age out at twice the TTL the same hash is recordable again + let aged_out = t0 + ttl * 2; + window.record_at(&[hash(3)], aged_out); + assert!(window.is_deferred_at(&hash(3), aged_out)); + assert_eq!(window.len(), 1); + } + + #[test] + fn cap_holds_when_one_record_exceeds_it() { + let t0 = Instant::now(); + let window = PeerBatchTxs::with_cap(PEER_BATCH_DEFER_TTL, 2); + window.record_at(&[hash(1), hash(2), hash(3), hash(4)], t0); + + assert_eq!(window.len(), 2, "a single oversized batch is capped too"); + // the first `cap` hashes of the record are kept and the rest are dropped + assert!(window.is_deferred_at(&hash(1), t0)); + assert!(window.is_deferred_at(&hash(2), t0)); + assert!(!window.is_deferred_at(&hash(3), t0)); + assert!(!window.is_deferred_at(&hash(4), t0)); + } + + /// A flood cannot evict a target's immune entry, so it cannot re-arm the target's deferral. + #[test] + fn full_window_cannot_re_arm_an_immune_entry() { + let t0 = Instant::now(); + let ttl = PEER_BATCH_DEFER_TTL; + let cap = 4; + let window = PeerBatchTxs::with_cap(ttl, cap); + // the target transaction is armed once, so its deferral ends at t0 + ttl + window.record_at(&[hash(1)], t0); + + // a byzantine validator floods the window with `cap` junk hashes carried by structurally + // valid batches, hoping to evict the target's expired-but-immune entry + let flood = t0 + ttl; + window.record_at(&[hash(2), hash(3), hash(4), hash(5)], flood); + assert_eq!(window.len(), cap, "the flood fills the window but evicts nothing"); + assert!(!window.is_deferred_at(&hash(1), flood), "the target's deferral already ended"); + + // the target is still remembered, so reporting it again cannot re-arm it + let re_report = flood + Duration::from_secs(1); + window.record_at(&[hash(1)], re_report); + assert!(!window.is_deferred_at(&hash(1), re_report)); + assert!(!window.is_deferred_at(&hash(1), t0 + ttl * 2 - Duration::from_millis(1))); + } +} diff --git a/crates/tn-reth/src/txn_pool.rs b/crates/tn-reth/src/txn_pool.rs index 21e4f3fbc..6e7b20647 100644 --- a/crates/tn-reth/src/txn_pool.rs +++ b/crates/tn-reth/src/txn_pool.rs @@ -44,7 +44,7 @@ use reth_rpc_eth_types::utils::recover_raw_transaction as reth_recover_raw_trans use reth_transaction_pool::{ error::{ Eip4844PoolTransactionError, Eip7702PoolTransactionError, InvalidPoolTransactionError, - PoolError, + PoolError, PoolTransactionError, }, AddedTransactionOutcome, BestTransactions, CanonicalStateUpdate, EthPooledTransaction, PoolSize, PoolTransaction, PoolUpdateKind, TransactionEvents, TransactionOrigin, @@ -59,8 +59,8 @@ use tokio_stream::wrappers::{errors::BroadcastStreamRecvError, BroadcastStream}; use tracing::{debug, info, trace, warn}; use crate::{ - error::TnRethResult, evm::TnEvmConfig, metrics::RETH_METRICS, traits::TelcoinNode, PoolTxn, - PoolTxnId, + error::TnRethResult, evm::TnEvmConfig, metrics::RETH_METRICS, peer_batch::PeerBatchTxs, + traits::TelcoinNode, PoolTxn, PoolTxnId, }; pub use reth_primitives_traits::InMemorySize as TxnSize; @@ -105,6 +105,14 @@ pub trait TxPool { /// only keep a sender's remaining transactions parked, never promote an unfunded one, and the /// engine's authoritative canonical update corrects it within the same consensus round. fn get_account_balance(&self, address: Address) -> U256; + /// Remember `hashes` as packed by a peer batch this node has just validated. + /// + /// The builder skips a remembered hash for + /// [`PEER_BATCH_DEFER_TTL`](crate::PEER_BATCH_DEFER_TTL) so this node does not pack a copy + /// of a transaction a peer is already proposing. + fn record_peer_batch(&self, hashes: &[TxHash]); + /// Return true if `hash` is still deferred by a validated peer batch. + fn is_peer_deferred(&self, hash: &TxHash) -> bool; } /// A telcoin network transaction pool. @@ -119,6 +127,9 @@ pub struct WorkerTxPool( /// The shared per-worker base-fee container: the single source of the pool's pending base /// fee (issue #1262). BaseFeeContainer, + /// The transactions this node has seen inside a validated peer batch, deferred by the + /// builder while that peer batch is in flight (issue #1329). + PeerBatchTxs, ); impl From @@ -213,7 +224,7 @@ impl WorkerTxPool { ); */ - Ok(Self(transaction_pool, blockchain_provider.clone(), base_fee)) + Ok(Self(transaction_pool, blockchain_provider.clone(), base_fee, PeerBatchTxs::default())) } /// Spawn the CRITICAL task that applies canonical-state updates to the pool. @@ -561,6 +572,14 @@ impl WorkerTxPool { pub fn pool_size(&self) -> PoolSize { self.0.pool_size() } + + /// The shared window of transactions seen inside a validated peer batch. + /// + /// The batch validator records into this window and the batch builder reads it, so a + /// transaction a peer is already proposing is not packed again here (issue #1329). + pub fn peer_batch_txs(&self) -> &PeerBatchTxs { + &self.3 + } } impl TxPool for WorkerTxPool { @@ -585,6 +604,14 @@ impl TxPool for WorkerTxPool { .map(|account| account.balance) .unwrap_or(U256::ZERO) } + + fn record_peer_batch(&self, hashes: &[TxHash]) { + self.3.record(hashes) + } + + fn is_peer_deferred(&self, hash: &TxHash) -> bool { + self.3.is_deferred(hash) + } } /// An iterator that produces the best transactions from a pool. @@ -642,6 +669,45 @@ impl BestTxns { ), ); } + + /// Skip a transaction a validated peer batch already carries. + /// + /// Marking it invalid for this build also skips the sender's later nonces: a nonce-gapped + /// copy would only be skipped at execution, after paying for batch space and a vote round. + /// The transaction stays in the pool and is packed normally once the deferral expires (or + /// leaves the pool with the peer batch's execution), so this is not a rejection. + pub fn peer_deferred(&mut self, pool_tx: &Arc) { + self.inner.mark_invalid( + pool_tx, + &InvalidPoolTransactionError::Other(Box::new(PeerBatchDeferred)), + ); + } +} + +/// The pool error reported when the builder skips a transaction already packed by a validated +/// peer batch (issue #1329). +/// +/// This is a local scheduling decision, not a judgement about the transaction: `is_bad_transaction` +/// is false, so no peer is penalized and the transaction stays poolable. +#[derive(Debug, Default, Clone, Copy)] +pub struct PeerBatchDeferred; + +impl std::fmt::Display for PeerBatchDeferred { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + write!(f, "transaction deferred: already packed by a validated peer batch") + } +} + +impl std::error::Error for PeerBatchDeferred {} + +impl PoolTransactionError for PeerBatchDeferred { + fn is_bad_transaction(&self) -> bool { + false + } + + fn as_any(&self) -> &dyn std::any::Any { + self + } } impl Iterator for BestTxns {