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

lez/sequencer/core/src/block_publisher.rs

@@ -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;
+pub use logos_blockchain_core::mantle::ops::channel::{Ed25519PublicKey, MsgId};
-use logos_blockchain_core::mantle::ops::channel::{ChannelId, inscribe::Inscription};
+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
+/// Commands the drive task executes with `&mut sequencer`.
-/// oneshot for the `MsgId` zone-sdk assigns.
+enum Command {
-type PublishSender = mpsc::Sender<(
+    /// Publish an inscription (+ atomic withdrawals); responds with the assigned `MsgId`.
-    Inscription,
+    Publish {
-    Vec<WithdrawArg>,
+        inscription: Inscription,
-    oneshot::Sender<Result<MsgId>>,
+        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
+                        // Drain external commands by calling the borrowing
-                        // borrowing handle — `&mut sequencer` is only
+                        // handle — `&mut sequencer` is only available here.
-                        // available here.
+                        Some(command) = command_rx.recv() => match command {
-                        Some((data_bounded, withdrawals, resp_tx)) = publish_rx.recv() => {
+                            Command::Publish { inscription: data_bounded, withdrawals, resp: resp_tx } => {
-                            let data_byte_size = data_bounded.len();
+                                let data_byte_size = data_bounded.len();
-                            let withdraw_count = withdrawals.len();
+                                let withdraw_count = withdrawals.len();
-                            let published = if withdrawals.is_empty() {
+                                let published = if withdrawals.is_empty() {
-                                sequencer.handle()
+                                    sequencer.handle()
-                                    .publish(data_bounded)
+                                        .publish(data_bounded)
-                                    .context("Failed to publish block")
+                                        .context("Failed to publish block")
-                            } else {
+                                } else {
-                                sequencer.handle()
+                                    sequencer.handle()
-                                    .publish_atomic_withdraw(data_bounded, withdrawals)
+                                        .publish_atomic_withdraw(data_bounded, withdrawals)
-                                    .context("Failed to publish block with withdrawals")
+                                        .context("Failed to publish block with withdrawals")
-                            };
+                                };
 
-                            let msg_result = published
+                                let msg_result = published
-                                .map(|(result, _checkpoint)| result.tx.inscription().this_msg);
+                                    .map(|(result, _checkpoint)| result.tx.inscription().this_msg);
-                            match &msg_result {
+                                match &msg_result {
-                                Ok(_) if withdraw_count == 0 => {
+                                    Ok(_) if withdraw_count == 0 => {
-                                    info!("Published block with the size of {data_byte_size} bytes");
+                                        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(_) => {
+                                let _dontcare = resp_tx.send(msg_result);
-                                    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);
+                            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
+        self.command_tx
-            .send((data_bounded, withdrawals, resp_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
     }

lez/sequencer/core/src/lib.rs

@@ -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>> {

lez/sequencer/core/src/mock.rs

@@ -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,
+        BlockPublisherTrait, CheckpointSink, Ed25519PublicKey, FinalizedBlockSink,
-        OnWithdrawEventSink, SequencerCheckpoint,
+        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
     }

lez/sequencer/core/src/tests.rs

@@ -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);
+}