logos-blockchain/logos-execution-zone
1
0
Fork 0
mirror of https://github.com/logos-blockchain/logos-execution-zone.git synced 2026-07-11 08:19:35 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat(sequencer): now keep orphaned in mempool + use with_state to avoid clone for method calls to V03State

Browse Source
This commit is contained in:
erhant 2026-07-10 15:54:37 +03:00
parent 2ba2cc676b
commit b21049bc28
3 changed files with 52 additions and 28 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

48
lez/sequencer/core/src/lib.rs
View File

@ -165,7 +165,7 @@ impl<BP: BlockPublisherTrait> SequencerCore<BP> {
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");
@ -346,18 +346,22 @@ impl<BP: BlockPublisherTrait> SequencerCore<BP> {
fn on_follow(
dbio: Arc<RocksDBIO>,
chain: Arc<Mutex<ChainState>>,
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 {
@ -379,7 +383,7 @@ impl<BP: BlockPublisherTrait> SequencerCore<BP> {
finalized.push(block);
}
}
(adopted, finalized, chain.head_state().clone())
(adopted, finalized, resubmit_txs, chain.head_state().clone())
};
for block in adopted {
@ -398,6 +402,14 @@ impl<BP: BlockPublisherTrait> SequencerCore<BP> {
);
}
}
// 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:#}");
}
}
})
})
}
@ -596,17 +608,14 @@ impl<BP: BlockPublisherTrait> SequencerCore<BP> {
})
}
/// 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<R>(&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 {
@ -837,6 +846,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<LeeTransaction> {
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<HashType> {
let LeeTransaction::Public(tx) = tx else {

14
lez/sequencer/core/src/tests.rs
View File

@ -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()

18
lez/sequencer/service/src/service.rs
View File

@ -124,8 +124,8 @@ impl<BC: BlockPublisherTrait + Send + 'static> sequencer_service_rpc::RpcServer
async fn get_account_balance(&self, account_id: AccountId) -> Result<u128, ErrorObjectOwned> {
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<BC: BlockPublisherTrait + Send + 'static> sequencer_service_rpc::RpcServer
account_ids: Vec<AccountId>,
) -> Result<Vec<Nonce>, 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<BC: BlockPublisherTrait + Send + 'static> sequencer_service_rpc::RpcServer
commitment: Commitment,
) -> Result<Option<MembershipProof>, 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<Account, ErrorObjectOwned> {
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<BTreeMap<String, ProgramId>, ErrorObjectOwned> {