diff --git a/lez/sequencer/core/src/block_publisher.rs b/lez/sequencer/core/src/block_publisher.rs index 21551131..f8a64c5d 100644 --- a/lez/sequencer/core/src/block_publisher.rs +++ b/lez/sequencer/core/src/block_publisher.rs @@ -12,10 +12,14 @@ use logos_blockchain_zone_sdk::{ adapter::NodeHttpClient, sequencer::{ DepositInfo, Event, FinalizedOp, InscriptionInfo, - SequencerConfig as ZoneSdkSequencerConfig, WithdrawArg, WithdrawInfo, ZoneSequencer, + SequencerConfig as ZoneSdkSequencerConfig, TurnNotification, WithdrawArg, WithdrawInfo, + ZoneSequencer, }, }; -use tokio::{sync::mpsc, task::JoinHandle}; +use tokio::{ + sync::{mpsc, watch}, + task::JoinHandle, +}; use crate::config::BedrockConfig; @@ -63,6 +67,9 @@ pub trait BlockPublisherTrait: Clone { async fn publish_block(&self, block: &Block, withdrawals: Vec) -> Result<()>; fn channel_id(&self) -> ChannelId; + + /// Whether this sequencer is currently authorized to write to the channel. + fn is_our_turn(&self) -> bool; } /// Real block publisher backed by zone-sdk's `ZoneSequencer`. @@ -70,6 +77,7 @@ pub trait BlockPublisherTrait: Clone { pub struct ZoneSdkPublisher { channel_id: ChannelId, publish_tx: mpsc::Sender<(Inscription, Vec)>, + turn_rx: watch::Receiver, // Aborts the drive task when the last clone is dropped. _drive_task: Arc, } @@ -112,6 +120,8 @@ impl BlockPublisherTrait for ZoneSdkPublisher { // Grab readiness receiver before moving the sequencer into the drive // task so we can await cold-start completion below. let mut ready_rx = sequencer.subscribe_ready(); + // 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); @@ -197,6 +207,7 @@ impl BlockPublisherTrait for ZoneSdkPublisher { Ok(Self { channel_id: config.channel_id, publish_tx, + turn_rx, _drive_task: Arc::new(DriveTaskGuard(drive_task)), }) } @@ -218,6 +229,10 @@ impl BlockPublisherTrait for ZoneSdkPublisher { fn channel_id(&self) -> ChannelId { self.channel_id } + + fn is_our_turn(&self) -> bool { + self.turn_rx.borrow().our_turn_to_write + } } /// Deserialize inscription payload as a `Block` and return it's`block_id`. diff --git a/lez/sequencer/core/src/lib.rs b/lez/sequencer/core/src/lib.rs index 75dcc1a9..c7e7e4a7 100644 --- a/lez/sequencer/core/src/lib.rs +++ b/lez/sequencer/core/src/lib.rs @@ -534,6 +534,12 @@ impl SequencerCore { self.block_publisher.clone() } + /// Whether this sequencer is currently authorized to write to the channel. + #[must_use] + pub fn is_our_turn(&self) -> bool { + self.block_publisher.is_our_turn() + } + 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 39f635f9..81508f7c 100644 --- a/lez/sequencer/core/src/mock.rs +++ b/lez/sequencer/core/src/mock.rs @@ -48,4 +48,8 @@ impl BlockPublisherTrait for MockBlockPublisher { fn channel_id(&self) -> ChannelId { self.channel_id } + + fn is_our_turn(&self) -> bool { + true + } } diff --git a/lez/sequencer/service/src/lib.rs b/lez/sequencer/service/src/lib.rs index 687ee424..d6779102 100644 --- a/lez/sequencer/service/src/lib.rs +++ b/lez/sequencer/service/src/lib.rs @@ -172,16 +172,15 @@ async fn main_loop(seq_core: Arc>, block_timeout: Duration) loop { tokio::time::sleep(block_timeout).await; - info!("Collecting transactions from mempool, block creation"); + let mut state = seq_core.lock().await; - let id = { - let mut state = seq_core.lock().await; - - state.produce_new_block().await? - }; + // Only produce on our turn. + if !state.is_our_turn() { + continue; + } + info!("Our turn: collecting transactions from mempool, creating block"); + let id = state.produce_new_block().await?; info!("Block with id {id} created"); - - info!("Waiting for new transactions"); } }