From 56acab3a16da49e2a62fe8eb21e392a2329a2349 Mon Sep 17 00:00:00 2001 From: erhant Date: Fri, 10 Jul 2026 15:54:37 +0300 Subject: [PATCH] feat(sequencer): now keep `orphaned` in mempool + use `with_state` to avoid clone for method calls to `V03State` --- lez/sequencer/core/src/lib.rs | 48 ++++++++++++++++++++-------- lez/sequencer/core/src/tests.rs | 14 ++++---- lez/sequencer/service/src/service.rs | 18 ++++++----- 3 files changed, 52 insertions(+), 28 deletions(-) diff --git a/lez/sequencer/core/src/lib.rs b/lez/sequencer/core/src/lib.rs index 59e2b23c..9e14f6df 100644 --- a/lez/sequencer/core/src/lib.rs +++ b/lez/sequencer/core/src/lib.rs @@ -164,7 +164,7 @@ impl SequencerCore { Self::on_finalized_block(store.dbio()), Self::on_deposit_event(store.dbio(), mempool_handle.clone()), Self::on_withdraw_event(store.dbio()), - Self::on_follow(store.dbio(), Arc::clone(&chain)), + Self::on_follow(store.dbio(), Arc::clone(&chain), mempool_handle.clone()), ) .await .expect("Failed to initialize Block Publisher"); @@ -327,18 +327,22 @@ impl SequencerCore { fn on_follow( dbio: Arc, chain: Arc>, + mempool_handle: MemPoolHandle<(TransactionOrigin, LeeTransaction)>, ) -> block_publisher::OnFollowSink { Box::new(move |update: block_publisher::FollowUpdate| { let chain = Arc::clone(&chain); let dbio = Arc::clone(&dbio); + let mempool_handle = mempool_handle.clone(); Box::pin(async move { // Apply under the lock and collect what to persist; take a single // head snapshot. Release the lock before touching disk so the // producer is never blocked on the follow path's I/O. - let (adopted, finalized, head_snapshot) = { + let (adopted, finalized, resubmit_txs, head_snapshot) = { let mut chain = chain.lock().expect("chain state mutex poisoned"); - for (this_msg, _) in &update.orphaned { + let mut resubmit_txs = Vec::new(); + for (this_msg, block) in &update.orphaned { chain.revert_orphan(*this_msg); + resubmit_txs.extend(resubmittable_txs(block)); } let mut adopted = Vec::new(); for (this_msg, block) in &update.adopted { @@ -360,7 +364,7 @@ impl SequencerCore { finalized.push(block); } } - (adopted, finalized, chain.head_state().clone()) + (adopted, finalized, resubmit_txs, chain.head_state().clone()) }; for block in adopted { @@ -379,6 +383,14 @@ impl SequencerCore { ); } } + + // Rebuild orphaned work: return its user txs to the mempool so the + // next on-turn production re-includes them on the new head. + for tx in resubmit_txs { + if let Err(err) = mempool_handle.push((TransactionOrigin::User, tx)).await { + error!("Failed to resubmit orphaned transaction: {err:#}"); + } + } }) }) } @@ -577,17 +589,14 @@ impl SequencerCore { }) } - /// A clone of the current head state. - /// - /// TODO: cloning the whole state per call is wasteful; add targeted - /// account/nonce/proof accessors so RPC reads don't clone the whole state. - #[must_use] - pub fn state(&self) -> lee::V03State { - self.chain + /// Reads the current head state under the lock without cloning it, so callers + /// reuse `V03State`'s own API (accounts, nonces, proofs) with no whole-state copy. + pub fn with_state(&self, f: impl FnOnce(&lee::V03State) -> R) -> R { + f(self + .chain .lock() .expect("chain state mutex poisoned") - .head_state() - .clone() + .head_state()) } pub const fn block_store(&self) -> &SequencerStore { @@ -818,6 +827,19 @@ fn build_bridge_deposit_tx_from_event(event: &PendingDepositEventRecord) -> Resu ))) } +/// User transactions of an orphaned block to return to the mempool: everything +/// except the trailing clock tx and sequencer-generated bridge deposits (those are +/// replayed from their own bedrock events, not the mempool). +fn resubmittable_txs(block: &Block) -> Vec { + let Some((_clock, rest)) = block.body.transactions.split_last() else { + return Vec::new(); + }; + rest.iter() + .filter(|tx| extract_bridge_deposit_id(tx).is_none()) + .cloned() + .collect() +} + #[must_use] fn extract_bridge_deposit_id(tx: &LeeTransaction) -> Option { let LeeTransaction::Public(tx) = tx else { diff --git a/lez/sequencer/core/src/tests.rs b/lez/sequencer/core/src/tests.rs index 237bcfb5..215d2e2e 100644 --- a/lez/sequencer/core/src/tests.rs +++ b/lez/sequencer/core/src/tests.rs @@ -139,8 +139,8 @@ async fn start_from_config() { let acc1_account_id = initial_public_user_accounts()[0].account_id; let acc2_account_id = initial_public_user_accounts()[1].account_id; - let balance_acc_1 = sequencer.state().get_account_by_id(acc1_account_id).balance; - let balance_acc_2 = sequencer.state().get_account_by_id(acc2_account_id).balance; + let balance_acc_1 = sequencer.with_state(|s| s.get_account_by_id(acc1_account_id).balance); + let balance_acc_2 = sequencer.with_state(|s| s.get_account_by_id(acc2_account_id).balance); assert_eq!(10000, balance_acc_1); assert_eq!(20000, balance_acc_2); @@ -371,8 +371,8 @@ async fn transaction_execute_native_transfer() { ) .unwrap(); - let bal_from = sequencer.state().get_account_by_id(acc1).balance; - let bal_to = sequencer.state().get_account_by_id(acc2).balance; + let bal_from = sequencer.with_state(|s| s.get_account_by_id(acc1).balance); + let bal_to = sequencer.with_state(|s| s.get_account_by_id(acc2).balance); assert_eq!(bal_from, 9900); assert_eq!(bal_to, 20100); @@ -563,8 +563,8 @@ async fn restart_from_storage() { // with the above transaction and update the state to reflect that. let (sequencer, _mempool_handle) = SequencerCoreWithMockClients::start_from_config(config.clone()).await; - let balance_acc_1 = sequencer.state().get_account_by_id(acc1_account_id).balance; - let balance_acc_2 = sequencer.state().get_account_by_id(acc2_account_id).balance; + let balance_acc_1 = sequencer.with_state(|s| s.get_account_by_id(acc1_account_id).balance); + let balance_acc_2 = sequencer.with_state(|s| s.get_account_by_id(acc2_account_id).balance); // Balances should be consistent with the stored block assert_eq!( @@ -788,7 +788,7 @@ async fn block_production_aborts_when_clock_account_data_is_corrupted() { // Corrupt the clock 01 account data so the clock program panics on deserialization. let clock_account_id = system_accounts::clock_account_ids()[0]; - let mut corrupted = sequencer.state().get_account_by_id(clock_account_id); + let mut corrupted = sequencer.with_state(|s| s.get_account_by_id(clock_account_id)); corrupted.data = vec![0xff; 3].try_into().unwrap(); sequencer .chain() diff --git a/lez/sequencer/service/src/service.rs b/lez/sequencer/service/src/service.rs index 3a48e7cc..f7771cd3 100644 --- a/lez/sequencer/service/src/service.rs +++ b/lez/sequencer/service/src/service.rs @@ -124,8 +124,8 @@ impl sequencer_service_rpc::RpcServer async fn get_account_balance(&self, account_id: AccountId) -> Result { let sequencer = self.sequencer.lock().await; - let account = sequencer.state().get_account_by_id(account_id); - Ok(account.balance) + let balance = sequencer.with_state(|state| state.get_account_by_id(account_id).balance); + Ok(balance) } async fn get_transaction( @@ -141,10 +141,12 @@ impl sequencer_service_rpc::RpcServer account_ids: Vec, ) -> Result, ErrorObjectOwned> { let sequencer = self.sequencer.lock().await; - let nonces = account_ids - .into_iter() - .map(|account_id| sequencer.state().get_account_by_id(account_id).nonce) - .collect(); + let nonces = sequencer.with_state(|state| { + account_ids + .into_iter() + .map(|account_id| state.get_account_by_id(account_id).nonce) + .collect() + }); Ok(nonces) } @@ -153,12 +155,12 @@ impl sequencer_service_rpc::RpcServer commitment: Commitment, ) -> Result, ErrorObjectOwned> { let sequencer = self.sequencer.lock().await; - Ok(sequencer.state().get_proof_for_commitment(&commitment)) + Ok(sequencer.with_state(|state| state.get_proof_for_commitment(&commitment))) } async fn get_account(&self, account_id: AccountId) -> Result { let sequencer = self.sequencer.lock().await; - Ok(sequencer.state().get_account_by_id(account_id)) + Ok(sequencer.with_state(|state| state.get_account_by_id(account_id))) } async fn get_program_ids(&self) -> Result, ErrorObjectOwned> {