From 10342a36f5f159547207c38586e5b0faa9624a92 Mon Sep 17 00:00:00 2001 From: erhant Date: Fri, 10 Jul 2026 12:10:47 +0300 Subject: [PATCH] fix(sequencer): apply the own-published block immediately to the head & dedup it later if received --- lez/sequencer/core/src/block_publisher.rs | 72 ++++++++++++++--------- lez/sequencer/core/src/lib.rs | 10 +++- lez/sequencer/core/src/mock.rs | 11 ++-- 3 files changed, 60 insertions(+), 33 deletions(-) diff --git a/lez/sequencer/core/src/block_publisher.rs b/lez/sequencer/core/src/block_publisher.rs index 5654dd34..9cd21b45 100644 --- a/lez/sequencer/core/src/block_publisher.rs +++ b/lez/sequencer/core/src/block_publisher.rs @@ -17,7 +17,7 @@ use logos_blockchain_zone_sdk::{ }, }; use tokio::{ - sync::{mpsc, watch}, + sync::{mpsc, oneshot, watch}, task::JoinHandle, }; @@ -58,6 +58,14 @@ pub struct FollowUpdate { pub type OnFollowSink = Box Pin + Send>> + Send + 'static>; +/// Publish request channel: the inscription, its bridge withdrawals, and a +/// oneshot for the `MsgId` zone-sdk assigns. +type PublishSender = mpsc::Sender<( + Inscription, + Vec, + oneshot::Sender>, +)>; + #[expect(async_fn_in_trait, reason = "We don't care about Send/Sync here")] pub trait BlockPublisherTrait: Clone { #[expect( @@ -76,9 +84,9 @@ pub trait BlockPublisherTrait: Clone { on_follow: OnFollowSink, ) -> Result; - /// Fire-and-forget publish. Zone-sdk drives the actual submission and - /// retries internally; this just hands the payload off. - async fn publish_block(&self, block: &Block, withdrawals: Vec) -> Result<()>; + /// Publish a block and return the `MsgId` zone-sdk assigned its inscription. + /// Zone-sdk drives the actual submission and retries internally. + async fn publish_block(&self, block: &Block, withdrawals: Vec) -> Result; fn channel_id(&self) -> ChannelId; @@ -90,7 +98,7 @@ pub trait BlockPublisherTrait: Clone { #[derive(Clone)] pub struct ZoneSdkPublisher { channel_id: ChannelId, - publish_tx: mpsc::Sender<(Inscription, Vec)>, + publish_tx: PublishSender, turn_rx: watch::Receiver, // Aborts the drive task when the last clone is dropped. _drive_task: Arc, @@ -138,8 +146,8 @@ impl BlockPublisherTrait for ZoneSdkPublisher { // Grab the turn watch before the move; the sdk actor keeps it current. let turn_rx = sequencer.subscribe_turn_to_write(); - let (publish_tx, mut publish_rx) = - mpsc::channel::<(Inscription, Vec)>(PUBLISH_INBOX_CAPACITY); + let (publish_tx, mut publish_rx): (PublishSender, _) = + mpsc::channel(PUBLISH_INBOX_CAPACITY); let drive_task = tokio::spawn(async move { loop { @@ -152,28 +160,33 @@ impl BlockPublisherTrait for ZoneSdkPublisher { // Drain external publish requests by calling the // borrowing handle — `&mut sequencer` is only // available here. - Some((data_bounded, withdrawals)) = publish_rx.recv() => { + Some((data_bounded, withdrawals, resp_tx)) = publish_rx.recv() => { let data_byte_size = data_bounded.len(); - if withdrawals.is_empty() { - if let Err(e) = sequencer.handle() - .publish(data_bounded) - .context("Failed to publish block") { - warn!("zone-sdk publish failed: {e:?}"); - } - - info!("Published block with the size of {data_byte_size} bytes"); + let withdraw_count = withdrawals.len(); + let published = if withdrawals.is_empty() { + sequencer.handle() + .publish(data_bounded) + .context("Failed to publish block") } else { - let withdraw_count = withdrawals.len(); - if let Err(e) = sequencer.handle() - .publish_atomic_withdraw(data_bounded, withdrawals) - .context("Failed to publish block with withdrawals") { - warn!("zone-sdk publish failed: {e:?}"); - } + sequencer.handle() + .publish_atomic_withdraw(data_bounded, withdrawals) + .context("Failed to publish block with withdrawals") + }; - info!( - "Published block with the size of {data_byte_size} bytes and {withdraw_count} bridge withdrawals", - ); + let msg_result = published + .map(|(result, _checkpoint)| result.tx.inscription().this_msg); + match &msg_result { + Ok(_) if withdraw_count == 0 => { + info!("Published block with the size of {data_byte_size} bytes"); + } + Ok(_) => { + info!( + "Published block with the size of {data_byte_size} bytes and {withdraw_count} bridge withdrawals", + ); + } + Err(e) => warn!("zone-sdk publish failed: {e:?}"), } + let _dontcare = resp_tx.send(msg_result); } event = sequencer.next_event() => { let Some(event) = event else { @@ -249,18 +262,21 @@ impl BlockPublisherTrait for ZoneSdkPublisher { }) } - async fn publish_block(&self, block: &Block, withdrawals: Vec) -> Result<()> { + async fn publish_block(&self, block: &Block, withdrawals: Vec) -> Result { let data = borsh::to_vec(block).context("Failed to serialize block")?; let data_bounded: Inscription = data .try_into() .context("Block data exceeds maximum allowed size")?; + let (resp_tx, resp_rx) = oneshot::channel(); self.publish_tx - .send((data_bounded, withdrawals)) + .send((data_bounded, withdrawals, resp_tx)) .await .map_err(|_closed| anyhow!("Drive task is no longer running"))?; - Ok(()) + resp_rx + .await + .map_err(|_closed| anyhow!("Drive task dropped the publish response"))? } fn channel_id(&self) -> ChannelId { diff --git a/lez/sequencer/core/src/lib.rs b/lez/sequencer/core/src/lib.rs index a6f154a8..45467d49 100644 --- a/lez/sequencer/core/src/lib.rs +++ b/lez/sequencer/core/src/lib.rs @@ -382,11 +382,19 @@ impl SequencerCore { .collect::>() .context("Failed to build reconciliation keys for block withdrawals")?; - self.block_publisher + let this_msg = self + .block_publisher .publish_block(&block, withdrawals) .await .context("Failed to publish block to Bedrock")?; + // Apply our own block to the head with the MsgId the publish assigned it, + // so the head advances and the later adopted redelivery dedups. + self.chain + .lock() + .expect("chain state mutex poisoned") + .apply_adopted(this_msg, &block); + self.store.update( &block, &deposit_event_ids, diff --git a/lez/sequencer/core/src/mock.rs b/lez/sequencer/core/src/mock.rs index a3922c22..37c43254 100644 --- a/lez/sequencer/core/src/mock.rs +++ b/lez/sequencer/core/src/mock.rs @@ -2,7 +2,7 @@ use std::time::Duration; use anyhow::Result; use common::block::Block; -use logos_blockchain_core::mantle::ops::channel::ChannelId; +use logos_blockchain_core::mantle::ops::channel::{ChannelId, MsgId}; use logos_blockchain_key_management_system_service::keys::Ed25519Key; use logos_blockchain_zone_sdk::sequencer::WithdrawArg; @@ -40,10 +40,13 @@ impl BlockPublisherTrait for MockBlockPublisher { async fn publish_block( &self, - _block: &Block, + block: &Block, _bridge_withdrawals: Vec, - ) -> Result<()> { - Ok(()) + ) -> Result { + // Deterministic per-block id so head dedup behaves in tests. + // + // TODO: should we allow more "mockability" here? + Ok(MsgId::from(block.header.hash.0)) } fn channel_id(&self) -> ChannelId {