feat(sequencer): read channel_update and respect adopted/orphaned blocks

Cargo.lock

@@ -9038,6 +9038,7 @@ dependencies = [
  "borsh",
  "bridge_core",
  "bytesize",
+ "chain_state",
  "chrono",
  "common",
  "cross_zone",

lez/sequencer/core/Cargo.toml

@@ -10,6 +10,7 @@ workspace = true
 [dependencies]
 lee.workspace = true
 lee_core.workspace = true
+chain_state.workspace = true
 common.workspace = true
 storage.workspace = true
 mempool.workspace = true

lez/sequencer/core/src/block_publisher.rs

@@ -11,7 +11,7 @@ use logos_blockchain_zone_sdk::{
     CommonHttpClient,
     adapter::NodeHttpClient,
     sequencer::{
-        DepositInfo, Event, FinalizedOp, InscriptionInfo,
+        DepositInfo, Event, FinalizedOp, InscriptionInfo, OrphanedTx,
         SequencerConfig as ZoneSdkSequencerConfig, TurnNotification, WithdrawArg, WithdrawInfo,
         ZoneSequencer,
     },
@@ -45,6 +45,19 @@ pub type OnDepositEventSink =
 pub type OnWithdrawEventSink =
     Box<dyn Fn(WithdrawInfo) -> Pin<Box<dyn Future<Output = ()> + Send>> + Send + 'static>;
 
+/// The channel delta the follow path consumes from one `Event::BlocksProcessed`,
+/// with inscription payloads decoded into `(MsgId, Block)` pairs.
+pub struct FollowUpdate {
+    pub adopted: Vec<(MsgId, Block)>,
+    pub orphaned: Vec<(MsgId, Block)>,
+    pub finalized: Vec<(MsgId, Block)>,
+}
+
+/// Sink for the follow path: apply adopted/finalized blocks to chain state and
+/// revert orphaned ones.
+pub type OnFollowSink =
+    Box<dyn Fn(FollowUpdate) -> Pin<Box<dyn Future<Output = ()> + Send>> + Send + 'static>;
+
 #[expect(async_fn_in_trait, reason = "We don't care about Send/Sync here")]
 pub trait BlockPublisherTrait: Clone {
     #[expect(
@@ -60,6 +73,7 @@ pub trait BlockPublisherTrait: Clone {
         on_finalized_block: FinalizedBlockSink,
         on_deposit_event: OnDepositEventSink,
         on_withdraw_event: OnWithdrawEventSink,
+        on_follow: OnFollowSink,
     ) -> Result<Self>;
 
     /// Fire-and-forget publish. Zone-sdk drives the actual submission and
@@ -100,6 +114,7 @@ impl BlockPublisherTrait for ZoneSdkPublisher {
         on_finalized_block: FinalizedBlockSink,
         on_deposit_event: OnDepositEventSink,
         on_withdraw_event: OnWithdrawEventSink,
+        on_follow: OnFollowSink,
     ) -> Result<Self> {
         let basic_auth = config.auth.clone().map(Into::into);
         let node = NodeHttpClient::new(CommonHttpClient::new(basic_auth), config.node_url.clone());
@@ -167,17 +182,32 @@ impl BlockPublisherTrait for ZoneSdkPublisher {
                             match event {
                                 Event::BlocksProcessed {
                                     checkpoint,
-                                    channel_update: _,
+                                    channel_update,
                                     finalized,
                                 } => {
                                     on_checkpoint(checkpoint);
+
+                                    let adopted = channel_update
+                                        .adopted
+                                        .iter()
+                                        .filter_map(block_from_inscription)
+                                        .collect();
+                                    let orphaned = channel_update
+                                        .orphaned
+                                        .iter()
+                                        .map(orphan_inscription)
+                                        .filter_map(block_from_inscription)
+                                        .collect();
+
+                                    let mut finalized_blocks = Vec::new();
                                     for op in finalized.into_iter().flat_map(|item| item.ops) {
                                         match op {
                                             FinalizedOp::Inscription(inscription) => {
-                                                if let Some(block_id) =
+                                                if let Some((msg, block)) =
-                                                    block_id_from_inscription(&inscription)
+                                                    block_from_inscription(&inscription)
                                                 {
-                                                    on_finalized_block(block_id);
+                                                    on_finalized_block(block.header.block_id);
+                                                    finalized_blocks.push((msg, block));
                                                 }
                                             }
                                             FinalizedOp::Deposit(deposit) => {
@@ -188,6 +218,13 @@ impl BlockPublisherTrait for ZoneSdkPublisher {
                                             }
                                         }
                                     }
+
+                                    on_follow(FollowUpdate {
+                                        adopted,
+                                        orphaned,
+                                        finalized: finalized_blocks,
+                                    })
+                                    .await;
                                 }
                                 Event::Ready | Event::TurnNotification { .. } => {}
                             }
@@ -235,13 +272,21 @@ impl BlockPublisherTrait for ZoneSdkPublisher {
     }
 }
 
-/// Deserialize inscription payload as a `Block` and return it's`block_id`.
+/// Deserialize an inscription payload into `(this_msg, Block)`. Bad payloads are
-/// Bad payloads are logged and skipped.
+/// logged and skipped.
-fn block_id_from_inscription(inscription: &InscriptionInfo) -> Option<u64> {
+fn block_from_inscription(inscription: &InscriptionInfo) -> Option<(MsgId, Block)> {
     borsh::from_slice::<Block>(&inscription.payload)
         .inspect_err(|err| {
             warn!("Failed to deserialize block from inscription: {err:?}");
         })
         .ok()
-        .map(|block| block.header.block_id)
+        .map(|block| (inscription.this_msg, block))
+}
+
+/// The inscription carried by an orphaned tx (plain or atomic-withdraw bundle).
+const fn orphan_inscription(orphan: &OrphanedTx) -> &InscriptionInfo {
+    match orphan {
+        OrphanedTx::Inscription(info) => info,
+        OrphanedTx::AtomicWithdraw(bundle) => &bundle.inscription,
+    }
 }

lez/sequencer/core/src/lib.rs

@@ -1,7 +1,12 @@
-use std::{path::Path, sync::Arc, time::Instant};
+use std::{
+    path::Path,
+    sync::{Arc, Mutex},
+    time::Instant,
+};
 
 use anyhow::{Context as _, Result, anyhow};
 use borsh::BorshDeserialize;
+use chain_state::{ChainState, Tip};
 use common::{
     HashType,
     block::{BedrockStatus, Block, HashableBlockData},
@@ -13,7 +18,10 @@ use lee::{AccountId, PublicTransaction, public_transaction::Message};
 use lee_core::GENESIS_BLOCK_ID;
 use log::{error, info, warn};
 use logos_blockchain_key_management_system_service::keys::{ED25519_SECRET_KEY_SIZE, Ed25519Key};
-use logos_blockchain_zone_sdk::sequencer::{DepositInfo, WithdrawArg};
+use logos_blockchain_zone_sdk::{
+    Slot,
+    sequencer::{DepositInfo, WithdrawArg},
+};
 use mempool::{MemPool, MemPoolHandle};
 #[cfg(feature = "mock")]
 pub use mock::SequencerCoreWithMockClients;
@@ -59,6 +67,9 @@ impl DepositMetadata {
 
 pub struct SequencerCore<BP: BlockPublisherTrait = ZoneSdkPublisher> {
     state: lee::V03State,
+    /// Two-tier follow state fed by the publisher's `on_follow` sink. Shared with
+    /// the drive task. Passive today; production moves onto its head next.
+    chain: Arc<Mutex<ChainState>>,
     store: SequencerStore,
     mempool: MemPool<(TransactionOrigin, LeeTransaction)>,
     sequencer_config: SequencerConfig,
@@ -132,6 +143,14 @@ impl<BP: BlockPublisherTrait> SequencerCore<BP> {
             .latest_block_meta()
             .expect("Failed to read latest block meta from store");
 
+        let chain = Arc::new(Mutex::new(ChainState::from_final(
+            state.clone(),
+            Some(Tip {
+                block_id: latest_block_meta.id,
+                hash: latest_block_meta.hash,
+            }),
+        )));
+
         let initial_checkpoint = store
             .get_zone_checkpoint()
             .expect("Failed to load zone-sdk checkpoint");
@@ -149,6 +168,7 @@ impl<BP: BlockPublisherTrait> SequencerCore<BP> {
             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)),
         )
         .await
         .expect("Failed to initialize Block Publisher");
@@ -195,6 +215,7 @@ impl<BP: BlockPublisherTrait> SequencerCore<BP> {
 
         let sequencer_core = Self {
             state,
+            chain,
             store,
             mempool,
             chain_height: latest_block_meta.id,
@@ -335,6 +356,28 @@ impl<BP: BlockPublisherTrait> SequencerCore<BP> {
         })
     }
 
+    /// Feed one channel delta into the follow state: revert orphaned, apply
+    /// adopted, then finalize. Passive today — production still uses `self.state`.
+    fn on_follow(chain: Arc<Mutex<ChainState>>) -> block_publisher::OnFollowSink {
+        Box::new(move |update: block_publisher::FollowUpdate| {
+            let chain = Arc::clone(&chain);
+            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);
+                }
+                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));
+                }
+            })
+        })
+    }
+
     /// Produces a new block from mempool transactions and publishes it via zone-sdk.
     pub async fn produce_new_block(&mut self) -> Result<u64> {
         let BlockWithMeta {
@@ -571,6 +614,12 @@ impl<BP: BlockPublisherTrait> SequencerCore<BP> {
         self.block_publisher.is_our_turn()
     }
 
+    /// Shared handle to the two-tier follow state.
+    #[must_use]
+    pub fn chain(&self) -> Arc<Mutex<ChainState>> {
+        Arc::clone(&self.chain)
+    }
+
     fn next_block_id(&self) -> u64 {
         self.chain_height
             .checked_add(1)

lez/sequencer/core/src/mock.rs

@@ -8,7 +8,7 @@ use logos_blockchain_zone_sdk::sequencer::WithdrawArg;
 
 use crate::{
     block_publisher::{
-        BlockPublisherTrait, CheckpointSink, FinalizedBlockSink, OnDepositEventSink,
+        BlockPublisherTrait, CheckpointSink, FinalizedBlockSink, OnDepositEventSink, OnFollowSink,
         OnWithdrawEventSink, SequencerCheckpoint,
     },
     config::BedrockConfig,
@@ -31,6 +31,7 @@ impl BlockPublisherTrait for MockBlockPublisher {
         _on_finalized_block: FinalizedBlockSink,
         _on_deposit_event: OnDepositEventSink,
         _on_withdraw_event: OnWithdrawEventSink,
+        _on_follow: OnFollowSink,
     ) -> Result<Self> {
         Ok(Self {
             channel_id: config.channel_id,