logos-blockchain/logos-execution-zone
1
0
Fork 0
mirror of https://github.com/logos-blockchain/logos-execution-zone.git synced 2026-07-15 10:20:01 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat(sequencer): channel roster configuration through the block publisher

Browse Source
This commit is contained in:
erhant 2026-07-13 23:25:38 +03:00
parent 9c8d0904ca
commit 0143a7edf1
4 changed files with 219 additions and 46 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

169
lez/sequencer/core/src/block_publisher.rs
View File

@ -3,8 +3,11 @@ use std::{pin::Pin, sync::Arc, time::Duration};
use anyhow::{Context as _, Result, anyhow};
use common::block::Block;
use log::{info, warn};
pub use logos_blockchain_core::mantle::ops::channel::MsgId;
use logos_blockchain_core::mantle::ops::channel::{ChannelId, inscribe::Inscription};
pub use logos_blockchain_core::mantle::ops::channel::{Ed25519PublicKey, MsgId};
use logos_blockchain_core::mantle::{
channel::{SlotTimeframe, SlotTimeout},
ops::channel::{ChannelId, config::Keys, inscribe::Inscription},
};
pub use logos_blockchain_key_management_system_service::keys::{Ed25519Key, ZkKey};
pub use logos_blockchain_zone_sdk::sequencer::SequencerCheckpoint;
use logos_blockchain_zone_sdk::{
@ -58,13 +61,26 @@ pub struct FollowUpdate {
pub type OnFollowSink =
Box<dyn Fn(FollowUpdate) -> Pin<Box<dyn Future<Output = ()> + 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<WithdrawArg>,
oneshot::Sender<Result<MsgId>>,
)>;
/// Commands the drive task executes with `&mut sequencer`.
enum Command {
/// Publish an inscription (+ atomic withdrawals); responds with the assigned `MsgId`.
Publish {
inscription: Inscription,
withdrawals: Vec<WithdrawArg>,
resp: oneshot::Sender<Result<MsgId>>,
},
/// Post a `ChannelConfig` op replacing the accredited keys / rotation params.
ConfigureChannel {
keys: Keys,
posting_timeframe: SlotTimeframe,
posting_timeout: SlotTimeout,
configuration_threshold: u16,
withdraw_threshold: u16,
resp: oneshot::Sender<Result<()>>,
},
}
type CommandSender = mpsc::Sender<Command>;
#[expect(async_fn_in_trait, reason = "We don't care about Send/Sync here")]
pub trait BlockPublisherTrait: Clone {
@ -88,6 +104,19 @@ pub trait BlockPublisherTrait: Clone {
/// Zone-sdk drives the actual submission and retries internally.
async fn publish_block(&self, block: &Block, withdrawals: Vec<WithdrawArg>) -> Result<MsgId>;
/// Update the channel's accredited key set and rotation parameters via a
/// `ChannelConfig` op. The sequencer's bedrock key must be the channel
/// admin (`keys[0]`); the L1 rejects non-admin signers, so this is not
/// re-validated here.
async fn configure_channel(
&self,
keys: Vec<Ed25519PublicKey>,
posting_timeframe: u32,
posting_timeout: u32,
configuration_threshold: u16,
withdraw_threshold: u16,
) -> Result<()>;
fn channel_id(&self) -> ChannelId;
/// Whether this sequencer is currently authorized to write to the channel.
@ -98,7 +127,7 @@ pub trait BlockPublisherTrait: Clone {
#[derive(Clone)]
pub struct ZoneSdkPublisher {
channel_id: ChannelId,
publish_tx: PublishSender,
command_tx: CommandSender,
turn_rx: watch::Receiver<TurnNotification>,
// Aborts the drive task when the last clone is dropped.
_drive_task: Arc<DriveTaskGuard>,
@ -146,7 +175,7 @@ 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): (PublishSender, _) =
let (command_tx, mut command_rx): (CommandSender, _) =
mpsc::channel(PUBLISH_INBOX_CAPACITY);
let drive_task = tokio::spawn(async move {
@ -157,37 +186,62 @@ impl BlockPublisherTrait for ZoneSdkPublisher {
)]
{
tokio::select! {
// Drain external publish requests by calling the
// borrowing handle — `&mut sequencer` is only
// available here.
Some((data_bounded, withdrawals, resp_tx)) = publish_rx.recv() => {
let data_byte_size = data_bounded.len();
let withdraw_count = withdrawals.len();
let published = if withdrawals.is_empty() {
sequencer.handle()
.publish(data_bounded)
.context("Failed to publish block")
} else {
sequencer.handle()
.publish_atomic_withdraw(data_bounded, withdrawals)
.context("Failed to publish block with withdrawals")
};
// Drain external commands by calling the borrowing
// handle — `&mut sequencer` is only available here.
Some(command) = command_rx.recv() => match command {
Command::Publish { inscription: data_bounded, withdrawals, resp: resp_tx } => {
let data_byte_size = data_bounded.len();
let withdraw_count = withdrawals.len();
let published = if withdrawals.is_empty() {
sequencer.handle()
.publish(data_bounded)
.context("Failed to publish block")
} else {
sequencer.handle()
.publish_atomic_withdraw(data_bounded, withdrawals)
.context("Failed to publish block with 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");
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:?}"),
}
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);
}
let _dontcare = resp_tx.send(msg_result);
}
Command::ConfigureChannel {
keys,
posting_timeframe,
posting_timeout,
configuration_threshold,
withdraw_threshold,
resp,
} => {
let result = sequencer
.handle()
.channel_config(
keys,
posting_timeframe,
posting_timeout,
configuration_threshold,
withdraw_threshold,
)
.map(|_queued| ())
.context("Failed to post channel config");
if let Err(err) = &result {
warn!("zone-sdk channel config failed: {err:?}");
}
let _dontcare = resp.send(result);
}
},
event = sequencer.next_event() => {
let Some(event) = event else {
continue;
@ -256,7 +310,7 @@ impl BlockPublisherTrait for ZoneSdkPublisher {
Ok(Self {
channel_id: config.channel_id,
publish_tx,
command_tx,
turn_rx,
_drive_task: Arc::new(DriveTaskGuard(drive_task)),
})
@ -269,8 +323,12 @@ impl BlockPublisherTrait for ZoneSdkPublisher {
.context("Block data exceeds maximum allowed size")?;
let (resp_tx, resp_rx) = oneshot::channel();
self.publish_tx
.send((data_bounded, withdrawals, resp_tx))
self.command_tx
.send(Command::Publish {
inscription: data_bounded,
withdrawals,
resp: resp_tx,
})
.await
.map_err(|_closed| anyhow!("Drive task is no longer running"))?;
@ -279,6 +337,33 @@ impl BlockPublisherTrait for ZoneSdkPublisher {
.map_err(|_closed| anyhow!("Drive task dropped the publish response"))?
}
async fn configure_channel(
&self,
keys: Vec<Ed25519PublicKey>,
posting_timeframe: u32,
posting_timeout: u32,
configuration_threshold: u16,
withdraw_threshold: u16,
) -> Result<()> {
let keys = Keys::try_from(keys)
.map_err(|_err| anyhow!("Channel key list must be non-empty and within bounds"))?;
let (resp_tx, resp_rx) = oneshot::channel();
self.command_tx
.send(Command::ConfigureChannel {
keys,
posting_timeframe: posting_timeframe.into(),
posting_timeout: posting_timeout.into(),
configuration_threshold,
withdraw_threshold,
resp: resp_tx,
})
.await
.map_err(|_closed| anyhow!("Drive task is no longer running"))?;
resp_rx
.await
.map_err(|_closed| anyhow!("Drive task dropped the config response"))?
}
fn channel_id(&self) -> ChannelId {
self.channel_id
}

23
lez/sequencer/core/src/lib.rs
View File

@ -33,7 +33,7 @@ use storage::sequencer::{
};
use crate::{
block_publisher::{BlockPublisherTrait, ZoneSdkPublisher},
block_publisher::{BlockPublisherTrait, Ed25519PublicKey, ZoneSdkPublisher},
block_store::SequencerStore,
};
@ -625,6 +625,27 @@ impl<BP: BlockPublisherTrait> SequencerCore<BP> {
self.block_publisher.is_our_turn()
}
/// Update the channel's accredited key set and rotation parameters.
/// This sequencer's bedrock key must be the channel admin (`keys[0]`).
pub async fn configure_channel(
&self,
keys: Vec<Ed25519PublicKey>,
posting_timeframe: u32,
posting_timeout: u32,
configuration_threshold: u16,
withdraw_threshold: u16,
) -> Result<()> {
self.block_publisher
.configure_channel(
keys,
posting_timeframe,
posting_timeout,
configuration_threshold,
withdraw_threshold,
)
.await
}
/// Shared handle to the two-tier follow state.
#[must_use]
pub fn chain(&self) -> Arc<Mutex<ChainState>> {

52
lez/sequencer/core/src/mock.rs
View File

@ -1,4 +1,7 @@
use std::time::Duration;
use std::{
sync::{Arc, Mutex},
time::Duration,
};
use anyhow::Result;
use common::block::Block;
@ -8,17 +11,38 @@ use logos_blockchain_zone_sdk::sequencer::WithdrawArg;
use crate::{
block_publisher::{
BlockPublisherTrait, CheckpointSink, FinalizedBlockSink, OnDepositEventSink, OnFollowSink,
OnWithdrawEventSink, SequencerCheckpoint,
BlockPublisherTrait, CheckpointSink, Ed25519PublicKey, FinalizedBlockSink,
OnDepositEventSink, OnFollowSink, OnWithdrawEventSink, SequencerCheckpoint,
},
config::BedrockConfig,
};
pub type SequencerCoreWithMockClients = crate::SequencerCore<MockBlockPublisher>;
/// One recorded `configure_channel` invocation.
#[derive(Clone)]
pub struct ConfigureChannelCall {
pub keys: Vec<Ed25519PublicKey>,
pub posting_timeframe: u32,
pub posting_timeout: u32,
pub configuration_threshold: u16,
pub withdraw_threshold: u16,
}
#[derive(Clone)]
pub struct MockBlockPublisher {
channel_id: ChannelId,
configure_channel_calls: Arc<Mutex<Vec<ConfigureChannelCall>>>,
}
impl MockBlockPublisher {
#[must_use]
pub fn configure_channel_calls(&self) -> Vec<ConfigureChannelCall> {
self.configure_channel_calls
.lock()
.expect("mock mutex poisoned")
.clone()
}
}
impl BlockPublisherTrait for MockBlockPublisher {
@ -35,6 +59,7 @@ impl BlockPublisherTrait for MockBlockPublisher {
) -> Result<Self> {
Ok(Self {
channel_id: config.channel_id,
configure_channel_calls: Arc::default(),
})
}
@ -49,6 +74,27 @@ impl BlockPublisherTrait for MockBlockPublisher {
Ok(MsgId::from(block.header.hash.0))
}
async fn configure_channel(
&self,
keys: Vec<Ed25519PublicKey>,
posting_timeframe: u32,
posting_timeout: u32,
configuration_threshold: u16,
withdraw_threshold: u16,
) -> Result<()> {
self.configure_channel_calls
.lock()
.expect("mock mutex poisoned")
.push(ConfigureChannelCall {
keys,
posting_timeframe,
posting_timeout,
configuration_threshold,
withdraw_threshold,
});
Ok(())
}
fn channel_id(&self) -> ChannelId {
self.channel_id
}

21
lez/sequencer/core/src/tests.rs
View File

@ -23,6 +23,7 @@ use lee_core::{
program::PdaSeed,
};
use logos_blockchain_core::mantle::ops::channel::{ChannelId, MsgId};
use logos_blockchain_key_management_system_service::keys::{ED25519_SECRET_KEY_SIZE, Ed25519Key};
use mempool::MemPoolHandle;
use storage::sequencer::sequencer_cells::PendingDepositEventRecord;
use tempfile::tempdir;
@ -1543,3 +1544,23 @@ async fn follow_finalized_backfill_block_is_applied_and_marked_finalized() {
assert_eq!(stored.header.hash, peer_block.header.hash);
assert!(matches!(stored.bedrock_status, BedrockStatus::Finalized));
}
#[tokio::test]
async fn configure_channel_delegates_to_publisher() {
let config = setup_sequencer_config();
let (sequencer, _mempool_handle) =
SequencerCoreWithMockClients::start_from_config(config).await;
let admin = Ed25519Key::from_bytes(&[0xA1; ED25519_SECRET_KEY_SIZE]).public_key();
let peer = Ed25519Key::from_bytes(&[0xB2; ED25519_SECRET_KEY_SIZE]).public_key();
sequencer
.configure_channel(vec![admin, peer], 20, 30, 1, 1)
.await
.unwrap();
let calls = sequencer.block_publisher().configure_channel_calls();
assert_eq!(calls.len(), 1);
assert_eq!(calls[0].keys.len(), 2);
assert_eq!(calls[0].posting_timeframe, 20);
assert_eq!(calls[0].posting_timeout, 30);
}