fix: prevent endless retry when CL times out validating large payloads

crates/engine/tree/src/tree/mod.rs

@@ -545,6 +545,20 @@ where
         self.emit_event(EngineApiEvent::BeaconConsensus(engine_event));
 
         let block_hash = num_hash.hash;
+        // if this block hash was previously marked invalid, return INVALID immediately.
+        // This prevents infinite loops when the CL retries a payload
+        // that Reth already validated and rejected, but the CL timed out before receiving invalid resp.
+        if let Some(invalid) = self.state.invalid_headers.get(&block_hash) {
+            warn!(
+                target: "engine::tree",
+                %block_hash,
+                block_number = %num_hash.number,
+                "reject known invalid block"
+            );
+            let status = self.prepare_invalid_response(invalid.parent)?;
+            return Ok(TreeOutcome::new(status))
+        }
+
         let mut lowest_buffered_ancestor = self.lowest_buffered_ancestor_or(block_hash);
         if lowest_buffered_ancestor == block_hash {
             lowest_buffered_ancestor = parent_hash;
@@ -2354,6 +2368,30 @@ where
             _ => {}
         };
 
+        // When the CL times out and retries with the same payload
+        // we look up the block by (number, parent_hash) to find the previously executed result,
+        // avoiding redundant re-execution.
+        if let Some(executed) = self.state.tree_state.executed_block_by_number_and_parent(
+            block_num_hash.number,
+            block_id.parent,
+        ) {
+            let executed_hash = executed.recovered_block().hash();
+
+            convert_to_block(self, input)?;
+
+            // Return the execution result using the post-execution hash, so the CL
+            // receives the correct `latest_valid_hash` for FCU.
+            let requests = self.state.tree_state
+                .execution_requests_by_hash(&executed_hash)
+                .unwrap_or_default()
+                .take();
+
+            return Ok(InsertPayloadOk::AlreadySeen(BlockStatus::Valid {
+                head: BlockNumHash::new(block_num_hash.number, executed_hash),
+                requests,
+            }))
+        }
+
         // Ensure that the parent state is available.
         match self.state_provider_builder(block_id.parent) {
             Err(err) => {

crates/engine/tree/src/tree/state.rs

@@ -101,6 +101,30 @@ impl<N: NodePrimitives> TreeState<N> {
         self.blocks_by_hash.get(hash).map(|b| b.execution_outcome().requests.first().unwrap_or(&Requests::default()).clone())
     }
 
+    /// Finds an already-executed block by (block_number, parent_hash).
+    ///
+    /// This is a fallback for the PBFT consensus model where delayed execution causes
+    /// the block hash to change after execution (because `gas_used` is updated from 0
+    /// to the actual value). In this scenario, the CL's proposal hash differs from the
+    /// EL's post-execution hash, so the standard hash-based lookup fails.
+    ///
+    /// Under PBFT, the (block_number, parent_hash) pair uniquely identifies a block:
+    /// - `block_number` is deterministic (parent_number + 1)
+    /// - `parent_hash` is agreed upon by consensus
+    /// - The proposer determines the transactions, which remain the same across retries
+    ///
+    /// Returns the executed block's hash if found (the post-execution hash).
+    pub(crate) fn executed_block_by_number_and_parent(
+        &self,
+        block_number: BlockNumber,
+        parent_hash: B256,
+    ) -> Option<&ExecutedBlockWithTrieUpdates<N>> {
+        self.blocks_by_number
+            .get(&block_number)?
+            .iter()
+            .find(|b| b.recovered_block().parent_hash() == parent_hash)
+    }
+
     /// Returns all available blocks for the given hash that lead back to the canonical chain, from
     /// newest to oldest. And the parent hash of the oldest block that is missing from the buffer.
     ///