diff --git a/zebra-state/src/lib.rs b/zebra-state/src/lib.rs index 88159fbd242..b5b3706bcfb 100644 --- a/zebra-state/src/lib.rs +++ b/zebra-state/src/lib.rs @@ -55,8 +55,8 @@ pub use request::{ pub use request::Spend; pub use response::{ - AnyTx, GetBlockTemplateChainInfo, KnownBlock, MinedTx, NonFinalizedBlocksListener, - ReadResponse, Response, + AnyTx, GetBlockTemplateChainInfo, HeaderRangeCommitOutcome, KnownBlock, MinedTx, + NonFinalizedBlocksListener, ReadResponse, Response, }; pub use service::{ chain_tip::{ChainTipBlock, ChainTipChange, ChainTipSender, LatestChainTip, TipAction}, diff --git a/zebra-state/src/response.rs b/zebra-state/src/response.rs index 80991e95d54..d0e7fe85d90 100644 --- a/zebra-state/src/response.rs +++ b/zebra-state/src/response.rs @@ -27,6 +27,21 @@ use crate::{ReadRequest, Request}; use crate::{service::read::AddressUtxos, NonFinalizedState, TransactionLocation, WatchReceiver}; +/// The outcome of a successful [`Request::CommitHeaderRange`]. +#[derive(Copy, Clone, Debug, PartialEq, Eq)] +pub struct HeaderRangeCommitOutcome { + /// Hash of the last header in the committed range. + pub tip_hash: block::Hash, + /// First height where the committed range replaced a previously stored + /// conflicting header suffix, when the commit reorged the stored header + /// chain. `None` when the range only extended or duplicated stored headers. + pub reorged_at: Option, + /// The new branch's hash at `reorged_at`, when the commit reorged the + /// stored header chain. Used by the switch orchestration to recognize the + /// stranded old-branch body suffix without re-reading state. + pub reorged_to_hash: Option, +} + #[derive(Clone, Debug, PartialEq, Eq)] /// A response to a [`StateService`](crate::service::StateService) [`Request`]. pub enum Response { @@ -34,6 +49,10 @@ pub enum Response { /// indicating that a block was successfully committed to the state. Committed(block::Hash), + /// Response to [`Request::CommitHeaderRange`] indicating that a header + /// range was successfully committed to the state. + CommittedHeaderRange(HeaderRangeCommitOutcome), + /// Response to [`Request::InvalidateBlock`] indicating that a block was found and /// invalidated in the state. Invalidated(block::Hash), diff --git a/zebra-state/src/service.rs b/zebra-state/src/service.rs index 4641208fc79..9b4b4279f84 100644 --- a/zebra-state/src/service.rs +++ b/zebra-state/src/service.rs @@ -59,7 +59,8 @@ use crate::{ watch_receiver::WatchReceiver, }, BoxError, CheckpointVerifiedBlock, CommitHeaderRangeError, CommitSemanticallyVerifiedError, - Config, KnownBlock, ReadRequest, ReadResponse, Request, Response, SemanticallyVerifiedBlock, + Config, HeaderRangeCommitOutcome, KnownBlock, ReadRequest, ReadResponse, Request, Response, + SemanticallyVerifiedBlock, }; pub mod block_iter; @@ -988,7 +989,7 @@ impl StateService { headers: Vec>, body_sizes: Vec, tree_aux_roots: Vec, - ) -> oneshot::Receiver> { + ) -> oneshot::Receiver> { let (rsp_tx, rsp_rx) = oneshot::channel(); let Some(sender) = &self.block_write_sender.non_finalized else { @@ -1255,7 +1256,7 @@ impl Service for StateService { .map_err(|_recv_error| CommitHeaderRangeError::CommitResponseDropped) .and_then(|result| result) .map_err(BoxError::from) - .map(Response::Committed) + .map(Response::CommittedHeaderRange) } .instrument(span) .boxed() diff --git a/zebra-state/src/service/finalized_state/zebra_db/block.rs b/zebra-state/src/service/finalized_state/zebra_db/block.rs index d82a2d48bd3..af93108ee32 100644 --- a/zebra-state/src/service/finalized_state/zebra_db/block.rs +++ b/zebra-state/src/service/finalized_state/zebra_db/block.rs @@ -51,7 +51,7 @@ use crate::{ FromDisk, IntoDisk, RawBytes, COMMITMENT_ROOTS_BY_HEIGHT, PRUNING_METADATA, VCT_SYNC_METADATA, VCT_UPGRADE_METADATA, }, - HashOrHeight, + HashOrHeight, HeaderRangeCommitOutcome, }; #[cfg(feature = "indexer")] @@ -75,6 +75,7 @@ pub(crate) struct AdvertisedBodySize(u32); /// Builds an [`AdvertisedBodySize`] for tests that plant raw rows. #[cfg(test)] +#[allow(dead_code)] // used by the coherence suite, dropped from canary picks pub(crate) fn test_body_size(size: u32) -> AdvertisedBodySize { AdvertisedBodySize(size) } @@ -2019,7 +2020,7 @@ impl DiskWriteBatch { anchor: block::Hash, headers: &[Arc], body_sizes: &[u32], - ) -> Result { + ) -> Result { let roots = inferred_header_range_roots(zebra_db, anchor, headers.len())?; self.prepare_header_range_batch_with_roots(zebra_db, anchor, headers, body_sizes, &roots) } @@ -2034,7 +2035,7 @@ impl DiskWriteBatch { headers: &[Arc], body_sizes: &[u32], tree_aux_roots: &[BlockCommitmentRoots], - ) -> Result { + ) -> Result { if headers.is_empty() { return Err(CommitHeaderRangeError::EmptyRange); } @@ -2328,9 +2329,17 @@ impl DiskWriteBatch { self.set_canonical_suffix(zebra_db, (fork_height, fork_hash), &new_rows)?; } - Ok(block::Hash::from( - &**headers.last().expect("headers is non-empty"), - )) + Ok(HeaderRangeCommitOutcome { + tip_hash: block::Hash::from(&**headers.last().expect("headers is non-empty")), + reorged_at: first_conflicting_height, + reorged_to_hash: first_conflicting_height.map(|reorged_at| { + validated_headers + .iter() + .find(|(height, ..)| *height == reorged_at) + .expect("the first conflicting height is inside the validated range") + .1 + }), + }) } /// Deletes the block header at `height`. diff --git a/zebra-state/src/service/finalized_state/zebra_db/block/tests/common.rs b/zebra-state/src/service/finalized_state/zebra_db/block/tests/common.rs index fb7c4689163..249be03aff6 100644 --- a/zebra-state/src/service/finalized_state/zebra_db/block/tests/common.rs +++ b/zebra-state/src/service/finalized_state/zebra_db/block/tests/common.rs @@ -125,13 +125,13 @@ pub(super) fn commit_header_range( ) -> block::Hash { let mut batch = DiskWriteBatch::new(); let body_sizes = vec![0; headers.len()]; - let committed_hash = batch + let outcome = batch .prepare_header_range_batch(state, anchor, headers, &body_sizes) .expect("header range is valid"); state .write_batch(batch) .expect("header range batch writes successfully"); - committed_hash + outcome.tip_hash } /// Commits `block`'s header and transaction data (the body-commit batch shape), diff --git a/zebra-state/src/service/finalized_state/zebra_db/block/tests/header_store_coherence/reads.rs b/zebra-state/src/service/finalized_state/zebra_db/block/tests/header_store_coherence/reads.rs index 84de63a5159..1fb00306321 100644 --- a/zebra-state/src/service/finalized_state/zebra_db/block/tests/header_store_coherence/reads.rs +++ b/zebra-state/src/service/finalized_state/zebra_db/block/tests/header_store_coherence/reads.rs @@ -61,7 +61,9 @@ fn redeliver_trunk( .collect(); let body_sizes = vec![0; headers.len()]; let mut batch = DiskWriteBatch::new(); - let result = batch.prepare_header_range_batch(state, anchor, &headers, &body_sizes); + let result = batch + .prepare_header_range_batch(state, anchor, &headers, &body_sizes) + .map(|outcome| outcome.tip_hash); if result.is_ok() { state .write_batch(batch) diff --git a/zebra-state/src/service/finalized_state/zebra_db/block/tests/vectors.rs b/zebra-state/src/service/finalized_state/zebra_db/block/tests/vectors.rs index 7bca301f624..bcc7c77f624 100644 --- a/zebra-state/src/service/finalized_state/zebra_db/block/tests/vectors.rs +++ b/zebra-state/src/service/finalized_state/zebra_db/block/tests/vectors.rs @@ -106,7 +106,7 @@ fn header_range_commit_keeps_body_availability_separate() { let (state, genesis, block1) = mainnet_state_with_genesis(); let mut batch = DiskWriteBatch::new(); - let committed_hash = batch + let outcome = batch .prepare_header_range_batch( &state, genesis.hash(), @@ -118,7 +118,7 @@ fn header_range_commit_keeps_body_availability_separate() { .write_batch(batch) .expect("header range batch writes successfully"); - assert_eq!(committed_hash, block1.hash()); + assert_eq!(outcome.tip_hash, block1.hash()); assert_eq!(state.best_header_tip(), Some((Height(1), block1.hash()))); assert_eq!(state.finalized_tip_height(), Some(Height(0))); assert_eq!(state.tip(), Some((Height(0), genesis.hash()))); diff --git a/zebra-state/src/service/non_finalized_state.rs b/zebra-state/src/service/non_finalized_state.rs index 337cf01ff1c..7c09aaf58b5 100644 --- a/zebra-state/src/service/non_finalized_state.rs +++ b/zebra-state/src/service/non_finalized_state.rs @@ -370,21 +370,25 @@ impl NonFinalizedState { Ok(()) } - /// Invalidate block with hash `block_hash` and all descendants from the non-finalized state. Insert - /// the new chain into the chain_set and discard the previous. + /// Removes the block with hash `block_hash` and all its descendants from + /// the non-finalized state, returning the removed blocks. Inserts the + /// truncated chain into the chain_set and discards the previous. #[allow(clippy::unwrap_in_result)] - pub fn invalidate_block(&mut self, block_hash: Hash) -> Result { + fn remove_block_and_descendants( + &mut self, + block_hash: Hash, + ) -> Result, InvalidateError> { let Some(chain) = self.find_chain(|chain| chain.contains_block_hash(block_hash)) else { return Err(InvalidateError::BlockNotFound(block_hash)); }; - let invalidated_blocks = if chain.non_finalized_root_hash() == block_hash { + let removed_blocks = if chain.non_finalized_root_hash() == block_hash { let chain_tip_hash = chain.non_finalized_tip_hash(); self.chain_set .retain(|chain| chain.non_finalized_tip_hash() != chain_tip_hash); chain.blocks.values().cloned().collect() } else { - let (new_chain, invalidated_blocks) = chain + let (new_chain, removed_blocks) = chain .invalidate_block(block_hash) .expect("already checked that chain contains hash"); @@ -394,9 +398,25 @@ impl NonFinalizedState { chain_set.retain(|c| !c.contains_block_hash(block_hash)) }); - invalidated_blocks + removed_blocks }; + self.update_metrics_for_chains(); + self.update_metrics_bars(); + + Ok(removed_blocks) + } + + /// Invalidate block with hash `block_hash` and all descendants from the non-finalized state. Insert + /// the new chain into the chain_set and discard the previous. + /// + /// The removed blocks are recorded in the invalidated list: they are + /// refused on re-delivery until explicitly reconsidered. For a reorg + /// rollback, use [`Self::rollback_block`] instead. + #[allow(clippy::unwrap_in_result)] + pub fn invalidate_block(&mut self, block_hash: Hash) -> Result { + let invalidated_blocks = self.remove_block_and_descendants(block_hash)?; + // TODO: Allow for invalidating multiple block hashes at a given height (#9552). self.invalidated_blocks.insert( invalidated_blocks @@ -411,9 +431,20 @@ impl NonFinalizedState { self.invalidated_blocks.shift_remove_index(0); } - self.update_metrics_for_chains(); - self.update_metrics_bars(); + Ok(block_hash) + } + /// Rolls back the block with hash `block_hash` and all its descendants + /// from the non-finalized state, without recording them as invalidated. + /// + /// A branch-switch rollback is a chain-selection outcome, not a verdict + /// on the blocks: the rolled-back branch must revalidate and recommit + /// normally the moment it wins the work race again. Recording it in the + /// invalidated list would refuse the re-delivery + /// (`BlockPreviouslyInvalidated`) and strand the node if the winning + /// branch later evaporates. + pub fn rollback_block(&mut self, block_hash: Hash) -> Result { + self.remove_block_and_descendants(block_hash)?; Ok(block_hash) } diff --git a/zebra-state/src/service/tests.rs b/zebra-state/src/service/tests.rs index 8f363c28cdb..316ee2ebc46 100644 --- a/zebra-state/src/service/tests.rs +++ b/zebra-state/src/service/tests.rs @@ -35,8 +35,8 @@ use crate::{ setup::{partial_nu5_chain_strategy, transaction_v4_from_coinbase}, FakeChainHelper, }, - BoxError, CheckpointVerifiedBlock, Config, ReadRequest, ReadResponse, Request, Response, - SemanticallyVerifiedBlock, + BoxError, CheckpointVerifiedBlock, Config, HeaderRangeCommitOutcome, ReadRequest, ReadResponse, + Request, Response, SemanticallyVerifiedBlock, }; const LAST_BLOCK_HEIGHT: u32 = 10; @@ -538,7 +538,11 @@ async fn header_only_service_requests_preserve_body_boundary() -> std::result::R tree_aux_roots: roots_from_height(Height(1), 2), }) .await?, - Response::Committed(block2_hash), + Response::CommittedHeaderRange(HeaderRangeCommitOutcome { + tip_hash: block2_hash, + reorged_at: None, + reorged_to_hash: None, + }), ); assert_eq!( @@ -793,7 +797,14 @@ async fn commit_header_range_completes_while_in_finalized_write_phase( .await .expect("CommitHeaderRange must not deadlock while in the finalized write phase")?; - assert_eq!(committed, Response::Committed(block2_hash)); + assert_eq!( + committed, + Response::CommittedHeaderRange(HeaderRangeCommitOutcome { + tip_hash: block2_hash, + reorged_at: None, + reorged_to_hash: None, + }) + ); assert_eq!( state diff --git a/zebra-state/src/service/write.rs b/zebra-state/src/service/write.rs index 178a97818d2..1ba12af6386 100644 --- a/zebra-state/src/service/write.rs +++ b/zebra-state/src/service/write.rs @@ -29,7 +29,7 @@ use crate::{ queued_blocks::{QueuedCheckpointVerified, QueuedSemanticallyVerified}, ChainTipBlock, ChainTipSender, InvalidateError, ReconsiderError, }, - SemanticallyVerifiedBlock, ValidateContextError, + HeaderRangeCommitOutcome, SemanticallyVerifiedBlock, ValidateContextError, }; // These types are used in doc links @@ -128,13 +128,41 @@ fn update_latest_chain_channels( tip_block_height } +/// Commits a validated header range, orchestrating the branch switch when the +/// range reorgs the stored header chain (REORG_PLAN Pillar 1a). +/// +/// The switch is sequenced so every crash intermediate is coherent: +/// +/// 1. **Evaluate**: the batch preparation validates the whole candidate range +/// in memory (linkage, checkpoints, contextual difficulty, the strictly- +/// greater cumulative-work gate) and decides the fork point, staging disk +/// writes only through the suffix-replacement primitive. A rejected +/// candidate leaves every store untouched. +/// 2. **Body rollback**: if the fork strands a committed non-finalized body +/// suffix (the old branch's blocks above the fork), invalidate it *before* +/// the header rewrite reaches disk, so the header store never names a +/// branch the body store still extends past the fork. Invalidation is a +/// rollback, not a ban: re-delivered old-branch blocks revalidate and +/// recommit normally if that branch later wins again. +/// 3. **Header rewrite**: write the prepared batch (one atomic suffix +/// replacement), then run the post-reorg store audit. +/// +/// A crash between steps 2 and 3 leaves the node "behind" on a coherent +/// store: the non-finalized backup restores or resync re-delivers whichever +/// branch currently wins, and the ordinary sync flow re-runs the switch — +/// recovery is never a special code path. +#[allow(clippy::too_many_arguments)] fn commit_header_range( finalized_state: &FinalizedState, + non_finalized_state: &mut NonFinalizedState, + chain_tip_sender: &mut ChainTipSender, + non_finalized_state_sender: &watch::Sender, + backup_dir_path: Option<&Path>, anchor: block::Hash, headers: Vec>, body_sizes: Vec, tree_aux_roots: Vec, - rsp_tx: oneshot::Sender>, + rsp_tx: oneshot::Sender>, ) { let mut batch = crate::service::finalized_state::DiskWriteBatch::new(); let result = batch @@ -145,7 +173,20 @@ fn commit_header_range( &body_sizes, &tree_aux_roots, ) - .and_then(|hash| { + .and_then(|outcome| { + if let (Some(reorged_at), Some(reorged_to_hash)) = + (outcome.reorged_at, outcome.reorged_to_hash) + { + invalidate_stranded_body_suffix( + non_finalized_state, + chain_tip_sender, + non_finalized_state_sender, + backup_dir_path, + reorged_at, + reorged_to_hash, + ); + } + let zakura_replaced_rows = batch.zakura_suffix_replaced_rows(); finalized_state .db @@ -154,7 +195,7 @@ fn commit_header_range( finalized_state .db .audit_zakura_header_store_after_reorg(zakura_replaced_rows); - hash + outcome }) .map_err(|error| { tracing::error!(?error, "failed to write validated header range"); @@ -168,6 +209,82 @@ fn commit_header_range( let _ = rsp_tx.send(result); } +/// Rolls back the non-finalized body suffix stranded by a header-chain reorg +/// (step 2 of the switch orchestration in [`commit_header_range`]). +/// +/// `reorged_at` is the first height where the winning header range replaces a +/// conflicting stored header; `reorged_to_hash` is the new branch's hash +/// there. If the best body chain still holds a different block at that +/// height, that block and its descendants are invalidated and the truncated +/// chain is published, so block-gap discovery reanchors at the fork and +/// re-downloads the new branch's bodies. Without this, body sync keeps +/// extending the stale branch above a header store that no longer names it. +fn invalidate_stranded_body_suffix( + non_finalized_state: &mut NonFinalizedState, + chain_tip_sender: &mut ChainTipSender, + non_finalized_state_sender: &watch::Sender, + backup_dir_path: Option<&Path>, + reorged_at: block::Height, + reorged_to_hash: block::Hash, +) { + let Some(old_hash) = non_finalized_state.best_hash(reorged_at) else { + // No committed body at the reorged height; nothing to roll back. + return; + }; + if old_hash == reorged_to_hash { + // The body chain is already on the new branch. + return; + } + + tracing::warn!( + ?reorged_at, + ?old_hash, + new_hash = ?reorged_to_hash, + "header reorg crossed the committed body suffix; rolling back the stranded branch so block sync re-downloads the new one" + ); + + match non_finalized_state.rollback_block(old_hash) { + Ok(hash) => { + metrics::counter!("sync.header.fork_recovery.body_suffix_invalidated").increment(1); + tracing::info!( + ?reorged_at, + ?hash, + "rolled back the stranded body suffix before the header rewrite" + ); + } + Err(error) => { + // Leave the header commit to proceed: the stranded suffix is + // rediscovered on the next reorging commit or at restart, and the + // ordinary flow re-runs this rollback. + tracing::warn!( + ?reorged_at, + ?old_hash, + ?error, + "failed to roll back the stranded body suffix after a header reorg" + ); + return; + } + } + + // Publish the truncated chain so the tip watchers see the rollback. + // Invalidating the non-finalized root can empty the chain set entirely; + // `update_latest_chain_channels` expects a best chain, so skip the + // publish in that case (the tip watch keeps the stale tip until the new + // branch's bodies commit — block sync follows the header store, not the + // tip watch, so re-download still reanchors correctly). + if non_finalized_state.best_chain().is_some() { + update_latest_chain_channels( + non_finalized_state, + chain_tip_sender, + non_finalized_state_sender, + backup_dir_path, + ); + } else { + let _ = non_finalized_state_sender.send(non_finalized_state.clone()); + let _ = chain_tip_sender; + } +} + /// A worker task that reads, validates, and writes blocks to the /// `finalized_state` or `non_finalized_state`. struct WriteBlockWorkerTask { @@ -204,7 +321,7 @@ pub enum NonFinalizedWriteMessage { headers: Vec>, body_sizes: Vec, tree_aux_roots: Vec, - rsp_tx: oneshot::Sender>, + rsp_tx: oneshot::Sender>, }, /// The hash of a block that should be invalidated and removed from /// the non-finalized state, if present. @@ -358,6 +475,10 @@ impl WriteBlockWorkerTask { }) => { commit_header_range( finalized_state, + non_finalized_state, + chain_tip_sender, + non_finalized_state_sender, + backup_dir_path.as_deref(), anchor, headers, body_sizes, @@ -551,6 +672,10 @@ impl WriteBlockWorkerTask { } => { commit_header_range( finalized_state, + non_finalized_state, + chain_tip_sender, + non_finalized_state_sender, + backup_dir_path.as_deref(), anchor, headers, body_sizes, @@ -838,4 +963,228 @@ mod tests { vec![(best_height, best_block.hash(), best_block.header.clone())], ); } + + /// Builds a three-block non-finalized chain plus the channel plumbing the + /// switch orchestration needs. + fn orchestration_fixture( + network: &Network, + ) -> ( + NonFinalizedState, + FinalizedState, + Vec>, + crate::service::ChainTipSender, + crate::LatestChainTip, + crate::ChainTipChange, + tokio::sync::watch::Sender, + tokio::sync::watch::Receiver, + ) { + let block1: Arc = Arc::new( + network + .test_block(653599, 583999) + .expect("test block exists"), + ); + let block2 = block1.make_fake_child().set_work(10); + let block3 = block2.make_fake_child().set_work(1); + + let mut non_finalized_state = NonFinalizedState::new(network); + let finalized_state = FinalizedState::new( + &Config::ephemeral(), + network, + #[cfg(feature = "elasticsearch")] + false, + ); + finalized_state.set_finalized_value_pool(ValueBalance::fake_populated_pool()); + + non_finalized_state + .commit_new_chain(block1.clone().prepare(), &finalized_state) + .expect("chain root commits"); + non_finalized_state + .commit_block(block2.clone().prepare(), &finalized_state) + .expect("child commits"); + non_finalized_state + .commit_block(block3.clone().prepare(), &finalized_state) + .expect("grandchild commits"); + + let (chain_tip_sender, latest_chain_tip, chain_tip_change) = + crate::service::ChainTipSender::new(None, network); + let (nf_sender, nf_receiver) = tokio::sync::watch::channel(NonFinalizedState::new(network)); + + ( + non_finalized_state, + finalized_state, + vec![block1, block2, block3], + chain_tip_sender, + latest_chain_tip, + chain_tip_change, + nf_sender, + nf_receiver, + ) + } + + /// A stranded suffix is rolled back from the reorged height up, keeping + /// the shared prefix, and the truncated chain is published. + #[test] + fn stranded_body_suffix_rolls_back_from_the_fork() { + let _init_guard = zebra_test::init(); + let network = Network::Mainnet; + let ( + mut non_finalized_state, + _finalized_state, + blocks, + mut chain_tip_sender, + _latest_chain_tip, + _chain_tip_change, + nf_sender, + nf_receiver, + ) = orchestration_fixture(&network); + + let reorged_at = blocks[1].coinbase_height().expect("fake child has height"); + super::invalidate_stranded_body_suffix( + &mut non_finalized_state, + &mut chain_tip_sender, + &nf_sender, + None, + reorged_at, + zebra_chain::block::Hash([0xAB; 32]), + ); + + let best_chain = non_finalized_state + .best_chain() + .expect("the shared prefix survives"); + assert!(best_chain.contains_block_hash(blocks[0].hash())); + assert!(!best_chain.contains_block_hash(blocks[1].hash())); + assert!(!best_chain.contains_block_hash(blocks[2].hash())); + + // The truncated chain was published to the watch channel. + assert!(nf_receiver + .borrow() + .best_chain() + .is_some_and(|chain| !chain.contains_block_hash(blocks[1].hash()))); + } + + /// Invalidation is a rollback, not a ban: the rolled-back block + /// revalidates and recommits if its branch wins again later. + #[test] + fn stranded_suffix_rollback_is_not_a_ban() { + let _init_guard = zebra_test::init(); + let network = Network::Mainnet; + let ( + mut non_finalized_state, + finalized_state, + blocks, + mut chain_tip_sender, + _latest_chain_tip, + _chain_tip_change, + nf_sender, + _nf_receiver, + ) = orchestration_fixture(&network); + + let reorged_at = blocks[1].coinbase_height().expect("fake child has height"); + super::invalidate_stranded_body_suffix( + &mut non_finalized_state, + &mut chain_tip_sender, + &nf_sender, + None, + reorged_at, + zebra_chain::block::Hash([0xAB; 32]), + ); + + non_finalized_state + .commit_block(blocks[1].clone().prepare(), &finalized_state) + .expect("a rolled-back block recommits when its branch wins again"); + assert!(non_finalized_state + .best_chain() + .expect("chain is non-empty") + .contains_block_hash(blocks[1].hash())); + } + + /// Rolling back from the non-finalized root empties the chain set: the + /// orchestration must skip the tip publish instead of panicking on the + /// empty state. + #[test] + fn whole_suffix_rollback_skips_publish_without_panicking() { + let _init_guard = zebra_test::init(); + let network = Network::Mainnet; + let ( + mut non_finalized_state, + _finalized_state, + blocks, + mut chain_tip_sender, + _latest_chain_tip, + _chain_tip_change, + nf_sender, + nf_receiver, + ) = orchestration_fixture(&network); + + let reorged_at = blocks[0].coinbase_height().expect("root has height"); + super::invalidate_stranded_body_suffix( + &mut non_finalized_state, + &mut chain_tip_sender, + &nf_sender, + None, + reorged_at, + zebra_chain::block::Hash([0xAB; 32]), + ); + + assert!(non_finalized_state.best_chain().is_none()); + // The emptied state was still published to the watch channel. + assert!(nf_receiver.borrow().best_chain().is_none()); + } + + /// Heights the body chain does not reach, and heights where the body + /// chain already matches the new branch, are left untouched. + #[test] + fn rollback_is_a_noop_when_bodies_match_or_are_absent() { + let _init_guard = zebra_test::init(); + let network = Network::Mainnet; + let ( + mut non_finalized_state, + _finalized_state, + blocks, + mut chain_tip_sender, + _latest_chain_tip, + _chain_tip_change, + nf_sender, + _nf_receiver, + ) = orchestration_fixture(&network); + + // The body chain already holds the "new" hash at the reorged height. + let reorged_at = blocks[1].coinbase_height().expect("fake child has height"); + super::invalidate_stranded_body_suffix( + &mut non_finalized_state, + &mut chain_tip_sender, + &nf_sender, + None, + reorged_at, + blocks[1].hash(), + ); + assert_eq!( + non_finalized_state + .best_chain() + .expect("chain untouched") + .blocks + .len(), + 3 + ); + + // A reorg entirely above the body tip has nothing to roll back. + let above_tip = + (blocks[2].coinbase_height().expect("has height") + 100).expect("height in range"); + super::invalidate_stranded_body_suffix( + &mut non_finalized_state, + &mut chain_tip_sender, + &nf_sender, + None, + above_tip, + zebra_chain::block::Hash([0xAB; 32]), + ); + assert_eq!( + non_finalized_state + .best_chain() + .expect("chain untouched") + .blocks + .len(), + 3 + ); + } } diff --git a/zebrad/src/commands/start/zakura/header_sync_driver.rs b/zebrad/src/commands/start/zakura/header_sync_driver.rs index 8f8b3daadd7..e8ad1bd4e54 100644 --- a/zebrad/src/commands/start/zakura/header_sync_driver.rs +++ b/zebrad/src/commands/start/zakura/header_sync_driver.rs @@ -673,7 +673,19 @@ pub(crate) async fn drive_zakura_header_sync_actions { + Ok(zebra_state::Response::CommittedHeaderRange(outcome)) => { + let tip_hash = outcome.tip_hash; + if let Some(reorged_at) = outcome.reorged_at { + // The stranded body suffix (if any) was already + // rolled back by the state's switch orchestration, + // before the header rewrite reached disk. + metrics::counter!("sync.header.reorg_commits").increment(1); + info!( + ?reorged_at, + ?tip_hash, + "header range commit reorged the stored header chain" + ); + } emit_commit_state( &trace, cs_trace::COMMIT_FINISH,