fix(sequencer): persist follow updates atomically in one write batch

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

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);
+}

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<BlockMeta> = 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<Item = DbResult<Block>> {

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