fix(sequencer, storage): persist under the chain lock, prune stale blocks on shortening reorgs, halt on persist failure

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

lez/sequencer/core/src/lib.rs

@@ -468,34 +468,30 @@ impl<BP: BlockPublisherTrait> SequencerCore<BP> {
         deposit_event_ids: &[HashType],
         withdrawal_reconciliation_keys: Vec<WithdrawalReconciliationKey>,
     ) -> 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<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())
+        .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<HashType> {
     let LeeTransaction::Public(tx) = tx else {

lez/sequencer/service/src/lib.rs

@@ -111,7 +111,7 @@ impl Drop for SequencerHandle {
     }
 }
 
-pub async fn run(config: SequencerConfig, port: u16) -> Result<SequencerHandle> {
+pub async fn run(config: SequencerConfig, listen_addr: SocketAddr) -> Result<SequencerHandle> {
     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<SequencerHandle>
     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<SequencerHandle>
 async fn run_server(
     sequencer: Arc<Mutex<SequencerCore>>,
     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")?;
 

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() => {

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<BlockMeta> = 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 {

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

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