From e2c4b486dcdaa351b4749dca819a650aedd730ec Mon Sep 17 00:00:00 2001 From: erhant Date: Tue, 14 Jul 2026 21:30:42 +0300 Subject: [PATCH] fix(sequencer): persist follow updates atomically in one write batch --- lez/sequencer/core/src/lib.rs | 37 +++++------ lez/sequencer/core/src/tests.rs | 50 +++++++++++++++ lez/storage/src/sequencer/mod.rs | 98 ++++++++++++++++++++++++------ lez/storage/src/sequencer/tests.rs | 30 +++++++++ 4 files changed, 176 insertions(+), 39 deletions(-) diff --git a/lez/sequencer/core/src/lib.rs b/lez/sequencer/core/src/lib.rs index cd693ee1..19bc7acf 100644 --- a/lez/sequencer/core/src/lib.rs +++ b/lez/sequencer/core/src/lib.rs @@ -719,23 +719,22 @@ async fn apply_follow_update( // 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, resubmit_txs, head_snapshot) = { + let (to_persist, resubmit_txs, head_snapshot) = { let mut chain = chain.lock().expect("chain state mutex poisoned"); 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(); + let mut to_persist = Vec::new(); for (this_msg, block) in &update.adopted { if matches!( chain.apply_adopted(*this_msg, block), AcceptOutcome::Applied ) { - adopted.push(block); + to_persist.push((block, false)); } } - let mut finalized = Vec::new(); for (this_msg, block) in &update.finalized { // TODO: thread the finalized inscription's L1 slot once the // sdk surfaces it; only used for the invalid-finalized stall. @@ -743,27 +742,23 @@ async fn apply_follow_update( chain.apply_finalized(*this_msg, block, Slot::from(0)), AcceptOutcome::Applied ) { - finalized.push(block); + to_persist.push((block, true)); } } - (adopted, finalized, resubmit_txs, chain.head_state().clone()) + (to_persist, resubmit_txs, chain.head_state().clone()) }; - for block in adopted { - if let Err(err) = dbio.store_followed_block(block, &head_snapshot, false) { - error!( - "Failed to persist adopted block {}: {err:#}", - block.header.block_id - ); - } - } - for block in finalized { - if let Err(err) = dbio.store_followed_block(block, &head_snapshot, true) { - error!( - "Failed to persist finalized block {}: {err:#}", - block.header.block_id - ); - } + // 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) { + error!("Failed to persist followed blocks: {err:#}"); } // Rebuild orphaned work: return its user txs to the mempool so the diff --git a/lez/sequencer/core/src/tests.rs b/lez/sequencer/core/src/tests.rs index 5c3630ce..965136ba 100644 --- a/lez/sequencer/core/src/tests.rs +++ b/lez/sequencer/core/src/tests.rs @@ -1621,3 +1621,53 @@ async fn record_produced_block_skips_persistence_when_block_no_longer_chains() { assert!(sequencer.store.get_block_at_id(2).unwrap().is_none()); assert_eq!(sequencer.chain_height(), 1, "head is unchanged"); } + +#[tokio::test] +async fn follow_update_persists_blocks_meta_and_state_atomically() { + let config = setup_sequencer_config(); + let (sequencer, mempool_handle) = SequencerCoreWithMockClients::start_from_config(config).await; + let genesis_meta = sequencer.store.latest_block_meta().unwrap(); + + let acc1 = initial_public_user_accounts()[0].account_id; + let acc2 = initial_public_user_accounts()[1].account_id; + let tx = common::test_utils::create_transaction_native_token_transfer( + acc1, + 0, + acc2, + 10, + &create_signing_key_for_account1(), + ); + let block2 = common::test_utils::produce_dummy_block(2, Some(genesis_meta.hash), vec![tx]); + let block3 = common::test_utils::produce_dummy_block(3, Some(block2.header.hash), vec![]); + + // One update carrying several blocks: both adopted, block 2 also finalized. + apply_follow_update( + sequencer.store.dbio(), + sequencer.chain(), + mempool_handle.clone(), + FollowUpdate { + adopted: vec![ + (MsgId::from([2; 32]), block2.clone()), + (MsgId::from([3; 32]), block3.clone()), + ], + orphaned: vec![], + finalized: vec![(MsgId::from([2; 32]), block2.clone())], + }, + ) + .await; + + // Blocks, tip meta and state all reflect the end of the batch: a late + // finalized entry for an earlier block must not drag the tip meta back. + let meta = sequencer.store.latest_block_meta().unwrap(); + assert_eq!(meta.id, 3); + assert_eq!(meta.hash, block3.header.hash); + let stored2 = sequencer.store.get_block_at_id(2).unwrap().unwrap(); + assert!(matches!(stored2.bedrock_status, BedrockStatus::Finalized)); + let stored_balance = sequencer + .store + .get_lee_state() + .unwrap() + .get_account_by_id(acc2) + .balance; + assert_eq!(stored_balance, 20010); +} diff --git a/lez/storage/src/sequencer/mod.rs b/lez/storage/src/sequencer/mod.rs index bd2364e9..3a7ab858 100644 --- a/lez/storage/src/sequencer/mod.rs +++ b/lez/storage/src/sequencer/mod.rs @@ -482,8 +482,6 @@ impl RocksDBIO { } pub fn put_block(&self, block: &Block, first: bool, batch: &mut WriteBatch) -> DbResult<()> { - let cf_block = self.block_column(); - if !first { let last_curr_block = self.get_meta_last_block_in_db()?; @@ -501,6 +499,12 @@ impl RocksDBIO { } } + self.put_block_payload(block, batch) + } + + /// 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(); batch.put_cf( &cf_block, borsh::to_vec(&block.header.block_id).map_err(|err| { @@ -615,31 +619,89 @@ impl RocksDBIO { } /// Persists a followed (peer) block so the store mirrors the canonical chain. - /// - /// Skips the write when the store already holds this block (matched by id and - /// hash) — covering our own blocks and re-deliveries — but still marks it - /// finalized when `finalized` is set. - /// - /// When `finalized`, the block is marked so backfilled blocks (which arrive - /// without a prior pending write) are not left pending. + /// One-block form of [`Self::store_followed_blocks`]. pub fn store_followed_block( &self, block: &Block, state: &V03State, finalized: bool, ) -> DbResult<()> { - let block_id = block.header.block_id; - let already_stored = self - .get_block(block_id)? - .is_some_and(|stored| stored.header.hash == block.header.hash); + self.store_followed_blocks(&[(block, finalized)], state) + } - if !already_stored { - self.atomic_update(block, &[], vec![], state)?; + /// Persists a **batch** of followed (peer) blocks and the state after the last + /// of them in one atomic write, so a crash can never leave the stored state + /// ahead of the stored blocks and tip meta. + /// + /// Per block, it: + /// - skips the payload write when the store already holds it (matched by id + /// and hash) thus covering our own blocks and re-deliveries + /// - stores it as finalized when its `finalized` flag is set + /// + /// When nothing needs writing, the state is left untouched too + /// + /// 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 by writing the checkpoint, blocks and + /// state in one batch, per the sdk's event contract is a follow-up. + pub fn store_followed_blocks( + &self, + blocks: &[(&Block, bool)], + state: &V03State, + ) -> 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 stored = self.get_block(block.header.block_id)?; + let already_stored = stored + .as_ref() + .is_some_and(|stored| stored.header.hash == block.header.hash); + + if already_stored && !finalized { + continue; + } + + let mut to_write = if already_stored { + stored.expect("`already_stored` implies a stored block") + } else { + (*block).clone() + }; + if *finalized { + 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 finalized { - self.mark_block_as_finalized(block_id)?; + + if batch.is_empty() { + return Ok(()); } - 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 + { + self.put_meta_last_block_in_db_batch(tip.id, &mut batch)?; + self.put_meta_latest_block_meta_batch(&tip, &mut batch)?; + } + self.put_lee_state_in_db_batch(state, &mut batch)?; + + self.db.write(batch).map_err(|rerr| { + DbError::rocksdb_cast_message( + rerr, + Some("Failed to write followed blocks batch".to_owned()), + ) + }) } pub fn get_all_blocks(&self) -> impl Iterator> { diff --git a/lez/storage/src/sequencer/tests.rs b/lez/storage/src/sequencer/tests.rs index ea9dc96a..2d86c494 100644 --- a/lez/storage/src/sequencer/tests.rs +++ b/lez/storage/src/sequencer/tests.rs @@ -87,6 +87,36 @@ fn store_followed_block_redelivery_is_a_noop_and_keeps_finalized() { ); } +#[test] +fn store_followed_blocks_batch_lands_meta_and_state_on_last_block() { + let temp_dir = tempdir().unwrap(); + let (dbio, genesis) = dbio_with_genesis(temp_dir.path()); + + // Block 2 is already stored (own production); one update then finalizes it + // and adopts block 3. + let block2 = produce_dummy_block(2, Some(genesis.header.hash), vec![]); + dbio.store_followed_block(&block2, &state_with_balance(200), false) + .unwrap(); + + let block3 = produce_dummy_block(3, Some(block2.header.hash), vec![]); + dbio.store_followed_blocks( + &[(&block2, true), (&block3, false)], + &state_with_balance(300), + ) + .unwrap(); + + let stored2 = dbio.get_block(2).unwrap().expect("block 2 is stored"); + assert!(matches!(stored2.bedrock_status, BedrockStatus::Finalized)); + let stored3 = dbio.get_block(3).unwrap().expect("block 3 is stored"); + assert!(matches!(stored3.bedrock_status, BedrockStatus::Pending)); + + // Meta and state land together on the last block of the batch. + let meta = dbio.latest_block_meta().unwrap(); + assert_eq!(meta.id, 3); + assert_eq!(meta.hash, block3.header.hash); + assert_eq!(stored_balance(&dbio), 300); +} + #[test] fn store_followed_block_overwrites_competing_block_at_same_id() { let temp_dir = tempdir().unwrap();