From 749113499443362b67d2c5eb0c7c582278c94f83 Mon Sep 17 00:00:00 2001 From: erhant Date: Fri, 10 Jul 2026 11:08:26 +0300 Subject: [PATCH] feat(sequencer): read `channel_update` and respect `adopted/orphaned` blocks --- Cargo.lock | 1 + lez/sequencer/core/Cargo.toml | 1 + lez/sequencer/core/src/block_publisher.rs | 63 +++++++++++++++++++---- lez/sequencer/core/src/lib.rs | 53 ++++++++++++++++++- lez/sequencer/core/src/mock.rs | 3 +- 5 files changed, 109 insertions(+), 12 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 65e0cde7..94adec54 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -8911,6 +8911,7 @@ dependencies = [ "borsh", "bridge_core", "bytesize", + "chain_state", "chrono", "common", "faucet_core", diff --git a/lez/sequencer/core/Cargo.toml b/lez/sequencer/core/Cargo.toml index 2931c833..61af29fa 100644 --- a/lez/sequencer/core/Cargo.toml +++ b/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 diff --git a/lez/sequencer/core/src/block_publisher.rs b/lez/sequencer/core/src/block_publisher.rs index f8a64c5d..5654dd34 100644 --- a/lez/sequencer/core/src/block_publisher.rs +++ b/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 Pin + 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 Pin + 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; /// 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 { 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) = - block_id_from_inscription(&inscription) + if let Some((msg, block)) = + 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`. -/// Bad payloads are logged and skipped. -fn block_id_from_inscription(inscription: &InscriptionInfo) -> Option { +/// Deserialize an inscription payload into `(this_msg, Block)`. Bad payloads are +/// logged and skipped. +fn block_from_inscription(inscription: &InscriptionInfo) -> Option<(MsgId, Block)> { borsh::from_slice::(&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, + } } diff --git a/lez/sequencer/core/src/lib.rs b/lez/sequencer/core/src/lib.rs index c7e7e4a7..928ffa1e 100644 --- a/lez/sequencer/core/src/lib.rs +++ b/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}, @@ -12,7 +17,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; @@ -57,6 +65,9 @@ impl DepositMetadata { pub struct SequencerCore { 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>, store: SequencerStore, mempool: MemPool<(TransactionOrigin, LeeTransaction)>, sequencer_config: SequencerConfig, @@ -130,6 +141,14 @@ impl SequencerCore { .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"); @@ -147,6 +166,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)), ) .await .expect("Failed to initialize Block Publisher"); @@ -164,6 +184,7 @@ impl SequencerCore { let sequencer_core = Self { state, + chain, store, mempool, chain_height: latest_block_meta.id, @@ -304,6 +325,28 @@ impl SequencerCore { }) } + /// 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>) -> 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 { let BlockWithMeta { @@ -540,6 +583,12 @@ impl SequencerCore { self.block_publisher.is_our_turn() } + /// Shared handle to the two-tier follow state. + #[must_use] + pub fn chain(&self) -> Arc> { + Arc::clone(&self.chain) + } + fn next_block_id(&self) -> u64 { self.chain_height .checked_add(1) diff --git a/lez/sequencer/core/src/mock.rs b/lez/sequencer/core/src/mock.rs index 81508f7c..a3922c22 100644 --- a/lez/sequencer/core/src/mock.rs +++ b/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 { Ok(Self { channel_id: config.channel_id,