refactor(stages): empty transactions range (paradigmxyz#19753)

Author: shekhirin

crates/stages/api/src/stage.rs

@@ -86,24 +86,22 @@ impl ExecInput {
     /// Return the next block range determined the number of transactions within it.
     /// This function walks the block indices until either the end of the range is reached or
     /// the number of transactions exceeds the threshold.
+    ///
+    /// Returns [`None`] if no transactions are found for the current execution input.
     #[instrument(level = "debug", target = "sync::stages", skip(provider), ret)]
     pub fn next_block_range_with_transaction_threshold<Provider>(
         &self,
         provider: &Provider,
         tx_threshold: u64,
-    ) -> Result<TransactionRangeOutput, StageError>
+    ) -> Result<Option<TransactionRangeOutput>, StageError>
     where
         Provider: StaticFileProviderFactory + BlockReader,
     {
         // Get lowest available block number for transactions
         let Some(lowest_transactions_block) =
             provider.static_file_provider().get_lowest_range_start(StaticFileSegment::Transactions)
         else {
-            return Ok(TransactionRangeOutput {
-                tx_range: 0..0,
-                block_range: 0..=0,
-                is_final_range: true,
-            });
+            return Ok(None)
         };
 
         // We can only process transactions that have associated static files, so we cap the start
@@ -115,6 +113,13 @@ impl ExecInput {
         let start_block = self.next_block().max(lowest_transactions_block);
         let target_block = self.target();
 
+        // If the start block is greater than the target, then there's no transactions to process
+        // and we return early. It's possible to trigger this scenario when running `reth
+        // stage run` manually for a range of transactions that doesn't exist.
+        if start_block > target_block {
+            return Ok(None)
+        }
+
         let start_block_body = provider
             .block_body_indices(start_block)?
             .ok_or(ProviderError::BlockBodyIndicesNotFound(start_block))?;
@@ -129,11 +134,7 @@ impl ExecInput {
 
         if all_tx_cnt == 0 {
             // if there is no more transaction return back.
-            return Ok(TransactionRangeOutput {
-                tx_range: first_tx_num..first_tx_num,
-                block_range: start_block..=target_block,
-                is_final_range: true,
-            })
+            return Ok(None)
         }
 
         // get block of this tx
@@ -154,11 +155,11 @@ impl ExecInput {
         };
 
         let tx_range = first_tx_num..next_tx_num;
-        Ok(TransactionRangeOutput {
+        Ok(Some(TransactionRangeOutput {
             tx_range,
             block_range: start_block..=end_block,
             is_final_range,
-        })
+        }))
     }
 }
 
@@ -355,9 +356,7 @@ mod tests {
             let range_output = exec_input
                 .next_block_range_with_transaction_threshold(&provider_factory, 10)
                 .unwrap();
-            assert_eq!(range_output.tx_range, 0..0);
-            assert_eq!(range_output.block_range, 0..=0);
-            assert!(range_output.is_final_range);
+            assert!(range_output.is_none());
         }
 
         // With checkpoint at block 10, without transactions in static files
@@ -368,9 +367,7 @@ mod tests {
             let range_output = exec_input
                 .next_block_range_with_transaction_threshold(&provider_factory, 10)
                 .unwrap();
-            assert_eq!(range_output.tx_range, 0..0);
-            assert_eq!(range_output.block_range, 0..=0);
-            assert!(range_output.is_final_range);
+            assert!(range_output.is_none());
         }
 
         // Without checkpoint, with transactions in static files starting from block 1
@@ -396,6 +393,7 @@ mod tests {
 
             let range_output = exec_input
                 .next_block_range_with_transaction_threshold(&provider_factory, 10)
+                .unwrap()
                 .unwrap();
             assert_eq!(range_output.tx_range, 0..2);
             assert_eq!(range_output.block_range, 1..=1);
@@ -424,6 +422,7 @@ mod tests {
 
             let range_output = exec_input
                 .next_block_range_with_transaction_threshold(&provider_factory, 10)
+                .unwrap()
                 .unwrap();
             assert_eq!(range_output.tx_range, 2..3);
             assert_eq!(range_output.block_range, 2..=2);
@@ -445,6 +444,7 @@ mod tests {
 
             let range_output = exec_input
                 .next_block_range_with_transaction_threshold(&provider_factory, 10)
+                .unwrap()
                 .unwrap();
             assert_eq!(range_output.tx_range, 2..3);
             assert_eq!(range_output.block_range, 2..=2);
@@ -473,6 +473,7 @@ mod tests {
 
             let range_output = exec_input
                 .next_block_range_with_transaction_threshold(&provider_factory, 10)
+                .unwrap()
                 .unwrap();
             assert_eq!(range_output.tx_range, 3..4);
             assert_eq!(range_output.block_range, 3..=3);

crates/stages/stages/src/stages/sender_recovery.rs

@@ -80,19 +80,17 @@ where
             return Ok(ExecOutput::done(input.checkpoint()))
         }
 
-        let range_output =
-            input.next_block_range_with_transaction_threshold(provider, self.commit_threshold)?;
-        let end_block = *range_output.block_range.end();
-
-        // No transactions to walk over
-        if range_output.tx_range.is_empty() {
-            info!(target: "sync::stages::sender_recovery", tx_range = ?range_output.tx_range, "Target transaction already reached");
+        let Some(range_output) =
+            input.next_block_range_with_transaction_threshold(provider, self.commit_threshold)?
+        else {
+            info!(target: "sync::stages::sender_recovery", "No transaction senders to recover");
             return Ok(ExecOutput {
-                checkpoint: StageCheckpoint::new(end_block)
+                checkpoint: StageCheckpoint::new(input.target())
                     .with_entities_stage_checkpoint(stage_checkpoint(provider)?),
-                done: range_output.is_final_range,
+                done: true,
             })
-        }
+        };
+        let end_block = *range_output.block_range.end();
 
         // Acquire the cursor for inserting elements
         let mut senders_cursor = provider.tx_ref().cursor_write::<tables::TransactionSenders>()?;

crates/stages/stages/src/stages/tx_lookup.rs

@@ -126,8 +126,15 @@ where
         );
 
         loop {
-            let range_output =
-                input.next_block_range_with_transaction_threshold(provider, self.chunk_size)?;
+            let Some(range_output) =
+                input.next_block_range_with_transaction_threshold(provider, self.chunk_size)?
+            else {
+                input.checkpoint = Some(
+                    StageCheckpoint::new(input.target())
+                        .with_entities_stage_checkpoint(stage_checkpoint(provider)?),
+                );
+                break;
+            };
 
             let end_block = *range_output.block_range.end();