From 6eb14350625b28f73911304ce12bd26bd58eef83 Mon Sep 17 00:00:00 2001 From: erhant Date: Mon, 20 Jul 2026 15:05:26 +0300 Subject: [PATCH] refactor(sequencer, storage): review nit-fixes, sync follow sink, validate-on-deserialize with turn bounds. --- Cargo.lock | 1 + lez/sequencer/core/src/block_publisher.rs | 6 +- lez/sequencer/core/src/lib.rs | 8 +- lez/sequencer/service/protocol/Cargo.toml | 3 + lez/sequencer/service/protocol/src/lib.rs | 89 +++++++++++++++++++++++ lez/sequencer/service/src/service.rs | 1 - lez/storage/src/sequencer/mod.rs | 19 ++--- 7 files changed, 106 insertions(+), 21 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 792f3856..d48601b9 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -9110,6 +9110,7 @@ dependencies = [ "lee", "lee_core", "serde", + "serde_json", "serde_with", ] diff --git a/lez/sequencer/core/src/block_publisher.rs b/lez/sequencer/core/src/block_publisher.rs index 0bf32719..47da1e44 100644 --- a/lez/sequencer/core/src/block_publisher.rs +++ b/lez/sequencer/core/src/block_publisher.rs @@ -60,8 +60,7 @@ pub struct FollowUpdate { /// Sink for the follow path: apply adopted/finalized blocks to chain state and /// revert orphaned ones. -pub type OnFollowSink = - Box Pin + Send>> + Send + 'static>; +pub type OnFollowSink = Box; /// Commands the drive task executes with `&mut sequencer`. enum Command { @@ -298,8 +297,7 @@ impl BlockPublisherTrait for ZoneSdkPublisher { adopted, orphaned, finalized: finalized_blocks, - }) - .await; + }); } Event::Ready => {} Event::TurnNotification { notification } => { diff --git a/lez/sequencer/core/src/lib.rs b/lez/sequencer/core/src/lib.rs index 580edc44..3586e811 100644 --- a/lez/sequencer/core/src/lib.rs +++ b/lez/sequencer/core/src/lib.rs @@ -417,7 +417,6 @@ impl SequencerCore { ) -> block_publisher::OnFollowSink { Box::new(move |update: block_publisher::FollowUpdate| { apply_follow_update(&dbio, &chain, &mempool_handle, update); - Box::pin(std::future::ready(())) }) } @@ -740,9 +739,10 @@ impl SequencerCore { .await } - /// Shared handle to the two-tier follow state. - #[must_use] - pub fn chain(&self) -> Arc> { + /// Shared handle to the two-tier follow state, for tests to drive the + /// follow path directly. + #[cfg(all(test, feature = "mock"))] + fn chain(&self) -> Arc> { Arc::clone(&self.chain) } } diff --git a/lez/sequencer/service/protocol/Cargo.toml b/lez/sequencer/service/protocol/Cargo.toml index 1eb413d0..5bbe0091 100644 --- a/lez/sequencer/service/protocol/Cargo.toml +++ b/lez/sequencer/service/protocol/Cargo.toml @@ -15,3 +15,6 @@ lee_core.workspace = true hex.workspace = true serde.workspace = true serde_with.workspace = true + +[dev-dependencies] +serde_json.workspace = true diff --git a/lez/sequencer/service/protocol/src/lib.rs b/lez/sequencer/service/protocol/src/lib.rs index d621e561..b117977d 100644 --- a/lez/sequencer/service/protocol/src/lib.rs +++ b/lez/sequencer/service/protocol/src/lib.rs @@ -33,6 +33,7 @@ impl FromStr for ChannelId { /// - `keys` are hex-encoded 32-byte Ed25519 public keys /// - `keys[0]` must be this sequencer's (admin) key. #[derive(Debug, Clone, serde::Serialize, serde::Deserialize)] +#[serde(try_from = "UncheckedConfigureChannelRequest")] pub struct ConfigureChannelRequest { pub keys: Vec, pub posting_timeframe: u32, @@ -41,6 +42,40 @@ pub struct ConfigureChannelRequest { pub withdraw_threshold: u16, } +/// Wire-shape mirror of [`ConfigureChannelRequest`]; deserialization goes +/// through it so [`ConfigureChannelRequest::validate`] runs on every decode. +#[derive(serde::Deserialize)] +struct UncheckedConfigureChannelRequest { + keys: Vec, + posting_timeframe: u32, + posting_timeout: u32, + configuration_threshold: u16, + withdraw_threshold: u16, +} + +impl TryFrom for ConfigureChannelRequest { + type Error = String; + + fn try_from(unchecked: UncheckedConfigureChannelRequest) -> Result { + let UncheckedConfigureChannelRequest { + keys, + posting_timeframe, + posting_timeout, + configuration_threshold, + withdraw_threshold, + } = unchecked; + let request = Self { + keys, + posting_timeframe, + posting_timeout, + configuration_threshold, + withdraw_threshold, + }; + request.validate()?; + Ok(request) + } +} + impl ConfigureChannelRequest { /// Structural sanity checks for the request. /// @@ -64,6 +99,18 @@ impl ConfigureChannelRequest { )); } } + if self.posting_timeframe == 0 || self.posting_timeout == 0 { + return Err(format!( + "posting_timeframe and posting_timeout must be nonzero, got {} and {}", + self.posting_timeframe, self.posting_timeout + )); + } + if self.posting_timeout < self.posting_timeframe { + return Err(format!( + "posting_timeout ({}) must not be shorter than posting_timeframe ({})", + self.posting_timeout, self.posting_timeframe + )); + } Ok(()) } } @@ -108,6 +155,30 @@ mod tests { .validate() .is_err() ); + assert!( + ConfigureChannelRequest { + posting_timeframe: 0, + ..base.clone() + } + .validate() + .is_err() + ); + assert!( + ConfigureChannelRequest { + posting_timeout: 0, + ..base.clone() + } + .validate() + .is_err() + ); + assert!( + ConfigureChannelRequest { + posting_timeout: 10, + ..base.clone() + } + .validate() + .is_err() + ); assert!( ConfigureChannelRequest { withdraw_threshold: 3, @@ -117,4 +188,22 @@ mod tests { .is_err() ); } + + #[test] + fn configure_channel_request_validates_on_deserialization() { + let valid = serde_json::json!({ + "keys": ["unparsed"], + "posting_timeframe": 20, + "posting_timeout": 30, + "configuration_threshold": 1, + "withdraw_threshold": 1, + }); + assert!(serde_json::from_value::(valid.clone()).is_ok()); + + let mut invalid = valid; + invalid["posting_timeout"] = serde_json::json!(10); + let err = serde_json::from_value::(invalid) + .expect_err("timeout below timeframe must fail to deserialize"); + assert!(err.to_string().contains("posting_timeout")); + } } diff --git a/lez/sequencer/service/src/service.rs b/lez/sequencer/service/src/service.rs index 63eac921..50b3a2fb 100644 --- a/lez/sequencer/service/src/service.rs +++ b/lez/sequencer/service/src/service.rs @@ -210,7 +210,6 @@ impl sequencer_service_rpc::Rpc &self, request: ConfigureChannelRequest, ) -> Result<(), ErrorObjectOwned> { - request.validate().map_err(invalid_params)?; let keys = request .keys .iter() diff --git a/lez/storage/src/sequencer/mod.rs b/lez/storage/src/sequencer/mod.rs index 8d3a467d..ed8130bf 100644 --- a/lez/storage/src/sequencer/mod.rs +++ b/lez/storage/src/sequencer/mod.rs @@ -677,19 +677,14 @@ impl RocksDBIO { let mut batch_tip: Option = None; for (block, finalized) in blocks { - let stored = self.get_block(block.header.block_id)?; - let already_stored = stored - .as_ref() - .is_some_and(|stored| stored.header.hash == block.header.hash); + let already_stored = self + .get_block(block.header.block_id)? + .filter(|stored| stored.header.hash == block.header.hash); - if already_stored && !finalized { - continue; - } - - let mut to_write = if already_stored { - stored.expect("`already_stored` implies a stored block") - } else { - (*block).clone() + let mut to_write = match already_stored { + Some(_) if !finalized => continue, + Some(stored) => stored, + None => (*block).clone(), }; if *finalized { to_write.bedrock_status = BedrockStatus::Finalized;