test(chain_state,sequencer): add unit tests for chain_state

Cargo.lock

@@ -1373,6 +1373,7 @@ name = "chain_state"
 version = "0.1.0"
 dependencies = [
  "anyhow",
+ "borsh",
  "common",
  "lee",
  "logos-blockchain-core",

lez/chain_state/Cargo.toml

@@ -19,4 +19,7 @@ thiserror.workspace = true
 
 [dev-dependencies]
 testnet_initial_state.workspace = true
+# we use borsh to compare byte-to-byte matching within tests
+# (it sorts hashmap's before serialization, so we can compare two states for equality)
+borsh.workspace = true
 serde_json.workspace = true

lez/chain_state/src/chain.rs

@@ -18,14 +18,16 @@ pub struct HeadEntry {
 }
 
 /// The head tier (reorg-able, from `adopted`/`orphaned`) over the final tier
-/// (irreversible, from `finalized`). `head_state == final_state` replayed
-/// through `head_blocks`.
+/// (irreversible, from `finalized`).
+///
+/// `head_state` is given by `final_state` replayed through `head_blocks`.
 pub struct ChainState {
     final_state: V03State,
     final_tip: Option<Tip>,
+    final_stall: Option<StallReason>,
+
     head_state: V03State,
     head_blocks: Vec<HeadEntry>,
-    final_stall: Option<StallReason>,
 }
 
 impl ChainState {
@@ -248,7 +250,10 @@ const fn stall_for(block: &Block, l1_slot: Slot, error: BlockIngestError) -> Sta
 
 #[cfg(test)]
 mod tests {
-    use common::test_utils::{create_transaction_native_token_transfer, produce_dummy_block};
+    use common::{
+        HashType,
+        test_utils::{create_transaction_native_token_transfer, produce_dummy_block},
+    };
     use testnet_initial_state::{initial_pub_accounts_private_keys, initial_state};
 
     use super::*;
@@ -261,6 +266,21 @@ mod tests {
         Slot::from(n)
     }
 
+    /// `head_state` equals `final_state` replayed through `head_blocks`.
+    fn assert_head_matches_replay(chain: &ChainState) {
+        let mut state = chain.final_state.clone();
+        let mut tip = chain.final_tip.clone();
+        for entry in &chain.head_blocks {
+            apply_block(tip.as_ref(), &entry.block, &mut state).expect("head blocks must replay");
+            tip = Some(tip_of(&entry.block));
+        }
+        assert_eq!(
+            borsh::to_vec(&state).expect("state serializes"),
+            borsh::to_vec(chain.head_state()).expect("state serializes"),
+            "head_state must equal final_state replayed through head_blocks"
+        );
+    }
+
     #[test]
     fn adopted_blocks_advance_head() {
         let mut chain = ChainState::new(initial_state());
@@ -403,6 +423,238 @@ mod tests {
         assert_eq!(stall.block_id, Some(3));
     }
 
+    #[test]
+    fn orphaning_a_suffix_rederives_head_state() {
+        let accounts = initial_pub_accounts_private_keys();
+        let from = accounts[0].account_id;
+        let to = accounts[1].account_id;
+        let sign_key = accounts[0].pub_sign_key.clone();
+
+        let mut chain = ChainState::new(initial_state());
+        let genesis = produce_dummy_block(1, None, vec![]);
+        chain.apply_adopted(msg(1), &genesis);
+
+        let tx2 = create_transaction_native_token_transfer(from, 0, to, 10, &sign_key);
+        let block2 = produce_dummy_block(2, Some(genesis.header.hash), vec![tx2]);
+        chain.apply_adopted(msg(2), &block2);
+        let tx3 = create_transaction_native_token_transfer(from, 1, to, 10, &sign_key);
+        let block3 = produce_dummy_block(3, Some(block2.header.hash), vec![tx3]);
+        chain.apply_adopted(msg(3), &block3);
+        let tx4 = create_transaction_native_token_transfer(from, 2, to, 10, &sign_key);
+        let block4 = produce_dummy_block(4, Some(block3.header.hash), vec![tx4]);
+        chain.apply_adopted(msg(4), &block4);
+
+        // Orphaning block 3 drops the whole suffix (3 and 4).
+        chain.revert_orphan(msg(3));
+
+        assert_eq!(chain.head_tip().expect("head tip").block_id, 2);
+        assert_eq!(chain.head_state().get_account_by_id(from).balance, 9990);
+        assert_eq!(chain.head_state().get_account_by_id(to).balance, 20010);
+        assert_head_matches_replay(&chain);
+    }
+
+    #[test]
+    fn channel_update_replaces_multi_block_suffix() {
+        let accounts = initial_pub_accounts_private_keys();
+        let from = accounts[0].account_id;
+        let to = accounts[1].account_id;
+        let sign_key = accounts[0].pub_sign_key.clone();
+
+        let mut chain = ChainState::new(initial_state());
+        let genesis = produce_dummy_block(1, None, vec![]);
+        chain.apply_adopted(msg(1), &genesis);
+
+        let tx2 = create_transaction_native_token_transfer(from, 0, to, 10, &sign_key);
+        let block2 = produce_dummy_block(2, Some(genesis.header.hash), vec![tx2]);
+        chain.apply_adopted(msg(2), &block2);
+        let tx3 = create_transaction_native_token_transfer(from, 1, to, 10, &sign_key);
+        let block3 = produce_dummy_block(3, Some(block2.header.hash), vec![tx3]);
+        chain.apply_adopted(msg(3), &block3);
+        let tx4 = create_transaction_native_token_transfer(from, 2, to, 10, &sign_key);
+        let block4 = produce_dummy_block(4, Some(block3.header.hash), vec![tx4]);
+        chain.apply_adopted(msg(4), &block4);
+
+        // A competing branch replaces blocks 3 and 4; orphans arrive unordered.
+        let tx3_prime = create_transaction_native_token_transfer(from, 1, to, 20, &sign_key);
+        let block3_prime = produce_dummy_block(3, Some(block2.header.hash), vec![tx3_prime]);
+        let tx4_prime = create_transaction_native_token_transfer(from, 2, to, 30, &sign_key);
+        let block4_prime = produce_dummy_block(4, Some(block3_prime.header.hash), vec![tx4_prime]);
+
+        let outcomes = chain.apply_channel_update(
+            &[msg(4), msg(3)],
+            &[(msg(13), block3_prime), (msg(14), block4_prime)],
+        );
+
+        assert!(matches!(
+            outcomes.as_slice(),
+            [AcceptOutcome::Applied, AcceptOutcome::Applied]
+        ));
+        assert_eq!(chain.head_tip().expect("head tip").block_id, 4);
+        assert_eq!(chain.head_state().get_account_by_id(from).balance, 9940);
+        assert_eq!(chain.head_state().get_account_by_id(to).balance, 20060);
+        assert_head_matches_replay(&chain);
+    }
+
+    #[test]
+    fn channel_update_ignores_unknown_orphan() {
+        let mut chain = ChainState::new(initial_state());
+        let genesis = produce_dummy_block(1, None, vec![]);
+        let block2 = produce_dummy_block(2, Some(genesis.header.hash), vec![]);
+        chain.apply_adopted(msg(1), &genesis);
+        chain.apply_adopted(msg(2), &block2);
+
+        let block3 = produce_dummy_block(3, Some(block2.header.hash), vec![]);
+        let outcomes = chain.apply_channel_update(&[msg(99)], &[(msg(3), block3)]);
+
+        assert!(matches!(outcomes.as_slice(), [AcceptOutcome::Applied]));
+        assert_eq!(chain.head_tip().expect("head tip").block_id, 3);
+        assert_head_matches_replay(&chain);
+    }
+
+    #[test]
+    fn finalized_reinscription_matches_by_block_hash() {
+        let mut chain = ChainState::new(initial_state());
+        let genesis = produce_dummy_block(1, None, vec![]);
+        let block2 = produce_dummy_block(2, Some(genesis.header.hash), vec![]);
+        let block3 = produce_dummy_block(3, Some(block2.header.hash), vec![]);
+        chain.apply_adopted(msg(1), &genesis);
+        chain.apply_adopted(msg(2), &block2);
+        chain.apply_adopted(msg(3), &block3);
+
+        // Block 2 finalizes re-inscribed under a fresh MsgId: matched by hash,
+        // finalized through, and the head above it survives.
+        assert!(matches!(
+            chain.apply_finalized(msg(42), &block2, slot(5)),
+            AcceptOutcome::Applied
+        ));
+        assert_eq!(chain.final_tip().expect("final tip").block_id, 2);
+        assert_eq!(chain.head_tip().expect("head tip").block_id, 3);
+        assert!(chain.final_stall().is_none());
+        assert_head_matches_replay(&chain);
+    }
+
+    #[test]
+    fn finalize_through_preserves_head_state_and_advances_final_state() {
+        let accounts = initial_pub_accounts_private_keys();
+        let from = accounts[0].account_id;
+        let to = accounts[1].account_id;
+        let sign_key = accounts[0].pub_sign_key.clone();
+
+        let mut chain = ChainState::new(initial_state());
+        let genesis = produce_dummy_block(1, None, vec![]);
+        chain.apply_adopted(msg(1), &genesis);
+        let tx2 = create_transaction_native_token_transfer(from, 0, to, 10, &sign_key);
+        let block2 = produce_dummy_block(2, Some(genesis.header.hash), vec![tx2]);
+        chain.apply_adopted(msg(2), &block2);
+        let tx3 = create_transaction_native_token_transfer(from, 1, to, 10, &sign_key);
+        let block3 = produce_dummy_block(3, Some(block2.header.hash), vec![tx3]);
+        chain.apply_adopted(msg(3), &block3);
+
+        chain.apply_finalized(msg(2), &block2, slot(10));
+
+        // Head still reflects both transfers
+        assert_eq!(chain.head_state().get_account_by_id(to).balance, 20020);
+        // ...while final reflects only the finalized prefix.
+        assert_eq!(chain.final_state().get_account_by_id(to).balance, 20010);
+        assert_head_matches_replay(&chain);
+    }
+
+    #[test]
+    fn head_self_heals_with_valid_competitor_after_park() {
+        let mut chain = ChainState::new(initial_state());
+        let genesis = produce_dummy_block(1, None, vec![]);
+        chain.apply_adopted(msg(1), &genesis);
+
+        // Correct id, wrong parent: parked, head frozen at 1, no stall.
+        let bad = produce_dummy_block(2, Some(HashType([9; 32])), vec![]);
+        assert!(matches!(
+            chain.apply_adopted(msg(2), &bad),
+            AcceptOutcome::Parked(BlockIngestError::BrokenChainLink { .. })
+        ));
+        assert_eq!(chain.head_tip().expect("head tip").block_id, 1);
+        assert!(chain.final_stall().is_none());
+
+        // A valid competitor at the same height applies without any reorg event.
+        let good = produce_dummy_block(2, Some(genesis.header.hash), vec![]);
+        assert!(matches!(
+            chain.apply_adopted(msg(12), &good),
+            AcceptOutcome::Applied
+        ));
+        assert_eq!(chain.head_tip().expect("head tip").block_id, 2);
+        assert_head_matches_replay(&chain);
+    }
+
+    #[test]
+    fn repeated_invalid_finalized_bumps_orphans_since() {
+        let mut chain = ChainState::new(initial_state());
+        let genesis = produce_dummy_block(1, None, vec![]);
+        chain.apply_finalized(msg(1), &genesis, slot(10));
+
+        let bad3 = produce_dummy_block(3, Some(genesis.header.hash), vec![]);
+        chain.apply_finalized(msg(3), &bad3, slot(20));
+        let bad5 = produce_dummy_block(5, Some(bad3.header.hash), vec![]);
+        assert!(matches!(
+            chain.apply_finalized(msg(5), &bad5, slot(30)),
+            AcceptOutcome::Parked(_)
+        ));
+
+        let stall = chain.final_stall().expect("final stall recorded");
+        assert_eq!(stall.block_id, Some(3), "first stall reason is preserved");
+        assert_eq!(stall.orphans_since, 1);
+    }
+
+    #[test]
+    fn valid_finalized_successor_clears_final_stall() {
+        let mut chain = ChainState::new(initial_state());
+        let genesis = produce_dummy_block(1, None, vec![]);
+        chain.apply_finalized(msg(1), &genesis, slot(10));
+
+        let bad = produce_dummy_block(3, Some(genesis.header.hash), vec![]);
+        chain.apply_finalized(msg(3), &bad, slot(20));
+        assert!(chain.final_stall().is_some());
+
+        // The valid successor of the frozen final tip finalizes: stall clears.
+        let block2 = produce_dummy_block(2, Some(genesis.header.hash), vec![]);
+        assert!(matches!(
+            chain.apply_finalized(msg(2), &block2, slot(30)),
+            AcceptOutcome::Applied
+        ));
+        assert!(chain.final_stall().is_none());
+        assert_eq!(chain.final_tip().expect("final tip").block_id, 2);
+        assert_head_matches_replay(&chain);
+    }
+
+    #[test]
+    fn finalized_unknown_block_rebases_head() {
+        let accounts = initial_pub_accounts_private_keys();
+        let from = accounts[0].account_id;
+        let to = accounts[1].account_id;
+        let sign_key = accounts[0].pub_sign_key.clone();
+
+        let mut chain = ChainState::new(initial_state());
+        let genesis = produce_dummy_block(1, None, vec![]);
+        chain.apply_adopted(msg(1), &genesis);
+        chain.apply_finalized(msg(1), &genesis, slot(10));
+
+        // Head advances on a competing branch…
+        let block2a = produce_dummy_block(2, Some(genesis.header.hash), vec![]);
+        chain.apply_adopted(msg(2), &block2a);
+
+        // …but a different block 2 finalizes. The finalized chain is
+        // authoritative, so head rebases onto it.
+        let tx = create_transaction_native_token_transfer(from, 0, to, 10, &sign_key);
+        let block2b = produce_dummy_block(2, Some(genesis.header.hash), vec![tx]);
+        assert!(matches!(
+            chain.apply_finalized(msg(22), &block2b, slot(20)),
+            AcceptOutcome::Applied
+        ));
+
+        assert_eq!(chain.final_tip().expect("final tip").block_id, 2);
+        assert_eq!(chain.head_tip().expect("head tip").block_id, 2);
+        assert_eq!(chain.head_state().get_account_by_id(to).balance, 20010);
+        assert_head_matches_replay(&chain);
+    }
+
     #[test]
     fn head_state_reflects_applied_transfers() {
         let accounts = initial_pub_accounts_private_keys();

lez/sequencer/core/src/lib.rs

@@ -352,83 +352,19 @@ impl<BP: BlockPublisherTrait> SequencerCore<BP> {
         })
     }
 
-    /// 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.
-    ///
-    /// TODO: unlike the indexer's ingest loop, this path does not retry
-    /// `is_retryable` (transient) apply failures — a failed block just parks and
-    /// relies on a valid successor or a restart. `ChainState` never emits
-    /// `AcceptOutcome::RetryableFailure` yet; adding retry parity here is a
-    /// follow-up.
+    /// Publisher sink adapter over [`apply_follow_update`].
     fn on_follow(
         dbio: Arc<RocksDBIO>,
         chain: Arc<Mutex<ChainState>>,
         mempool_handle: MemPoolHandle<(TransactionOrigin, LeeTransaction)>,
     ) -> block_publisher::OnFollowSink {
         Box::new(move |update: block_publisher::FollowUpdate| {
-            let chain = Arc::clone(&chain);
-            let dbio = Arc::clone(&dbio);
-            let mempool_handle = mempool_handle.clone();
-            Box::pin(async move {
-                // 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 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();
-                    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, 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
-                        );
-                    }
-                }
-
-                // Rebuild orphaned work: return its user txs to the mempool so the
-                // next on-turn production re-includes them on the new head.
-                for tx in resubmit_txs {
-                    if let Err(err) = mempool_handle.push((TransactionOrigin::User, tx)).await {
-                        error!("Failed to resubmit orphaned transaction: {err:#}");
-                    }
-                }
-            })
+            Box::pin(apply_follow_update(
+                Arc::clone(&dbio),
+                Arc::clone(&chain),
+                mempool_handle.clone(),
+                update,
+            ))
         })
     }
 
@@ -702,6 +638,81 @@ struct BlockWithMeta {
     withdrawals: Vec<WithdrawArg>,
 }
 
+/// 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. Wired to the publisher via
+/// [`SequencerCore::on_follow`]; a free function so tests can drive it directly.
+///
+/// TODO: unlike the indexer's ingest loop, this path does not retry
+/// `is_retryable` (transient) apply failures — a failed block just parks and
+/// relies on a valid successor or a restart. `ChainState` never emits
+/// `AcceptOutcome::RetryableFailure` yet; adding retry parity here is a
+/// follow-up.
+async fn apply_follow_update(
+    dbio: Arc<RocksDBIO>,
+    chain: Arc<Mutex<ChainState>>,
+    mempool_handle: MemPoolHandle<(TransactionOrigin, LeeTransaction)>,
+    update: block_publisher::FollowUpdate,
+) {
+    // 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 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();
+        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, 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
+            );
+        }
+    }
+
+    // Rebuild orphaned work: return its user txs to the mempool so the
+    // next on-turn production re-includes them on the new head.
+    for tx in resubmit_txs {
+        if let Err(err) = mempool_handle.push((TransactionOrigin::User, tx)).await {
+            error!("Failed to resubmit orphaned transaction: {err:#}");
+        }
+    }
+}
+
 /// Checks the database for any pending deposit events that have not yet been marked as submitted in
 /// a block, and re-queues them in the mempool in a separate async task for inclusion in the next
 /// block.

lez/sequencer/core/src/tests.rs

@@ -4,7 +4,7 @@ use std::{pin::pin, time::Duration};
 
 use common::{
     HashType,
-    block::HashableBlockData,
+    block::{BedrockStatus, HashableBlockData},
     test_utils::sequencer_sign_key_for_testing,
     transaction::{LeeTransaction, clock_invocation},
 };
@@ -22,19 +22,21 @@ use lee_core::{
     account::{AccountWithMetadata, Nonce},
     program::PdaSeed,
 };
-use logos_blockchain_core::mantle::ops::channel::ChannelId;
+use logos_blockchain_core::mantle::ops::channel::{ChannelId, MsgId};
 use mempool::MemPoolHandle;
 use storage::sequencer::sequencer_cells::PendingDepositEventRecord;
 use tempfile::tempdir;
 use testnet_initial_state::{initial_pub_accounts_private_keys, initial_public_user_accounts};
 
 use crate::{
-    TransactionOrigin,
+    TransactionOrigin, apply_follow_update,
+    block_publisher::FollowUpdate,
     block_store::SequencerStore,
-    build_genesis_state,
+    build_bridge_deposit_tx_from_event, build_genesis_state,
     config::{BedrockConfig, SequencerConfig},
     is_sequencer_only_program,
     mock::SequencerCoreWithMockClients,
+    resubmittable_txs,
 };
 
 #[derive(borsh::BorshSerialize)]
@@ -1271,3 +1273,273 @@ fn pda_mechanism_with_pinata_token_program() {
         expected_winner_token_holding_post
     );
 }
+
+#[test]
+fn resubmittable_txs_drops_clock_and_bridge_deposits() {
+    let user_tx = common::test_utils::produce_dummy_empty_transaction();
+    let deposit_tx = build_bridge_deposit_tx_from_event(&PendingDepositEventRecord {
+        deposit_op_id: HashType([13; 32]),
+        source_tx_hash: HashType([7; 32]),
+        amount: 1,
+        metadata: borsh::to_vec(&DepositMetadataForEncoding {
+            recipient_id: initial_public_user_accounts()[0].account_id,
+        })
+        .unwrap(),
+        submitted_in_block_id: None,
+    })
+    .unwrap();
+    let withdraw_tx = {
+        let message = lee::public_transaction::Message::try_new(
+            programs::bridge().id(),
+            vec![system_accounts::bridge_account_id()],
+            vec![],
+            bridge_core::Instruction::Withdraw {
+                amount: 1,
+                bedrock_account_pk: [0; 32],
+            },
+        )
+        .unwrap();
+        LeeTransaction::Public(PublicTransaction::new(
+            message,
+            lee::public_transaction::WitnessSet::from_raw_parts(vec![]),
+        ))
+    };
+
+    let block = common::test_utils::produce_dummy_block(
+        2,
+        Some(HashType([1; 32])),
+        vec![user_tx.clone(), deposit_tx, withdraw_tx.clone()],
+    );
+
+    // The trailing clock tx and the sequencer-generated deposit are dropped;
+    // user txs (withdrawals included) are returned.
+    assert_eq!(resubmittable_txs(&block), vec![user_tx, withdraw_tx]);
+}
+
+#[test]
+fn resubmittable_txs_of_blocks_without_user_txs_is_empty() {
+    // No transactions at all (not even the mandatory clock tx).
+    let empty = HashableBlockData {
+        block_id: 1,
+        prev_block_hash: HashType([0; 32]),
+        timestamp: 0,
+        transactions: vec![],
+    }
+    .into_pending_block(&sequencer_sign_key_for_testing());
+    assert!(resubmittable_txs(&empty).is_empty());
+
+    let clock_only = common::test_utils::produce_dummy_block(1, None, vec![]);
+    assert!(resubmittable_txs(&clock_only).is_empty());
+}
+
+#[tokio::test]
+async fn follow_adopted_peer_block_applies_and_persists() {
+    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 peer_block = common::test_utils::produce_dummy_block(2, Some(genesis_meta.hash), vec![tx]);
+
+    apply_follow_update(
+        sequencer.store.dbio(),
+        sequencer.chain(),
+        mempool_handle.clone(),
+        FollowUpdate {
+            adopted: vec![(MsgId::from([1; 32]), peer_block.clone())],
+            orphaned: vec![],
+            finalized: vec![],
+        },
+    )
+    .await;
+
+    assert_eq!(sequencer.chain_height(), 2);
+    let stored = sequencer
+        .store
+        .get_block_at_id(2)
+        .unwrap()
+        .expect("adopted peer block should be persisted");
+    assert_eq!(stored.header.hash, peer_block.header.hash);
+    assert_eq!(
+        sequencer.with_state(|s| s.get_account_by_id(acc2).balance),
+        20010
+    );
+}
+
+#[tokio::test]
+async fn follow_redelivery_of_own_block_is_deduped() {
+    let config = setup_sequencer_config();
+    let (mut sequencer, mempool_handle) =
+        SequencerCoreWithMockClients::start_from_config(config).await;
+
+    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(),
+    );
+    mempool_handle
+        .push((TransactionOrigin::User, tx))
+        .await
+        .unwrap();
+    sequencer.produce_new_block().await.unwrap();
+    let block2 = sequencer.store.get_block_at_id(2).unwrap().unwrap();
+
+    // The channel redelivers our own block under the MsgId the mock publisher
+    // assigned at publish time.
+    apply_follow_update(
+        sequencer.store.dbio(),
+        sequencer.chain(),
+        mempool_handle.clone(),
+        FollowUpdate {
+            adopted: vec![(MsgId::from(block2.header.hash.0), block2.clone())],
+            orphaned: vec![],
+            finalized: vec![],
+        },
+    )
+    .await;
+
+    assert_eq!(sequencer.chain_height(), 2);
+    assert_eq!(
+        sequencer.with_state(|s| s.get_account_by_id(acc2).balance),
+        20010,
+        "the transfer must not be double-applied"
+    );
+}
+
+#[tokio::test]
+async fn follow_orphan_reverts_head_and_requeues_user_txs() {
+    let config = setup_sequencer_config();
+    let (mut sequencer, mempool_handle) =
+        SequencerCoreWithMockClients::start_from_config(config).await;
+
+    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(),
+    );
+    mempool_handle
+        .push((TransactionOrigin::User, tx.clone()))
+        .await
+        .unwrap();
+    sequencer.produce_new_block().await.unwrap();
+    let block2 = sequencer.store.get_block_at_id(2).unwrap().unwrap();
+
+    apply_follow_update(
+        sequencer.store.dbio(),
+        sequencer.chain(),
+        mempool_handle.clone(),
+        FollowUpdate {
+            adopted: vec![],
+            orphaned: vec![(MsgId::from(block2.header.hash.0), block2.clone())],
+            finalized: vec![],
+        },
+    )
+    .await;
+
+    assert_eq!(sequencer.chain_height(), 1);
+    assert_eq!(
+        sequencer.with_state(|s| s.get_account_by_id(acc1).balance),
+        10000,
+        "the orphaned transfer must be reverted from the head"
+    );
+    let (origin, requeued) = sequencer
+        .mempool
+        .pop()
+        .expect("orphaned user tx should be requeued");
+    assert!(matches!(origin, TransactionOrigin::User));
+    assert_eq!(requeued, tx);
+    assert!(
+        sequencer.mempool.pop().is_none(),
+        "the clock tx must not be requeued"
+    );
+}
+
+#[tokio::test]
+async fn follow_finalized_own_block_moves_final_tier_and_marks_store() {
+    let config = setup_sequencer_config();
+    let (mut sequencer, mempool_handle) =
+        SequencerCoreWithMockClients::start_from_config(config).await;
+
+    let tx = common::test_utils::produce_dummy_empty_transaction();
+    mempool_handle
+        .push((TransactionOrigin::User, tx))
+        .await
+        .unwrap();
+    sequencer.produce_new_block().await.unwrap();
+    let block2 = sequencer.store.get_block_at_id(2).unwrap().unwrap();
+
+    apply_follow_update(
+        sequencer.store.dbio(),
+        sequencer.chain(),
+        mempool_handle.clone(),
+        FollowUpdate {
+            adopted: vec![],
+            orphaned: vec![],
+            finalized: vec![(MsgId::from(block2.header.hash.0), block2.clone())],
+        },
+    )
+    .await;
+
+    let final_tip = sequencer
+        .chain()
+        .lock()
+        .expect("chain mutex poisoned")
+        .final_tip()
+        .expect("final tip set");
+    assert_eq!(final_tip.block_id, 2);
+    assert_eq!(sequencer.chain_height(), 2, "head is unchanged");
+    let stored = sequencer.store.get_block_at_id(2).unwrap().unwrap();
+    assert!(matches!(stored.bedrock_status, BedrockStatus::Finalized));
+}
+
+#[tokio::test]
+async fn follow_finalized_backfill_block_is_applied_and_marked_finalized() {
+    let config = setup_sequencer_config();
+    let (sequencer, mempool_handle) = SequencerCoreWithMockClients::start_from_config(config).await;
+    let genesis_meta = sequencer.store.latest_block_meta().unwrap();
+
+    // A peer block we never saw as adopted arrives straight from the
+    // finalized (backfill) stream.
+    let peer_block = common::test_utils::produce_dummy_block(2, Some(genesis_meta.hash), vec![]);
+
+    apply_follow_update(
+        sequencer.store.dbio(),
+        sequencer.chain(),
+        mempool_handle.clone(),
+        FollowUpdate {
+            adopted: vec![],
+            orphaned: vec![],
+            finalized: vec![(MsgId::from([2; 32]), peer_block.clone())],
+        },
+    )
+    .await;
+
+    assert_eq!(
+        sequencer.chain_height(),
+        2,
+        "head mirrors final on backfill"
+    );
+    let stored = sequencer
+        .store
+        .get_block_at_id(2)
+        .unwrap()
+        .expect("backfilled block should be persisted");
+    assert_eq!(stored.header.hash, peer_block.header.hash);
+    assert!(matches!(stored.bedrock_status, BedrockStatus::Finalized));
+}

lez/storage/src/sequencer/mod.rs

@@ -689,3 +689,6 @@ impl RocksDBIO {
         })
     }
 }
+
+#[cfg(test)]
+mod tests;

lez/storage/src/sequencer/tests.rs

@@ -0,0 +1,108 @@
+use common::test_utils::produce_dummy_block;
+use lee::{Account, AccountId};
+use tempfile::tempdir;
+
+use super::*;
+
+fn marker_id() -> AccountId {
+    AccountId::new([1; 32])
+}
+
+/// A state distinguishable by the marker account's balance, so tests can tell
+/// which snapshot a write persisted.
+///
+/// TODO: is this a bit too much of a hot-fix for test snapshot?
+fn state_with_balance(balance: u128) -> V03State {
+    V03State::new().with_public_accounts([(
+        marker_id(),
+        Account {
+            balance,
+            ..Account::default()
+        },
+    )])
+}
+
+fn dbio_with_genesis(path: &Path) -> (RocksDBIO, Block) {
+    let genesis = produce_dummy_block(1, None, vec![]);
+    let dbio = RocksDBIO::create(path, &genesis, &state_with_balance(100)).unwrap();
+    (dbio, genesis)
+}
+
+fn stored_balance(dbio: &RocksDBIO) -> u128 {
+    dbio.get_lee_state()
+        .unwrap()
+        .get_account_by_id(marker_id())
+        .balance
+}
+
+#[test]
+fn store_followed_block_persists_new_block_and_state() {
+    let temp_dir = tempdir().unwrap();
+    let (dbio, genesis) = dbio_with_genesis(temp_dir.path());
+
+    let block2 = produce_dummy_block(2, Some(genesis.header.hash), vec![]);
+    dbio.store_followed_block(&block2, &state_with_balance(200), false)
+        .unwrap();
+
+    let stored = dbio.get_block(2).unwrap().expect("block 2 is stored");
+    assert_eq!(stored.header.hash, block2.header.hash);
+    assert!(matches!(stored.bedrock_status, BedrockStatus::Pending));
+    assert_eq!(dbio.latest_block_meta().unwrap().id, 2);
+    assert_eq!(stored_balance(&dbio), 200);
+}
+
+#[test]
+fn store_followed_block_finalized_marks_block() {
+    let temp_dir = tempdir().unwrap();
+    let (dbio, genesis) = dbio_with_genesis(temp_dir.path());
+
+    let block2 = produce_dummy_block(2, Some(genesis.header.hash), vec![]);
+    dbio.store_followed_block(&block2, &state_with_balance(200), true)
+        .unwrap();
+
+    let stored = dbio.get_block(2).unwrap().expect("block 2 is stored");
+    assert!(matches!(stored.bedrock_status, BedrockStatus::Finalized));
+}
+
+#[test]
+fn store_followed_block_redelivery_is_a_noop_and_keeps_finalized() {
+    let temp_dir = tempdir().unwrap();
+    let (dbio, genesis) = dbio_with_genesis(temp_dir.path());
+
+    let block2 = produce_dummy_block(2, Some(genesis.header.hash), vec![]);
+    dbio.store_followed_block(&block2, &state_with_balance(200), true)
+        .unwrap();
+    dbio.store_followed_block(&block2, &state_with_balance(300), false)
+        .unwrap();
+
+    let stored = dbio.get_block(2).unwrap().expect("block 2 is stored");
+    assert!(
+        matches!(stored.bedrock_status, BedrockStatus::Finalized),
+        "re-delivery must not demote a finalized block"
+    );
+    assert_eq!(
+        stored_balance(&dbio),
+        200,
+        "re-delivery must not overwrite the persisted state"
+    );
+}
+
+#[test]
+fn store_followed_block_overwrites_competing_block_at_same_id() {
+    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();
+
+    // A reorg replaces block 2: the competing block wins the slot.
+    let block2b = produce_dummy_block(2, Some(HashType([9; 32])), vec![]);
+    dbio.store_followed_block(&block2b, &state_with_balance(300), false)
+        .unwrap();
+
+    let stored = dbio.get_block(2).unwrap().expect("block 2 is stored");
+    assert_eq!(stored.header.hash, block2b.header.hash);
+    assert!(matches!(stored.bedrock_status, BedrockStatus::Pending));
+    assert_eq!(stored_balance(&dbio), 300);
+}