From c8dd76a94b9b554a9bb77faca70ee2dd792bb66f Mon Sep 17 00:00:00 2001 From: erhant Date: Mon, 20 Jul 2026 21:59:24 +0300 Subject: [PATCH] fix(sequencer, storage): persist under the chain lock, prune stale blocks on shortening reorgs, halt on persist failure --- lez/chain_state/src/chain.rs | 4 + lez/sequencer/core/src/lib.rs | 127 +++++++++++++++-------------- lez/sequencer/service/src/lib.rs | 8 +- lez/sequencer/service/src/main.rs | 14 +++- lez/storage/src/sequencer/mod.rs | 46 +++++++---- lez/storage/src/sequencer/tests.rs | 45 ++++++++++ test_fixtures/src/setup.rs | 3 +- 7 files changed, 160 insertions(+), 87 deletions(-) diff --git a/lez/chain_state/src/chain.rs b/lez/chain_state/src/chain.rs index b65ece7b..fe0ebbbb 100644 --- a/lez/chain_state/src/chain.rs +++ b/lez/chain_state/src/chain.rs @@ -252,6 +252,10 @@ impl ChainState { self.final_state = scratch; self.final_tip = Some(tip_of(block)); self.final_stall = None; + // Any head suffix dropped here was already reverted as + // `orphaned` earlier in the same channel update (the sdk + // orders orphans before their finalized replacement), so its + // txs are back in the caller's mempool. self.head_blocks.clear(); self.head_state = self.final_state.clone(); AcceptOutcome::Applied diff --git a/lez/sequencer/core/src/lib.rs b/lez/sequencer/core/src/lib.rs index 3586e811..790fad1f 100644 --- a/lez/sequencer/core/src/lib.rs +++ b/lez/sequencer/core/src/lib.rs @@ -468,34 +468,30 @@ impl SequencerCore { deposit_event_ids: &[HashType], withdrawal_reconciliation_keys: Vec, ) -> Result<()> { - let head_state = { - let mut chain = self.chain.lock().expect("chain state mutex poisoned"); - match chain.apply_adopted(this_msg, block) { - AcceptOutcome::Applied => Some(chain.head_state().clone()), - AcceptOutcome::AlreadyApplied => { - warn!( - "Produced block {} lost a competing-write race, skipping persistence", - block.header.block_id - ); - None - } - AcceptOutcome::Parked(err) | AcceptOutcome::RetryableFailure(err) => { - warn!( - "Produced block {} no longer chains on the head, skipping persistence: {err}", - block.header.block_id - ); - None - } + let mut chain = self.chain.lock().expect("chain state mutex poisoned"); + match chain.apply_adopted(this_msg, block) { + AcceptOutcome::Applied => { + // Persisted under the lock so disk writes land in apply order + // with the follow path. + self.store.update( + block, + deposit_event_ids, + withdrawal_reconciliation_keys, + chain.head_state(), + )?; + } + AcceptOutcome::AlreadyApplied => { + warn!( + "Produced block {} lost a competing-write race, skipping persistence", + block.header.block_id + ); + } + AcceptOutcome::Parked(err) | AcceptOutcome::RetryableFailure(err) => { + warn!( + "Produced block {} no longer chains on the head, skipping persistence: {err}", + block.header.block_id + ); } - }; - - if let Some(head_state) = head_state { - self.store.update( - block, - deposit_event_ids, - withdrawal_reconciliation_keys, - &head_state, - )?; } Ok(()) @@ -775,9 +771,9 @@ fn apply_follow_update( finalized, } = update; - // Apply under the lock and collect what to persist; release it before - // touching disk so the producer is never blocked on follow I/O. - let (to_persist, resubmit_txs, head_snapshot, final_snapshot) = { + // The lock is held across the persist below so disk writes land in apply + // order — the produce path persists under this same lock. + let resubmit_txs = { let mut chain = chain.lock().expect("chain state mutex poisoned"); // User txs of orphaned blocks, returned to the mempool below. @@ -808,41 +804,39 @@ fn apply_follow_update( } } // Snapshot the advanced final tier so a restart re-anchors on it. - let final_snapshot = final_advanced.then(|| { + let final_meta = final_advanced.then(|| { let tip = chain.final_tip().expect("advanced final tier has a tip"); - ( - chain.final_state().clone(), - BlockMeta { - id: tip.block_id, - hash: tip.hash, - }, - ) + BlockMeta { + id: tip.block_id, + hash: tip.hash, + } + }); + let head_tip = chain.head_tip().map(|tip| BlockMeta { + id: tip.block_id, + hash: tip.hash, }); - ( - to_persist, - resubmit_txs, - chain.head_state().clone(), - final_snapshot, + // One atomic write for the whole update: blocks, tip meta and the + // state after the last block land together, so a crash can never + // leave the stored state ahead of the stored blocks. A persist + // failure is fatal: the in-memory chain has already advanced, and + // continuing would leave a permanent gap in the store. + // + // TODO: the zone-sdk checkpoint is persisted by `on_checkpoint` + // before this write; a crash in between resumes past these blocks + // without them ever landing in the store. Full `BlocksProcessed` + // atomicity (checkpoint + blocks + state in one batch, per the sdk's + // event contract) is a follow-up. + dbio.store_followed_blocks( + &to_persist, + head_tip.as_ref(), + chain.head_state(), + final_meta.as_ref().map(|meta| (chain.final_state(), meta)), ) - }; + .unwrap_or_else(|err| panic!("Failed to persist followed blocks: {err:#}")); - // One atomic write for the whole update: blocks, tip meta and the state - // after the last block land together, so a crash can never leave the - // stored state ahead of the stored blocks. - // - // TODO: the zone-sdk checkpoint is persisted by `on_checkpoint` before - // this write; a crash in between resumes past these blocks without them - // ever landing in the store. Full `BlocksProcessed` atomicity (checkpoint - // + blocks + state in one batch, per the sdk's event contract) is a - // follow-up. - if let Err(err) = dbio.store_followed_blocks( - &to_persist, - &head_snapshot, - final_snapshot.as_ref().map(|(state, meta)| (state, meta)), - ) { - error!("Failed to persist followed blocks: {err:#}"); - } + resubmit_txs + }; // Rebuild orphaned work: return its user txs to the mempool so the // next on-turn production re-includes them on the new head. @@ -1061,18 +1055,25 @@ 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). +/// except the trailing clock tx, sequencer-generated bridge deposits (replayed +/// from their own bedrock events) and sequencer-only cross-zone txs (replayed +/// by the watcher; the ingress guard rejects them as `User`). 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()) + .filter(|tx| extract_bridge_deposit_id(tx).is_none() && !is_sequencer_only_tx(tx)) .cloned() .collect() } +#[must_use] +fn is_sequencer_only_tx(tx: &LeeTransaction) -> bool { + matches!(tx, LeeTransaction::Public(tx) + if is_sequencer_only_program(tx.message().program_id)) +} + #[must_use] fn extract_bridge_deposit_id(tx: &LeeTransaction) -> Option { let LeeTransaction::Public(tx) = tx else { diff --git a/lez/sequencer/service/src/lib.rs b/lez/sequencer/service/src/lib.rs index d6779102..2275102d 100644 --- a/lez/sequencer/service/src/lib.rs +++ b/lez/sequencer/service/src/lib.rs @@ -111,7 +111,7 @@ impl Drop for SequencerHandle { } } -pub async fn run(config: SequencerConfig, port: u16) -> Result { +pub async fn run(config: SequencerConfig, listen_addr: SocketAddr) -> Result { let block_timeout = config.block_create_timeout; let max_block_size = config.max_block_size; @@ -125,7 +125,7 @@ pub async fn run(config: SequencerConfig, port: u16) -> Result let (server_handle, addr) = run_server( Arc::clone(&seq_core_wrapped), mempool_handle_for_server, - port, + listen_addr, max_block_size.as_u64(), ) .await?; @@ -142,7 +142,7 @@ pub async fn run(config: SequencerConfig, port: u16) -> Result async fn run_server( sequencer: Arc>, mempool_handle: MemPoolHandle<(TransactionOrigin, LeeTransaction)>, - port: u16, + listen_addr: SocketAddr, max_block_size: u64, ) -> Result<(ServerHandle, SocketAddr)> { let server = jsonrpsee::server::ServerBuilder::with_config( @@ -153,7 +153,7 @@ async fn run_server( ) .build(), ) - .build(SocketAddr::from(([0, 0, 0, 0], port))) + .build(listen_addr) .await .context("Failed to build RPC server")?; diff --git a/lez/sequencer/service/src/main.rs b/lez/sequencer/service/src/main.rs index 4dede6d2..d73271d3 100644 --- a/lez/sequencer/service/src/main.rs +++ b/lez/sequencer/service/src/main.rs @@ -1,4 +1,7 @@ -use std::path::PathBuf; +use std::{ + net::{IpAddr, SocketAddr}, + path::PathBuf, +}; use anyhow::Result; use clap::Parser; @@ -12,6 +15,11 @@ struct Args { config_path: PathBuf, #[clap(short, long, default_value = "3040")] port: u16, + /// Interface the RPC server binds to. The RPC has no caller auth + /// (including `adminConfigureChannel`) — bind loopback unless the port + /// is firewalled. + #[clap(long, default_value = "0.0.0.0")] + listen_address: IpAddr, /// Override the config's home directory (`RocksDB` + bedrock signing key), /// so multiple instances can share one config file. #[clap(long)] @@ -29,6 +37,7 @@ async fn main() -> Result<()> { let Args { config_path, port, + listen_address, home, } = Args::parse(); @@ -40,7 +49,8 @@ async fn main() -> Result<()> { if let Some(home) = home { config.home = home; } - let sequencer_handle = sequencer_service::run(config, port).await?; + let sequencer_handle = + sequencer_service::run(config, SocketAddr::new(listen_address, port)).await?; tokio::select! { () = cancellation_token.cancelled() => { diff --git a/lez/storage/src/sequencer/mod.rs b/lez/storage/src/sequencer/mod.rs index ed8130bf..0ff2d525 100644 --- a/lez/storage/src/sequencer/mod.rs +++ b/lez/storage/src/sequencer/mod.rs @@ -507,6 +507,18 @@ impl RocksDBIO { self.put_block_payload(block, batch) } + /// Stages deletion of a block payload into `batch`. + fn delete_block_payload(&self, block_id: u64, batch: &mut WriteBatch) -> DbResult<()> { + let cf_block = self.block_column(); + batch.delete_cf( + &cf_block, + borsh::to_vec(&block_id).map_err(|err| { + DbError::borsh_cast_message(err, Some("Failed to serialize block id".to_owned())) + })?, + ); + Ok(()) + } + /// Stages just the block payload into `batch`, without touching the tip meta. fn put_block_payload(&self, block: &Block, batch: &mut WriteBatch) -> DbResult<()> { let cf_block = self.block_column(); @@ -644,19 +656,28 @@ impl RocksDBIO { } /// Persists a followed (peer) block so the store mirrors the canonical chain. - /// One-block form of [`Self::store_followed_blocks`], without a final snapshot. + /// One-block form of [`Self::store_followed_blocks`], with the block as the + /// head tip and no final snapshot. pub fn store_followed_block( &self, block: &Block, state: &V03State, finalized: bool, ) -> DbResult<()> { - self.store_followed_blocks(&[(block, finalized)], state, None) + let head_tip = BlockMeta { + id: block.header.block_id, + hash: block.header.hash, + }; + self.store_followed_blocks(&[(block, finalized)], Some(&head_tip), state, None) } /// Persists a batch of followed blocks, the caller's head-tip `state`, and /// the optional final-tier snapshot in one atomic write. /// + /// The tip meta is pinned to `head_tip`, and blocks stored above it (left + /// behind by a net-shortening reorg) are deleted in the same write, so + /// restart replay never walks past the tip. + /// /// Per block: skips the payload write when the store already holds it (by /// id and hash), unless `finalized` is set, which rewrites it with the /// finalized status. When nothing needs writing, nothing is written — @@ -669,12 +690,12 @@ impl RocksDBIO { pub fn store_followed_blocks( &self, blocks: &[(&Block, bool)], + head_tip: Option<&BlockMeta>, state: &V03State, final_snapshot: Option<(&V03State, &BlockMeta)>, ) -> DbResult<()> { let last_block_in_db = self.get_meta_last_block_in_db()?; let mut batch = WriteBatch::default(); - let mut batch_tip: Option = None; for (block, finalized) in blocks { let already_stored = self @@ -690,26 +711,17 @@ impl RocksDBIO { to_write.bedrock_status = BedrockStatus::Finalized; } self.put_block_payload(&to_write, &mut batch)?; - - let id = to_write.header.block_id; - if batch_tip.as_ref().is_none_or(|tip| id >= tip.id) { - batch_tip = Some(BlockMeta { - id, - hash: to_write.header.hash, - }); - } } if batch.is_empty() && final_snapshot.is_none() { return Ok(()); } - // Meta is staged once for the batch's highest block. - // (`>=` so a same-height overwrite refreshes the tip hash, matching `put_block`) - if let Some(tip) = batch_tip - && tip.id >= last_block_in_db - { + if let Some(tip) = head_tip { + for stale_id in tip.id.saturating_add(1)..=last_block_in_db { + self.delete_block_payload(stale_id, &mut batch)?; + } self.put_meta_last_block_in_db_batch(tip.id, &mut batch)?; - self.put_meta_latest_block_meta_batch(&tip, &mut batch)?; + self.put_meta_latest_block_meta_batch(tip, &mut batch)?; } self.put_lee_state_in_db_batch(state, &mut batch)?; if let Some((final_state, final_meta)) = final_snapshot { diff --git a/lez/storage/src/sequencer/tests.rs b/lez/storage/src/sequencer/tests.rs index 1938684e..0202241a 100644 --- a/lez/storage/src/sequencer/tests.rs +++ b/lez/storage/src/sequencer/tests.rs @@ -99,8 +99,13 @@ fn store_followed_blocks_batch_lands_meta_and_state_on_last_block() { .unwrap(); let block3 = produce_dummy_block(3, Some(block2.header.hash), vec![]); + let head_tip = BlockMeta { + id: 3, + hash: block3.header.hash, + }; dbio.store_followed_blocks( &[(&block2, true), (&block3, false)], + Some(&head_tip), &state_with_balance(300), None, ) @@ -134,6 +139,7 @@ fn final_snapshot_round_trips_and_is_absent_on_fresh_store() { }; dbio.store_followed_blocks( &[(&block2, true)], + Some(&final_meta), &state_with_balance(300), Some((&state_with_balance(200), &final_meta)), ) @@ -175,3 +181,42 @@ fn store_followed_block_overwrites_competing_block_at_same_id() { assert_eq!(meta.id, 2); assert_eq!(meta.hash, block2b.header.hash); } + +#[test] +fn net_shortening_reorg_drops_stale_blocks() { + let temp_dir = tempdir().unwrap(); + let (dbio, genesis) = dbio_with_genesis(temp_dir.path()); + + let block2a = produce_dummy_block(2, Some(genesis.header.hash), vec![]); + dbio.store_followed_block(&block2a, &state_with_balance(200), false) + .unwrap(); + let block3 = produce_dummy_block(3, Some(block2a.header.hash), vec![]); + dbio.store_followed_block(&block3, &state_with_balance(300), false) + .unwrap(); + + // A shorter competing chain wins: block 2 is replaced, block 3 gets no + // replacement. + let block2b = produce_dummy_block(2, Some(HashType([9; 32])), vec![]); + let head_tip = BlockMeta { + id: 2, + hash: block2b.header.hash, + }; + dbio.store_followed_blocks( + &[(&block2b, false)], + Some(&head_tip), + &state_with_balance(400), + None, + ) + .unwrap(); + + let stored2 = dbio.get_block(2).unwrap().expect("block 2 is stored"); + assert_eq!(stored2.header.hash, block2b.header.hash); + assert!( + dbio.get_block(3).unwrap().is_none(), + "stale block above the new head must be deleted, or restart replay panics on its broken link" + ); + let meta = dbio.latest_block_meta().unwrap(); + assert_eq!(meta.id, 2); + assert_eq!(meta.hash, block2b.header.hash); + assert_eq!(stored_balance(&dbio), 400); +} diff --git a/test_fixtures/src/setup.rs b/test_fixtures/src/setup.rs index d5094e14..e2081bc7 100644 --- a/test_fixtures/src/setup.rs +++ b/test_fixtures/src/setup.rs @@ -247,7 +247,8 @@ async fn setup_sequencer_inner( ) .context("Failed to create Sequencer config")?; - let sequencer_handle = sequencer_service::run(config, 0).await?; + let sequencer_handle = + sequencer_service::run(config, SocketAddr::from(([127, 0, 0, 1], 0))).await?; Ok((sequencer_handle, temp_sequencer_dir)) }