refactor(sync): enforce block range invariant in StageExecutionInput (#308)

Strengthens the `Stage` trait contract by enforcing that `StageExecutionInput.to >= StageExecutionInput.from`. `Stage` implementors can now rely on receiving valid block ranges.

---------

Co-authored-by: Claude <noreply@anthropic.com>

Author: kariy

crates/sync/pipeline/src/lib.rs

@@ -317,7 +317,7 @@ impl<P: StageCheckpointProvider> Pipeline<P> {
             info!(target: "pipeline", %id, from = %checkpoint, %to, "Executing stage.");
 
             // plus 1 because the checkpoint is inclusive
-            let input = StageExecutionInput { from: checkpoint + 1, to };
+            let input = StageExecutionInput::new(checkpoint + 1, to);
             stage.execute(&input).await.map_err(|error| Error::StageExecution { id, error })?;
             self.provider.set_checkpoint(id, to)?;
 

crates/sync/pipeline/tests/pipeline.rs

@@ -57,11 +57,11 @@ impl Stage for TrackingStage {
     }
 
     fn execute<'a>(&'a mut self, input: &'a StageExecutionInput) -> BoxFuture<'a, StageResult> {
-        let from = input.from;
-        let to = input.to;
-        let executions = self.executions.clone();
         Box::pin(async move {
-            executions.lock().unwrap().push(ExecutionRecord { from, to });
+            self.executions
+                .lock()
+                .unwrap()
+                .push(ExecutionRecord { from: input.from(), to: input.to() });
             Ok(())
         })
     }

crates/sync/stage/src/blocks/mod.rs

@@ -47,13 +47,9 @@ where
 
     fn execute<'a>(&'a mut self, input: &'a StageExecutionInput) -> BoxFuture<'a, StageResult> {
         Box::pin(async move {
-            // TODO: Implement range validation in the `Pipeline` level - or maybe in each stage as
-            // well?
-            debug_assert!(input.from <= input.to);
-
             let blocks = self
                 .downloader
-                .download_blocks(input.from, input.to)
+                .download_blocks(input.from(), input.to())
                 .await
                 .map_err(Error::Gateway)?;
 

crates/sync/stage/src/classes.rs

@@ -66,7 +66,7 @@ where
 
     fn execute<'a>(&'a mut self, input: &'a StageExecutionInput) -> BoxFuture<'a, StageResult> {
         Box::pin(async move {
-            let declared_classes = self.get_declared_classes(input.from, input.to)?;
+            let declared_classes = self.get_declared_classes(input.from(), input.to())?;
 
             if !declared_classes.is_empty() {
                 // fetch the classes artifacts

crates/sync/stage/src/lib.rs

@@ -16,10 +16,41 @@ pub use sequencing::Sequencing;
 /// The result type of a stage execution. See [Stage::execute].
 pub type StageResult = Result<(), Error>;
 
-#[derive(Debug, Default, Clone)]
+/// Input parameters for stage execution.
+///
+/// # Invariant
+///
+/// The `to` field must always be greater than or equal to the `from` field (`to >= from`).
+/// This invariant is enforced at construction time via the [`new`](Self::new) method and
+/// maintained by keeping the fields private.
+#[derive(Debug, Clone, Default)]
 pub struct StageExecutionInput {
-    pub from: BlockNumber,
-    pub to: BlockNumber,
+    from: BlockNumber,
+    to: BlockNumber,
+}
+
+impl StageExecutionInput {
+    /// Creates a new [`StageExecutionInput`] with the given range.
+    ///
+    /// # Panics
+    ///
+    /// Panics if `to < from`, as this violates the type's invariant.
+    pub fn new(from: BlockNumber, to: BlockNumber) -> Self {
+        assert!(to >= from, "Invalid block range: `to` ({to}) must be >= `from` ({from})");
+        Self { from, to }
+    }
+
+    /// Returns the starting block number (inclusive).
+    #[inline]
+    pub fn from(&self) -> BlockNumber {
+        self.from
+    }
+
+    /// Returns the ending block number (inclusive).
+    #[inline]
+    pub fn to(&self) -> BlockNumber {
+        self.to
+    }
 }
 
 #[derive(Debug, Default)]
@@ -50,6 +81,10 @@ pub enum Error {
 /// in the synchronization process (e.g., downloading blocks, downloading classes, executing
 /// transactions).
 ///
+/// Stages are responsible for processing a range of blocks. Each stage implementation can assume
+/// that the block range provided in [`StageExecutionInput`] is valid (i.e., `input.to >=
+/// input.from`).
+///
 /// # Implementation Note
 ///
 /// The [`execute`](Stage::execute) method returns a [`BoxFuture`] instead of `impl Future` to
@@ -62,7 +97,7 @@ pub trait Stage: Send + Sync {
     /// Returns the id which uniquely identifies the stage.
     fn id(&self) -> &'static str;
 
-    /// Executes the stage.
+    /// Executes the stage for the given block range.
     ///
     /// # Arguments
     ///
@@ -71,5 +106,26 @@ pub trait Stage: Send + Sync {
     /// # Returns
     ///
     /// A [`BoxFuture`] that resolves to a [`StageResult`] upon completion.
+    ///
+    /// # Block Range
+    ///
+    /// Implementors can rely on the following guarantees:
+    /// - The `input.to` field will always be greater than or equal to `input.from`
+    /// - The block range `[input.from, input.to]` represents an inclusive range
+    ///
+    /// Implementors are expected to perform any necessary processings on all blocks in the range
+    /// `[input.from, input.to]`.
     fn execute<'a>(&'a mut self, input: &'a StageExecutionInput) -> BoxFuture<'a, StageResult>;
 }
+
+#[cfg(test)]
+mod tests {
+    use crate::StageExecutionInput;
+
+    #[tokio::test]
+    #[should_panic(expected = "Invalid block range")]
+    async fn invalid_range_panics() {
+        // When from > to, the range is invalid and should panic at construction time
+        let _ = StageExecutionInput::new(100, 99);
+    }
+}

crates/sync/stage/tests/block.rs

@@ -223,7 +223,7 @@ async fn download_and_store_blocks(
     }
 
     let mut stage = Blocks::new(provider.clone(), downloader.clone());
-    let input = StageExecutionInput { from: from_block, to: to_block };
+    let input = StageExecutionInput::new(from_block, to_block);
 
     let result = stage.execute(&input).await;
     assert!(result.is_ok());
@@ -244,7 +244,7 @@ async fn download_failure_returns_error() {
     let provider = MockBlockWriter::new();
 
     let mut stage = Blocks::new(provider.clone(), downloader.clone());
-    let input = StageExecutionInput { from: block_number, to: block_number };
+    let input = StageExecutionInput::new(block_number, block_number);
 
     let result = stage.execute(&input).await;
 
@@ -274,7 +274,7 @@ async fn storage_failure_returns_error() {
     let provider = MockBlockWriter::new().with_insert_error(error_msg.clone());
 
     let mut stage = Blocks::new(provider.clone(), downloader.clone());
-    let input = StageExecutionInput { from: block_number, to: block_number };
+    let input = StageExecutionInput::new(block_number, block_number);
 
     let result = stage.execute(&input).await;
 
@@ -294,29 +294,6 @@ async fn storage_failure_returns_error() {
     assert_eq!(provider.stored_block_count(), 0);
 }
 
-// This test is only testing the debug sanity check in Blocks::execute(). Becase the
-// `BlockDownloader` implementation could theoretically return whatever based on the block input
-// because the input of `BlockDownloader::download_blocks` doesn't prohibit invalid block range.
-// Maybe that's a good reason to change its method signature to `fn download_blocks(&self, range:
-// Range<BlockNumber>)` ??
-#[tokio::test]
-#[ignore = "Stage input validation should be done on the `Pipeline` level"]
-async fn empty_range_downloads_nothing() {
-    // When from > to, the range is empty
-    let downloader = MockBlockDownloader::new();
-    let provider = MockBlockWriter::new();
-
-    let mut stage = Blocks::new(provider.clone(), downloader.clone());
-    let input = StageExecutionInput { from: 100, to: 99 };
-
-    let result = stage.execute(&input).await;
-    assert!(result.is_ok());
-
-    // No downloads should occur for empty range
-    assert_eq!(downloader.requested_blocks().len(), 0);
-    assert_eq!(provider.stored_block_count(), 0);
-}
-
 #[tokio::test]
 async fn partial_download_failure_stops_execution() {
     let from_block = 100;
@@ -332,7 +309,7 @@ async fn partial_download_failure_stops_execution() {
     let provider = MockBlockWriter::new();
     let mut stage = Blocks::new(provider.clone(), downloader.clone());
 
-    let input = StageExecutionInput { from: from_block, to: to_block };
+    let input = StageExecutionInput::new(from_block, to_block);
     let result = stage.execute(&input).await;
 
     // Should fail on block 103
@@ -355,7 +332,7 @@ async fn fetch_blocks_from_gateway() {
 
     let mut stage = Blocks::new(&provider, downloader);
 
-    let input = StageExecutionInput { from: from_block, to: to_block };
+    let input = StageExecutionInput::new(from_block, to_block);
     stage.execute(&input).await.expect("failed to execute stage");
 
     // check provider storage