diff --git a/lez/sequencer/core/src/lib.rs b/lez/sequencer/core/src/lib.rs index 779aa690..cfeb060d 100644 --- a/lez/sequencer/core/src/lib.rs +++ b/lez/sequencer/core/src/lib.rs @@ -6,7 +6,7 @@ use std::{ use anyhow::{Context as _, Result, anyhow}; use borsh::BorshDeserialize; -use chain_state::{ChainState, Tip}; +use chain_state::{AcceptOutcome, ChainState, Tip}; use common::{ HashType, block::{BedrockStatus, Block, HashableBlockData}, @@ -165,7 +165,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(Arc::clone(&chain)), + Self::on_follow(store.dbio(), Arc::clone(&chain)), ) .await .expect("Failed to initialize Block Publisher"); @@ -340,27 +340,63 @@ impl SequencerCore { }) } - /// Feed one channel delta into the follow state: revert orphaned, apply - /// adopted, then finalize. Production builds on this same head. - /// - /// TODO: persist adopted/finalized peer blocks to the store here so restart and RPC see the - /// full canonical chain, not just our own blocks. Open question: will peers communicate in - /// another way? - fn on_follow(chain: Arc>) -> block_publisher::OnFollowSink { + /// Feed one channel delta into the follow state and mirror it to the store: + /// revert orphaned, then apply and persist adopted and finalized blocks. + /// Production builds on this same head. + fn on_follow( + dbio: Arc, + chain: Arc>, + ) -> block_publisher::OnFollowSink { Box::new(move |update: block_publisher::FollowUpdate| { let chain = Arc::clone(&chain); + let dbio = Arc::clone(&dbio); Box::pin(async move { - let mut chain = chain.lock().expect("chain state mutex poisoned"); - for (this_msg, _) in &update.orphaned { - chain.revert_orphan(*this_msg); + // 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 mut chain = chain.lock().expect("chain state mutex poisoned"); + for (this_msg, _) in &update.orphaned { + chain.revert_orphan(*this_msg); + } + let mut adopted = Vec::new(); + for (this_msg, block) in &update.adopted { + if matches!( + chain.apply_adopted(*this_msg, block), + AcceptOutcome::Applied + ) { + adopted.push(block); + } + } + 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. + if matches!( + chain.apply_finalized(*this_msg, block, Slot::from(0)), + AcceptOutcome::Applied + ) { + finalized.push(block); + } + } + (adopted, finalized, 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 (this_msg, block) in &update.adopted { - chain.apply_adopted(*this_msg, block); - } - for (this_msg, block) in &update.finalized { - // TODO: thread the finalized inscription's L1 slot once the sdk - // surfaces it; only used for the rare invalid-finalized stall. - chain.apply_finalized(*this_msg, block, Slot::from(0)); + 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 + ); + } } }) }) diff --git a/lez/storage/src/sequencer/mod.rs b/lez/storage/src/sequencer/mod.rs index 1a3a1886..1e2feca2 100644 --- a/lez/storage/src/sequencer/mod.rs +++ b/lez/storage/src/sequencer/mod.rs @@ -612,6 +612,34 @@ impl RocksDBIO { Ok(()) } + /// 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. + 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); + + if !already_stored { + self.atomic_update(block, &[], vec![], state)?; + } + if finalized { + self.mark_block_as_finalized(block_id)?; + } + Ok(()) + } + pub fn get_all_blocks(&self) -> impl Iterator> { let cf_block = self.block_column(); self.db