fix(staged-sync): prevent StaticFileProducer from running with an unwinded target on legacy engine (#11717)

crates/consensus/beacon/src/engine/hooks/static_file.rs

@@ -9,7 +9,8 @@ use futures::FutureExt;
 use reth_errors::RethResult;
 use reth_primitives::static_file::HighestStaticFiles;
 use reth_provider::{
-    BlockReader, DatabaseProviderFactory, StageCheckpointReader, StaticFileProviderFactory,
+    BlockReader, ChainStateBlockReader, DatabaseProviderFactory, StageCheckpointReader,
+    StaticFileProviderFactory,
 };
 use reth_static_file::{StaticFileProducer, StaticFileProducerWithResult};
 use reth_tasks::TaskSpawner;
@@ -31,8 +32,9 @@ pub struct StaticFileHook<Provider> {
 impl<Provider> StaticFileHook<Provider>
 where
     Provider: StaticFileProviderFactory
-        + DatabaseProviderFactory<Provider: StageCheckpointReader + BlockReader>
+        + DatabaseProviderFactory<
-        + 'static,
+            Provider: StageCheckpointReader + BlockReader + ChainStateBlockReader,
+        > + 'static,
 {
     /// Create a new instance
     pub fn new(
@@ -104,6 +106,11 @@ where
                     return Ok(None)
                 };
 
+                let finalized_block_number = locked_static_file_producer
+                    .last_finalized_block()?
+                    .map(|on_disk| finalized_block_number.min(on_disk))
+                    .unwrap_or(finalized_block_number);
+
                 let targets =
                     locked_static_file_producer.get_static_file_targets(HighestStaticFiles {
                         headers: Some(finalized_block_number),
@@ -137,8 +144,9 @@ where
 impl<Provider> EngineHook for StaticFileHook<Provider>
 where
     Provider: StaticFileProviderFactory
-        + DatabaseProviderFactory<Provider: StageCheckpointReader + BlockReader>
+        + DatabaseProviderFactory<
-        + 'static,
+            Provider: StageCheckpointReader + BlockReader + ChainStateBlockReader,
+        > + 'static,
 {
     fn name(&self) -> &'static str {
         "StaticFile"

crates/stages/api/src/pipeline/mod.rs

@@ -276,6 +276,10 @@ impl<N: ProviderNodeTypes> Pipeline<N> {
         // Unwind stages in reverse order of execution
         let unwind_pipeline = self.stages.iter_mut().rev();
 
+        // Legacy Engine: This prevents a race condition in which the `StaticFileProducer` could
+        // attempt to proceed with a finalized block which has been unwinded
+        let _locked_sf_producer = self.static_file_producer.lock();
+
         let mut provider_rw = self.provider_factory.database_provider_rw()?;
 
         for stage in unwind_pipeline {

crates/static-file/static-file/src/static_file_producer.rs

@@ -5,8 +5,8 @@ use alloy_primitives::BlockNumber;
 use parking_lot::Mutex;
 use rayon::prelude::*;
 use reth_provider::{
-    providers::StaticFileWriter, BlockReader, DBProvider, DatabaseProviderFactory,
+    providers::StaticFileWriter, BlockReader, ChainStateBlockReader, DBProvider,
-    StageCheckpointReader, StaticFileProviderFactory,
+    DatabaseProviderFactory, StageCheckpointReader, StaticFileProviderFactory,
 };
 use reth_prune_types::PruneModes;
 use reth_stages_types::StageId;
@@ -106,6 +106,16 @@ impl<Provider> StaticFileProducerInner<Provider> {
     }
 }
 
+impl<Provider> StaticFileProducerInner<Provider>
+where
+    Provider: StaticFileProviderFactory + DatabaseProviderFactory<Provider: ChainStateBlockReader>,
+{
+    /// Returns the last finalized block number on disk.
+    pub fn last_finalized_block(&self) -> ProviderResult<Option<BlockNumber>> {
+        self.provider.database_provider_ro()?.last_finalized_block_number()
+    }
+}
+
 impl<Provider> StaticFileProducerInner<Provider>
 where
     Provider: StaticFileProviderFactory