diff --git a/crates/consensus/beacon/src/engine/invalid_headers.rs b/crates/consensus/beacon/src/engine/invalid_headers.rs
index fbe6bf462b..b93138901d 100644
--- a/crates/consensus/beacon/src/engine/invalid_headers.rs
+++ b/crates/consensus/beacon/src/engine/invalid_headers.rs
@@ -67,7 +67,7 @@ impl InvalidHeaderCache {
     }
 
     /// Inserts an invalid ancestor into the map.
-    pub(crate) fn insert(&mut self, invalid_ancestor: SealedHeader) {
+    pub fn insert(&mut self, invalid_ancestor: SealedHeader) {
         if self.get(&invalid_ancestor.hash()).is_none() {
             let hash = invalid_ancestor.hash();
             let header = invalid_ancestor.unseal();
diff --git a/crates/engine/tree/src/tree/mod.rs b/crates/engine/tree/src/tree/mod.rs
index 9fa1f247f6..c3c1aa8185 100644
--- a/crates/engine/tree/src/tree/mod.rs
+++ b/crates/engine/tree/src/tree/mod.rs
@@ -35,6 +35,7 @@ use reth_rpc_types::{
     },
     ExecutionPayload,
 };
+use reth_stages_api::ControlFlow;
 use reth_trie::HashedPostState;
 use std::{
     collections::{BTreeMap, HashMap, HashSet},
@@ -340,9 +341,8 @@ where
                         debug!(target: "consensus::engine", "received backfill sync started event");
                         self.is_backfill_active = true;
                     }
-                    FromOrchestrator::BackfillSyncFinished(_ctrl) => {
-                        debug!(target: "consensus::engine", "received backfill sync finished event");
-                        self.is_backfill_active = false;
+                    FromOrchestrator::BackfillSyncFinished(ctrl) => {
+                        self.on_backfill_sync_finished(ctrl);
                     }
                 },
                 FromEngine::Request(request) => match request {
@@ -409,6 +409,63 @@ where
         }
     }
 
+    /// Invoked if the backfill sync has finished to target.
+    ///
+    /// Checks the tracked finalized block against the block on disk and restarts backfill if
+    /// needed.
+    ///
+    /// This will also try to connect the buffered blocks.
+    fn on_backfill_sync_finished(&mut self, ctrl: ControlFlow) {
+        debug!(target: "consensus::engine", "received backfill sync finished event");
+        self.is_backfill_active = false;
+
+        // Pipeline unwound, memorize the invalid block and wait for CL for next sync target.
+        if let ControlFlow::Unwind { bad_block, .. } = ctrl {
+            warn!(target: "consensus::engine", invalid_hash=?bad_block.hash(), invalid_number=?bad_block.number, "Bad block detected in unwind");
+            // update the `invalid_headers` cache with the new invalid header
+            self.state.invalid_headers.insert(*bad_block);
+            return
+        }
+
+        let Some(sync_target_state) = self.state.forkchoice_state_tracker.sync_target_state()
+        else {
+            return
+        };
+
+        if sync_target_state.finalized_block_hash.is_zero() {
+            return
+        }
+
+        // get the block number of the finalized block, if we have it
+        let newest_finalized = self
+            .state
+            .buffer
+            .block(&sync_target_state.finalized_block_hash)
+            .map(|block| block.number);
+
+        // TODO: state housekeeping
+
+        // The block number that the backfill finished at - if the progress or newest
+        // finalized is None then we can't check the distance anyways.
+        //
+        // If both are Some, we perform another distance check and return the desired
+        // backfill target
+        let Some(backfill_target) =
+            ctrl.block_number().zip(newest_finalized).and_then(|(progress, finalized_number)| {
+                // Determines whether or not we should run backfill again, in case
+                // the new gap is still large enough and requires running backfill again
+                self.backfill_sync_target(progress, finalized_number, None)
+            })
+        else {
+            return
+        };
+
+        // request another backfill run
+        self.emit_event(EngineApiEvent::BackfillAction(BackfillAction::Start(
+            backfill_target.into(),
+        )));
+    }
+
     /// Attempts to make the given target canonical.
     ///
     /// This will update the tracked canonical in memory state and do the necessary housekeeping.
@@ -723,12 +780,10 @@ where
             }
         }
 
-        // if the number of missing blocks is greater than the max, run the
-        // pipeline
+        // if the number of missing blocks is greater than the max, trigger backfill
         if exceeds_backfill_threshold {
             if let Some(state) = sync_target_state {
-                // if we have already canonicalized the finalized block, we should
-                // skip the pipeline run
+                // if we have already canonicalized the finalized block, we should skip backfill
                 match self.provider.header_by_hash_or_number(state.finalized_block_hash.into()) {
                     Err(err) => {
                         warn!(target: "consensus::engine", %err, "Failed to get finalized block header");
@@ -753,7 +808,7 @@ where
                         //
                         // However, optimism chains will do this. The risk of a reorg is however
                         // low.
-                        debug!(target: "consensus::engine", hash=?state.head_block_hash, "Setting head hash as an optimistic pipeline target.");
+                        debug!(target: "consensus::engine", hash=?state.head_block_hash, "Setting head hash as an optimistic backfill target.");
                         return Some(state.head_block_hash)
                     }
                     Ok(Some(_)) => {
@@ -790,7 +845,7 @@ where
         //    * this case represents a missing block on a fork that is shorter than the canonical
         //      chain
         //  * the missing parent block num >= canonical tip num, but the number of missing blocks is
-        //    less than the pipeline threshold
+        //    less than the backfill threshold
         //    * this case represents a potentially long range of blocks to download and execute
         let request = if let Some(distance) =
             self.distance_from_local_tip(head.number, missing_parent.number)