From f50b97a94d51fff748228fb3ae085b880bc48fc1 Mon Sep 17 00:00:00 2001 From: Francis Li Date: Fri, 13 Nov 2026 20:44:02 -0800 Subject: [PATCH 1/7] perf(stages): commit incremental merkle per chunk --- crates/stages/stages/src/stages/merkle.rs | 87 ++++++++++++++--------- 1 file changed, 55 insertions(+), 32 deletions(-) diff --git a/crates/stages/stages/src/stages/merkle.rs b/crates/stages/stages/src/stages/merkle.rs index 5c7b7729bc8..9bd2f518e70 100644 --- a/crates/stages/stages/src/stages/merkle.rs +++ b/crates/stages/stages/src/stages/merkle.rs @@ -313,33 +313,32 @@ where } } else { debug!(target: "sync::stages::merkle::exec", current = ?current_block_number, target = ?to_block, "Updating trie in chunks"); - let mut final_root = None; - for start_block in range.step_by(incremental_threshold as usize) { - let chunk_to = std::cmp::min(start_block + incremental_threshold - 1, to_block); - let chunk_range = start_block..=chunk_to; - debug!( - target: "sync::stages::merkle::exec", - current = ?current_block_number, - target = ?to_block, - incremental_threshold, - chunk_range = ?chunk_range, - "Processing chunk" - ); - let (root, updates) = reth_trie_db::with_adapter!(provider, |A| { - DbStateRoot::<_, A>::incremental_root_with_updates(provider, chunk_range) - }) - .map_err(|e| { - error!(target: "sync::stages::merkle", %e, ?current_block_number, ?to_block, "Incremental state root failed! {INVALID_STATE_ROOT_ERROR_MESSAGE}"); - StageError::Fatal(Box::new(e)) - })?; - provider.write_trie_updates(updates)?; - final_root = Some(root); - } - - // if we had no final root, we must have not looped above, which should not be possible - let final_root = final_root.ok_or(StageError::Fatal( - "Incremental merkle hashing did not produce a final root".into(), - ))?; + // One chunk per call: the pipeline commits after each call, so the uncommitted trie + // updates stay bounded by one chunk and a restart resumes after the last committed one. + // + // Hashed state is already at the pipeline target, so a chunk's root is not the root of + // its last block and only the final chunk can be validated. The committed trie holds + // target values for every key changed up to the checkpoint; the remaining range's + // prefix sets cover every key that changes after it, so resuming forward, even toward a + // newer target, still ends at the right root. + let chunk = input.next_block_range_with_threshold(incremental_threshold); + let chunk_to = *chunk.block_range.end(); + debug!( + target: "sync::stages::merkle::exec", + current = ?current_block_number, + target = ?to_block, + incremental_threshold, + chunk_range = ?chunk.block_range, + "Processing chunk" + ); + let (root, updates) = reth_trie_db::with_adapter!(provider, |A| { + DbStateRoot::<_, A>::incremental_root_with_updates(provider, chunk.block_range) + }) + .map_err(|e| { + error!(target: "sync::stages::merkle", %e, ?current_block_number, ?to_block, "Incremental state root failed! {INVALID_STATE_ROOT_ERROR_MESSAGE}"); + StageError::Fatal(Box::new(e)) + })?; + provider.write_trie_updates(updates)?; let total_hashed_entries = (provider.count_entries::()? + provider.count_entries::()?) @@ -352,8 +351,16 @@ where processed: total_hashed_entries, total: total_hashed_entries, }; + + if !chunk.is_final_range { + return Ok(ExecOutput::in_progress( + StageCheckpoint::new(chunk_to) + .with_entities_stage_checkpoint(entities_checkpoint), + )) + } + // Save the checkpoint - (final_root, entities_checkpoint) + (root, entities_checkpoint) }; // Reset the checkpoint @@ -579,16 +586,32 @@ mod tests { let mut runner = MerkleTestRunner { db: TestStageDB::default(), clean_threshold, incremental_threshold }; - let input = ExecInput { + let mut input = ExecInput { target: Some(previous_stage), checkpoint: Some(StageCheckpoint::new(stage_progress)), }; runner.seed_execution(input).expect("failed to seed execution"); - let rx = runner.execute(input); - // Assert the successful result - let result = rx.await.unwrap(); + // Each call commits one chunk and stops at its last block until the target is reached. + let mut chunk_ends = Vec::new(); + let result = loop { + let result = runner.execute(input).await.unwrap(); + match result { + Ok(ExecOutput { checkpoint, done: false }) => { + chunk_ends.push(checkpoint.block_number); + input.checkpoint = Some(checkpoint); + } + result => break result, + } + }; + assert_eq!( + chunk_ends, + (stage_progress + incremental_threshold..previous_stage) + .step_by(incremental_threshold as usize) + .collect::>() + ); + assert_matches!( result, Ok(ExecOutput { From e86b435e4c1ba4984d96e561899b16fe0784b0a4 Mon Sep 17 00:00:00 2001 From: Francis Li Date: Fri, 13 Nov 2026 21:13:06 -0800 Subject: [PATCH 2/7] fix(stages): make chunked merkle safe to retry and unwind Finish an interrupted rebuild instead of updating its partial trie incrementally, and have MerkleUnwind recompute from MerkleExecute's checkpoint when it stopped short of the unwind target. Tests cover retrying an uncommitted chunk, resuming toward a newer target, an interrupted rebuild, and unwinding on either side of the last committed chunk. --- crates/stages/stages/src/stages/merkle.rs | 365 ++++++++++++++++++++-- 1 file changed, 334 insertions(+), 31 deletions(-) diff --git a/crates/stages/stages/src/stages/merkle.rs b/crates/stages/stages/src/stages/merkle.rs index 9bd2f518e70..a097fb34455 100644 --- a/crates/stages/stages/src/stages/merkle.rs +++ b/crates/stages/stages/src/stages/merkle.rs @@ -213,11 +213,13 @@ where .ok_or_else(|| ProviderError::HeaderNotFound(to_block.into()))?; let target_block_root = target_block.state_root(); + // A rebuild clears the trie tables before it starts, so if one was interrupted, finish it: + // updating the partial trie incrementally would recompute every missing node in memory. + let mut checkpoint = self.get_execution_checkpoint(provider)?; + let (trie_root, entities_checkpoint) = if range.is_empty() { (target_block_root, input.checkpoint().entities_stage_checkpoint().unwrap_or_default()) - } else if to_block - from_block > threshold || from_block == 1 { - let mut checkpoint = self.get_execution_checkpoint(provider)?; - + } else if to_block - from_block > threshold || from_block == 1 || checkpoint.is_some() { // if there are more blocks than threshold it is faster to rebuild the trie let mut entities_checkpoint = if let Some(checkpoint) = checkpoint.as_ref().filter(|c| c.target_block == to_block) @@ -380,11 +382,18 @@ where input: UnwindInput, ) -> Result { let tx = provider.tx_ref(); - let range = input.unwind_block_range(); if matches!(self, Self::Execution { .. }) { info!(target: "sync::stages::merkle::unwind", "Stage is always skipped"); return Ok(UnwindOutput { checkpoint: StageCheckpoint::new(input.unwind_to) }) } + // MerkleExecute is unwound first, so its checkpoint is `unwind_to` unless it had not + // reached that block. Incremental chunks commit trie updates only up to that + // checkpoint; keys changed after it can still have stale nodes, so they must be + // recomputed too. + let merkle_execute = provider + .get_stage_checkpoint(StageId::MerkleExecute)? + .map_or(input.unwind_to, |checkpoint| checkpoint.block_number); + let range = merkle_execute.min(input.unwind_to) + 1..=input.checkpoint.block_number; let walk_all_changed_branch_children = match self { Self::Unwind { walk_all_changed_branch_children } => *walk_all_changed_branch_children, #[cfg(any(test, feature = "test-utils"))] @@ -479,18 +488,26 @@ mod tests { stage_test_suite_ext, ExecuteStageTestRunner, StageTestRunner, StorageKind, TestRunnerError, TestStageDB, UnwindStageTestRunner, }; - use alloy_primitives::{keccak256, U256}; + use alloy_primitives::{keccak256, Address, U256}; use assert_matches::assert_matches; use reth_db_api::cursor::{DbCursorRO, DbCursorRW, DbDupCursorRO}; - use reth_primitives_traits::{SealedBlock, StorageEntry}; - use reth_provider::{providers::StaticFileWriter, StaticFileProviderFactory}; + use reth_primitives_traits::{Account, SealedBlock, StorageEntry}; + use reth_provider::{ + providers::StaticFileWriter, BlockNumReader, ProviderResult, StaticFileProviderFactory, + }; use reth_stages_api::StageUnitCheckpoint; use reth_static_file_types::StaticFileSegment; use reth_testing_utils::generators::{ self, random_block, random_block_range, random_changeset_range, random_contract_account_range, BlockParams, BlockRangeParams, }; - use reth_trie::test_utils::{state_root, state_root_prehashed}; + use reth_trie::{ + test_utils::{state_root, state_root_prehashed}, + trie_cursor::{TrieCursor, TrieCursorFactory}, + updates::{StorageTrieUpdates, TrieUpdates}, + BranchNodeCompact, Nibbles, + }; + use reth_trie_db::DatabaseTrieCursorFactory; use std::collections::BTreeMap; stage_test_suite_ext!(MerkleTestRunner, merkle); @@ -654,6 +671,314 @@ mod tests { ); } + /// A crash before a chunk commits loses only that chunk: retrying the same input finishes, and + /// the stage validates the final root against the target header. + #[tokio::test] + async fn incremental_merkle_retries_uncommitted_chunk() { + let (runner, mut input) = start_round().await; + + for _ in 0..2 { + let output = runner.execute(input).await.unwrap().unwrap(); + input.checkpoint = Some(output.checkpoint); + } + let provider = runner.db.factory.provider_rw().unwrap(); + let uncommitted = runner.stage().execute(&*provider, input).unwrap(); + assert_eq!(uncommitted.checkpoint.block_number, ROUND_START + 30); + drop(provider); + + let output = execute_until_done(&runner, input).await.unwrap(); + assert_eq!(output.checkpoint.block_number, ROUND_TARGET); + } + + /// A round interrupted after some chunks resumes from its checkpoint when the next round + /// targets a later block, and the stage validates the root at that block. + #[tokio::test] + async fn incremental_merkle_resumes_toward_newer_target() { + let (runner, mut input) = start_round().await; + + input.target = Some(150); + for _ in 0..3 { + let output = runner.execute(input).await.unwrap().unwrap(); + assert!(!output.done); + input.checkpoint = Some(output.checkpoint); + } + assert_eq!(input.checkpoint().block_number, ROUND_START + 30); + + input.target = Some(ROUND_TARGET); + let output = execute_until_done(&runner, input).await.unwrap(); + assert_eq!(output.checkpoint.block_number, ROUND_TARGET); + } + + /// A short range after an interrupted rebuild finishes the rebuild instead of updating the + /// partial trie incrementally. + #[tokio::test] + async fn interrupted_rebuild_is_finished_on_short_range() { + let (target, round_start) = (110, 100); + let mut runner = MerkleTestRunner { + db: TestStageDB::default(), + clean_threshold: 100, + incremental_threshold: 10, + }; + let input = + ExecInput { target: Some(target), checkpoint: Some(StageCheckpoint::new(round_start)) }; + runner.seed_execution(input).expect("failed to seed execution"); + + // The interrupted rebuild left its inner checkpoint and a storage trie node for an account + // that does not exist. Incremental updates never visit that storage trie; a rebuild + // clears it. + let stray_account = B256::repeat_byte(0xab); + let stray_path = Nibbles::from_nibbles([0xf, 0xf]); + let provider = runner.db.factory.provider_rw().unwrap(); + let mut storage_updates = StorageTrieUpdates::default(); + storage_updates.storage_nodes.insert( + stray_path, + BranchNodeCompact::new( + 0b11, + 0, + 0b11, + vec![B256::repeat_byte(1), B256::repeat_byte(2)], + None, + ), + ); + let mut updates = TrieUpdates::default(); + updates.storage_tries.insert(stray_account, storage_updates); + provider.write_trie_updates(updates).unwrap(); + runner + .stage() + .save_execution_checkpoint( + &*provider, + Some(MerkleCheckpoint::new(target + 1, B256::ZERO, Vec::new(), Default::default())), + ) + .unwrap(); + provider.commit().unwrap(); + + let output = execute_until_done(&runner, input).await.unwrap(); + assert_eq!(output.checkpoint.block_number, target); + + let provider = runner.db.factory.provider().unwrap(); + assert!(runner.stage().get_execution_checkpoint(&provider).unwrap().is_none()); + let stray = reth_trie_db::with_adapter!(provider, |A| { + DatabaseTrieCursorFactory::<_, A>::new(provider.tx_ref()) + .storage_trie_cursor(stray_account) + .unwrap() + .seek_exact(stray_path) + .unwrap() + }); + assert!(stray.is_none(), "the rebuild clears the partial trie"); + } + + #[tokio::test] + async fn unwind_before_last_committed_chunk() { + unwind_after_partial_incremental_merkle(105).await; + } + + #[tokio::test] + async fn unwind_past_last_committed_chunk() { + unwind_after_partial_incremental_merkle(190).await; + } + + /// Unwinds to `unwind_to` after `MerkleExecute` committed one chunk, through block 110, of the + /// round from [`start_round`]. + async fn unwind_after_partial_incremental_merkle(unwind_to: BlockNumber) { + let last_chunk_end = ROUND_START + 10; + let (runner, mut input) = start_round().await; + while input.checkpoint().block_number < last_chunk_end { + let output = runner.execute(input).await.unwrap().unwrap(); + input.checkpoint = Some(output.checkpoint); + } + + // The pipeline unwinds MerkleExecute first, which only moves its checkpoint down. + let provider = runner.db.factory.provider_rw().unwrap(); + provider + .save_stage_checkpoint( + StageId::MerkleExecute, + StageCheckpoint::new(last_chunk_end.min(unwind_to)), + ) + .unwrap(); + provider.commit().unwrap(); + + let unwind = UnwindInput { + checkpoint: StageCheckpoint::new(ROUND_TARGET), + unwind_to, + bad_block: None, + }; + runner.before_unwind(unwind).unwrap(); + set_header_state_root(&runner.db, unwind_to, hashed_state_root(&runner.db).unwrap()); + + // The stage validates the recomputed root against the header of `unwind_to`. + let output = runner.unwind(unwind).await.unwrap(); + assert_eq!(output.checkpoint.block_number, unwind_to); + } + + const ROUND_START: BlockNumber = 100; + const ROUND_TARGET: BlockNumber = 200; + /// Block at which every tenth extra account in [`start_round`] changes, and nothing else. + const LATE_CHANGE_BLOCK: BlockNumber = 151; + + /// Sets up a pipeline round as `MerkleExecute` finds it: the trie built at block 100, hashed + /// state at block 200, and `MerkleUnwind` already at 200. The state is wide enough that the + /// trie stores branch hashes below the root, so trie nodes the stage fails to update stay + /// stale and break the next validated root. + async fn start_round() -> (MerkleTestRunner, ExecInput) { + let mut runner = MerkleTestRunner { + db: TestStageDB::default(), + clean_threshold: 1000, + incremental_threshold: 10, + }; + let input = ExecInput { + target: Some(ROUND_TARGET), + checkpoint: Some(StageCheckpoint::new(ROUND_START)), + }; + runner.seed_execution(input).expect("failed to seed execution"); + + let extra_accounts = (1..=2000u64) + .map(|i| { + let address = Address::from_word(B256::from(U256::from(i))); + (address, Account { nonce: 1, balance: U256::from(i), bytecode_hash: None }) + }) + .collect::>(); + runner + .db + .insert_accounts_and_storages( + extra_accounts + .iter() + .map(|(address, account)| (*address, (*account, std::iter::empty()))), + ) + .unwrap(); + let late_changes = extra_accounts.iter().step_by(10); + runner + .db + .insert_changesets( + [late_changes + .clone() + .map(|(address, account)| (*address, *account, Vec::new())) + .collect::>()], + Some(LATE_CHANGE_BLOCK), + ) + .unwrap(); + runner + .db + .insert_accounts_and_storages(late_changes.map(|(address, account)| { + let account = Account { balance: account.balance + U256::from(1), ..*account }; + (*address, (account, std::iter::empty())) + })) + .unwrap(); + + // Give the start and target headers the roots of the widened state. + set_header_state_root(&runner.db, ROUND_TARGET, hashed_state_root(&runner.db).unwrap()); + let target_hashed_state = hashed_state(&runner.db); + let to_round_start = UnwindInput { + checkpoint: StageCheckpoint::new(ROUND_TARGET), + unwind_to: ROUND_START, + bad_block: None, + }; + runner.before_unwind(to_round_start).unwrap(); + set_header_state_root(&runner.db, ROUND_START, hashed_state_root(&runner.db).unwrap()); + restore_hashed_state(&runner.db, target_hashed_state.clone()); + + // Build the trie at the round start: merkelize the target, then unwind to the start. + execute_until_done(&runner, input).await.unwrap(); + runner.before_unwind(to_round_start).unwrap(); + runner.unwind(to_round_start).await.unwrap(); + restore_hashed_state(&runner.db, target_hashed_state); + + (runner, input) + } + + /// Drives the stage to completion one committed call at a time, as the pipeline does. + async fn execute_until_done( + runner: &MerkleTestRunner, + mut input: ExecInput, + ) -> Result { + loop { + match runner.execute(input).await.unwrap()? { + ExecOutput { checkpoint, done: false } => input.checkpoint = Some(checkpoint), + output => return Ok(output), + } + } + } + + type HashedState = (Vec<(B256, Account)>, Vec<(B256, StorageEntry)>); + + fn hashed_state(db: &TestStageDB) -> HashedState { + db.query(|tx| { + let accounts = tx + .cursor_read::()? + .walk(None)? + .collect::, _>>()?; + let storages = tx + .cursor_dup_read::()? + .walk(None)? + .collect::, _>>()?; + Ok((accounts, storages)) + }) + .unwrap() + } + + fn restore_hashed_state(db: &TestStageDB, (accounts, storages): HashedState) { + db.commit(|tx| { + tx.clear::()?; + tx.clear::()?; + for (hashed_address, account) in accounts { + tx.put::(hashed_address, account)?; + } + for (hashed_address, entry) in storages { + tx.put::(hashed_address, entry)?; + } + Ok(()) + }) + .unwrap(); + } + + /// Computes the state root of the hashed state tables from scratch. + fn hashed_state_root(db: &TestStageDB) -> ProviderResult { + db.query(|tx| { + let mut accounts = BTreeMap::default(); + let mut accounts_cursor = tx.cursor_read::()?; + let mut storage_cursor = tx.cursor_dup_read::()?; + for entry in accounts_cursor.walk_range(..)? { + let (key, account) = entry?; + let mut storage_entries = Vec::new(); + let mut entry = storage_cursor.seek_exact(key)?; + while let Some((_, storage)) = entry { + storage_entries.push(storage); + entry = storage_cursor.next_dup()?; + } + let storage = storage_entries + .into_iter() + .filter(|v| !v.value.is_zero()) + .map(|v| (v.key, v.value)) + .collect::>(); + accounts.insert(key, (account, storage)); + } + + Ok(state_root_prehashed(accounts)) + }) + } + + /// Rewrites the header of `block` with `state_root`, re-appending the headers after it. + fn set_header_state_root(db: &TestStageDB, block: BlockNumber, state_root: B256) { + let provider = db.factory.provider().unwrap(); + let last = provider.last_block_number().unwrap(); + let headers = (block..=last) + .map(|number| provider.header_by_number(number).unwrap().unwrap()) + .collect::>(); + drop(provider); + + let static_file_provider = db.factory.static_file_provider(); + let mut writer = static_file_provider.latest_writer(StaticFileSegment::Headers).unwrap(); + writer.prune_headers(headers.len() as u64).unwrap(); + writer.commit().unwrap(); + for (i, mut header) in headers.into_iter().enumerate() { + if i == 0 { + header.state_root = state_root; + } + let hash = header.hash_slow(); + writer.append_header(&header, &hash).unwrap(); + } + writer.commit().unwrap(); + } + struct MerkleTestRunner { db: TestStageDB, clean_threshold: u64, @@ -758,29 +1083,7 @@ mod tests { self.db.insert_changesets(transitions, Some(start))?; self.db.insert_accounts_and_storages(final_state)?; - // Calculate state root - let root = self.db.query(|tx| { - let mut accounts = BTreeMap::default(); - let mut accounts_cursor = tx.cursor_read::()?; - let mut storage_cursor = tx.cursor_dup_read::()?; - for entry in accounts_cursor.walk_range(..)? { - let (key, account) = entry?; - let mut storage_entries = Vec::new(); - let mut entry = storage_cursor.seek_exact(key)?; - while let Some((_, storage)) = entry { - storage_entries.push(storage); - entry = storage_cursor.next_dup()?; - } - let storage = storage_entries - .into_iter() - .filter(|v| !v.value.is_zero()) - .map(|v| (v.key, v.value)) - .collect::>(); - accounts.insert(key, (account, storage)); - } - - Ok(state_root_prehashed(accounts)) - })?; + let root = hashed_state_root(&self.db)?; let static_file_provider = self.db.factory.static_file_provider(); let mut writer = From 948e600bdaa153563f1856f0bf25206ee4d475ca Mon Sep 17 00:00:00 2001 From: Francis Li Date: Fri, 13 Nov 2026 21:25:58 -0800 Subject: [PATCH 3/7] fix(stages): unwind merkle root errors to the last validated block A chunked MerkleExecute round commits checkpoints whose state root was never validated, so a validation error must not unwind to one. Finish only advances after MerkleExecute validated a range, so unwind to the lower of the previous checkpoint and the Finish checkpoint. --- crates/stages/api/src/pipeline/mod.rs | 67 +++++++++++++++++++++-- crates/stages/stages/src/stages/merkle.rs | 11 ++++ 2 files changed, 74 insertions(+), 4 deletions(-) diff --git a/crates/stages/api/src/pipeline/mod.rs b/crates/stages/api/src/pipeline/mod.rs index 72d65de6dc7..5acd83ed4c6 100644 --- a/crates/stages/api/src/pipeline/mod.rs +++ b/crates/stages/api/src/pipeline/mod.rs @@ -664,12 +664,20 @@ impl Pipeline { } } + // Incremental MerkleExecute commits its checkpoint after each chunk without validating + // that block's state root, so unwind to the last block whose root was validated: Finish + // only advances after MerkleExecute validated the end of a range. + let mut target = prev_checkpoint.unwrap_or_default().block_number; + if stage_id == StageId::MerkleExecute && + let Some(finish) = + self.provider_factory.provider()?.get_stage_checkpoint(StageId::Finish)? + { + target = target.min(finish.block_number); + } + // We unwind because of a block error. If the unwind itself fails, we bail entirely; // otherwise, we restart the execution loop from the beginning. - Ok(Some(ControlFlow::Unwind { - target: prev_checkpoint.unwrap_or_default().block_number, - bad_block: block, - })) + Ok(Some(ControlFlow::Unwind { target, bad_block: block })) } else if let StageError::MissingStaticFileData { block, segment } = err { error!( target: "sync::pipeline", @@ -1248,6 +1256,57 @@ mod tests { ); } + /// A state-root error after `MerkleExecute` committed unvalidated chunks unwinds to the last + /// validated block, the Finish checkpoint, not to the last committed chunk. + #[tokio::test] + async fn merkle_execute_validation_error_unwinds_to_finish_checkpoint() { + let provider_factory = create_test_provider_factory(); + let provider_rw = provider_factory.provider_rw().unwrap(); + provider_rw.save_stage_checkpoint(StageId::Finish, StageCheckpoint::new(100)).unwrap(); + provider_rw + .save_stage_checkpoint(StageId::MerkleExecute, StageCheckpoint::new(100)) + .unwrap(); + provider_rw.commit().unwrap(); + + let mut pipeline = Pipeline::::builder() + .add_stage( + TestStage::new(StageId::MerkleExecute) + .add_exec(Ok(ExecOutput::in_progress(StageCheckpoint::new(110)))) + .add_exec(Err(StageError::Block { + block: Box::new(random_block_with_parent( + &mut generators::rng(), + 200, + Default::default(), + )), + error: BlockErrorKind::Validation(ConsensusError::BaseFeeMissing), + })) + .add_unwind(Ok(UnwindOutput { checkpoint: StageCheckpoint::new(100) })) + .add_exec(Ok(ExecOutput::done(StageCheckpoint::new(200)))), + ) + .with_max_block(200) + .build( + provider_factory.clone(), + StaticFileProducer::new(provider_factory.clone(), PruneModes::default()), + ); + let events = pipeline.events(); + tokio::spawn(async move { + pipeline.run().await.expect("Could not run pipeline"); + }); + + let events = events.collect::>().await; + assert!( + events.contains(&PipelineEvent::Unwind { + stage_id: StageId::MerkleExecute, + input: UnwindInput { + checkpoint: StageCheckpoint::new(110), + unwind_to: 100, + bad_block: Some(200), + }, + }), + "{events:?}" + ); + } + /// Checks that the pipeline re-runs stages on non-fatal errors and stops on fatal ones. #[tokio::test] async fn pipeline_error_handling() { diff --git a/crates/stages/stages/src/stages/merkle.rs b/crates/stages/stages/src/stages/merkle.rs index a097fb34455..a7948e08eb4 100644 --- a/crates/stages/stages/src/stages/merkle.rs +++ b/crates/stages/stages/src/stages/merkle.rs @@ -696,6 +696,15 @@ mod tests { async fn incremental_merkle_resumes_toward_newer_target() { let (runner, mut input) = start_round().await; + // The interrupted round targeted block 150, so hashing had only reached that block. + let target_hashed_state = hashed_state(&runner.db); + runner + .before_unwind(UnwindInput { + checkpoint: StageCheckpoint::new(ROUND_TARGET), + unwind_to: 150, + bad_block: None, + }) + .unwrap(); input.target = Some(150); for _ in 0..3 { let output = runner.execute(input).await.unwrap().unwrap(); @@ -704,6 +713,8 @@ mod tests { } assert_eq!(input.checkpoint().block_number, ROUND_START + 30); + // The next round's hashing reaches block 200 before `MerkleExecute` resumes. + restore_hashed_state(&runner.db, target_hashed_state); input.target = Some(ROUND_TARGET); let output = execute_until_done(&runner, input).await.unwrap(); assert_eq!(output.checkpoint.block_number, ROUND_TARGET); From 9dd95a3268c52091ac75f33cc364ea9a2e40ca5f Mon Sep 17 00:00:00 2001 From: Francis Li Date: Fri, 13 Nov 2026 21:37:49 -0800 Subject: [PATCH 4/7] fix(stages): keep merkle checkpoint at the last validated block Advancing the MerkleExecute block checkpoint per chunk made unvalidated blocks look validated: a root error unwound to one, and with_max_block or a lagging Finish could treat them as done. Keep the block checkpoint at the last validated block, record committed chunks in the execution checkpoint, and have MerkleUnwind clear that progress when it rewrites the trie. Reverts the Finish-based unwind target. --- crates/stages/api/src/pipeline/mod.rs | 67 +-------- crates/stages/stages/src/stages/merkle.rs | 161 ++++++++++++++++------ 2 files changed, 123 insertions(+), 105 deletions(-) diff --git a/crates/stages/api/src/pipeline/mod.rs b/crates/stages/api/src/pipeline/mod.rs index 5acd83ed4c6..72d65de6dc7 100644 --- a/crates/stages/api/src/pipeline/mod.rs +++ b/crates/stages/api/src/pipeline/mod.rs @@ -664,20 +664,12 @@ impl Pipeline { } } - // Incremental MerkleExecute commits its checkpoint after each chunk without validating - // that block's state root, so unwind to the last block whose root was validated: Finish - // only advances after MerkleExecute validated the end of a range. - let mut target = prev_checkpoint.unwrap_or_default().block_number; - if stage_id == StageId::MerkleExecute && - let Some(finish) = - self.provider_factory.provider()?.get_stage_checkpoint(StageId::Finish)? - { - target = target.min(finish.block_number); - } - // We unwind because of a block error. If the unwind itself fails, we bail entirely; // otherwise, we restart the execution loop from the beginning. - Ok(Some(ControlFlow::Unwind { target, bad_block: block })) + Ok(Some(ControlFlow::Unwind { + target: prev_checkpoint.unwrap_or_default().block_number, + bad_block: block, + })) } else if let StageError::MissingStaticFileData { block, segment } = err { error!( target: "sync::pipeline", @@ -1256,57 +1248,6 @@ mod tests { ); } - /// A state-root error after `MerkleExecute` committed unvalidated chunks unwinds to the last - /// validated block, the Finish checkpoint, not to the last committed chunk. - #[tokio::test] - async fn merkle_execute_validation_error_unwinds_to_finish_checkpoint() { - let provider_factory = create_test_provider_factory(); - let provider_rw = provider_factory.provider_rw().unwrap(); - provider_rw.save_stage_checkpoint(StageId::Finish, StageCheckpoint::new(100)).unwrap(); - provider_rw - .save_stage_checkpoint(StageId::MerkleExecute, StageCheckpoint::new(100)) - .unwrap(); - provider_rw.commit().unwrap(); - - let mut pipeline = Pipeline::::builder() - .add_stage( - TestStage::new(StageId::MerkleExecute) - .add_exec(Ok(ExecOutput::in_progress(StageCheckpoint::new(110)))) - .add_exec(Err(StageError::Block { - block: Box::new(random_block_with_parent( - &mut generators::rng(), - 200, - Default::default(), - )), - error: BlockErrorKind::Validation(ConsensusError::BaseFeeMissing), - })) - .add_unwind(Ok(UnwindOutput { checkpoint: StageCheckpoint::new(100) })) - .add_exec(Ok(ExecOutput::done(StageCheckpoint::new(200)))), - ) - .with_max_block(200) - .build( - provider_factory.clone(), - StaticFileProducer::new(provider_factory.clone(), PruneModes::default()), - ); - let events = pipeline.events(); - tokio::spawn(async move { - pipeline.run().await.expect("Could not run pipeline"); - }); - - let events = events.collect::>().await; - assert!( - events.contains(&PipelineEvent::Unwind { - stage_id: StageId::MerkleExecute, - input: UnwindInput { - checkpoint: StageCheckpoint::new(110), - unwind_to: 100, - bad_block: Some(200), - }, - }), - "{events:?}" - ); - } - /// Checks that the pipeline re-runs stages on non-fatal errors and stops on fatal ones. #[tokio::test] async fn pipeline_error_handling() { diff --git a/crates/stages/stages/src/stages/merkle.rs b/crates/stages/stages/src/stages/merkle.rs index a7948e08eb4..51b2e0d5576 100644 --- a/crates/stages/stages/src/stages/merkle.rs +++ b/crates/stages/stages/src/stages/merkle.rs @@ -12,8 +12,9 @@ use reth_provider::{ StageCheckpointWriter, StatsReader, StorageChangeSetReader, StorageSettingsCache, TrieWriter, }; use reth_stages_api::{ - BlockErrorKind, EntitiesCheckpoint, ExecInput, ExecOutput, MerkleCheckpoint, Stage, - StageCheckpoint, StageError, StageId, StorageRootMerkleCheckpoint, UnwindInput, UnwindOutput, + BlockErrorKind, CheckpointBlockRange, EntitiesCheckpoint, ExecInput, ExecOutput, + ExecutionCheckpoint, MerkleCheckpoint, Stage, StageCheckpoint, StageError, StageId, + StorageRootMerkleCheckpoint, UnwindInput, UnwindOutput, }; use reth_trie::{IntermediateStateRootState, StateRoot, StateRootProgress, StoredSubNode}; use reth_trie_db::DatabaseStateRoot; @@ -319,22 +320,30 @@ where // updates stay bounded by one chunk and a restart resumes after the last committed one. // // Hashed state is already at the pipeline target, so a chunk's root is not the root of - // its last block and only the final chunk can be validated. The committed trie holds - // target values for every key changed up to the checkpoint; the remaining range's - // prefix sets cover every key that changes after it, so resuming forward, even toward a - // newer target, still ends at the right root. - let chunk = input.next_block_range_with_threshold(incremental_threshold); - let chunk_to = *chunk.block_range.end(); + // its last block and only the final chunk can be validated. The block checkpoint stays + // at the last validated block, which keeps it a safe unwind target, and the execution + // checkpoint records the last committed chunk. The committed trie holds target values + // for every key changed up to that chunk; the remaining range's prefix sets cover every + // key that changes after it, so resuming forward, even toward a newer target, still + // ends at the right root. `MerkleUnwind` clears the progress when it rewrites the trie. + let chunk_from = input + .checkpoint() + .execution_stage_checkpoint() + .filter(|progress| { + progress.block_range.from == from_block && progress.block_range.to < to_block + }) + .map_or(from_block, |progress| progress.block_range.to + 1); + let chunk_to = to_block.min(chunk_from.saturating_add(incremental_threshold - 1)); debug!( target: "sync::stages::merkle::exec", current = ?current_block_number, target = ?to_block, incremental_threshold, - chunk_range = ?chunk.block_range, + chunk_range = ?(chunk_from..=chunk_to), "Processing chunk" ); let (root, updates) = reth_trie_db::with_adapter!(provider, |A| { - DbStateRoot::<_, A>::incremental_root_with_updates(provider, chunk.block_range) + DbStateRoot::<_, A>::incremental_root_with_updates(provider, chunk_from..=chunk_to) }) .map_err(|e| { error!(target: "sync::stages::merkle", %e, ?current_block_number, ?to_block, "Incremental state root failed! {INVALID_STATE_ROOT_ERROR_MESSAGE}"); @@ -342,6 +351,18 @@ where })?; provider.write_trie_updates(updates)?; + if chunk_to < to_block { + return Ok(ExecOutput::in_progress( + input.checkpoint().with_execution_stage_checkpoint(ExecutionCheckpoint { + block_range: CheckpointBlockRange { from: from_block, to: chunk_to }, + progress: EntitiesCheckpoint { + processed: chunk_to - current_block_number, + total: to_block - current_block_number, + }, + }), + )) + } + let total_hashed_entries = (provider.count_entries::()? + provider.count_entries::()?) as u64; @@ -354,13 +375,6 @@ where total: total_hashed_entries, }; - if !chunk.is_final_range { - return Ok(ExecOutput::in_progress( - StageCheckpoint::new(chunk_to) - .with_entities_stage_checkpoint(entities_checkpoint), - )) - } - // Save the checkpoint (root, entities_checkpoint) }; @@ -386,13 +400,21 @@ where info!(target: "sync::stages::merkle::unwind", "Stage is always skipped"); return Ok(UnwindOutput { checkpoint: StageCheckpoint::new(input.unwind_to) }) } - // MerkleExecute is unwound first, so its checkpoint is `unwind_to` unless it had not - // reached that block. Incremental chunks commit trie updates only up to that - // checkpoint; keys changed after it can still have stale nodes, so they must be - // recomputed too. - let merkle_execute = provider - .get_stage_checkpoint(StageId::MerkleExecute)? - .map_or(input.unwind_to, |checkpoint| checkpoint.block_number); + // `MerkleExecute` is unwound first, so its block checkpoint is `unwind_to` unless it had + // not validated that block yet. Its committed chunks only cover keys changed up to their + // last block; keys changed after its checkpoint can still have stale nodes, so they must + // be recomputed too. The recomputed trie no longer matches its chunk progress, so clear it. + let merkle_execute = provider.get_stage_checkpoint(StageId::MerkleExecute)?; + if let Some(checkpoint) = + merkle_execute.filter(|checkpoint| checkpoint.execution_stage_checkpoint().is_some()) + { + provider.save_stage_checkpoint( + StageId::MerkleExecute, + StageCheckpoint::new(checkpoint.block_number), + )?; + } + let merkle_execute = + merkle_execute.map_or(input.unwind_to, |checkpoint| checkpoint.block_number); let range = merkle_execute.min(input.unwind_to) + 1..=input.checkpoint.block_number; let walk_all_changed_branch_children = match self { Self::Unwind { walk_all_changed_branch_children } => *walk_all_changed_branch_children, @@ -610,13 +632,15 @@ mod tests { runner.seed_execution(input).expect("failed to seed execution"); - // Each call commits one chunk and stops at its last block until the target is reached. + // Each call commits one chunk and records it, keeping the block checkpoint at the last + // validated block until the target is reached. let mut chunk_ends = Vec::new(); let result = loop { let result = runner.execute(input).await.unwrap(); match result { Ok(ExecOutput { checkpoint, done: false }) => { - chunk_ends.push(checkpoint.block_number); + assert_eq!(checkpoint.block_number, stage_progress); + chunk_ends.push(last_committed_chunk(checkpoint).unwrap()); input.checkpoint = Some(checkpoint); } result => break result, @@ -683,7 +707,7 @@ mod tests { } let provider = runner.db.factory.provider_rw().unwrap(); let uncommitted = runner.stage().execute(&*provider, input).unwrap(); - assert_eq!(uncommitted.checkpoint.block_number, ROUND_START + 30); + assert_eq!(last_committed_chunk(uncommitted.checkpoint), Some(ROUND_START + 30)); drop(provider); let output = execute_until_done(&runner, input).await.unwrap(); @@ -711,7 +735,7 @@ mod tests { assert!(!output.done); input.checkpoint = Some(output.checkpoint); } - assert_eq!(input.checkpoint().block_number, ROUND_START + 30); + assert_eq!(last_committed_chunk(input.checkpoint()), Some(ROUND_START + 30)); // The next round's hashing reaches block 200 before `MerkleExecute` resumes. restore_hashed_state(&runner.db, target_hashed_state); @@ -720,6 +744,40 @@ mod tests { assert_eq!(output.checkpoint.block_number, ROUND_TARGET); } + /// A wrong root at the end of a chunked range fails the stage with its block checkpoint still + /// at the last validated block, so the pipeline unwinds to a block whose root holds. + #[tokio::test] + async fn final_root_error_keeps_checkpoint_at_last_validated_block() { + let (runner, mut input) = start_round().await; + set_header_state_root(&runner.db, ROUND_TARGET, B256::repeat_byte(0x42)); + + let error = loop { + match runner.execute(input).await.unwrap() { + Ok(output) => { + assert!(!output.done); + assert_eq!(output.checkpoint.block_number, ROUND_START); + input.checkpoint = Some(output.checkpoint); + } + Err(error) => break error, + } + }; + assert_matches!(error, StageError::Block { .. }); + assert_eq!(last_committed_chunk(input.checkpoint()), Some(ROUND_TARGET - 10)); + + // On a block error the pipeline keeps the previous checkpoint and unwinds to its block. + let provider = runner.db.factory.provider_rw().unwrap(); + provider.save_stage_checkpoint(StageId::MerkleExecute, input.checkpoint()).unwrap(); + provider.commit().unwrap(); + let unwind = UnwindInput { + checkpoint: StageCheckpoint::new(ROUND_TARGET), + unwind_to: input.checkpoint().block_number, + bad_block: None, + }; + runner.before_unwind(unwind).unwrap(); + let output = runner.unwind(unwind).await.unwrap(); + assert_eq!(output.checkpoint.block_number, ROUND_START); + } + /// A short range after an interrupted rebuild finishes the rebuild instead of updating the /// partial trie incrementally. #[tokio::test] @@ -789,23 +847,18 @@ mod tests { } /// Unwinds to `unwind_to` after `MerkleExecute` committed one chunk, through block 110, of the - /// round from [`start_round`]. + /// round from [`start_round`], then runs the next round to the target. async fn unwind_after_partial_incremental_merkle(unwind_to: BlockNumber) { - let last_chunk_end = ROUND_START + 10; - let (runner, mut input) = start_round().await; - while input.checkpoint().block_number < last_chunk_end { - let output = runner.execute(input).await.unwrap().unwrap(); - input.checkpoint = Some(output.checkpoint); - } + let (runner, input) = start_round().await; + let target_hashed_state = hashed_state(&runner.db); + let output = runner.execute(input).await.unwrap().unwrap(); + assert_eq!(output.checkpoint.block_number, ROUND_START); + assert_eq!(last_committed_chunk(output.checkpoint), Some(ROUND_START + 10)); - // The pipeline unwinds MerkleExecute first, which only moves its checkpoint down. + // The pipeline saves each call's checkpoint. It then unwinds `MerkleExecute` first, which + // leaves it alone: its block checkpoint, the round start, is not above `unwind_to`. let provider = runner.db.factory.provider_rw().unwrap(); - provider - .save_stage_checkpoint( - StageId::MerkleExecute, - StageCheckpoint::new(last_chunk_end.min(unwind_to)), - ) - .unwrap(); + provider.save_stage_checkpoint(StageId::MerkleExecute, output.checkpoint).unwrap(); provider.commit().unwrap(); let unwind = UnwindInput { @@ -819,6 +872,30 @@ mod tests { // The stage validates the recomputed root against the header of `unwind_to`. let output = runner.unwind(unwind).await.unwrap(); assert_eq!(output.checkpoint.block_number, unwind_to); + let merkle_execute = runner + .db + .factory + .provider() + .unwrap() + .get_stage_checkpoint(StageId::MerkleExecute) + .unwrap(); + assert_eq!(merkle_execute, Some(StageCheckpoint::new(ROUND_START))); + + // The next round hashes up to the target again and resumes from `MerkleExecute`'s + // checkpoint; the stage validates the root at the target. + restore_hashed_state(&runner.db, target_hashed_state); + let output = execute_until_done( + &runner, + ExecInput { target: Some(ROUND_TARGET), checkpoint: merkle_execute }, + ) + .await + .unwrap(); + assert_eq!(output.checkpoint.block_number, ROUND_TARGET); + } + + /// Returns the last block of the chunks an in-progress incremental range has committed. + fn last_committed_chunk(checkpoint: StageCheckpoint) -> Option { + checkpoint.execution_stage_checkpoint().map(|progress| progress.block_range.to) } const ROUND_START: BlockNumber = 100; From ec44cb331a5c2345ffd2d45df324f5c7b6e91670 Mon Sep 17 00:00:00 2001 From: Francis Li Date: Fri, 13 Nov 2026 21:50:43 -0800 Subject: [PATCH 5/7] fix(stages): keep the merkle unwind range while a rebuild is in progress --- crates/stages/stages/src/stages/merkle.rs | 18 ++++++++++++++---- 1 file changed, 14 insertions(+), 4 deletions(-) diff --git a/crates/stages/stages/src/stages/merkle.rs b/crates/stages/stages/src/stages/merkle.rs index 51b2e0d5576..c2ce2c78c73 100644 --- a/crates/stages/stages/src/stages/merkle.rs +++ b/crates/stages/stages/src/stages/merkle.rs @@ -413,9 +413,16 @@ where StageCheckpoint::new(checkpoint.block_number), )?; } - let merkle_execute = - merkle_execute.map_or(input.unwind_to, |checkpoint| checkpoint.block_number); - let range = merkle_execute.min(input.unwind_to) + 1..=input.checkpoint.block_number; + // A rebuild in progress cleared the trie and writes target values, so it needs no wider + // range, and widening it to a rebuild's range start can load the whole chain's changesets. + let range = if self.get_execution_checkpoint(provider)?.is_some() { + input.unwind_block_range() + } else { + merkle_execute + .map_or(input.unwind_to, |checkpoint| checkpoint.block_number) + .min(input.unwind_to) + + 1..=input.checkpoint.block_number + }; let walk_all_changed_branch_children = match self { Self::Unwind { walk_all_changed_branch_children } => *walk_all_changed_branch_children, #[cfg(any(test, feature = "test-utils"))] @@ -751,9 +758,12 @@ mod tests { let (runner, mut input) = start_round().await; set_header_state_root(&runner.db, ROUND_TARGET, B256::repeat_byte(0x42)); + // Like the pipeline, commit each successful call and drop the transaction of a failed one. let error = loop { - match runner.execute(input).await.unwrap() { + let provider = runner.db.factory.provider_rw().unwrap(); + match runner.stage().execute(&*provider, input) { Ok(output) => { + provider.commit().unwrap(); assert!(!output.done); assert_eq!(output.checkpoint.block_number, ROUND_START); input.checkpoint = Some(output.checkpoint); From c00b81c4c49952e0d86d3e2bef2b529c00463194 Mon Sep 17 00:00:00 2001 From: Francis Li Date: Fri, 13 Nov 2026 22:23:23 -0800 Subject: [PATCH 6/7] test(stages): default remaining account fields in merkle tests --- crates/stages/stages/src/stages/merkle.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/crates/stages/stages/src/stages/merkle.rs b/crates/stages/stages/src/stages/merkle.rs index c2ce2c78c73..48e5c2e84eb 100644 --- a/crates/stages/stages/src/stages/merkle.rs +++ b/crates/stages/stages/src/stages/merkle.rs @@ -932,7 +932,7 @@ mod tests { let extra_accounts = (1..=2000u64) .map(|i| { let address = Address::from_word(B256::from(U256::from(i))); - (address, Account { nonce: 1, balance: U256::from(i), bytecode_hash: None }) + (address, Account { nonce: 1, balance: U256::from(i), ..Default::default() }) }) .collect::>(); runner From 1b511fb59b0f5eb646c65541f84c23efca76b1aa Mon Sep 17 00:00:00 2001 From: Francis Li Date: Fri, 13 Nov 2026 23:03:15 -0800 Subject: [PATCH 7/7] test(stages): clone accounts so merkle tests build with account-ext --- crates/stages/stages/src/stages/merkle.rs | 9 ++++++--- 1 file changed, 6 insertions(+), 3 deletions(-) diff --git a/crates/stages/stages/src/stages/merkle.rs b/crates/stages/stages/src/stages/merkle.rs index 48e5c2e84eb..6b323201206 100644 --- a/crates/stages/stages/src/stages/merkle.rs +++ b/crates/stages/stages/src/stages/merkle.rs @@ -917,6 +917,8 @@ mod tests { /// state at block 200, and `MerkleUnwind` already at 200. The state is wide enough that the /// trie stores branch hashes below the root, so trie nodes the stage fails to update stay /// stale and break the next validated root. + // `Account` is only `Copy` without the `account-ext` feature. + #[allow(clippy::clone_on_copy)] async fn start_round() -> (MerkleTestRunner, ExecInput) { let mut runner = MerkleTestRunner { db: TestStageDB::default(), @@ -940,7 +942,7 @@ mod tests { .insert_accounts_and_storages( extra_accounts .iter() - .map(|(address, account)| (*address, (*account, std::iter::empty()))), + .map(|(address, account)| (*address, (account.clone(), std::iter::empty()))), ) .unwrap(); let late_changes = extra_accounts.iter().step_by(10); @@ -949,7 +951,7 @@ mod tests { .insert_changesets( [late_changes .clone() - .map(|(address, account)| (*address, *account, Vec::new())) + .map(|(address, account)| (*address, account.clone(), Vec::new())) .collect::>()], Some(LATE_CHANGE_BLOCK), ) @@ -957,7 +959,8 @@ mod tests { runner .db .insert_accounts_and_storages(late_changes.map(|(address, account)| { - let account = Account { balance: account.balance + U256::from(1), ..*account }; + let account = + Account { balance: account.balance + U256::from(1), ..account.clone() }; (*address, (account, std::iter::empty())) })) .unwrap();