diff --git a/live-tests/e2e/tests/test_vectors.rs b/live-tests/e2e/tests/test_vectors.rs index 83dc4ac47..7896ae3bc 100644 --- a/live-tests/e2e/tests/test_vectors.rs +++ b/live-tests/e2e/tests/test_vectors.rs @@ -399,7 +399,7 @@ async fn create_200_block_regtest_chain_vectors() { // Update parent block parent_block_sapling_tree_size = chain_block.commitment_tree_data().sizes().sapling(); parent_block_orchard_tree_size = chain_block.commitment_tree_data().sizes().orchard(); - parent_chain_work = Some(*chain_block.chainwork()); + parent_chain_work = *chain_block.chainwork(); data.push((height, zebra_block, block_roots, block_treestate)); } diff --git a/live-tests/zaino-testutils/src/lib.rs b/live-tests/zaino-testutils/src/lib.rs index 6ebb6298f..42b7beb27 100644 --- a/live-tests/zaino-testutils/src/lib.rs +++ b/live-tests/zaino-testutils/src/lib.rs @@ -223,10 +223,12 @@ pub trait PollableTip: Status + Sync { impl PollableTip for FetchServiceSubscriber { async fn tip_height(&self) -> u64 { + // A not-yet-ready backend reports tip 0 (poll keeps waiting) rather + // than panicking. self.get_latest_block() .await - .expect("PollableTip: FetchServiceSubscriber::get_latest_block failed") - .height + .map(|block| block.height) + .unwrap_or(0) } } @@ -234,23 +236,20 @@ impl PollableTip for StateServiceSubscriber { async fn tip_height(&self) -> u64 { self.get_latest_block() .await - .expect("PollableTip: StateServiceSubscriber::get_latest_block failed") - .height + .map(|block| block.height) + .unwrap_or(0) } } impl PollableTip for NodeBackedChainIndexSubscriber { async fn tip_height(&self) -> u64 { - let snapshot = self - .snapshot_nonfinalized_state() - .await - .expect("PollableTip: chain-index snapshot_nonfinalized_state failed"); - u64::from(u32::from( - self.best_chaintip(&snapshot) - .await - .expect("PollableTip: chain-index best_chaintip failed") - .height, - )) + let Ok(snapshot) = self.snapshot_nonfinalized_state().await else { + return 0; + }; + match self.best_chaintip(&snapshot).await { + Ok(tip) => u64::from(u32::from(tip.height)), + Err(_) => 0, + } } } diff --git a/packages/zaino-state/CHANGELOG.md b/packages/zaino-state/CHANGELOG.md index 2ed018bd9..046a7681c 100644 --- a/packages/zaino-state/CHANGELOG.md +++ b/packages/zaino-state/CHANGELOG.md @@ -5,6 +5,16 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/), and this library adheres to Rust's notion of [Semantic Versioning](https://semver.org/spec/v2.0.0.html). +## [Unreleased not landing in 0.4.0] +### Added + +### Changed +- `chain_index::types::BlockContext`'s chainwork field is now optional +- `chain_index::ChainIndexSnapshot` has been completely overhauled, is now + a struct instead of an enum, with completely private internals. Callers + must now interface with it only by passing it to a ChainIndex implementor, + as intended. + ## [Unreleased] ### Added @@ -47,6 +57,7 @@ and this library adheres to Rust's notion of can no longer inflate each other's peak memory. ### Deprecated ### Removed +- `NonFinalizedSnapshot::max_serviceable_height` ### Fixed - The finalised-state txout-set accumulator rebuild at chain tip no longer OOM-crashes on memory-constrained hosts. It auto-shards its in-memory spent set diff --git a/packages/zaino-state/src/backends/fetch.rs b/packages/zaino-state/src/backends/fetch.rs index 058004802..a5813be2c 100644 --- a/packages/zaino-state/src/backends/fetch.rs +++ b/packages/zaino-state/src/backends/fetch.rs @@ -74,10 +74,7 @@ use crate::{ utils::{get_build_info, ServiceMetadata}, BackendType, }; -use crate::{ - chain_index::{non_finalised_state::ChainIndexSnapshot, NonFinalizedSnapshot}, - ChainIndex, NodeBackedChainIndex, NodeBackedChainIndexSubscriber, -}; +use crate::{ChainIndex, NodeBackedChainIndex, NodeBackedChainIndexSubscriber}; /// Chain fetch service backed by Zcashd's JsonRPC engine. /// @@ -499,7 +496,7 @@ impl ZcashIndexer for FetchServiceSubscriber { async fn get_chain_tips(&self) -> Result { let snapshot = self.indexer.snapshot_nonfinalized_state().await?; - let Some(non_finalized_snapshot) = snapshot.get_nfs_snapshot() else { + let Some(non_finalized_snapshot) = snapshot.resolved_nfs_snapshot() else { return Ok(self.fetcher.get_chain_tips().await?); }; @@ -758,7 +755,9 @@ impl ZcashIndexer for FetchServiceSubscriber { let (height, confirmations, block_hash, in_best_chain) = match best_chain_location { Some(types::BestChainLocation::Block(block_hash, height)) => { let confirmations = snapshot - .max_serviceable_height() + .get_nfs_snapshot() + .best_tip + .height .0 .saturating_sub(height.0) .saturating_add(1); @@ -815,7 +814,9 @@ impl ZcashIndexer for FetchServiceSubscriber { self.indexer .snapshot_nonfinalized_state() .await? - .max_serviceable_height() + .get_nfs_snapshot() + .best_tip + .height .0, )) } @@ -896,18 +897,8 @@ impl ZcashIndexer for FetchServiceSubscriber { impl LightWalletIndexer for FetchServiceSubscriber { /// Return the height of the tip of the best chain async fn get_latest_block(&self) -> Result { - match self.indexer.snapshot_nonfinalized_state().await? { - ChainIndexSnapshot::NonFinalizedStateExists { - non_finalized_snapshot, - } => Ok(non_finalized_snapshot.best_tip.to_wire()), - ChainIndexSnapshot::StillSyncingFinalizedState { .. } => { - // TODO: This probably shouldn't be an error. - // this is an improvement over previous behaviour of reporting - // the genesis block - Err(FetchServiceError::UnavailableNotSyncedEnough) - } - } - // dbg!(&tip); + let snapshot = self.indexer.snapshot_nonfinalized_state().await?; + Ok(snapshot.get_nfs_snapshot().best_tip.to_wire()) } /// Return the compact block corresponding to the given block identifier @@ -938,12 +929,7 @@ impl LightWalletIndexer for FetchServiceSubscriber { } }; - let Some(non_finalized_snapshot) = snapshot.get_nfs_snapshot() else { - // TODO: This probably shouldn't be an error. - // this is an improvement over previous behaviour of - // acting as if we are only synced to the genesis block - return Err(FetchServiceError::UnavailableNotSyncedEnough); - }; + let non_finalized_snapshot = snapshot.get_nfs_snapshot(); match self .indexer @@ -1018,12 +1004,7 @@ impl LightWalletIndexer for FetchServiceSubscriber { } } }; - let Some(non_finalized_snapshot) = snapshot.get_nfs_snapshot() else { - // TODO: This probably shouldn't be an error. - // this is an improvement over previous behaviour of - // acting as if we are only synced to the genesis block - return Err(FetchServiceError::UnavailableNotSyncedEnough); - }; + let non_finalized_snapshot = snapshot.get_nfs_snapshot(); match self .indexer .get_compact_block( @@ -1119,20 +1100,7 @@ impl LightWalletIndexer for FetchServiceSubscriber { let timeout_result = timeout( time::Duration::from_secs((service_timeout * 4) as u64), async { - let Some(non_finalized_snapshot) = snapshot.get_nfs_snapshot() else { - // TODO: This probably shouldn't be an error. - // this is an improvement over previous behaviour of - // acting as if we are only synced to the genesis block - if let Err(e) = channel_tx - .send(Err(tonic::Status::failed_precondition( - "zaino not yet synced".to_string(), - ))) - .await - { - warn!(%e, "GetBlockRange channel closed unexpectedly"); - }; - return; - }; + let non_finalized_snapshot = snapshot.get_nfs_snapshot(); // Use the snapshot tip directly, as this function doesn't support passthrough let chain_height = non_finalized_snapshot.best_tip.height.0; @@ -1252,20 +1220,7 @@ impl LightWalletIndexer for FetchServiceSubscriber { let timeout_result = timeout( time::Duration::from_secs((service_timeout * 4) as u64), async { - let Some(non_finalized_snapshot) = snapshot.get_nfs_snapshot() else { - // TODO: This probably shouldn't be an error. - // this is an improvement over previous behaviour of - // acting as if we are only synced to the genesis block - if let Err(e) = channel_tx - .send(Err(tonic::Status::failed_precondition( - "zaino not yet synced".to_string(), - ))) - .await - { - warn!(%e, "GetBlockRangeNullifiers channel closed unexpectedly"); - }; - return; - }; + let non_finalized_snapshot = snapshot.get_nfs_snapshot(); // Use the snapshot tip directly, as this function doesn't support passthrough let chain_height = non_finalized_snapshot.best_tip.height.0; @@ -1389,12 +1344,7 @@ impl LightWalletIndexer for FetchServiceSubscriber { // Zebra returns None for mempool transactions, convert to `Mempool Height`. None => { let snapshot = self.indexer.snapshot_nonfinalized_state().await?; - let Some(non_finalized_snapshot) = snapshot.get_nfs_snapshot() else { - // TODO: This probably shouldn't be an error. - // this is an improvement over previous behaviour of - // acting as if we are only synced to the genesis block - return Err(FetchServiceError::UnavailableNotSyncedEnough); - }; + let non_finalized_snapshot = snapshot.get_nfs_snapshot(); non_finalized_snapshot.best_tip.height.0 as u64 } }; @@ -1745,20 +1695,7 @@ impl LightWalletIndexer for FetchServiceSubscriber { let timeout = timeout( time::Duration::from_secs((service_timeout * 6) as u64), async { - let Some(non_finalized_snapshot) = snapshot.get_nfs_snapshot() else { - // TODO: This probably shouldn't be an error. - // this is an improvement over previous behaviour of - // acting as if we are only synced to the genesis block - if let Err(e) = channel_tx - .send(Err(tonic::Status::failed_precondition( - "zaino not yet synced".to_string(), - ))) - .await - { - warn!(%e, "GetMempoolStream channel closed unexpectedly"); - }; - return; - }; + let non_finalized_snapshot = snapshot.get_nfs_snapshot(); let mempool_height = non_finalized_snapshot.best_tip.height.0; match indexer.get_mempool_stream(None) { Some(mut mempool_stream) => { diff --git a/packages/zaino-state/src/backends/state.rs b/packages/zaino-state/src/backends/state.rs index 62945c1df..2cfc392de 100644 --- a/packages/zaino-state/src/backends/state.rs +++ b/packages/zaino-state/src/backends/state.rs @@ -1,5 +1,6 @@ //! Zcash chain fetch and tx submission service backed by Zebras [`ReadStateService`]. +use crate::{chain_index::types::BestChainLocation, TransactionHash}; #[allow(deprecated)] use crate::{ chain_index::{ @@ -20,10 +21,6 @@ use crate::{ utils::{get_build_info, ServiceMetadata}, BackendType, NodeBackedChainIndex, NodeBackedChainIndexSubscriber, State, }; -use crate::{ - chain_index::{types::BestChainLocation, NonFinalizedSnapshot}, - TransactionHash, -}; use tokio_stream::StreamExt as _; use zaino_fetch::{ chain::{transaction::FullTransaction, utils::ParseFromSlice}, @@ -578,9 +575,8 @@ impl StateServiceSubscriber { let timeout_result = timeout( time::Duration::from_secs((service_timeout * 4) as u64), async { - // This method does not support passthrough. Just return. - let Some(non_finalized_snapshot) = snapshot.get_nfs_snapshot() else {return}; - let chain_height = non_finalized_snapshot.best_tip.height.0; + let nfs_snapshot = snapshot.get_nfs_snapshot(); + let chain_height = nfs_snapshot.best_tip.height.0; match state_service_clone .indexer @@ -688,7 +684,7 @@ impl StateServiceSubscriber { height: u32, ) -> Result { let snapshot = self.indexer.snapshot_nonfinalized_state().await?; - let chain_height = snapshot.max_serviceable_height().0; + let chain_height = snapshot.get_nfs_snapshot().best_tip.height.0; Err(if height >= chain_height { StateServiceError::TonicStatusError(tonic::Status::out_of_range(format!( "Error: Height out of range [{height}]. Height requested \ @@ -1567,19 +1563,14 @@ impl ZcashIndexer for StateServiceSubscriber { /// tags: blockchain async fn get_block_count(&self) -> Result { let snapshot = self.indexer.snapshot_nonfinalized_state().await?; - let Some(non_finalized_snapshot) = snapshot.get_nfs_snapshot() else { - // TODO: This probably shouldn't be an error. - // this is an improvement over previous behaviour of - // acting as if we are only synced to the genesis block - return Err(StateServiceError::UnavailableNotSyncedEnough); - }; - let h = non_finalized_snapshot.best_tip.height; + let nfs_snapshot = snapshot.get_nfs_snapshot(); + let h = nfs_snapshot.best_tip.height; Ok(h.into()) } async fn get_chain_tips(&self) -> Result { let snapshot = self.indexer.snapshot_nonfinalized_state().await?; - let Some(non_finalized_snapshot) = snapshot.get_nfs_snapshot() else { + let Some(non_finalized_snapshot) = snapshot.resolved_nfs_snapshot() else { return Ok(self.rpc_client.get_chain_tips().await?); }; @@ -1790,7 +1781,9 @@ impl ZcashIndexer for StateServiceSubscriber { let (height, confirmations, block_hash, in_best_chain) = match best_chain_location { Some(BestChainLocation::Block(block_hash, height)) => { let confirmations = snapshot - .max_serviceable_height() + .get_nfs_snapshot() + .best_tip + .height .0 .saturating_sub(height.0) .saturating_add(1); @@ -1905,13 +1898,8 @@ impl LightWalletIndexer for StateServiceSubscriber { /// Return the height of the tip of the best chain async fn get_latest_block(&self) -> Result { let snapshot = self.indexer.snapshot_nonfinalized_state().await?; - let Some(non_finalized_snapshot) = snapshot.get_nfs_snapshot() else { - // TODO: This probably shouldn't be an error. - // this is an improvement over previous behaviour of - // acting as if we are only synced to the genesis block - return Err(StateServiceError::UnavailableNotSyncedEnough); - }; - Ok(non_finalized_snapshot.best_tip.to_wire()) + let nfs_snapshot = snapshot.get_nfs_snapshot(); + Ok(nfs_snapshot.best_tip.to_wire()) } /// Return the compact block corresponding to the given block identifier @@ -1946,13 +1934,8 @@ impl LightWalletIndexer for StateServiceSubscriber { { Ok(Some(block)) => Ok(block), Ok(None) => { - let Some(non_finalized_snapshot) = snapshot.get_nfs_snapshot() else { - // TODO: This probably shouldn't be an error. - // this is an improvement over previous behaviour of - // acting as if we are only synced to the genesis block - return Err(StateServiceError::UnavailableNotSyncedEnough); - }; - let chain_height = non_finalized_snapshot.best_tip.height.0; + let nfs_snapshot = snapshot.get_nfs_snapshot(); + let chain_height = nfs_snapshot.best_tip.height.0; match hash_or_height { HashOrHeight::Height(Height(height)) if height >= chain_height => Err( StateServiceError::TonicStatusError(tonic::Status::out_of_range(format!( @@ -1966,13 +1949,8 @@ impl LightWalletIndexer for StateServiceSubscriber { } } Err(e) => { - let Some(non_finalized_snapshot) = snapshot.get_nfs_snapshot() else { - // TODO: This probably shouldn't be an error. - // this is an improvement over previous behaviour of - // acting as if we are only synced to the genesis block - return Err(StateServiceError::UnavailableNotSyncedEnough); - }; - let chain_height = non_finalized_snapshot.best_tip.height.0; + let nfs_snapshot = snapshot.get_nfs_snapshot(); + let chain_height = nfs_snapshot.best_tip.height.0; match hash_or_height { HashOrHeight::Height(Height(height)) if height >= chain_height => Err( StateServiceError::TonicStatusError(tonic::Status::out_of_range(format!( @@ -2400,13 +2378,8 @@ impl LightWalletIndexer for StateServiceSubscriber { let (channel_tx, channel_rx) = mpsc::channel(self.config.common.service.channel_size as usize); let snapshot = self.indexer.snapshot_nonfinalized_state().await?; - let Some(non_finalized_snapshot) = snapshot.get_nfs_snapshot() else { - // TODO: This probably shouldn't be an error. - // this is an improvement over previous behaviour of - // acting as if we are only synced to the genesis block - return Err(StateServiceError::UnavailableNotSyncedEnough); - }; - let mempool_height = non_finalized_snapshot.best_tip.height.0; + let nfs_snapshot = snapshot.get_nfs_snapshot(); + let mempool_height = nfs_snapshot.best_tip.height.0; tokio::spawn(async move { let timeout = timeout( time::Duration::from_secs((service_timeout * 6) as u64), diff --git a/packages/zaino-state/src/chain_index.rs b/packages/zaino-state/src/chain_index.rs index 182fd6107..37d42daa8 100644 --- a/packages/zaino-state/src/chain_index.rs +++ b/packages/zaino-state/src/chain_index.rs @@ -11,7 +11,7 @@ //! - b. Build trasparent tx indexes efficiently //! - NOTE: Full transaction and block data is served from the backend finalizer. -use crate::chain_index::non_finalised_state::ChainIndexSnapshot; +use crate::chain_index::non_finalised_state::{ChainIndexSnapshot, SnapshotAvailability}; use crate::chain_index::source::GetTransactionLocation; use crate::chain_index::types::db::metadata::MempoolInfo; use crate::chain_index::types::helpers::{BlockMetadata, BlockWithMetadata, TreeRootData}; @@ -22,14 +22,13 @@ use crate::error::{ChainIndexError, ChainIndexErrorKind, FinalisedStateError}; use crate::metric_names::*; use crate::status::Status; use crate::{ - CompactBlockStream, NamedAtomicStatus, NonFinalizedState, StatusType, SyncError, TxOutCompact, + CompactBlockStream, IndexedBlock, NamedAtomicStatus, Outpoint, StatusType, SyncError, + TransactionHash, TxOutCompact, }; -use crate::{IndexedBlock, Outpoint, TransactionHash}; use std::collections::HashSet; use std::str::FromStr; use std::{sync::Arc, time::Duration}; -use arc_swap::ArcSwapOption; use futures::{FutureExt, Stream}; use hex::FromHex as _; use non_finalised_state::NonfinalizedBlockCacheSnapshot; @@ -78,6 +77,9 @@ mod tests; #[cfg(not(test))] pub(crate) const NON_FINALIZED_DEPTH: u32 = zebra_state::MAX_BLOCK_REORG_HEIGHT + 1; +/// The ceiling of the finalized (immutable) chain for a given chain tip: +/// `NON_FINALIZED_DEPTH` below `best_tip`. A block is finalized — and therefore +/// reorg-safe — exactly when its height is at or below this value. /// In-crate unit tests pin the depth at the pre-zebra-10 value (`100`). /// /// Zebra 10 raised `MAX_BLOCK_REORG_HEIGHT` from 99 to 1000, so the @@ -94,13 +96,24 @@ pub(crate) const NON_FINALIZED_DEPTH: u32 = 100; /// Lower bound on zaino's finalized-DB tip, derived from the current /// best-known chain tip. /// -/// After a chain-shortening reorg this floor can move backwards while -/// the on-disk `finalized_height` does not — finalized blocks are -/// never evicted. Callers comparing this floor against -/// `finalized_height` should account for the asymmetry -/// (see zingolabs/zaino#1128). -pub(crate) fn finalized_height_floor(chain_tip: u32) -> crate::Height { - crate::Height(chain_tip.saturating_sub(NON_FINALIZED_DEPTH)) +/// Reorg-safety and finality are the same notion: a reorg is bounded to the +/// non-finalized window and can never evict a finalized block, so the ceiling +/// monotonically increases and is unaffected by reorgs. +/// +/// This is the single boundary the passthrough path needs: a height at or below +/// it is reorg-safe to fetch from the backing validator by height (see +/// [`is_finalized`]). It also drives the finalized DB's sync target. +pub(crate) fn finalization_ceiling(best_tip: u32) -> crate::Height { + crate::Height(best_tip.saturating_sub(NON_FINALIZED_DEPTH)) +} + +/// Whether `height` is finalized given the chain `best_tip` — at or below the +/// [`finalization_ceiling`], in the immutable range, and therefore reorg-safe +/// to fetch from the backing validator by height. Heights above it are in the +/// (reorg-mutable) non-finalized window and must be served only from the NFS's +/// own validated view. +pub(crate) fn is_finalized(best_tip: types::Height, height: types::Height) -> bool { + height <= finalization_ceiling(best_tip.0) } /// Current wall-clock time as a Unix timestamp in fractional seconds, for @@ -700,9 +713,8 @@ pub trait ChainIndex { /// - Snapshot-based consistency for queries #[derive(Debug)] pub struct NodeBackedChainIndex { - #[allow(dead_code)] mempool: std::sync::Arc>, - non_finalized_state: Arc>>, + non_finalized_state: Arc>, finalized_db: std::sync::Arc>, sync_loop_handle: Option>>, status: NamedAtomicStatus, @@ -796,13 +808,20 @@ impl NodeBackedChainIndex { .map_err(crate::InitError::MempoolInitialzationError) .await?; + let network = config.network.to_zebra_network(); + // The non-finalized state is built eagerly so it always exists. It + // seeds Provisional from the validator (genesis); the sync loop extends + // it, and the finalized DB catching up to the seam flips it to Resolved. + let non_finalized_state = + Arc::new(crate::NonFinalizedState::initialize(source.clone(), network.clone()).await?); + let mut chain_index = Self { mempool: std::sync::Arc::new(mempool_state), - non_finalized_state: Arc::new(ArcSwapOption::empty()), + non_finalized_state, finalized_db, sync_loop_handle: None, status: NamedAtomicStatus::new("ChainIndex", StatusType::Spawning), - network: config.network.to_zebra_network(), + network, source, sync_timings, cancel_token: CancellationToken::new(), @@ -865,7 +884,6 @@ impl NodeBackedChainIndex { let fs = self.finalized_db.clone(); let status = self.status.clone(); let source = self.source.clone(); - let network = self.network.clone(); let timings = self.sync_timings; let cancel_token = self.cancel_token.clone(); @@ -883,7 +901,6 @@ impl NodeBackedChainIndex { loop { let source = source.clone(); - let network = network.clone(); if cancel_token.is_cancelled() { return Ok(()); } @@ -920,9 +937,19 @@ impl NodeBackedChainIndex { "node returned no best block height", )) })?; + // The NFS leads: validator-sourced and anchored at the + // finalization ceiling, it walks to the iter-committed + // `chain_height` without waiting for the finalized DB. + // Passing `chain_height` rather than letting the NFS extend + // until `get_block` returns None bounds the iter against + // mid-iter source advances (#1126). The finalized DB then + // catches up toward the ceiling; until it reaches the seam + // the published snapshot is Provisional. + nfs.sync(fs.clone(), chain_height.into()).await?; + #[cfg(feature = "prometheus")] metrics::gauge!("zaino.chain.tip_height").set(chain_height.0 as f64); - let finalised_height = finalized_height_floor(chain_height.0); + let finalised_height = finalization_ceiling(chain_height.0); #[cfg(feature = "prometheus")] { metrics::gauge!(CHAIN_TIP_HEIGHT).set(chain_height.0 as f64); @@ -934,41 +961,6 @@ impl NodeBackedChainIndex { .await .map_err(source_error)?; - let intermediate_nfs_for_scoping = nfs.load(); - let non_finalized_state = match *intermediate_nfs_for_scoping { - Some(ref nfs) => nfs, - None => { - // Anchor the non-finalised state at `finalised_height` - // (= chain tip − NON_FINALIZED_DEPTH), never at genesis: a missing - // anchor used to fall through to genesis and then re-anchor up to the - // lagging finalised tip, grinding millions of blocks one at a time - // (#1261). `resolve_anchor_block` serves the anchor from the finalised - // DB / passthrough or builds it from the validator. - let anchor = NonFinalizedState::resolve_anchor_block( - &source, - &fs.to_reader(), - &network, - finalised_height, - ) - .await?; - nfs.store(Some(Arc::new( - NonFinalizedState::initialize(source, network, Some(anchor)) - .await - .map_err(source_error)?, - ))); - &nfs.load_full().expect("just set to Some") - } - }; - - // Sync nfs to the iter-committed `chain_height`, trimming - // blocks to finalized tip. Passing `chain_height` rather - // than letting NFS extend until `get_block` returns None - // bounds the iter against mid-iter source advances (#1126). - non_finalized_state - .sync(fs.clone(), chain_height.into()) - .await?; - std::mem::drop(intermediate_nfs_for_scoping); - Ok(()) } => r, }; @@ -1080,7 +1072,7 @@ impl Drop for NodeBackedChainIndex { #[derive(Clone, Debug)] pub struct NodeBackedChainIndexSubscriber { mempool: mempool::MempoolSubscriber, - non_finalized_state: Arc>>, + non_finalized_state: Arc>, finalized_state: finalised_state::reader::DbReader, status: NamedAtomicStatus, network: ZebraNetwork, @@ -1265,7 +1257,7 @@ impl NodeBackedChainIndexSubscriber { .values() .find(|h| **h == hash) // Canonical height is None for blocks not on the best chain - .map(|_| block.context.index.height)), + .map(|_| block.height())), None => self // ChainIndex step 4: .finalized_state @@ -1318,46 +1310,6 @@ impl NodeBackedChainIndexSubscriber { .chain(non_finalized_blocks_containing_transaction)) } - async fn get_block_height_passthrough( - &self, - max_serviceable_height: &types::Height, - hash: types::BlockHash, - ) -> Result, ChainIndexError> { - //ChainIndex step 5: - match self - .source() - .get_block(HashOrHeight::Hash(hash.into())) - .await - { - Ok(Some(block)) => { - // At this point, we know that - // the block is in the VALIDATOR. - match block.coinbase_height() { - None => { - // the block is in the VALIDATOR. but doesnt have a height. That would imply a bug. - Err(ChainIndexError::validator_data_error_block_coinbase_height_missing()) - } - Some(height) => { - // The VALIDATOR returned a block with a height. - // However, there is as of yet no guaranteed the Block is FINALIZED - if height <= *max_serviceable_height { - Ok(Some(types::Height::from(height))) - } else { - // non-finalized block - // no passthrough - Ok(None) - } - } - } - } - Ok(None) => { - // the block is neither in the INDEXER nor VALIDATOR - Ok(None) - } - Err(e) => Err(ChainIndexError::backing_validator(e)), - } - } - /// Returns true when the block hash is present in the local chain index. /// /// During finalized-state sync, a hash is considered known when it is in @@ -1368,36 +1320,14 @@ impl NodeBackedChainIndexSubscriber { snapshot: &ChainIndexSnapshot, hash: &types::BlockHash, ) -> Result { - match snapshot { - ChainIndexSnapshot::NonFinalizedStateExists { - non_finalized_snapshot, - } => { - if non_finalized_snapshot.blocks.contains_key(hash) { - return Ok(true); - } - Ok(self - .finalized_state - .get_block_height(*hash) - .await? - .is_some()) - } - ChainIndexSnapshot::StillSyncingFinalizedState { - validator_finalized_height, - } => { - if self - .finalized_state - .get_block_height(*hash) - .await? - .is_some() - { - return Ok(true); - } - Ok(self - .get_block_height_passthrough(validator_finalized_height, *hash) - .await? - .is_some()) - } + if snapshot.get_nfs_snapshot().blocks.contains_key(hash) { + return Ok(true); } + Ok(self + .finalized_state + .get_block_height(*hash) + .await? + .is_some()) } /// Returns true when the hash-or-height string refers to a block known to @@ -1429,14 +1359,10 @@ impl NodeBackedChainIndexSubscriber { // Get the height of the mempool fn get_mempool_height(&self, snapshot: &ChainIndexSnapshot) -> Option { - let ChainIndexSnapshot::NonFinalizedStateExists { - non_finalized_snapshot, - } = snapshot - else { - return None; - }; - - non_finalized_snapshot + // The mempool tip is a recent block; if it's in the always-present NFS + // window we can report its height (no absolute chainwork needed). + snapshot + .get_nfs_snapshot() .blocks .iter() .find(|(hash, _block)| **hash == self.mempool.mempool_chain_tip()) @@ -1457,6 +1383,9 @@ impl Status for NodeBackedChainIndexSubscriber } } +// Interim (#1096): `get_indexed_block_by_*` return the in-crate `ChainBlock` +// (collapses to `IndexedBlock` once resolution promotion lands). See the trait. +#[allow(private_interfaces)] impl ChainIndex for NodeBackedChainIndexSubscriber { type Snapshot = ChainIndexSnapshot; type Error = ChainIndexError; @@ -1467,26 +1396,11 @@ impl ChainIndex for NodeBackedChainIndexSubscriber Result { - match self.non_finalized_state.load().as_ref() { - Some(non_finalised_state) => Ok(ChainIndexSnapshot::NonFinalizedStateExists { - non_finalized_snapshot: non_finalised_state.get_snapshot(), - }), - None => { - let height = self - .source - .get_best_block_height() - .await - .map_err(ChainIndexError::backing_validator)? - .ok_or(ChainIndexError::database_hole( - "validator has no best block", - None, - ))?; - let validator_finalized_height = finalized_height_floor(height.0); - Ok(ChainIndexSnapshot::StillSyncingFinalizedState { - validator_finalized_height, - }) - } - } + // The non-finalized state always exists; a snapshot is infallible. Its + // own `availability` says whether the window is validated yet. + Ok(ChainIndexSnapshot::new( + self.non_finalized_state.get_snapshot(), + )) } // ********** Block methods ********** @@ -1505,20 +1419,36 @@ impl ChainIndex for NodeBackedChainIndexSubscriber { - self.get_indexed_block_height(non_finalized_snapshot, hash) - .await - } - ChainIndexSnapshot::StillSyncingFinalizedState { - validator_finalized_height, - } => { - self.get_block_height_passthrough(validator_finalized_height, hash) - .await - } // ChainIndex step 5 + // ChainIndex steps 2-4: serve the snapshot's own data first (NFS window + // ∪ finalised DB). + if let Some(height) = self + .get_indexed_block_height(snapshot.get_nfs_snapshot(), hash) + .await? + { + return Ok(Some(height)); + } + + // ChainIndex step 5: the block is in neither the NFS window nor the + // finalised DB. While the snapshot is provisional the finalised DB may + // not yet have ingested a finalized block the validator already has; + // that catch-up gap lies at or below the finalization ceiling and is + // reorg-safe to serve by passthrough. A block above the ceiling is in + // the reorg-mutable window and must not be served from the validator. + let best_tip = snapshot.get_nfs_snapshot().best_tip.height; + match self + .source() + .get_block(HashOrHeight::Hash(hash.into())) + .await + .map_err(ChainIndexError::backing_validator)? + { + Some(block) => match block.coinbase_height() { + None => Err(ChainIndexError::validator_data_error_block_coinbase_height_missing()), + Some(height) => { + let height = types::Height::from(height); + Ok(is_finalized(best_tip, height).then_some(height)) + } + }, + None => Ok(None), } } @@ -1530,55 +1460,46 @@ impl ChainIndex for NodeBackedChainIndexSubscriber Result, Self::Error> { - // First check non-finalised state. - match snapshot { - ChainIndexSnapshot::NonFinalizedStateExists { - non_finalized_snapshot, - } => match non_finalized_snapshot - .heights_to_hashes - .get(&height) - .copied() - { - Some(block_hash) => Ok(Some(block_hash)), - // If not found check finalised state. - None => self - .finalized_state - .get_block_hash(height) - .await - .map_err(Into::into), - }, + // Serve the snapshot's own data first: NFS window, then finalised DB. + if let Some(block_hash) = snapshot + .get_nfs_snapshot() + .heights_to_hashes + .get(&height) + .copied() + { + return Ok(Some(block_hash)); + } + if let Some(block_hash) = self + .finalized_state + .get_block_hash(height) + .await + .map_err(ChainIndexError::from)? + { + return Ok(Some(block_hash)); + } - ChainIndexSnapshot::StillSyncingFinalizedState { - validator_finalized_height, - } => { - if height <= *validator_finalized_height { - // If still syncing try to fetch from backing validator (*passthrough*). - // - // Note this requires fetching the full block from the backing node. - match self - .source() - .get_block(HashOrHeight::Height(height.into())) - .await - .map_err(ChainIndexError::backing_validator)? - { - Some(block) => Ok(Some(block.hash().into())), - None => Ok(None), - } - } else { - // The requested block is non-finalized - // We can't safely serve it via passthrough - Ok(None) - } - } + // Neither holds it. A finalized-range height (at or below the + // finalization ceiling) absent from the finalised DB is a provisional + // catch-up gap: it is reorg-safe to fetch by height from the validator. + // Heights above the ceiling are reorg-mutable — return None. + let best_tip = snapshot.get_nfs_snapshot().best_tip.height; + if !is_finalized(best_tip, height) { + return Ok(None); + } + match self + .source() + .get_block(HashOrHeight::Height(height.into())) + .await + .map_err(ChainIndexError::backing_validator)? + { + Some(block) => Ok(Some(block.hash().into())), + None => Ok(None), } } /// Returns Some(IndexedBlock) for the given block hash. /// /// Returns None if the specified block is not found. - /// - /// **NOTE: This Method is currently not "passthrough aware", cumulative - /// chain work must be made optional to enable this.** async fn get_indexed_block_by_hash( &self, snapshot: &Self::Snapshot, @@ -1600,9 +1521,6 @@ impl ChainIndex for NodeBackedChainIndexSubscriber ChainIndex for NodeBackedChainIndexSubscriber ChainIndex for NodeBackedChainIndexSubscriber ChainIndex for NodeBackedChainIndexSubscriber self - // usually getting by height is not reorg-safe, but here, height is known to be below or equal to validator_finalized_height. + // Absent from both the finalized DB and the + // NFS window: a finalized-range catch-up gap. + // The height is at or below the finalization + // ceiling, so a by-height fetch is reorg-safe. .get_fullblock_bytes_from_node(HashOrHeight::Height( zebra_chain::block::Height(height), )) @@ -1707,41 +1634,33 @@ impl ChainIndex for NodeBackedChainIndexSubscriber Result, Self::Error> { - match snapshot { - ChainIndexSnapshot::NonFinalizedStateExists { - non_finalized_snapshot, - } => { - if height <= non_finalized_snapshot.best_tip.height { - Ok(Some(match snapshot.get_chainblock_by_height(&height) { - Some(block) => compact_block_with_pool_types( - block.to_compact_block(), - &pool_types.to_pool_types_vector(), - ), - None => { - match self - .finalized_state - .get_compact_block(height, pool_types.clone()) - .await - { - Ok(block) => block, - Err(_) => self - .get_compact_block_from_node(height, &pool_types) - .await? - .ok_or(ChainIndexError::database_hole(height, None))?, - } - } - })) - } else { - Ok(None) - } - } - - ChainIndexSnapshot::StillSyncingFinalizedState { - validator_finalized_height: _, - //TODO: Once we make chainwork an option field we should be able to - // support passthrougth for this - } => Ok(None), + // Compact blocks carry no absolute chainwork, so this serves from the + // always-present NFS regardless of availability. Only a height below + // the NFS floor that the finalized DB has not ingested yet (possible + // while Provisional) falls through to the finalized DB. + let non_finalized_snapshot = snapshot.get_nfs_snapshot(); + if height > non_finalized_snapshot.best_tip.height { + return Ok(None); } + Ok(Some(match snapshot.get_chainblock_by_height(&height) { + Some(block) => compact_block_with_pool_types( + block.to_compact_block(), + &pool_types.to_pool_types_vector(), + ), + None => match self + .finalized_state + .get_compact_block(height, pool_types.clone()) + .await + { + Ok(block) => block, + // Finalized-gap while Provisional: #1066 — fall back to the + // validator here once passthrough lands. + Err(e) => self + .get_compact_block_from_node(height, &pool_types) + .await? + .ok_or(ChainIndexError::database_hole(height, Some(Box::new(e))))?, + }, + })) } /// Streams *compact* blocks for an inclusive height range. @@ -2020,19 +1939,20 @@ impl ChainIndex for NodeBackedChainIndexSubscriber block.context.index.height.into(), + Some(block) => block.height().into(), // As above Ok(None) None => return Ok(None), } @@ -2060,113 +1980,113 @@ impl ChainIndex for NodeBackedChainIndexSubscriber Result<(Option, HashSet), ChainIndexError> { - match snapshot { - ChainIndexSnapshot::NonFinalizedStateExists { - non_finalized_snapshot, - } => { - let blocks_containing_transaction = self - .blocks_containing_transaction(non_finalized_snapshot, txid.0) - .await? - .collect::>(); - let Some(start_of_nonfinalized) = - non_finalized_snapshot.heights_to_hashes.keys().min() - else { - return Err(ChainIndexError::database_hole("no blocks", None)); - }; - let mut best_chain_block = blocks_containing_transaction - .iter() - .find(|block| { - non_finalized_snapshot - .heights_to_hashes - .get(&block.height()) - == Some(block.hash()) - || block.height() < *start_of_nonfinalized - // this block is either in the best chain ``heights_to_hashes`` or finalized. - }) - .map(|block| BestChainLocation::Block(*block.hash(), block.height())); - let mut non_best_chain_blocks: HashSet = - blocks_containing_transaction - .iter() - .filter(|block| { - non_finalized_snapshot - .heights_to_hashes - .get(&block.height()) - != Some(block.hash()) - && block.height() >= *start_of_nonfinalized - }) - .map(|block| NonBestChainLocation::Block(*block.hash(), block.height())) - .collect(); - let in_mempool = self - .mempool - .contains_txid(&mempool::MempoolKey { - txid: txid.to_rpc_hex(), - }) - .await; - if in_mempool { - let mempool_tip_hash = self.mempool.mempool_chain_tip(); - if mempool_tip_hash == non_finalized_snapshot.best_tip.hash { - if best_chain_block.is_some() { - return Err(ChainIndexError { + // The non-finalised snapshot serves its own data (own blocks ∪ + // finalised state) regardless of availability; it never passes through + // to the backing validator. + let non_finalized_snapshot = snapshot.get_nfs_snapshot(); + let blocks_containing_transaction = self + .blocks_containing_transaction(non_finalized_snapshot, txid.0) + .await? + .collect::>(); + let Some(start_of_nonfinalized) = non_finalized_snapshot.heights_to_hashes.keys().min() + else { + return Err(ChainIndexError::database_hole("no blocks", None)); + }; + let mut best_chain_block = blocks_containing_transaction + .iter() + .find(|block| { + non_finalized_snapshot + .heights_to_hashes + .get(&block.height()) + == Some(block.hash()) + || block.height() < *start_of_nonfinalized + // this block is either in the best chain ``heights_to_hashes`` or finalized. + }) + .map(|block| BestChainLocation::Block(*block.hash(), block.height())); + let mut non_best_chain_blocks: HashSet = + blocks_containing_transaction + .iter() + .filter(|block| { + non_finalized_snapshot + .heights_to_hashes + .get(&block.height()) + != Some(block.hash()) + && block.height() >= *start_of_nonfinalized + }) + .map(|block| NonBestChainLocation::Block(*block.hash(), block.height())) + .collect(); + // If the snapshot's own data (NFS window ∪ finalised DB) doesn't contain + // the transaction, it may sit in a finalized-range block the finalised + // DB hasn't ingested yet (a provisional catch-up gap). Such a block, at + // or below the finalization ceiling, is reorg-safe to resolve via the + // validator. (A gap block is finalized, so it can't also be in the + // mempool — this is mutually exclusive with the mempool logic below.) + if best_chain_block.is_none() && non_best_chain_blocks.is_empty() { + if let Some((_transaction, GetTransactionLocation::BestChain(height))) = self + .source() + .get_transaction(*txid) + .await + .map_err(ChainIndexError::backing_validator)? + { + let best_tip = snapshot.get_nfs_snapshot().best_tip.height; + if is_finalized(best_tip, types::Height::from(height)) { + if let Some(block) = self + .source() + .get_block(HashOrHeight::Height(height)) + .await + .map_err(ChainIndexError::backing_validator)? + { + best_chain_block = Some(BestChainLocation::Block( + block.hash().into(), + types::Height::from(height), + )); + } + } + } + } + let in_mempool = self + .mempool + .contains_txid(&mempool::MempoolKey { + txid: txid.to_rpc_hex(), + }) + .await; + if in_mempool { + let mempool_tip_hash = self.mempool.mempool_chain_tip(); + if mempool_tip_hash == non_finalized_snapshot.best_tip.hash { + if best_chain_block.is_some() { + return Err(ChainIndexError { kind: ChainIndexErrorKind::InvalidSnapshot, message: "Best chain and up-to-date mempool both contain the same transaction" .to_string(), source: None, }); - } else { - best_chain_block = Some(BestChainLocation::Mempool( - non_finalized_snapshot.best_tip.height + 1, - )); - } - } else { - // the best chain and the mempool have divergent tip hashes - // get a new snapshot and use it to find the height of the mempool - if let ChainIndexSnapshot::NonFinalizedStateExists { - non_finalized_snapshot: new_snapshot, - } = self.snapshot_nonfinalized_state().await? - { - let target_height = - new_snapshot.blocks.iter().find_map(|(hash, block)| { - if *hash == mempool_tip_hash { - Some(block.height() + 1) - // found the block that is the tip that the mempool is hanging on to - } else { - None - } - }); - non_best_chain_blocks - .insert(NonBestChainLocation::Mempool(target_height)); - } - } + } else { + best_chain_block = Some(BestChainLocation::Mempool( + non_finalized_snapshot.best_tip.height + 1, + )); } - Ok((best_chain_block, non_best_chain_blocks)) - } - ChainIndexSnapshot::StillSyncingFinalizedState { - validator_finalized_height, - } => { - if let Some((_transaction, GetTransactionLocation::BestChain(height))) = self - .source() - .get_transaction(*txid) - .await - .map_err(ChainIndexError::backing_validator)? + } else { + // the best chain and the mempool have divergent tip hashes + // get a new snapshot and use it to find the height of the mempool { - if height <= *validator_finalized_height { - if let Some(block) = self - .source() - .get_block(HashOrHeight::Height(height)) - .await - .map_err(ChainIndexError::backing_validator)? - { - return Ok(( - Some(BestChainLocation::Block(block.hash().into(), height.into())), - HashSet::new(), - )); + // The NFS always exists; take a fresh snapshot to + // locate the mempool's tip block. + let fresh_snapshot = self.snapshot_nonfinalized_state().await?; + let new_snapshot = fresh_snapshot.get_nfs_snapshot(); + let target_height = new_snapshot.blocks.iter().find_map(|(hash, block)| { + if *hash == mempool_tip_hash { + Some(block.height() + 1) + // found the block that is the tip that the mempool is hanging on to + } else { + None } - } + }); + non_best_chain_blocks.insert(NonBestChainLocation::Mempool(target_height)); } - Ok((None, HashSet::new())) } } + Ok((best_chain_block, non_best_chain_blocks)) } /// Returns all txids currently in the mempool. @@ -2215,13 +2135,11 @@ impl ChainIndex for NodeBackedChainIndexSubscriber, ) -> Option, Self::Error>>> { let non_finalized_snapshot = match snapshot { - Some(s) => match s { - ChainIndexSnapshot::NonFinalizedStateExists { - non_finalized_snapshot, - } => Some(non_finalized_snapshot), - // If we're still syncing the finalized state, the chain tip - // is newer than the snapshot's tip. Return None. - ChainIndexSnapshot::StillSyncingFinalizedState { .. } => return None, + Some(s) => match s.availability() { + SnapshotAvailability::Reified => Some(s.get_nfs_snapshot()), + // While Provisional the chain tip is newer than the snapshot's + // tip, so there's no settled tip to anchor the stream to. + SnapshotAvailability::Provisional => return None, }, None => None, }; @@ -2297,89 +2215,69 @@ impl ChainIndex for NodeBackedChainIndexSubscriber { - match non_finalized_snapshot.get_chainblock_by_hash(hash) { - Some(block) => { - // At this point, we know that - // The block is non-FINALIZED in the INDEXER - // ChainIndex step 3: - if non_finalized_snapshot - .heights_to_hashes - .get(&block.height()) - == Some(block.hash()) - { - // The block is in the best chain. - Ok(Some((*block.hash(), block.height()))) - } else { - // Otherwise, it's non-best chain! Grab its parent, and recurse - Box::pin(self.find_fork_point(snapshot, &block.context.parent_hash)) - .await - // gotta pin recursive async functions to prevent infinite-sized - // Future-implementing types - } - } - None => { - // At this point, we know that - // the block is NOT non-FINALIZED in the INDEXER. - // as the non finalzed state is known to be populated, - // we now check the finalized state - match self.finalized_state.get_block_height(*hash).await { - Ok(Some(height)) => { - // the block is FINALIZED in the INDEXER - Ok(Some((*hash, height))) - } - Err(e) => Err(ChainIndexError::database_hole(hash, Some(Box::new(e)))), - Ok(None) => Ok(None), - } - } + // The non-finalised snapshot serves its own data (own blocks ∪ + // finalised state) regardless of availability; it never passes through + // to the backing validator. + let non_finalized_snapshot = snapshot.get_nfs_snapshot(); + match non_finalized_snapshot.get_chainblock_by_hash(hash) { + Some(block) => { + // At this point, we know that + // The block is non-FINALIZED in the INDEXER + // ChainIndex step 3: + if non_finalized_snapshot + .heights_to_hashes + .get(&block.height()) + == Some(block.hash()) + { + // The block is in the best chain. + Ok(Some((*block.hash(), block.height()))) + } else { + // Otherwise, it's non-best chain! Grab its parent, and recurse. + // NOTE: walks the (provisional-stage UNTRUSTED) prev-hash linkage; + // the Availability step should gate this trust on Resolved (#1096). + Box::pin(self.find_fork_point(snapshot, block.context.parent_hash())).await + // gotta pin recursive async functions to prevent infinite-sized + // Future-implementing types } } - ChainIndexSnapshot::StillSyncingFinalizedState { - validator_finalized_height, - } => { - // We're not fully synced, so we pass through. - // Now, we ask the VALIDATOR. - // ChainIndex step 5 - match self - .source() - .get_block(HashOrHeight::Hash(zebra_chain::block::Hash::from(*hash))) - .await - { - Ok(Some(block)) => { - // At this point, we know that - // the block is in the VALIDATOR. - match block.coinbase_height() { - None => { - // the block is in the VALIDATOR. but doesnt have a height. That would imply a bug. - Err(ChainIndexError::validator_data_error_block_coinbase_height_missing()) - } - Some(height) => { - // The VALIDATOR returned a block with a height. - // However, there is as of yet no guaranteed the Block is FINALIZED - if height <= *validator_finalized_height { - Ok(Some(( - types::BlockHash::from(block.hash()), - types::Height::from(height), - ))) - } else { - // non-finalized block - // no passthrough - Ok(None) - } - } - } + None => { + // At this point, we know that + // the block is NOT non-FINALIZED in the INDEXER. + // as the non finalzed state is known to be populated, + // we now check the finalized state + match self.finalized_state.get_block_height(*hash).await { + Ok(Some(height)) => { + // the block is FINALIZED in the INDEXER + Ok(Some((*hash, height))) } - + Err(e) => Err(ChainIndexError::database_hole(hash, Some(Box::new(e)))), Ok(None) => { - // At this point, we know that - // the block is NOT FINALIZED in the VALIDATOR. - // Return Ok(None) = no block found. - Ok(None) + // Neither the NFS window nor the finalised DB holds it. + // While provisional, a finalized-range block the + // finalised DB hasn't ingested yet (a catch-up gap, at + // or below the finalization ceiling) is reorg-safe to + // fetch by hash from the validator, and is its own fork + // point. Above the ceiling the block is reorg-mutable; + // we must not passthrough. + let best_tip = snapshot.get_nfs_snapshot().best_tip.height; + match self + .source() + .get_block(HashOrHeight::Hash(zebra_chain::block::Hash::from(*hash))) + .await + .map_err(ChainIndexError::backing_validator)? + { + Some(block) => match block.coinbase_height() { + None => Err( + ChainIndexError::validator_data_error_block_coinbase_height_missing(), + ), + Some(height) => { + let height = types::Height::from(height); + Ok(is_finalized(best_tip, height).then_some((*hash, height))) + } + }, + None => Ok(None), + } } - Err(e) => Err(ChainIndexError::backing_validator(e)), } } } @@ -2556,36 +2454,9 @@ impl ChainIndex for NodeBackedChainIndexSubscriber Result { - Ok(match snapshot { - ChainIndexSnapshot::NonFinalizedStateExists { - non_finalized_snapshot, - } => non_finalized_snapshot.best_tip, - - ChainIndexSnapshot::StillSyncingFinalizedState { - validator_finalized_height, - } => { - BlockIndex { - height: *validator_finalized_height, - hash: self - .source() - // TODO: do something more efficient than getting the whole block - .get_block(HashOrHeight::Height((*validator_finalized_height).into())) - .await - .map_err(|e| { - ChainIndexError::database_hole( - validator_finalized_height, - Some(Box::new(e)), - ) - })? - .ok_or(ChainIndexError::database_hole( - validator_finalized_height, - None, - ))? - .hash() - .into(), - } - } - }) + // The best tip is read directly from the always-present NFS in every + // availability state — no validator round-trip, so no database_hole. + Ok(snapshot.get_nfs_snapshot().best_tip) } async fn get_tx_out_set_info(&self) -> Result { @@ -2598,11 +2469,9 @@ impl ChainIndex for NodeBackedChainIndexSubscriber non_finalized_snapshot, - ChainIndexSnapshot::StillSyncingFinalizedState { .. } => { + let non_finalized_snapshot = match snapshot.availability() { + SnapshotAvailability::Reified => snapshot.get_nfs_snapshot(), + SnapshotAvailability::Provisional => { // Accumulator invariants are not established until the finalised state catches // up. Match zcashd's "stats collection failed" empty-object shape. return Ok(GetTxOutSetInfoResponse::Empty(EmptyTxOutSetInfo {})); @@ -2805,10 +2674,6 @@ where fn get_chainblock_by_height(&self, target_height: &types::Height) -> Option<&IndexedBlock> { self.as_ref().get_chainblock_by_height(target_height) } - - fn max_serviceable_height(&self) -> &types::Height { - self.as_ref().max_serviceable_height() - } } /// A snapshot of the non-finalized state, for consistent queries @@ -2817,8 +2682,6 @@ pub trait NonFinalizedSnapshot { fn get_chainblock_by_hash(&self, target_hash: &types::BlockHash) -> Option<&IndexedBlock>; /// Height -> block fn get_chainblock_by_height(&self, target_height: &types::Height) -> Option<&IndexedBlock>; - /// The maximum height that this snapshot can serve data for. - fn max_serviceable_height(&self) -> &types::Height; } impl NonFinalizedSnapshot for NonfinalizedBlockCacheSnapshot { @@ -2840,42 +2703,16 @@ impl NonFinalizedSnapshot for NonfinalizedBlockCacheSnapshot { } }) } - - fn max_serviceable_height(&self) -> &types::Height { - &self.best_tip.height - } } impl NonFinalizedSnapshot for ChainIndexSnapshot { fn get_chainblock_by_hash(&self, target_hash: &types::BlockHash) -> Option<&IndexedBlock> { - match self { - ChainIndexSnapshot::NonFinalizedStateExists { - non_finalized_snapshot, - } => non_finalized_snapshot.get_chainblock_by_hash(target_hash), - - ChainIndexSnapshot::StillSyncingFinalizedState { .. } => None, - } + // The NFS always exists; block lookup needs no absolute chainwork. + self.get_nfs_snapshot().get_chainblock_by_hash(target_hash) } fn get_chainblock_by_height(&self, target_height: &types::Height) -> Option<&IndexedBlock> { - match self { - ChainIndexSnapshot::NonFinalizedStateExists { - non_finalized_snapshot, - } => non_finalized_snapshot.get_chainblock_by_height(target_height), - - ChainIndexSnapshot::StillSyncingFinalizedState { .. } => None, - } - } - - fn max_serviceable_height(&self) -> &types::Height { - match self { - ChainIndexSnapshot::NonFinalizedStateExists { - non_finalized_snapshot, - } => non_finalized_snapshot.max_serviceable_height(), - - ChainIndexSnapshot::StillSyncingFinalizedState { - validator_finalized_height, - } => validator_finalized_height, - } + self.get_nfs_snapshot() + .get_chainblock_by_height(target_height) } } diff --git a/packages/zaino-state/src/chain_index/finalised_state.rs b/packages/zaino-state/src/chain_index/finalised_state.rs index 456df8ddf..0af96fff2 100644 --- a/packages/zaino-state/src/chain_index/finalised_state.rs +++ b/packages/zaino-state/src/chain_index/finalised_state.rs @@ -682,6 +682,15 @@ impl FinalisedState { where T: Send + Sync + 'static, { + // Honour a source-imposed sync cap (a test/backpressure hook): the + // finalized DB may be held below the chain's finalization ceiling, + // which — since the non-finalized state leads — holds the snapshot + // Provisional with a deterministic catch-up gap. + #[cfg(test)] + let height = match source.finalized_sync_cap() { + Some(cap) => height.min(cap), + None => height, + }; if self.db.primary_is_ephemeral() { return Ok(()); } @@ -1082,7 +1091,7 @@ impl FinalisedState { format!("error building block data at height {height}"), )) })?; - parent_chainwork = Some(chain_block.context.chainwork); + parent_chainwork = chain_block.context.chainwork; db.write_block_v1_0_0(chain_block).await?; } diff --git a/packages/zaino-state/src/chain_index/finalised_state/finalised_source/v1/write_core.rs b/packages/zaino-state/src/chain_index/finalised_state/finalised_source/v1/write_core.rs index 016e00d64..a827be165 100644 --- a/packages/zaino-state/src/chain_index/finalised_state/finalised_source/v1/write_core.rs +++ b/packages/zaino-state/src/chain_index/finalised_state/finalised_source/v1/write_core.rs @@ -93,7 +93,7 @@ impl DbWrite for DbV1 { Err(e) => Err(FinalisedStateError::LmdbError(e)), } })?; - (tip.0 + 1, chainwork) + (tip.0 + 1, chainwork.flatten()) } }; @@ -162,7 +162,7 @@ impl DbWrite for DbV1 { #[cfg(feature = "prometheus")] metrics::histogram!(SYNC_BLOCK_BUILD_SECONDS) .record(build_start.elapsed().as_secs_f64()); - parent_chainwork = Some(block.context.chainwork); + parent_chainwork = block.context.chainwork; batch_bytes = batch_bytes.saturating_add(approx_indexed_block_bytes(&block)); batch.push(block); next += 1; diff --git a/packages/zaino-state/src/chain_index/non_finalised_state.rs b/packages/zaino-state/src/chain_index/non_finalised_state.rs index 5e473e977..d8e4a9bb8 100644 --- a/packages/zaino-state/src/chain_index/non_finalised_state.rs +++ b/packages/zaino-state/src/chain_index/non_finalised_state.rs @@ -1,16 +1,15 @@ -use super::{ - finalised_state::{reader::DbReader, FinalisedState}, - source::BlockchainSource, - NON_FINALIZED_DEPTH, -}; +use super::{finalised_state::FinalisedState, source::BlockchainSource, NON_FINALIZED_DEPTH}; #[cfg(feature = "prometheus")] use crate::metric_names::*; use crate::{ - chain_index::types::{ - self, BlockHash, BlockIndex, BlockMetadata, BlockWithMetadata, Height, TreeRootData, + chain_index::{ + finalization_ceiling, + types::{ + self, BlockHash, BlockIndex, BlockMetadata, BlockWithMetadata, Height, TreeRootData, + }, }, error::FinalisedStateError, - ChainWork, IndexedBlock, + BlockContext, ChainWork, ChainWorkError, IndexedBlock, }; use arc_swap::ArcSwap; use futures::lock::Mutex; @@ -20,21 +19,6 @@ use tracing::{info, instrument, warn}; use zebra_chain::{parameters::Network, serialization::BytesInDisplayOrder}; use zebra_state::HashOrHeight; -/// Hard cap on how many blocks below the tip the non-finalised state retains in memory. -/// -/// [`NonFinalizedState::update`] normally trims everything below the finalised database height, -/// but that height can lag far behind the tip while the finalised DB syncs in the background, and -/// is pinned at `0` in ephemeral mode. Without an independent floor the snapshot would grow by one -/// block per new block indefinitely. This caps retention to a fixed window regardless, a small -/// margin above [`NON_FINALIZED_DEPTH`] so it never trims inside the reorg-possible range. -/// -/// It also bounds the non-finalised ancestry walkers ([`NonFinalizedState::handle_reorg`] and -/// [`NonFinalizedState::add_nonbest_block`]): neither should recurse further back than the window -/// they maintain. The bound is load-bearing for `add_nonbest_block` on the state backend, where -/// `source.get_block` serves *any* block by hash (including finalised blocks below the window), so -/// without it a side chain rooted below the anchor would recurse to genesis and overflow the stack. -const MAX_NFS_DEPTH: u32 = NON_FINALIZED_DEPTH + 10; - /// Holds the block cache #[derive(Debug)] pub struct NonFinalizedState { @@ -57,43 +41,74 @@ pub struct NonFinalizedState { } #[derive(Debug, Clone)] -/// A snapshot of the chain index -/// -/// If zaino has synced above the validator's finalized tip, -/// this contains a snapshot of the non-finalized state. +/// A consistent snapshot of the chain index's non-finalized state. /// -/// If zaino is still syncing, this contains only the height -/// of the validator's finalized tip as of snapshot creation, -/// which is used to determine how high we can pass through -/// calls to the backing validator without serving nonfinalized -/// data. -pub enum ChainIndexSnapshot { - /// Zaino is ready to serve non-finalized data. - NonFinalizedStateExists { - /// The snapshot of the non_finalized state. - #[allow(private_interfaces)] - // Rust doesn't support private fields of enum variants - // The type of this field being private gives us something like it, though - non_finalized_snapshot: Arc, - }, - /// Zaino is not ready to serve non-finalized data. - StillSyncingFinalizedState { - /// The height the validater had last finalized as of snapshot creation. - validator_finalized_height: Height, - }, +/// The non-finalized state always exists — it is built eagerly at chain-index +/// creation — so a snapshot always carries a [`NonfinalizedBlockCacheSnapshot`]. +/// Whether that window has been validated against the finalized chain yet (the +/// finalized DB has reached its seam) is its [`SnapshotAvailability`]: while +/// `Provisional`, reads that need finalized data pass through to the validator. +pub struct ChainIndexSnapshot { + non_finalized_snapshot: Arc, } impl ChainIndexSnapshot { - /// Convenience fn to go from ChainIndexSnapshot to Option, - /// throwing away the validator_finalized_height in the None case. For ease of mapping, etc. - pub(crate) fn get_nfs_snapshot(&self) -> Option<&NonfinalizedBlockCacheSnapshot> { - match self { - ChainIndexSnapshot::NonFinalizedStateExists { - non_finalized_snapshot, - } => Some(non_finalized_snapshot), - ChainIndexSnapshot::StillSyncingFinalizedState { .. } => None, + pub(crate) fn new(non_finalized_snapshot: Arc) -> Self { + Self { + non_finalized_snapshot, } } + + /// The non-finalized snapshot. Always present: the NFS is eagerly + /// constructed and never absent. + pub(crate) fn get_nfs_snapshot(&self) -> &NonfinalizedBlockCacheSnapshot { + &self.non_finalized_snapshot + } + + /// Whether the finalized DB has caught up to this window's seam, and (when + /// it has) the seam's absolute-chainwork base. + pub(crate) fn availability(&self) -> SnapshotAvailability { + self.non_finalized_snapshot.availability + } + + /// True once a sync pass has validated this window against the finalized + /// chain (the finalized DB reached the seam). While false, reads needing + /// finalized data must pass through to the validator. + pub(crate) fn is_resolved(&self) -> bool { + matches!(self.availability(), SnapshotAvailability::Reified) + } + + /// The non-finalized snapshot, but only once it is `Resolved` (validated + /// against the finalized chain). `None` while `Provisional`, so callers + /// that need authoritative data fall back (e.g. to the validator) then. + /// Distinct from [`Self::get_nfs_snapshot`], which is unconditional. + pub(crate) fn resolved_nfs_snapshot(&self) -> Option<&NonfinalizedBlockCacheSnapshot> { + self.is_resolved().then(|| self.get_nfs_snapshot()) + } +} + +/// Whether a published snapshot's non-finalized window has been validated +/// against the finalized chain yet. +/// +/// This is purely about *absolute cumulative chainwork*: the snapshot always +/// exists and always serves its own block data regardless. While `Provisional` +/// the window carries only relative work; once the finalized DB reaches the +/// seam it is `Resolved` and absolute work is recoverable. Flipped to +/// `Resolved` inside `update`, atomically with the `compare_and_swap` that +/// publishes the snapshot (the value rides in the snapshot, so the flip is not +/// separable from the block contents). It carries no validator/passthrough +/// height — the NFS never passes through; it serves its own data. +#[derive(Debug, Clone, Copy)] +pub(crate) enum SnapshotAvailability { + /// The finalized DB has not reached the seam: the window's prev-hash + /// linkage is unvalidated and its blocks carry only relative chainwork. + Provisional, + /// The finalized DB has reached the seam: the window is validated against + /// the finalized chain. (The seam's absolute-chainwork base — needed to + /// resolve window blocks' *absolute* chainwork = `base + relative` — is + /// reattached by the resolution-promotion step, #1096, alongside its + /// reader; it isn't carried yet.) + Reified, } #[derive(Debug, Clone)] @@ -112,6 +127,38 @@ pub(crate) struct NonfinalizedBlockCacheSnapshot { // best_tip is a BestTip, which contains // a Height, and a BlockHash as named fields. pub best_tip: BlockIndex, + /// Whether the finalized DB has caught up to this window's seam, and (when + /// it has) the seam's absolute-chainwork base. Set atomically with the + /// snapshot publish in `update`. + pub availability: SnapshotAvailability, +} + +/// Cumulative work measured *relative to the seam*, header-derived. +/// +/// A distinct type from [`ChainWork`] (which is ABSOLUTE) precisely so the two +/// cannot be confused at a call site: passing relative work where absolute is +/// required — or writing it into an [`IndexedBlock`]'s `chainwork` field — is a +/// type error, not merely a naming convention. This is the misattribution +/// guard in the type system. +/// +/// Relative work is a sound best-tip ordering within the non-finalized window: +/// under the assumption that no reorg exceeds [`NON_FINALIZED_DEPTH`], the seam +/// is a stable common ancestor of every competing chain, so relative ordering +/// matches absolute ordering. +#[derive(Clone, Copy, Debug, PartialEq, Eq, PartialOrd, Ord)] +pub(crate) struct ProvisionalCumulativeWork(ChainWork); + +impl ProvisionalCumulativeWork { + pub(crate) fn new(&work: &ChainWork) -> Self { + Self(work) + } + /// Accumulate one block's own (header-derived) work onto the running + /// relative total. + /// Override the normal Chainwork behaviour of never + /// adding 0 + pub(crate) fn add_block_work(&self, block_work: &ChainWork) -> Result { + Ok(Self(self.0.add(block_work)?)) + } } #[derive(Debug)] @@ -182,10 +229,6 @@ impl From for SyncError { } } -#[derive(thiserror::Error, Debug)] -#[error("Genesis block missing in validator")] -struct MissingGenesisBlock; - #[derive(thiserror::Error, Debug)] #[error("data from validator invalid: {0}")] struct InvalidData(String); @@ -202,39 +245,58 @@ pub enum InitError { #[error(transparent)] /// The finalized state failed to initialize FinalisedStateInitialzationError(#[from] FinalisedStateError), - /// the initial block provided was not on the best chain + /// The non-finalized state failed to initialize #[error("initial block not on best chain")] - InitalBlockMissingHeight, -} - -/// This is the core of the concurrent block cache. -impl BlockIndex { - /// Create a BlockID from an IndexedBlock - fn from_block(block: &IndexedBlock) -> Self { - let height = block.height(); - let hash = *block.hash(); - Self { height, hash } - } + NonFinalizedStateInitError(#[from] SyncError), } impl NonfinalizedBlockCacheSnapshot { - /// Create initial snapshot from a single block - fn from_initial_block(block: IndexedBlock) -> Self { - let best_tip = BlockIndex::from_block(&block); - let hash = *block.hash(); - let height = best_tip.height; + async fn init_from_blockchain_source( + source: &impl BlockchainSource, + network: Network, + ) -> Result { + let missing_block_error = |target_block| { + SyncError::ValidatorConnectionError(NodeConnectionError::UnrecoverableError(Box::new( + MissingBlockError(format!("source missing block: {target_block}")), + ))) + }; + let tip_height = source + .get_best_block_height() + .await + .map_err(|e| SyncError::ErrorFromSource(Box::new(e)))? + .ok_or(missing_block_error("no chaintip"))?; + let start_height = finalization_ceiling(tip_height.0); + let start_block = source + .get_block(HashOrHeight::Height(start_height.into())) + .await + .map_err(|e| SyncError::ErrorFromSource(Box::new(e)))? + .ok_or(missing_block_error( + "100 blocks below self-reported chaintip height", + ))?; + + let indexed_block = start_block + .to_indexed_block(source, network.clone()) + .await?; let mut blocks = HashMap::new(); let mut heights_to_hashes = HashMap::new(); - blocks.insert(hash, block); - heights_to_hashes.insert(height, hash); + let block_index = BlockIndex { + height: indexed_block.height(), + hash: *indexed_block.hash(), + }; - Self { + blocks.insert(block_index.hash, indexed_block); + heights_to_hashes.insert(block_index.height, block_index.hash); + + Ok(Self { blocks, heights_to_hashes, - best_tip, - } + best_tip: block_index, + // Newly seeded from the seam block: the finalized DB has not yet + // caught up to it. `update` flips this to `Resolved` once it has. + availability: SnapshotAvailability::Provisional, + }) } fn add_block_new_chaintip(&mut self, block: IndexedBlock) { @@ -284,19 +346,13 @@ impl NonFinalizedState { /// TODO: Currently, we can't initate without an snapshot, we need to create a cache /// of at least one block. Should this be tied to the instantiation of the data structure /// itself? - #[instrument(name = "NonFinalizedState::initialize", skip(source, start_block), fields(network = %network))] - pub async fn initialize( - source: Source, - network: Network, - start_block: Option, - ) -> Result { + #[instrument(name = "NonFinalizedState::initialize", skip(source), fields(network = %network))] + pub async fn initialize(source: Source, network: Network) -> Result { info!(network = %network, "Initializing non-finalized state"); - // Resolve the initial block (provided or genesis) - let initial_block = Self::resolve_initial_block(&source, &network, start_block).await?; - - // Create initial snapshot from the block - let snapshot = NonfinalizedBlockCacheSnapshot::from_initial_block(initial_block); + let snapshot = + NonfinalizedBlockCacheSnapshot::init_from_blockchain_source(&source, network.clone()) + .await?; // Set up optional listener let nfs_change_listener = Self::setup_listener(&source).await; @@ -309,96 +365,6 @@ impl NonFinalizedState { }) } - /// Fetch the genesis block and convert it to IndexedBlock - async fn get_genesis_indexed_block( - source: &Source, - network: &Network, - ) -> Result { - let genesis_block = source - .get_block(HashOrHeight::Height(zebra_chain::block::Height(0))) - .await - .map_err(|e| InitError::InvalidNodeData(Box::new(e)))? - .ok_or_else(|| InitError::InvalidNodeData(Box::new(MissingGenesisBlock)))?; - - let (sapling_root_and_len, orchard_root_and_len) = source - .get_commitment_tree_roots(genesis_block.hash().into()) - .await - .map_err(|e| InitError::InvalidNodeData(Box::new(e)))?; - - let tree_roots = TreeRootData { - sapling: sapling_root_and_len, - orchard: orchard_root_and_len, - }; - - // Genesis has no parent — pass None so create_block_context computes - // chainwork as just the genesis block's own work. - Self::create_indexed_block_with_optional_roots( - genesis_block.as_ref(), - &tree_roots, - None, - network.clone(), - ) - .map_err(|e| InitError::InvalidNodeData(Box::new(InvalidData(e)))) - } - - /// Resolve the initial block - either use provided block or fetch genesis - async fn resolve_initial_block( - source: &Source, - network: &Network, - start_block: Option, - ) -> Result { - match start_block { - Some(block) => Ok(block), - None => Self::get_genesis_indexed_block(source, network).await, - } - } - - /// Resolve the non-finalised state's anchor (root) block at `anchor_height`. - /// - /// Prefers the finalised reader, which serves the block from the persistent DB when the height - /// is in range, or from the validator via the ReadOnly ephemeral passthrough while the finalised - /// DB is catching up in the background. Falls back to building the block directly from the - /// validator source when the reader cannot serve it yet — e.g. the first worker iteration, before - /// any passthrough is installed — so the anchor never silently drops to genesis (issue #1261). - /// - /// The anchor sits below the reorg-possible range, so its chainwork is irrelevant to best-chain - /// selection; it is set to zero, matching the ephemeral passthrough's own anchor build. - pub(super) async fn resolve_anchor_block( - source: &Source, - reader: &DbReader, - network: &Network, - anchor_height: Height, - ) -> Result { - if let Some(block) = reader.get_chain_block_by_height(anchor_height).await? { - return Ok(block); - } - - let block = source - .get_block(HashOrHeight::Height(zebra_chain::block::Height( - anchor_height.0, - ))) - .await? - .ok_or_else(|| { - FinalisedStateError::DataUnavailable(format!( - "anchor block {} unavailable from validator", - anchor_height.0 - )) - })?; - - let (sapling, orchard) = source - .get_commitment_tree_roots(block.hash().into()) - .await?; - let tree_roots = TreeRootData { sapling, orchard }; - - Self::create_indexed_block_with_optional_roots( - block.as_ref(), - &tree_roots, - None, - network.clone(), - ) - .map_err(FinalisedStateError::Custom) - } - /// Set up the optional non-finalized change listener async fn setup_listener( source: &Source, @@ -431,33 +397,28 @@ impl NonFinalizedState { finalized_db: Arc>, chain_height: Height, ) -> Result<(), SyncError> { + // The NFS is validator-sourced and *leads* the finalized DB: it holds + // only the non-finalized window `[ceiling, tip]`, anchored at the + // finalization ceiling (`chain_height - NON_FINALIZED_DEPTH`). Whenever + // the current floor sits below the ceiling — at cold start (seeded from + // genesis) or after a deep rollback — re-anchor at the ceiling so the + // NFS never walks below it (in particular, never from genesis). + // + // The seam (ceiling) block comes from the finalized DB if it has already + // reached the ceiling, otherwise straight from the source: the NFS does + // not wait for the finalized DB to catch up. + let anchor_height = super::finalization_ceiling(chain_height.0); let mut initial_state = self.get_snapshot(); - let local_finalized_tip = finalized_db.to_reader().db_height().await?; - // Anchor floor: the non-finalised state must never start more than `NON_FINALIZED_DEPTH` - // blocks below the chain tip, even when the finalised DB tip lags far behind during - // background catch-up. Without this floor a freshly-initialised (or genesis-fallback) - // snapshot would try to bridge the entire gap from the finalised tip up to the chain tip one - // block at a time — millions of sequential validator fetches that never converge (#1261). - // When the floor sits above the finalised tip the anchor block isn't in the persistent DB; - // `resolve_anchor_block` serves it via the passthrough or builds it from the validator. - let anchor_height = Height( - local_finalized_tip - .map(|height| height.0) - .unwrap_or(0) - .max(u32::from(chain_height).saturating_sub(NON_FINALIZED_DEPTH)), - ); - if initial_state.best_tip.height.0 < anchor_height.0 { - let anchor_block = Self::resolve_anchor_block( - &self.source, - &finalized_db.to_reader(), - &self.network, - anchor_height, - ) - .await?; + if initial_state.best_tip.height < anchor_height { self.current.swap(Arc::new( - NonfinalizedBlockCacheSnapshot::from_initial_block(anchor_block), + NonfinalizedBlockCacheSnapshot::init_from_blockchain_source( + &self.source, + self.network.clone(), + ) + .await + .expect("todo error handling"), )); - initial_state = self.get_snapshot() + initial_state = self.get_snapshot(); } let mut working_snapshot = initial_state.as_ref().clone(); @@ -492,24 +453,11 @@ impl NonFinalizedState { let parent_hash = BlockHash::from(block.header.previous_block_hash); if parent_hash == working_snapshot.best_tip.hash { // Normal chain progression - let prev_block = working_snapshot - .blocks - .get(&working_snapshot.best_tip.hash) - .ok_or_else(|| { - SyncError::ReorgFailure(format!( - "found blocks {:?}, expected block {:?}", - working_snapshot - .blocks - .values() - .map(|block| (block.context.index.hash, block.context.index.height)) - .collect::>(), - working_snapshot.best_tip - )) - })?; - let chainblock = self.block_to_chainblock(prev_block, &block).await?; + let chainblock = + block_to_indexed_block(&block, &self.source, self.network.clone()).await?; info!( height = (working_snapshot.best_tip.height + 1).0, - hash = %chainblock.context.index.hash, + hash = %chainblock.hash(), "Syncing block" ); working_snapshot.add_block_new_chaintip(chainblock); @@ -550,12 +498,12 @@ impl NonFinalizedState { ) -> Result { // We should never recurse back more than the non-finalised window, assuming a complete // reorg of the entire nonfinalized state. `MAX_NFS_DEPTH` adds a small safety margin. - if u32::from(recursion_count) > MAX_NFS_DEPTH { + if u32::from(recursion_count) > NON_FINALIZED_DEPTH { return Err(SyncError::ReorgFailure( "reorg handling recursed beyond reason".to_string(), )); } - let prev_block = match working_snapshot + match working_snapshot .get_block_by_hash_bytes_in_serialized_order(block.prev_hash_bytes_serialized_order()) .cloned() { @@ -594,7 +542,9 @@ impl NonFinalizedState { .await? } }; - let indexed_block = block.to_indexed_block(&prev_block, self).await?; + let indexed_block = block + .to_indexed_block(&self.source, self.network.clone()) + .await?; working_snapshot.add_block_new_chaintip(indexed_block.clone()); Ok(indexed_block) } @@ -653,9 +603,18 @@ impl NonFinalizedState { return Err(SyncError::CompetingSyncProcess); }; + // The NFS holds only the non-finalized window `[seam, tip]`. The seam is + // the lowest height it tracks; listener blocks below it are finalized + // and not the NFS's concern. Processing one would walk its ancestry off + // the bottom of the window — recursing past genesis in `add_nonbest_block` + // and erroring with `MissingBlockError`. + let seam = super::finalization_ceiling(working_snapshot.best_tip.height.0); loop { match listener.try_recv() { Ok((hash, block)) => { + if block.coinbase_height().is_some_and(|h| Height(h.0) < seam) { + continue; + } if !self .current .load() @@ -692,28 +651,52 @@ impl NonFinalizedState { .map_err(|_e| UpdateError::FinalizedStateCorruption)? .unwrap_or(Height(0)); - // Trim below the finalised height, but never retain more than `MAX_NFS_DEPTH` blocks below - // the tip even when `db_height` under-reports (background sync) or is `0` (ephemeral mode). - // This bounds NFS memory to a fixed window; the `max` keeps the normal finalised-height - // floor in healthy operation, where it sits above the tip-relative cap. - let tip_height = new_snapshot.best_tip.height.0; - let trim_height = Height( - finalized_height - .0 - .max(tip_height.saturating_sub(MAX_NFS_DEPTH)), - ); - - new_snapshot.remove_finalized_blocks(trim_height); + if new_snapshot + .heights_to_hashes + .contains_key(&finalized_height) + { + new_snapshot.availability = SnapshotAvailability::Reified; + } + + let seam = match new_snapshot.availability { + SnapshotAvailability::Provisional => { + // While provisional, we keep the blocks we know to be finalized + super::finalization_ceiling(new_snapshot.best_tip.height.0) + } + // Once we're reified, we keep blocks until the FS syncs them, + // to ensure we never allow a gap + SnapshotAvailability::Reified => finalized_height, + }; + + new_snapshot.remove_finalized_blocks(seam); let best_block = &new_snapshot .blocks .values() - .max_by_key(|block| block.chainwork()) + .max_by_key(|block| block.provisional_cumulative_work(&new_snapshot)) .cloned() .expect("empty snapshot impossible"); self.handle_reorg(&mut new_snapshot, best_block, 0) .await .map_err(|_e| UpdateError::DatabaseHole)?; + // Resolve availability atomically with the publish below. The finalized + // DB has caught up iff the trimmed window still contains a block at the + // FS tip (the seam overlaps); when it does, the FS holds that block's + // absolute cumulative work, which is the base for the window's + // relative work. Until then the window floats free of the finalized + // chain — Provisional. + new_snapshot.availability = if new_snapshot + .heights_to_hashes + .contains_key(&finalized_height) + { + // Seam overlap: the finalized DB has reached the window's floor. + // (The seam's absolute-chainwork base is fetched by the resolution + // promotion, #1096, when its reader exists.) + SnapshotAvailability::Reified + } else { + SnapshotAvailability::Provisional + }; + // Need to get best hash at some point in this process let stored = self .current @@ -777,90 +760,13 @@ impl NonFinalizedState { self.current.load_full() } - async fn block_to_chainblock( - &self, - prev_block: &IndexedBlock, - block: &zebra_chain::block::Block, - ) -> Result { - let tree_roots = self - .get_tree_roots_from_source(block.hash().into()) - .await - .map_err(|e| { - SyncError::ValidatorConnectionError(NodeConnectionError::UnrecoverableError( - Box::new(InvalidData(format!("{}", e))), - )) - })?; - - Self::create_indexed_block_with_optional_roots( - block, - &tree_roots, - Some(*prev_block.chainwork()), - self.network.clone(), - ) - .map_err(|e| { - SyncError::ValidatorConnectionError(NodeConnectionError::UnrecoverableError(Box::new( - InvalidData(e), - ))) - }) - } - - /// Get commitment tree roots from the blockchain source - async fn get_tree_roots_from_source( - &self, - block_hash: BlockHash, - ) -> Result { - let (sapling_root_and_len, orchard_root_and_len) = - self.source.get_commitment_tree_roots(block_hash).await?; - - Ok(TreeRootData { - sapling: sapling_root_and_len, - orchard: orchard_root_and_len, - }) - } - - /// Create IndexedBlock with optional tree roots (for genesis/sync cases) - /// - /// TODO: Issue #604 - This uses `unwrap_or_default()` uniformly for both Sapling and Orchard, - /// but they have different activation heights. This masks potential bugs and prevents proper - /// validation based on network upgrade activation. - fn create_indexed_block_with_optional_roots( - block: &zebra_chain::block::Block, - tree_roots: &TreeRootData, - parent_chainwork: Option, - network: Network, - ) -> Result { - let (sapling_root, sapling_size, orchard_root, orchard_size) = - tree_roots.clone().extract_with_defaults(); - - let metadata = BlockMetadata::new( - sapling_root, - sapling_size as u32, - orchard_root, - orchard_size as u32, - parent_chainwork, - network, - ); - - let block_with_metadata = BlockWithMetadata::new(block, metadata); - IndexedBlock::try_from(block_with_metadata) - } - - /// Cache a non-best (side-chain) block, recursively resolving any ancestors not already in the - /// working snapshot. - /// - /// Returns `Ok(None)` when the block cannot be placed within the non-finalised window: the walk - /// back to a known ancestor exceeded [`MAX_NFS_DEPTH`], so the side chain is rooted in finalised - /// history. Skipping it is safe and intentional — zaino does not guarantee knowledge of all - /// sidechain data (see `ChainIndexReader::find_fork_point`). Without this bound the walk would - /// follow `source.get_block` down into finalised history (on the state backend `get_block` - /// serves any block by hash) and overflow the worker stack. async fn add_nonbest_block( &self, working_snapshot: &mut NonfinalizedBlockCacheSnapshot, block: &impl Block, recursion_count: u8, ) -> Result, SyncError> { - if u32::from(recursion_count) > MAX_NFS_DEPTH { + if u32::from(recursion_count) > NON_FINALIZED_DEPTH { warn!( depth = recursion_count, "non-best block ancestry walk exceeded the non-finalised window; \ @@ -868,45 +774,39 @@ impl NonFinalizedState { ); return Ok(None); } - let prev_block = match working_snapshot + if working_snapshot .get_block_by_hash_bytes_in_serialized_order(block.prev_hash_bytes_serialized_order()) .cloned() + .is_none() { - Some(block) => block, - None => { - let prev_block = self - .source - .get_block(HashOrHeight::Hash( - zebra_chain::block::Hash::from_bytes_in_serialized_order( - block.prev_hash_bytes_serialized_order(), - ), - )) - .await - .map_err(|e| { - SyncError::ValidatorConnectionError( - NodeConnectionError::UnrecoverableError(Box::new(e)), - ) - })? - .ok_or(SyncError::ValidatorConnectionError( - NodeConnectionError::UnrecoverableError(Box::new(MissingBlockError( - "zebrad missing block".to_string(), - ))), - ))?; - match Box::pin(self.add_nonbest_block( - working_snapshot, - &*prev_block, - recursion_count + 1, + let prev_block = self + .source + .get_block(HashOrHeight::Hash( + zebra_chain::block::Hash::from_bytes_in_serialized_order( + block.prev_hash_bytes_serialized_order(), + ), )) + .await + .map_err(|e| { + SyncError::ValidatorConnectionError(NodeConnectionError::UnrecoverableError( + Box::new(e), + )) + })? + .ok_or(SyncError::ValidatorConnectionError( + NodeConnectionError::UnrecoverableError(Box::new(MissingBlockError( + "zebrad missing block".to_string(), + ))), + ))?; + if Box::pin(self.add_nonbest_block(working_snapshot, &*prev_block, recursion_count + 1)) .await? - { - Some(prev_block) => prev_block, - // The parent could not be resolved within the window, so this block can't be - // placed either. Skip it (best-effort), matching the ancestor's decision. - None => return Ok(None), - } + .is_none() + { + return Ok(None); } }; - let indexed_block = block.to_indexed_block(&prev_block, self).await?; + let indexed_block = block + .to_indexed_block(&self.source, self.network.clone()) + .await?; working_snapshot .blocks .insert(*indexed_block.hash(), indexed_block.clone()); @@ -914,6 +814,85 @@ impl NonFinalizedState { } } +/// Build an [`IndexedBlock`] from a source block +/// Chainwork will always be provisional +async fn block_to_indexed_block( + block: &zebra_chain::block::Block, + source: &impl BlockchainSource, + network: Network, +) -> Result { + let tree_roots = get_tree_roots_from_source(block.hash().into(), source) + .await + .map_err(|e| { + SyncError::ValidatorConnectionError(NodeConnectionError::UnrecoverableError(Box::new( + InvalidData(format!("{}", e)), + ))) + })?; + + indexed_block_from_parts(block, &tree_roots, network).map_err(|e| { + SyncError::ValidatorConnectionError(NodeConnectionError::UnrecoverableError(Box::new( + InvalidData(e), + ))) + }) +} +/// Assemble an [`IndexedBlock`] from already-fetched parts. Reuses the +/// shared `BlockWithMetadata` extractors (`extract_block_data`, +/// `extract_transactions`, `create_commitment_tree_data`, `block_work`); +fn indexed_block_from_parts( + block: &zebra_chain::block::Block, + tree_roots: &TreeRootData, + network: Network, +) -> Result { + let (sapling_root, sapling_size, orchard_root, orchard_size) = + tree_roots.clone().extract_with_defaults(); + + let metadata = BlockMetadata::new( + sapling_root, + sapling_size as u32, + orchard_root, + orchard_size as u32, + None, + network, + ); + let block_with_metadata = BlockWithMetadata::new(block, metadata); + + let data = block_with_metadata.extract_block_data()?; + let transactions = block_with_metadata.extract_transactions()?; + let commitment_tree_data = block_with_metadata.create_commitment_tree_data(); + let chainwork = None; + + let hash = BlockHash::from(block.hash()); + let parent_hash = BlockHash::from(block.header.previous_block_hash); + let height = block + .coinbase_height() + .map(|height| Height(height.0)) + .ok_or_else(|| String::from("Any valid block has a coinbase height"))?; + + Ok(IndexedBlock { + context: BlockContext { + index: BlockIndex { height, hash }, + parent_hash, + chainwork, + }, + data, + transactions, + commitment_tree_data, + }) +} + +/// Get commitment tree roots from the blockchain source +async fn get_tree_roots_from_source( + block_hash: BlockHash, + source: &impl BlockchainSource, +) -> Result { + let (sapling_root_and_len, orchard_root_and_len) = + source.get_commitment_tree_roots(block_hash).await?; + + Ok(TreeRootData { + sapling: sapling_root_and_len, + orchard: orchard_root_and_len, + }) +} /// Errors that occur during a snapshot update pub enum UpdateError { /// The block reciever disconnected. This should only happen during shutdown. @@ -935,12 +914,12 @@ pub enum UpdateError { trait Block { fn hash_bytes_serialized_order(&self) -> [u8; 32]; - fn prev_hash_bytes_serialized_order(&self) -> [u8; 32]; async fn to_indexed_block( &self, - prev_block: &IndexedBlock, - nfs: &NonFinalizedState, + source: &Source, + network: Network, ) -> Result; + fn prev_hash_bytes_serialized_order(&self) -> [u8; 32]; } impl Block for IndexedBlock { @@ -949,17 +928,18 @@ impl Block for IndexedBlock { } fn prev_hash_bytes_serialized_order(&self) -> [u8; 32] { - self.context.parent_hash.0 + self.context.parent_hash().0 } async fn to_indexed_block( &self, - _prev_block: &IndexedBlock, - _nfs: &NonFinalizedState, + _source: &Source, + _network: Network, ) -> Result { Ok(self.clone()) } } + impl Block for zebra_chain::block::Block { fn hash_bytes_serialized_order(&self) -> [u8; 32] { self.hash().bytes_in_serialized_order() @@ -968,12 +948,11 @@ impl Block for zebra_chain::block::Block { fn prev_hash_bytes_serialized_order(&self) -> [u8; 32] { self.header.previous_block_hash.bytes_in_serialized_order() } - async fn to_indexed_block( &self, - prev_block: &IndexedBlock, - nfs: &NonFinalizedState, + source: &Source, + network: Network, ) -> Result { - nfs.block_to_chainblock(prev_block, self).await + block_to_indexed_block(self, source, network).await } } diff --git a/packages/zaino-state/src/chain_index/source.rs b/packages/zaino-state/src/chain_index/source.rs index 73e3e13de..d15eccf4d 100644 --- a/packages/zaino-state/src/chain_index/source.rs +++ b/packages/zaino-state/src/chain_index/source.rs @@ -229,6 +229,20 @@ pub trait BlockchainSource: Clone + Send + Sync + 'static { fn subscribe_to_blocks_received(&self) -> Option> { None } + + /// An optional upper bound on the height the finalized DB may sync to this + /// iteration, independent of the chain tip. + /// + /// Default `None` imposes no cap — real validators let the finalized DB + /// climb to the finalization ceiling. Test mockchains override it to hold + /// the finalized DB at a known height below the ceiling: since the + /// non-finalized state is validator-sourced and *leads*, capping the + /// finalized DB deterministically holds the snapshot Provisional with a + /// known catch-up gap — without slowing any source call. + #[cfg(test)] + fn finalized_sync_cap(&self) -> Option { + None + } } /// Sleep up to `duration`, but return early if `change_rx` resolves first. diff --git a/packages/zaino-state/src/chain_index/tests.rs b/packages/zaino-state/src/chain_index/tests.rs index d2cd1d58e..0233d654a 100644 --- a/packages/zaino-state/src/chain_index/tests.rs +++ b/packages/zaino-state/src/chain_index/tests.rs @@ -31,7 +31,7 @@ use zaino_common::{network::ActivationHeights, DatabaseConfig, Network, StorageC use crate::{ chain_index::{ finalised_state::FinalisedState, - finalized_height_floor, + finalization_ceiling, source::mockchain_source::MockchainSource, tests::vectors::{ build_active_mockchain_source, build_mockchain_source, copy_dir_recursive, @@ -48,12 +48,12 @@ use crate::{ /// /// - `Active` → `build_active_mockchain_source(150, blocks)`: source has a /// separately-tracked `active_height = 150` that tests can advance via -/// `mockchain.mine_blocks(N)`. Indexer's finalised sync target is -/// `finalized_height_floor(150) = 50`. +/// `mockchain.mine_blocks(N)`. Indexer's finalised sync target is the +/// finalization ceiling for tip 150 = 50. /// - `Static` → `build_mockchain_source(blocks)`: every loaded block is /// immediately active (`active_height = tip_height = 200` for the 201-block /// vector); the tip doesn't move during the test. Indexer's finalised sync -/// target is `finalized_height_floor(200) = 100`. +/// target is the finalization ceiling for tip 200 = 100. #[derive(Clone, Copy, Debug, PartialEq, Eq)] enum MockchainMode { Active, @@ -144,7 +144,7 @@ async fn load_with_settings( // Wait until the indexer's non-finalised state has been built and its // best tip matches the source. The previous form checked only - // `finalized_state.db_height() == finalized_height_floor(active_height)`, + // `finalized_state.db_height() == finalization ceiling for active_height`, // which the seed copy makes true *before* the sync loop has had a chance // to initialise NFS. Tests that read the NFS immediately after the // helper returns (`nfs_lowest_block_matches_finalized_db_tip`, @@ -164,7 +164,7 @@ async fn load_with_settings( loop { let nfs_ready = match index_reader.snapshot_nonfinalized_state().await { Ok(snap) => snap - .get_nfs_snapshot() + .resolved_nfs_snapshot() .is_some_and(|nfs| nfs.best_tip.height.0 == expected_nfs_tip), Err(_) => false, }; @@ -209,7 +209,7 @@ async fn v1_finalised_seed_dir(mode: MockchainMode) -> &'static Path { MockchainMode::Active => build_active_mockchain_source(150, blocks.clone()), MockchainMode::Static => build_mockchain_source(blocks.clone()), }; - let target = finalized_height_floor(source.active_height()).0; + let target = finalization_ceiling(source.active_height()).0; let temp_dir: TempDir = tempfile::tempdir().unwrap(); let config = ChainIndexConfig { diff --git a/packages/zaino-state/src/chain_index/tests/mockchain_tests.rs b/packages/zaino-state/src/chain_index/tests/mockchain_tests.rs index 31518411e..fbc6897d5 100644 --- a/packages/zaino-state/src/chain_index/tests/mockchain_tests.rs +++ b/packages/zaino-state/src/chain_index/tests/mockchain_tests.rs @@ -39,7 +39,7 @@ async fn wait_for_indexer_tip( .snapshot_nonfinalized_state() .await .ok()? - .get_nfs_snapshot()? + .resolved_nfs_snapshot()? .best_tip .height .0; @@ -176,7 +176,7 @@ async fn sync_blocks_after_startup() { .snapshot_nonfinalized_state() .await .unwrap() - .get_nfs_snapshot() + .resolved_nfs_snapshot() .unwrap() .best_tip ) @@ -195,7 +195,7 @@ async fn sync_blocks_after_startup() { .snapshot_nonfinalized_state() .await .unwrap() - .get_nfs_snapshot() + .resolved_nfs_snapshot() .unwrap() .best_tip ) diff --git a/packages/zaino-state/src/chain_index/tests/non_finalised_state.rs b/packages/zaino-state/src/chain_index/tests/non_finalised_state.rs index f39068b16..04889613a 100644 --- a/packages/zaino-state/src/chain_index/tests/non_finalised_state.rs +++ b/packages/zaino-state/src/chain_index/tests/non_finalised_state.rs @@ -16,12 +16,14 @@ //! absence (the slot does not flip back to "still syncing"). //! - **G**: `shutdown()` causes the sync loop to terminate cleanly. //! -//! Tests of the cold-start "still-syncing" variant are deliberately -//! omitted: that variant is being eliminated, and pinning its shape -//! would create immediate test churn at the refactor PR. +//! The cold-start "still-syncing" variant has been eliminated (#1096): the NFS +//! is always present and carries a `Provisional`/`Resolved` availability. The +//! trailing `best_chaintip_derives_tip_from_nfs_snapshot` test — formerly the +//! red driver for that elimination — is now a passing regression that pins +//! `best_chaintip` reading the snapshot's `best_tip` directly. use super::{load_test_vectors_and_sync_chain_index, poll::poll_until, MockchainMode}; -use crate::chain_index::{finalized_height_floor, ChainIndex}; +use crate::chain_index::{finalization_ceiling, ChainIndex}; use std::time::Duration; use tokio::time::sleep; @@ -36,10 +38,10 @@ async fn nfs_lowest_block_matches_finalized_db_tip() { let snapshot = index_reader.snapshot_nonfinalized_state().await.unwrap(); let nfs = snapshot - .get_nfs_snapshot() + .resolved_nfs_snapshot() .expect("NFS exists after harness completes finalized sync"); - let seam_height = finalized_height_floor(mockchain.active_height()); + let seam_height = finalization_ceiling(mockchain.active_height()); let nfs_seam_hash = nfs .heights_to_hashes .get(&seam_height) @@ -61,6 +63,50 @@ async fn nfs_lowest_block_matches_finalized_db_tip() { ); } +/// Converged full-coverage check, the Resolved counterpart to the Provisional +/// `passthrough_*` tests: once the indexer has resolved over the real-vector +/// chain, every height from genesis to the best tip — across the FS/NFS seam — +/// is served from the snapshot's own data (finalized DB below the seam, NFS +/// window above it), with no catch-up gap, and height ↔ hash round-trips +/// through all three query methods. +#[tokio::test(flavor = "multi_thread")] +async fn resolved_snapshot_serves_every_block() { + let (_blocks, _indexer, index_reader, mockchain) = + load_test_vectors_and_sync_chain_index(MockchainMode::Active).await; + + let snapshot = index_reader.snapshot_nonfinalized_state().await.unwrap(); + assert!( + snapshot.is_resolved(), + "harness syncs the finalized DB to the seam, so the snapshot is Resolved", + ); + + for height in 0..=mockchain.active_height() { + let height = crate::Height(height); + let hash = index_reader + .get_block_hash(&snapshot, height) + .await + .unwrap() + .unwrap_or_else(|| panic!("no block served at height {}", height.0)); + + assert_eq!( + index_reader + .get_block_height(&snapshot, hash) + .await + .unwrap(), + Some(height), + "get_block_height round-trip at height {}", + height.0, + ); + + let fork_point = index_reader + .find_fork_point(&snapshot, &hash) + .await + .unwrap() + .unwrap_or_else(|| panic!("no fork point served at height {}", height.0)); + assert_eq!(fork_point, (hash, height)); + } +} + /// **D**: A block in the NFS is evicted once the finalized DB advances /// past its height. Pins the trim step inside `update` /// (`non_finalised_state.rs:remove_finalized_blocks`, which retains @@ -70,11 +116,11 @@ async fn block_is_evicted_from_nfs_when_finalized_advances_past_it() { let (_blocks, _indexer, index_reader, mockchain) = load_test_vectors_and_sync_chain_index(MockchainMode::Active).await; - let initial_seam_height = finalized_height_floor(mockchain.active_height()); + let initial_seam_height = finalization_ceiling(mockchain.active_height()); let initial_snapshot = index_reader.snapshot_nonfinalized_state().await.unwrap(); let initial_nfs = initial_snapshot - .get_nfs_snapshot() + .resolved_nfs_snapshot() .expect("NFS exists after harness"); let target_hash = *initial_nfs .heights_to_hashes @@ -86,7 +132,6 @@ async fn block_is_evicted_from_nfs_when_finalized_advances_past_it() { ); mockchain.mine_blocks(20); - let post_mine_active_height = mockchain.active_height(); // Poll the *NFS tip*, not `finalized_state.db_height()`: // `fs.sync_to_height` advances the finalized DB BEFORE @@ -95,20 +140,20 @@ async fn block_is_evicted_from_nfs_when_finalized_advances_past_it() { // NFS reaching the post-mine chain tip is only observable after // `update` has published the trimmed snapshot. poll_until( - "NFS tip to catch up to the mined chain (post-trim state)", + "NFS tip to catch up to the mined chain (post-trim state) and FS to sync to finalized tip also", Duration::from_secs(10), Duration::from_millis(25), || async { let snapshot = index_reader.snapshot_nonfinalized_state().await.ok()?; - let nfs = snapshot.get_nfs_snapshot()?; - (nfs.best_tip.height.0 == post_mine_active_height).then_some(()) + let nfs = snapshot.resolved_nfs_snapshot()?; + (!nfs.blocks.contains_key(&target_hash)).then_some(()) }, ) .await; let later_snapshot = index_reader.snapshot_nonfinalized_state().await.unwrap(); let later_nfs = later_snapshot - .get_nfs_snapshot() + .resolved_nfs_snapshot() .expect("NFS still exists after advance"); assert!( @@ -136,7 +181,7 @@ async fn nfs_slot_is_monotonic_post_init() { for i in 0..10 { let snapshot = index_reader.snapshot_nonfinalized_state().await.unwrap(); assert!( - snapshot.get_nfs_snapshot().is_some(), + snapshot.resolved_nfs_snapshot().is_some(), "iteration {i}: post-init snapshot must contain an NFS", ); sleep(Duration::from_millis(100)).await; @@ -170,7 +215,7 @@ async fn shutdown_terminates_sync_loop_cleanly() { Duration::from_millis(50), || async { let snapshot = index_reader.snapshot_nonfinalized_state().await.ok()?; - let nfs = snapshot.get_nfs_snapshot()?; + let nfs = snapshot.resolved_nfs_snapshot()?; (nfs.best_tip.height.0 == target_tip).then_some(()) }, ) @@ -210,13 +255,13 @@ async fn shutdown_terminates_sync_loop_cleanly() { /// [`MockchainSource::arm_one_shot_get_block_hook`]. The hook fires the /// *first* time the worker requests `get_block(Height(_))`, which is the /// first call inside iter N's NFS-sync while loop *after* iter N has already -/// committed to `chain_height = initial_active` and called -/// `fs.sync_to_height(finalized_height_floor(initial_active))` as a no-op. +/// committed to `chain_height = initial_active` and synced the finalized DB to +/// `finalization_ceiling(initial_active)` as a no-op. /// From inside the hook the test mines 20 blocks; the same `get_block` call /// then reads the *new* `active_chain_height = initial_active + 20` and /// returns block `initial_active + 1`, which the worker's loop happily /// extends past the iter's commitment all the way to `initial_active + 20`. -/// The iter's `update` step uses `finalized_height_floor(initial_active)` +/// The iter's `update` step uses the finalization ceiling for `initial_active` /// for the trim and publishes a snapshot whose lowest height is *below* the /// post-mine seam. /// @@ -234,11 +279,11 @@ async fn race_pre_mine_finalized_height_block_is_evicted_when_source_advances_mi load_test_vectors_and_sync_chain_index(MockchainMode::Active).await; let initial_active = mockchain.active_height(); - let pre_mine_finalized_height = dbg!(finalized_height_floor(initial_active)); + let pre_mine_finalized_height = finalization_ceiling(initial_active); let initial_snapshot = index_reader.snapshot_nonfinalized_state().await.unwrap(); let initial_nfs = initial_snapshot - .get_nfs_snapshot() + .resolved_nfs_snapshot() .expect("NFS exists after harness"); let target_hash = *initial_nfs .heights_to_hashes @@ -260,22 +305,21 @@ async fn race_pre_mine_finalized_height_block_is_evicted_when_source_advances_mi let mc = mockchain.clone(); mockchain.arm_one_shot_get_block_hook(Box::new(move || mc.mine_blocks(advance))); - let post_mine_active = initial_active + advance; poll_until( "NFS tip to reach post-mine height (race window forced)", Duration::from_secs(10), Duration::from_millis(25), || async { let snapshot = index_reader.snapshot_nonfinalized_state().await.ok()?; - let nfs = snapshot.get_nfs_snapshot()?; - (nfs.best_tip.height.0 == post_mine_active).then_some(()) + let nfs = snapshot.resolved_nfs_snapshot()?; + (!nfs.blocks.contains_key(&target_hash)).then_some(()) }, ) .await; let later_snapshot = index_reader.snapshot_nonfinalized_state().await.unwrap(); let later_nfs = later_snapshot - .get_nfs_snapshot() + .resolved_nfs_snapshot() .expect("NFS still exists after advance"); assert!( @@ -286,3 +330,29 @@ async fn race_pre_mine_finalized_height_block_is_evicted_when_source_advances_mi pre_mine_finalized_height.0, ); } + +/// Regression for #1096: `best_chaintip` derives the tip from the +/// always-present NFS snapshot, never via a validator passthrough. +/// +/// Before #1096, the cold-start `StillSyncingFinalizedState` variant had no +/// snapshot tip, so `best_chaintip` round-tripped to the validator and reported +/// the finalized *floor* (and could surface `database_hole`). Now the snapshot +/// always carries an NFS `best_tip` and `best_chaintip` reads it directly. This +/// pins that `best_chaintip` equals the snapshot's `best_tip` — which, after the +/// harness's sync, is the real chain tip — with no validator call. +/// +/// multi_thread: depends on the harness's background sync loop advancing the +/// NFS concurrently with the setup's poll-until-ready loop. +#[tokio::test(flavor = "multi_thread")] +async fn best_chaintip_derives_tip_from_nfs_snapshot_not_validator_passthrough() { + let (_blocks, _indexer, index_reader, mockchain) = + load_test_vectors_and_sync_chain_index(MockchainMode::Active).await; + + let snapshot = index_reader.snapshot_nonfinalized_state().await.unwrap(); + let tip = index_reader.best_chaintip(&snapshot).await.unwrap(); + + // best_chaintip reads the snapshot's best_tip directly (no passthrough)... + assert_eq!(tip, snapshot.get_nfs_snapshot().best_tip); + // ...which the harness has synced to the real chain tip. + assert_eq!(tip.height.0, mockchain.active_height()); +} diff --git a/packages/zaino-state/src/chain_index/tests/proptest_blockgen.rs b/packages/zaino-state/src/chain_index/tests/proptest_blockgen.rs index 8dde1281c..56e81f269 100644 --- a/packages/zaino-state/src/chain_index/tests/proptest_blockgen.rs +++ b/packages/zaino-state/src/chain_index/tests/proptest_blockgen.rs @@ -26,7 +26,7 @@ use zebra_state::{FromDisk, HashOrHeight, IntoDisk as _}; use crate::{ chain_index::{ - finalized_height_floor, + finalization_ceiling, non_finalised_state::ChainIndexSnapshot, source::{BlockchainSourceResult, GetTransactionLocation}, tests::{init_tracing, poll::poll_until, proptest_blockgen::proptest_helpers::add_segment}, @@ -37,6 +37,13 @@ use crate::{ NodeBackedChainIndexSubscriber, TransactionHash, }; +/// The finalization ceiling for a snapshot's own best tip: the highest height +/// it serves from its own data, and the boundary below which the validator may +/// be consulted by passthrough. +fn snapshot_finalization_ceiling(snapshot: &ChainIndexSnapshot) -> crate::Height { + finalization_ceiling(snapshot.get_nfs_snapshot().best_tip.height.0) +} + /// Handle all the boilerplate for a passthrough fn passthrough_test( // The actual assertions. Takes as args: @@ -65,12 +72,15 @@ fn passthrough_test( let mockchain = ProptestMockchain { genesis_segment, branching_segments, - // This number can be played with. We want to slow down - // sync enough to trigger passthrough without - // slowing down passthrough more than we need to - delay: Some(Duration::from_millis(100)), + // Hold the finalized DB at genesis so the snapshot stays + // Provisional: the always-leading NFS reaches the tip while the + // finalized DB lags, so every block below the finalization + // ceiling is served through the validator-passthrough gap. No + // artificial per-call delay — the gap is deterministic. + finalized_sync_cap: Arc::new(std::sync::atomic::AtomicU32::new(0)), best_branch_cache: Arc::new(std::sync::OnceLock::new()), tx_index: Arc::new(std::sync::OnceLock::new()), + commitment_roots_cache: Arc::new(std::sync::OnceLock::new()), }; let temp_dir: tempfile::TempDir = tempfile::tempdir().unwrap(); let db_path: std::path::PathBuf = temp_dir.path().to_path_buf(); @@ -98,27 +108,30 @@ fn passthrough_test( // serviceable cutoff is the finalized floor at that tip — mirror // production's `finalized_height_floor` exactly. let tip_height = (2 * segment_length - 1) as u32; - let expected_max_serviceable_height = finalized_height_floor(tip_height).0 as usize; + let expected_finalization_ceiling = finalization_ceiling(tip_height).0 as usize; // Poll rather than sleeping a fixed 5 s: the indexer discovers the // chain topology as soon as the sync task has walked enough of the // source to identify the finalized-state cutoff. With a 1 s // per-block source delay (above) that's well under 5 s in practice, // but can be longer under parallel-suite scheduler pressure. poll_until( - "indexer to reach expected max_serviceable_height", + "indexer to reach the expected finalization ceiling", Duration::from_secs(30), Duration::from_millis(50), || async { let snapshot = index_reader.snapshot_nonfinalized_state().await.ok()?; - (snapshot.max_serviceable_height().0 as usize - == expected_max_serviceable_height) + (snapshot_finalization_ceiling(&snapshot).0 as usize + == expected_finalization_ceiling) .then_some(()) }, ) .await; let snapshot = index_reader.snapshot_nonfinalized_state().await.unwrap(); - assert_eq!(snapshot.max_serviceable_height().0 as usize, expected_max_serviceable_height); - assert!(matches!(snapshot, ChainIndexSnapshot::StillSyncingFinalizedState { .. })); + assert_eq!( + snapshot_finalization_ceiling(&snapshot).0 as usize, + expected_finalization_ceiling + ); + assert!(!snapshot.is_resolved()); test(&mockchain, index_reader, &snapshot).await; @@ -139,7 +152,7 @@ fn passthrough_find_fork_point() { // as this lets us call all the get_raw_transaction requests // at the same time and wait for them in parallel // - // This allows the artificial delays to happen in parallel + // This lets the per-block passthrough source calls run concurrently. let mut parallel = FuturesUnordered::new(); // As we only have one branch, arbitrary branch order is fine for (height, hash) in mockchain @@ -154,14 +167,11 @@ fn passthrough_find_fork_point() { .await .unwrap(); - if height <= *snapshot.max_serviceable_height() { - // passthrough fork point can only ever be the requested block - // as we don't passthrough to nonfinalized state - assert_eq!(hash, fork_point.unwrap().0); - assert_eq!(height, fork_point.unwrap().1); - } else { - assert!(fork_point.is_none()); - } + // Single branch: every block is on the best chain, so its fork + // point is itself — served from the NFS window, or (below the + // ceiling) via the validator-passthrough gap. Never None. + assert_eq!(hash, fork_point.unwrap().0); + assert_eq!(height, fork_point.unwrap().1); }) } while let Some(_success) = parallel.next().await {} @@ -175,7 +185,7 @@ fn passthrough_get_transaction_status() { // as this lets us call all the get_raw_transaction requests // at the same time and wait for them in parallel // - // This allows the artificial delays to happen in parallel + // This lets the per-block passthrough source calls run concurrently. let mut parallel = FuturesUnordered::new(); // As we only have one branch, arbitrary branch order is fine for (height, txid) in mockchain.all_blocks_arb_branch_order().flat_map(|block| { @@ -193,18 +203,14 @@ fn passthrough_get_transaction_status() { .await .unwrap(); - if height <= *snapshot.max_serviceable_height() { - // passthrough transaction status can only ever be on the best - // chain as we don't passthrough to nonfinalized state - let Some(BestChainLocation::Block(_block_hash, transaction_height)) = - transaction_status.0 - else { - panic!("expected best chain location") - }; - assert_eq!(height, transaction_height); - } else { - assert!(transaction_status.0.is_none()); - } + // Single branch: every transaction is on the best chain, served + // from the NFS window or (below the ceiling) the passthrough gap. + let Some(BestChainLocation::Block(_block_hash, transaction_height)) = + transaction_status.0 + else { + panic!("expected best chain location") + }; + assert_eq!(height, transaction_height); assert!(transaction_status.1.is_empty()); }) } @@ -219,7 +225,7 @@ fn passthrough_get_raw_transaction() { // as this lets us call all the get_raw_transaction requests // at the same time and wait for them in parallel // - // This allows the artificial delays to happen in parallel + // This lets the per-block passthrough source calls run concurrently. let mut parallel = FuturesUnordered::new(); // As we only have one branch, arbitrary branch order is fine for (expected_transaction, height) in @@ -258,6 +264,8 @@ fn passthrough_get_raw_transaction() { fn passthrough_best_chaintip() { passthrough_test(async |mockchain, index_reader, snapshot| { let tip = index_reader.best_chaintip(snapshot).await.unwrap(); + // best_chaintip derives from the always-leading NFS, which sits at the + // full chain tip even while the finalized DB lags behind. assert_eq!( tip.height.0, mockchain @@ -265,8 +273,8 @@ fn passthrough_best_chaintip() { .last() .unwrap() .coinbase_height() - .map(|h| finalized_height_floor(h.0).0) .unwrap() + .0 ); }) } @@ -278,7 +286,7 @@ fn passthrough_get_block_height() { // as this lets us call all the get_raw_transaction requests // at the same time and wait for them in parallel // - // This allows the artificial delays to happen in parallel + // This lets the per-block passthrough source calls run concurrently. let mut parallel = FuturesUnordered::new(); for (expected_height, hash) in mockchain @@ -292,11 +300,9 @@ fn passthrough_get_block_height() { .get_block_height(&snapshot, hash.into()) .await .unwrap(); - if expected_height <= *snapshot.max_serviceable_height() { - assert_eq!(height, Some(expected_height.into())); - } else { - assert_eq!(height, None); - } + // Every block is served: the NFS window, or (below the ceiling) + // the validator-passthrough gap. + assert_eq!(height, Some(expected_height.into())); }); } while let Some(_success) = parallel.next().await {} @@ -310,7 +316,7 @@ fn passthrough_get_block_range() { // as this lets us call all the get_raw_transaction requests // at the same time and wait for them in parallel // - // This allows the artificial delays to happen in parallel + // This lets the per-block passthrough source calls run concurrently. let mut parallel = FuturesUnordered::new(); for expected_start_height in mockchain @@ -327,36 +333,33 @@ fn passthrough_get_block_range() { expected_start_height.into(), Some(expected_end_height.into()), ); - if expected_start_height <= *snapshot.max_serviceable_height() { - let mut block_range_stream = Box::pin(block_range_stream.unwrap()); - let mut num_blocks_in_stream = 0; - while let Some(block) = block_range_stream.next().await { - let expected_block = mockchain - .all_blocks_arb_branch_order() - .nth(expected_start_height.0 as usize + num_blocks_in_stream) - .unwrap() - .zcash_serialize_to_vec() - .unwrap(); - assert_eq!(block.unwrap(), expected_block); - num_blocks_in_stream += 1; - } - assert_eq!( - num_blocks_in_stream, - // expect 10 blocks - 10.min( - // unless the provided range overlaps the finalized boundary. - // in that case, expect all blocks between start height - // and finalized height, (+1 for inclusive range) - snapshot - .max_serviceable_height() - .0 - .saturating_sub(expected_start_height.0) - + 1 - ) as usize - ); - } else { - assert!(block_range_stream.is_none()) + // Every height up to the tip is served (NFS window ∪ + // passthrough gap), so the range is always servable. + let mut block_range_stream = Box::pin(block_range_stream.unwrap()); + let mut num_blocks_in_stream = 0; + while let Some(block) = block_range_stream.next().await { + let expected_block = mockchain + .all_blocks_arb_branch_order() + .nth(expected_start_height.0 as usize + num_blocks_in_stream) + .unwrap() + .zcash_serialize_to_vec() + .unwrap(); + assert_eq!(block.unwrap(), expected_block); + num_blocks_in_stream += 1; } + assert_eq!( + num_blocks_in_stream, + // 10 blocks, unless the range runs past the best tip. + 10.min( + snapshot + .get_nfs_snapshot() + .best_tip + .height + .0 + .saturating_sub(expected_start_height.0) + + 1 + ) as usize + ); }); } } @@ -390,9 +393,12 @@ fn make_chain() { let mockchain = ProptestMockchain { genesis_segment, branching_segments, - delay: None, + // No cap: the finalized DB syncs all the way to the ceiling, so + // the snapshot resolves. + finalized_sync_cap: Arc::new(std::sync::atomic::AtomicU32::new(u32::MAX)), best_branch_cache: Arc::new(std::sync::OnceLock::new()), tx_index: Arc::new(std::sync::OnceLock::new()), + commitment_roots_cache: Arc::new(std::sync::OnceLock::new()), }; let temp_dir: tempfile::TempDir = tempfile::tempdir().unwrap(); let db_path: std::path::PathBuf = temp_dir.path().to_path_buf(); @@ -422,19 +428,22 @@ fn make_chain() { Duration::from_millis(25), || async { let snapshot = index_reader.snapshot_nonfinalized_state().await.ok()?; - (snapshot.get_nfs_snapshot()?.blocks.len() == expected_block_count) + (snapshot.resolved_nfs_snapshot()?.blocks.len() == expected_block_count) .then_some(snapshot) }, ) .await; - let non_finalized_snapshot = snapshot.get_nfs_snapshot().expect("not synced"); + let non_finalized_snapshot = snapshot.resolved_nfs_snapshot().expect("not synced"); let best_tip_hash = non_finalized_snapshot.best_tip.hash; let best_tip_block = non_finalized_snapshot .get_chainblock_by_hash(&best_tip_hash) .unwrap(); for (hash, block) in &non_finalized_snapshot.blocks { if hash != &best_tip_hash { - assert!(block.chainwork() <= best_tip_block.chainwork()); + assert!( + block.provisional_cumulative_work(non_finalized_snapshot) + <= best_tip_block.provisional_cumulative_work(non_finalized_snapshot) + ); if non_finalized_snapshot.heights_to_hashes.get(&block.height()) == Some(block.hash()) { assert_eq!(index_reader.find_fork_point(&snapshot, hash).await.unwrap().unwrap().0, *hash); } else { @@ -451,11 +460,75 @@ fn make_chain() { }); } +/// Sapling and Orchard commitment-tree `(root, tree_size)` for a block, as +/// returned by [`BlockchainSource::get_commitment_tree_roots`]. +type CommitmentRoots = ( + Option<(zebra_chain::sapling::tree::Root, u64)>, + Option<(zebra_chain::orchard::tree::Root, u64)>, +); + +type SaplingFrontier = incrementalmerkletree::frontier::Frontier; +type OrchardFrontier = + incrementalmerkletree::frontier::Frontier; + +/// Append each block's note commitments to the running frontiers, recording the +/// resulting `(root, tree_size)` per block hash into `map`. Returns the final +/// frontiers so a branch can continue from where the genesis segment left off. +fn fold_commitment_roots<'a>( + blocks: impl Iterator>, + mut sapling: Option, + mut orchard: Option, + map: &mut std::collections::HashMap, +) -> (Option, Option) { + sapling.get_or_insert_with(SaplingFrontier::empty); + orchard.get_or_insert_with(OrchardFrontier::empty); + for block in blocks { + for transaction in &block.transactions { + for sap in transaction.sapling_note_commitments() { + let node = sapling_crypto::Node::from_bytes(sap.to_bytes()).unwrap(); + sapling + .get_or_insert_with(SaplingFrontier::empty) + .append(node); + } + for orc in transaction.orchard_note_commitments() { + let node = zebra_chain::orchard::tree::Node::from(*orc); + orchard + .get_or_insert_with(OrchardFrontier::empty) + .append(node); + } + } + map.insert( + BlockHash::from(block.hash()), + ( + sapling.as_ref().map(|f| { + ( + zebra_chain::sapling::tree::Root::from_bytes(f.root().to_bytes()), + f.tree_size(), + ) + }), + orchard.as_ref().map(|f| { + ( + zebra_chain::orchard::tree::Root::from_bytes(f.root().as_bytes()), + f.tree_size(), + ) + }), + ), + ); + } + (sapling, orchard) +} + #[derive(Clone)] struct ProptestMockchain { genesis_segment: ChainSegment, branching_segments: Vec, - delay: Option, + /// Caps the height the finalized DB may sync to (`u32::MAX` = no cap), + /// surfaced via [`BlockchainSource::finalized_sync_cap`]. Holding it below + /// the finalization ceiling keeps the snapshot deterministically Provisional + /// — the always-leading NFS reaches the tip while the finalized DB stays + /// behind — so the catch-up gap (served by passthrough) is exercised without + /// any artificial per-call delay. + finalized_sync_cap: Arc, /// Cached result of `best_branch()`. The best branch is pure function of /// the other fields (which are never mutated after construction), so it's /// safe to memoize. Shared via `Arc` so `mockchain.clone()` — which @@ -478,6 +551,12 @@ struct ProptestMockchain { >, >, >, + /// Cached commitment-tree roots keyed by block hash, built lazily in one + /// incremental pass. The finalized DB requires real roots (it rejects + /// `None`), and recomputing them by folding from genesis on every call was + /// O(N) crypto per call → O(N²) across a sync; this makes lookups O(1). + commitment_roots_cache: + Arc>>, } impl ProptestMockchain { @@ -545,33 +624,6 @@ impl ProptestMockchain { .flat_map(|branch| branch.iter()), ) } - - fn get_block_and_all_preceeding( - &self, - // This probably doesn't need to allow FnMut closures (Fn should suffice) - // but there's no cost to allowing it - mut block_identifier: impl FnMut(&zebra_chain::block::Block) -> bool, - ) -> std::option::Option>> { - let mut blocks = Vec::new(); - for block in self.genesis_segment.iter() { - blocks.push(block); - if block_identifier(block) { - return Some(blocks); - } - } - for branch in self.branching_segments.iter() { - let mut branch_blocks = Vec::new(); - for block in branch.iter() { - branch_blocks.push(block); - if block_identifier(block) { - blocks.extend_from_slice(&branch_blocks); - return Some(blocks); - } - } - } - - None - } } impl BlockchainSource for ProptestMockchain { @@ -580,9 +632,6 @@ impl BlockchainSource for ProptestMockchain { &self, id: HashOrHeight, ) -> BlockchainSourceResult>> { - if let Some(delay) = self.delay { - tokio::time::sleep(delay).await; - } match id { HashOrHeight::Hash(hash) => { let matches_hash = |block: &&Arc| block.hash() == hash; @@ -616,72 +665,29 @@ impl BlockchainSource for ProptestMockchain { } } - /// Returns the block commitment tree data by hash + /// Returns the block commitment tree data by hash. + /// + /// The NFS sync calls this once per block as it walks the window, so roots + /// are precomputed for every block in one incremental pass and cached by + /// hash (O(1) lookups). The previous implementation folded the frontier + /// from genesis on every call — O(N) cryptographic hashing per call, O(N²) + /// across the window walk, which dominated these tests' runtime. async fn get_commitment_tree_roots( &self, id: BlockHash, - ) -> BlockchainSourceResult<( - Option<(zebra_chain::sapling::tree::Root, u64)>, - Option<(zebra_chain::orchard::tree::Root, u64)>, - )> { - if let Some(delay) = self.delay { - tokio::time::sleep(delay).await; - } - let Some(chain_up_to_block) = - self.get_block_and_all_preceeding(|block| block.hash().0 == id.0) - else { - return Ok((None, None)); - }; - - let (sapling, orchard) = - chain_up_to_block - .iter() - .fold((None, None), |(mut sapling, mut orchard), block| { - for transaction in &block.transactions { - for sap_commitment in transaction.sapling_note_commitments() { - let sap_commitment = - sapling_crypto::Node::from_bytes(sap_commitment.to_bytes()) - .unwrap(); - - sapling = Some(sapling.unwrap_or_else(|| { - incrementalmerkletree::frontier::Frontier::<_, 32>::empty() - })); - - sapling = sapling.map(|mut tree| { - tree.append(sap_commitment); - tree - }); - } - for orc_commitment in transaction.orchard_note_commitments() { - let orc_commitment = - zebra_chain::orchard::tree::Node::from(*orc_commitment); - - orchard = Some(orchard.unwrap_or_else(|| { - incrementalmerkletree::frontier::Frontier::<_, 32>::empty() - })); - - orchard = orchard.map(|mut tree| { - tree.append(orc_commitment); - tree - }); - } - } - (sapling, orchard) - }); - Ok(( - sapling.map(|sap_front| { - ( - zebra_chain::sapling::tree::Root::from_bytes(sap_front.root().to_bytes()), - sap_front.tree_size(), - ) - }), - orchard.map(|orc_front| { - ( - zebra_chain::orchard::tree::Root::from_bytes(orc_front.root().as_bytes()), - orc_front.tree_size(), - ) - }), - )) + ) -> BlockchainSourceResult { + let roots = self.commitment_roots_cache.get_or_init(|| { + let mut map = std::collections::HashMap::new(); + // Genesis segment, then each branch continuing from the genesis-end + // frontier — one incremental fold over the whole tree. + let (sapling, orchard) = + fold_commitment_roots(self.genesis_segment.iter(), None, None, &mut map); + for branch in self.branching_segments.iter() { + fold_commitment_roots(branch.iter(), sapling.clone(), orchard.clone(), &mut map); + } + map + }); + Ok(roots.get(&id).cloned().unwrap_or((None, None))) } /// Returns the sapling and orchard treestate by hash @@ -697,9 +703,6 @@ impl BlockchainSource for ProptestMockchain { async fn get_mempool_txids( &self, ) -> BlockchainSourceResult>> { - if let Some(delay) = self.delay { - tokio::time::sleep(delay).await; - } Ok(Some(Vec::new())) } @@ -713,9 +716,6 @@ impl BlockchainSource for ProptestMockchain { GetTransactionLocation, )>, > { - if let Some(delay) = self.delay { - tokio::time::sleep(delay).await; - } Ok(self.tx_index().get(&txid.into()).cloned()) } @@ -723,9 +723,6 @@ impl BlockchainSource for ProptestMockchain { async fn get_best_block_hash( &self, ) -> BlockchainSourceResult> { - if let Some(delay) = self.delay { - tokio::time::sleep(delay).await; - } Ok(Some(self.best_branch().last().unwrap().hash())) } @@ -733,9 +730,6 @@ impl BlockchainSource for ProptestMockchain { async fn get_best_block_height( &self, ) -> BlockchainSourceResult> { - if let Some(delay) = self.delay { - tokio::time::sleep(delay).await; - } Ok(Some( self.best_branch() .last() @@ -745,6 +739,16 @@ impl BlockchainSource for ProptestMockchain { )) } + fn finalized_sync_cap(&self) -> Option { + match self + .finalized_sync_cap + .load(std::sync::atomic::Ordering::Relaxed) + { + u32::MAX => None, + cap => Some(crate::Height(cap)), + } + } + /// Get a listener for new nonfinalized blocks, /// if supported async fn nonfinalized_listener( diff --git a/packages/zaino-state/src/chain_index/tests/sync_loop.rs b/packages/zaino-state/src/chain_index/tests/sync_loop.rs index 7037f7e8e..5cfa5b2a7 100644 --- a/packages/zaino-state/src/chain_index/tests/sync_loop.rs +++ b/packages/zaino-state/src/chain_index/tests/sync_loop.rs @@ -120,7 +120,7 @@ async fn tip_converges_after_burst_mine() { .snapshot_nonfinalized_state() .await .ok()? - .get_nfs_snapshot()? + .resolved_nfs_snapshot()? .best_tip .height .0; @@ -133,7 +133,7 @@ async fn tip_converges_after_burst_mine() { .snapshot_nonfinalized_state() .await .unwrap() - .get_nfs_snapshot() + .resolved_nfs_snapshot() .unwrap() .best_tip .height diff --git a/packages/zaino-state/src/chain_index/tests/types.rs b/packages/zaino-state/src/chain_index/tests/types.rs index 2da3fab70..facd376ef 100644 --- a/packages/zaino-state/src/chain_index/tests/types.rs +++ b/packages/zaino-state/src/chain_index/tests/types.rs @@ -26,7 +26,7 @@ pub(crate) fn canonical_blockheaderdata() -> BlockHeaderData { let solution = EquihashSolution::Standard([6u8; 1344]); let bits = CompactDifficulty::try_from_bits(TEST_VALID_NBITS).expect("valid nBits"); - let bctx = BlockContext::new(hash, parent_hash, chainwork, height); + let bctx = BlockContext::new(hash, parent_hash, Some(chainwork), height); let bdata = BlockData::new(1, 2, [3u8; 32], [4u8; 32], bits, [5u8; 32], solution); BlockHeaderData::new(bctx, bdata) } diff --git a/packages/zaino-state/src/chain_index/tests/vectors.rs b/packages/zaino-state/src/chain_index/tests/vectors.rs index b0b5115fc..6d44123d1 100644 --- a/packages/zaino-state/src/chain_index/tests/vectors.rs +++ b/packages/zaino-state/src/chain_index/tests/vectors.rs @@ -118,7 +118,7 @@ pub(crate) fn indexed_block_chain( ); let chain_block = IndexedBlock::try_from(BlockWithMetadata::new(&vector.zebra_block, metadata)).unwrap(); - parent_chain_work = Some(chain_block.context.chainwork); + parent_chain_work = chain_block.context.chainwork; chain_block }) } diff --git a/packages/zaino-state/src/chain_index/types/block_context.rs b/packages/zaino-state/src/chain_index/types/block_context.rs index bb43a15d5..55b40486d 100644 --- a/packages/zaino-state/src/chain_index/types/block_context.rs +++ b/packages/zaino-state/src/chain_index/types/block_context.rs @@ -17,7 +17,7 @@ pub struct BlockContext { pub parent_hash: BlockHash, /// The cumulative proof-of-work of the blockchain up to this block, /// used for chain selection. - pub chainwork: ChainWork, + pub chainwork: Option, } impl BlockContext { @@ -26,7 +26,7 @@ impl BlockContext { pub fn new( hash: BlockHash, parent_hash: BlockHash, - chainwork: ChainWork, + chainwork: Option, height: Height, ) -> Self { Self { @@ -47,7 +47,7 @@ impl BlockContext { } /// Returns the cumulative chainwork up to this block. - pub fn chainwork(&self) -> &ChainWork { + pub fn chainwork(&self) -> &Option { &self.chainwork } diff --git a/packages/zaino-state/src/chain_index/types/db/block.rs b/packages/zaino-state/src/chain_index/types/db/block.rs index 186abcebc..f90c34ce0 100644 --- a/packages/zaino-state/src/chain_index/types/db/block.rs +++ b/packages/zaino-state/src/chain_index/types/db/block.rs @@ -41,14 +41,16 @@ use crate::chain_index::{ pub(super) struct PersistentChainWork([u8; 32]); impl PersistentChainWork { - pub(super) fn from_business(cw: &ChainWork) -> Self { + pub(super) fn from_business(cw: &Option) -> Self { let mut buf = [0u8; 32]; - // Big-endian: the value occupies the low-order (last) 16 bytes. - buf[16..].copy_from_slice(&cw.as_non_zero_u128().get().to_be_bytes()); + if let Some(work) = cw { + // Big-endian: the value occupies the low-order (last) 16 bytes. + buf[16..].copy_from_slice(&work.as_non_zero_u128().get().to_be_bytes()); + } Self(buf) } - pub(super) fn into_business(self) -> io::Result { + pub(super) fn into_business(self) -> io::Result> { // Big-endian: the high-order 16 bytes must be zero for the value to fit u128. if self.0[..16] != [0u8; 16] { return Err(io::Error::new( @@ -59,9 +61,7 @@ impl PersistentChainWork { let mut be_bytes = [0u8; 16]; be_bytes.copy_from_slice(&self.0[16..]); let value = u128::from_be_bytes(be_bytes); - NonZeroU128::new(value) - .map(ChainWork::new) - .ok_or_else(|| io::Error::new(io::ErrorKind::InvalidData, "chainwork is zero")) + Ok(NonZeroU128::new(value).map(ChainWork::new)) } } @@ -258,7 +258,9 @@ mod tests { let bctx = BlockContext::new( BlockHash::from([0x11; 32]), BlockHash::from([0x22; 32]), - ChainWork::new(NonZeroU128::new(0x0123_4567u128).expect("nonzero")), + Some(ChainWork::new( + NonZeroU128::new(0x0123_4567u128).expect("nonzero"), + )), Height(0x0dec_0de0), ); let persisted = PersistentBlockContext::from_business(&bctx); @@ -281,7 +283,7 @@ mod tests { let cw = PersistentChainWork(on_disk) .into_business() .expect("big-endian on-disk chainwork must decode"); - assert_eq!(cw.as_non_zero_u128().get(), 17); + assert_eq!(cw.unwrap().as_non_zero_u128().get(), 17); } /// Verbatim recovery of the pre-#1313 on-disk `ChainWork` encoder — the @@ -334,7 +336,7 @@ mod tests { // Encode: the current encoder reproduces the original's big-endian bytes. assert_eq!( - PersistentChainWork::from_business(&cw).0, + PersistentChainWork::from_business(&Some(cw)).0, original_bytes, "encode mismatch at {value:#034x}", ); @@ -345,7 +347,7 @@ mod tests { PersistentChainWork(original_bytes) .into_business() .expect("original on-disk bytes must decode"), - cw, + Some(cw), "decode mismatch at {value:#034x}", ); assert_eq!(original.to_u256(), primitive_types::U256::from(value)); diff --git a/packages/zaino-state/src/chain_index/types/db/legacy.rs b/packages/zaino-state/src/chain_index/types/db/legacy.rs index ee4074d5e..3b401f196 100644 --- a/packages/zaino-state/src/chain_index/types/db/legacy.rs +++ b/packages/zaino-state/src/chain_index/types/db/legacy.rs @@ -40,7 +40,11 @@ use crate::chain_index::encoding::{ read_vec, version, write_fixed_le, write_i64_le, write_option, write_u16_be, write_u32_be, write_u32_le, write_u64_le, write_vec, FixedEncodedLen, ZainoVersionedSerde, }; +use crate::chain_index::non_finalised_state::{ + NonfinalizedBlockCacheSnapshot, ProvisionalCumulativeWork, +}; use crate::chain_index::types::{BlockContext, ChainWork, CompactDifficulty}; +use crate::chain_index::NonFinalizedSnapshot as _; use super::commitment::{CommitmentTreeData, CommitmentTreeRoots, CommitmentTreeSizes}; @@ -718,8 +722,6 @@ impl FixedEncodedLen for Outpoint { const ENCODED_LEN: usize = 32 + 4; } -// *** Block Level Objects *** - /// Essential block header fields required for chain validation and serving block header data. /// /// NOTE: Optional fields may be added for: @@ -1040,7 +1042,7 @@ impl IndexedBlock { } /// Returns the cumulative chainwork. - pub fn chainwork(&self) -> &ChainWork { + pub fn chainwork(&self) -> &Option { self.context.chainwork() } @@ -1049,6 +1051,23 @@ impl IndexedBlock { self.data.bits.to_work() } + /// Cumulative work relative to the seam. Use this — never an absolute + /// chainwork — for best-tip selection within the non-finalized window. + pub(crate) fn provisional_cumulative_work( + &self, + snapshot: &NonfinalizedBlockCacheSnapshot, + ) -> ProvisionalCumulativeWork { + let mut work = ProvisionalCumulativeWork::new(&ChainWork::from(self.data.bits.to_work())); + let mut parent_hash = self.context.parent_hash; + while let Some(prev_block) = snapshot.get_chainblock_by_hash(&parent_hash) { + work = work + .add_block_work(&ChainWork::from(prev_block.data.bits.to_work())) + .expect("chainwork integer overflow"); + parent_hash = prev_block.context.parent_hash; + } + work + } + /// Converts this `IndexedBlock` into a CompactBlock protobuf message using proto v4 format. /// /// NOTE: This method currently includes transparent tx data in the compact block produced, @@ -1058,33 +1077,53 @@ impl IndexedBlock { /// with tx data being added selectively here. pub fn to_compact_block(&self) -> zaino_proto::proto::compact_formats::CompactBlock { // NOTE: Returns u64::MAX if the block is not in the best chain. - let height: u64 = self.height().0.into(); - - let hash = self.hash().0.to_vec(); - let prev_hash = self.context.parent_hash.0.to_vec(); - - let vtx: Vec = self - .transactions() - .iter() - .map(|tx| tx.to_compact_tx(None)) - .collect(); - - let sapling_commitment_tree_size = self.commitment_tree_data().sizes().sapling(); - let orchard_commitment_tree_size = self.commitment_tree_data().sizes().orchard(); - - zaino_proto::proto::compact_formats::CompactBlock { - proto_version: 0, - height, - hash, - prev_hash, - time: self.data().time() as u32, - header: vec![], - vtx, - chain_metadata: Some(zaino_proto::proto::compact_formats::ChainMetadata { - sapling_commitment_tree_size, - orchard_commitment_tree_size, - }), - } + compact_block_from_parts( + self.height(), + self.hash(), + &self.context.parent_hash, + self.data().time(), + self.transactions(), + self.commitment_tree_data(), + ) + } +} + +/// Build a proto-v4 `CompactBlock` from the parts shared by [`IndexedBlock`] +/// and the non-finalized `ProvisionalBlock`. Compact-block construction needs +/// no cumulative chainwork, so both block types produce identical output via +/// this one assembly — no duplication. +pub(crate) fn compact_block_from_parts( + height: Height, + hash: &BlockHash, + parent_hash: &BlockHash, + time: i64, + transactions: &[CompactTxData], + commitment_tree_data: &CommitmentTreeData, +) -> zaino_proto::proto::compact_formats::CompactBlock { + let height: u64 = height.0.into(); + let hash = hash.0.to_vec(); + let prev_hash = parent_hash.0.to_vec(); + + let vtx: Vec = transactions + .iter() + .map(|tx| tx.to_compact_tx(None)) + .collect(); + + let sapling_commitment_tree_size = commitment_tree_data.sizes().sapling(); + let orchard_commitment_tree_size = commitment_tree_data.sizes().orchard(); + + zaino_proto::proto::compact_formats::CompactBlock { + proto_version: 0, + height, + hash, + prev_hash, + time: time as u32, + header: vec![], + vtx, + chain_metadata: Some(zaino_proto::proto::compact_formats::ChainMetadata { + sapling_commitment_tree_size, + orchard_commitment_tree_size, + }), } } @@ -1260,7 +1299,7 @@ impl let context = BlockContext::new( BlockHash::from(hash), BlockHash::from(parent_hash), - chainwork, + Some(chainwork), height, ); diff --git a/packages/zaino-state/src/chain_index/types/helpers.rs b/packages/zaino-state/src/chain_index/types/helpers.rs index d4e63db94..3ce27175b 100644 --- a/packages/zaino-state/src/chain_index/types/helpers.rs +++ b/packages/zaino-state/src/chain_index/types/helpers.rs @@ -140,7 +140,7 @@ impl<'a> BlockWithMetadata<'a> { } /// Extract block header data - fn extract_block_data(&self) -> Result { + pub(crate) fn extract_block_data(&self) -> Result { let block = self.block; let network = &self.metadata.network; @@ -165,7 +165,7 @@ impl<'a> BlockWithMetadata<'a> { } /// Extract and process all transactions in the block - fn extract_transactions(&self) -> Result, String> { + pub(crate) fn extract_transactions(&self) -> Result, String> { let mut transactions = Vec::new(); for (i, txn) in self.block.transactions.iter().enumerate() { @@ -305,11 +305,16 @@ impl<'a> BlockWithMetadata<'a> { None => block_work, }; - Ok(BlockContext::new(hash, parent_hash, chainwork, height)) + Ok(BlockContext::new( + hash, + parent_hash, + Some(chainwork), + height, + )) } /// Create commitment tree data from metadata - fn create_commitment_tree_data(&self) -> super::db::CommitmentTreeData { + pub(crate) fn create_commitment_tree_data(&self) -> super::db::CommitmentTreeData { let commitment_tree_roots = super::db::CommitmentTreeRoots::new( <[u8; 32]>::from(self.metadata.sapling_root), <[u8; 32]>::from(self.metadata.orchard_root),