refactor(sequencer, storage): review nit-fixes, sync follow sink, validate-on-deserialize with turn bounds.

Cargo.lock

@@ -9110,6 +9110,7 @@ dependencies = [
  "lee",
  "lee_core",
  "serde",
+ "serde_json",
  "serde_with",
 ]
 

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<dyn Fn(FollowUpdate) -> Pin<Box<dyn Future<Output = ()> + Send>> + Send + 'static>;
+pub type OnFollowSink = Box<dyn Fn(FollowUpdate) + Send + 'static>;
 
 /// 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 } => {

lez/sequencer/core/src/lib.rs

@@ -417,7 +417,6 @@ impl<BP: BlockPublisherTrait> SequencerCore<BP> {
     ) -> 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<BP: BlockPublisherTrait> SequencerCore<BP> {
             .await
     }
 
-    /// Shared handle to the two-tier follow state.
-    #[must_use]
-    pub fn chain(&self) -> Arc<Mutex<ChainState>> {
+    /// 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<Mutex<ChainState>> {
         Arc::clone(&self.chain)
     }
 }

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

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<String>,
     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<String>,
+    posting_timeframe: u32,
+    posting_timeout: u32,
+    configuration_threshold: u16,
+    withdraw_threshold: u16,
+}
+
+impl TryFrom<UncheckedConfigureChannelRequest> for ConfigureChannelRequest {
+    type Error = String;
+
+    fn try_from(unchecked: UncheckedConfigureChannelRequest) -> Result<Self, Self::Error> {
+        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::<ConfigureChannelRequest>(valid.clone()).is_ok());
+
+        let mut invalid = valid;
+        invalid["posting_timeout"] = serde_json::json!(10);
+        let err = serde_json::from_value::<ConfigureChannelRequest>(invalid)
+            .expect_err("timeout below timeframe must fail to deserialize");
+        assert!(err.to_string().contains("posting_timeout"));
+    }
 }

lez/sequencer/service/src/service.rs

@@ -210,7 +210,6 @@ impl<BC: BlockPublisherTrait + Send + Sync + 'static> sequencer_service_rpc::Rpc
         &self,
         request: ConfigureChannelRequest,
     ) -> Result<(), ErrorObjectOwned> {
-        request.validate().map_err(invalid_params)?;
         let keys = request
             .keys
             .iter()

lez/storage/src/sequencer/mod.rs

@@ -677,19 +677,14 @@ impl RocksDBIO {
         let mut batch_tip: Option<BlockMeta> = 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;