Skip to main content

zebra_state/service/
non_finalized_state.rs

1//! Non-finalized chain state management as defined by [RFC0005]
2//!
3//! [RFC0005]: https://zebra.zfnd.org/dev/rfcs/0005-state-updates.html
4
5use std::{
6    collections::{BTreeSet, HashMap},
7    mem,
8    path::{Path, PathBuf},
9    sync::Arc,
10};
11
12use indexmap::IndexMap;
13use tokio::sync::watch;
14use zebra_chain::{
15    block::{self, Block, Hash, Height},
16    parameters::Network,
17    sprout::{self},
18    transparent,
19};
20
21use crate::{
22    constants::{MAX_INVALIDATED_BLOCKS, MAX_NON_FINALIZED_CHAIN_FORKS},
23    error::ReconsiderError,
24    request::{ContextuallyVerifiedBlock, FinalizableBlock},
25    service::{
26        check,
27        finalized_state::{calculate_deferred_pool_balance_change, ZebraDb},
28        InvalidateError,
29    },
30    SemanticallyVerifiedBlock, ValidateContextError, WatchReceiver,
31};
32
33mod backup;
34mod chain;
35
36#[cfg(test)]
37pub(crate) use backup::MIN_DURATION_BETWEEN_BACKUP_UPDATES;
38
39#[cfg(test)]
40mod tests;
41
42pub(crate) use chain::{Chain, SpendingTransactionId};
43
44/// The state of the chains in memory, including queued blocks.
45///
46/// Clones of the non-finalized state contain independent copies of the chains.
47/// This is different from `FinalizedState::clone()`,
48/// which returns a shared reference to the database.
49///
50/// Most chain data is clone-on-write using [`Arc`].
51pub struct NonFinalizedState {
52    // Chain Data
53    //
54    /// Verified, non-finalized chains, in ascending work order.
55    ///
56    /// The best chain is [`NonFinalizedState::best_chain()`], or `chain_iter().next()`.
57    /// Using `chain_set.last()` or `chain_set.iter().next_back()` is deprecated,
58    /// callers should migrate to `chain_iter().next()`.
59    chain_set: BTreeSet<Arc<Chain>>,
60
61    /// Blocks that have been invalidated in, and removed from, the non finalized
62    /// state.
63    invalidated_blocks: IndexMap<Height, Arc<Vec<ContextuallyVerifiedBlock>>>,
64
65    // Configuration
66    //
67    /// The configured Zcash network.
68    pub network: Network,
69
70    // Diagnostics
71    //
72    /// Configures the non-finalized state to count metrics.
73    ///
74    /// Used for skipping metrics and progress bars when testing block proposals
75    /// with a commit to a cloned non-finalized state.
76    //
77    // TODO: make this field private and set it via an argument to NonFinalizedState::new()
78    should_count_metrics: bool,
79
80    /// Number of chain forks transmitter.
81    #[cfg(feature = "progress-bar")]
82    chain_count_bar: Option<howudoin::Tx>,
83
84    /// A chain fork length transmitter for each [`Chain`] in [`chain_set`](Self.chain_set).
85    ///
86    /// Because `chain_set` contains `Arc<Chain>`s, it is difficult to update the metrics state
87    /// on each chain. ([`Arc`]s are read-only, and we don't want to clone them just for metrics.)
88    #[cfg(feature = "progress-bar")]
89    chain_fork_length_bars: Vec<howudoin::Tx>,
90}
91
92impl std::fmt::Debug for NonFinalizedState {
93    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
94        let mut f = f.debug_struct("NonFinalizedState");
95
96        f.field("chain_set", &self.chain_set)
97            .field("network", &self.network);
98
99        f.field("should_count_metrics", &self.should_count_metrics);
100
101        f.finish()
102    }
103}
104
105impl Clone for NonFinalizedState {
106    fn clone(&self) -> Self {
107        Self {
108            chain_set: self.chain_set.clone(),
109            network: self.network.clone(),
110            invalidated_blocks: self.invalidated_blocks.clone(),
111            should_count_metrics: self.should_count_metrics,
112            // Don't track progress in clones.
113            #[cfg(feature = "progress-bar")]
114            chain_count_bar: None,
115            #[cfg(feature = "progress-bar")]
116            chain_fork_length_bars: Vec::new(),
117        }
118    }
119}
120
121impl NonFinalizedState {
122    /// Returns a new non-finalized state for `network`.
123    pub fn new(network: &Network) -> NonFinalizedState {
124        NonFinalizedState {
125            chain_set: Default::default(),
126            network: network.clone(),
127            invalidated_blocks: Default::default(),
128            should_count_metrics: true,
129            #[cfg(feature = "progress-bar")]
130            chain_count_bar: None,
131            #[cfg(feature = "progress-bar")]
132            chain_fork_length_bars: Vec::new(),
133        }
134    }
135
136    /// Writes the current non-finalized state to the backup directory at `backup_dir_path`.
137    ///
138    /// Reads the existing backup directory contents, writes any blocks that are in the
139    /// non-finalized state but missing from the backup, and deletes any backup files that
140    /// are no longer present in the non-finalized state.
141    ///
142    /// This method performs blocking I/O and should only be called from a blocking context.
143    pub(crate) fn write_to_backup(&self, backup_dir_path: &Path) {
144        let backup_blocks: HashMap<block::Hash, PathBuf> =
145            backup::list_backup_dir_entries(backup_dir_path).collect();
146        backup::update_non_finalized_state_backup(backup_dir_path, self, backup_blocks);
147    }
148
149    /// Accepts an optional path to the non-finalized state backup directory and a handle to the database.
150    ///
151    /// If a backup directory path is provided:
152    /// - Creates a new backup directory at the provided path if none exists,
153    /// - Restores non-finalized blocks from the backup directory, if any, and
154    /// - Unless `skip_backup_task` is true, spawns a task that updates the non-finalized
155    ///   backup cache with the latest non-finalized state sent to the returned watch channel.
156    ///
157    /// Returns the non-finalized state with a watch channel sender and receiver.
158    pub async fn with_backup(
159        self,
160        backup_dir_path: Option<PathBuf>,
161        finalized_state: &ZebraDb,
162        should_restore_backup: bool,
163        skip_backup_task: bool,
164    ) -> (
165        Self,
166        watch::Sender<NonFinalizedState>,
167        WatchReceiver<NonFinalizedState>,
168    ) {
169        let with_watch_channel = |non_finalized_state: NonFinalizedState| {
170            let (sender, receiver) = watch::channel(non_finalized_state.clone());
171            (non_finalized_state, sender, WatchReceiver::new(receiver))
172        };
173
174        let Some(backup_dir_path) = backup_dir_path else {
175            return with_watch_channel(self);
176        };
177
178        if !should_restore_backup {
179            tracing::info!(
180                ?backup_dir_path,
181                spawning_backup_task = !skip_backup_task,
182                "not restoring non-finalized blocks from backup, any backed up blocks that are \
183                 missing from the non-finalized state will be deleted"
184            );
185        } else if skip_backup_task {
186            tracing::info!(
187                ?backup_dir_path,
188                "restoring non-finalized blocks from backup (sync write mode, backup task skipped)"
189            );
190        } else {
191            tracing::info!(
192                ?backup_dir_path,
193                "restoring non-finalized blocks from backup and spawning backup task"
194            );
195        }
196
197        let non_finalized_state = {
198            let backup_dir_path = backup_dir_path.clone();
199            let finalized_state = finalized_state.clone();
200            tokio::task::spawn_blocking(move || {
201                // Create a new backup directory if none exists
202                std::fs::create_dir_all(&backup_dir_path)
203                    .expect("failed to create non-finalized state backup directory");
204
205                if should_restore_backup {
206                    backup::restore_backup(self, &backup_dir_path, &finalized_state)
207                } else {
208                    self
209                }
210            })
211            .await
212            .expect("failed to join blocking task")
213        };
214
215        let (non_finalized_state, sender, receiver) = with_watch_channel(non_finalized_state);
216
217        if !skip_backup_task {
218            tokio::spawn(backup::run_backup_task(receiver.clone(), backup_dir_path));
219        }
220
221        if !non_finalized_state.is_chain_set_empty() {
222            let num_blocks_restored = non_finalized_state
223                .best_chain()
224                .expect("must have best chain if chain set is not empty")
225                .len();
226
227            tracing::info!(
228                ?num_blocks_restored,
229                "restored blocks from non-finalized backup cache"
230            );
231        } else if should_restore_backup {
232            tracing::info!("no blocks were restored from the non-finalized backup cache");
233        }
234
235        (non_finalized_state, sender, receiver)
236    }
237
238    /// Is the internal state of `self` the same as `other`?
239    ///
240    /// [`Chain`] has a custom [`Eq`] implementation based on proof of work,
241    /// which is used to select the best chain. So we can't derive [`Eq`] for [`NonFinalizedState`].
242    ///
243    /// Unlike the custom trait impl, this method returns `true` if the entire internal state
244    /// of two non-finalized states is equal.
245    ///
246    /// If the internal states are different, it returns `false`,
247    /// even if the chains and blocks are equal.
248    #[cfg(any(test, feature = "proptest-impl"))]
249    #[allow(dead_code)]
250    pub fn eq_internal_state(&self, other: &NonFinalizedState) -> bool {
251        // this method must be updated every time a consensus-critical field is added to NonFinalizedState
252        // (diagnostic fields can be ignored)
253
254        self.chain_set.len() == other.chain_set.len()
255            && self
256                .chain_set
257                .iter()
258                .zip(other.chain_set.iter())
259                .all(|(self_chain, other_chain)| self_chain.eq_internal_state(other_chain))
260            && self.network == other.network
261    }
262
263    /// Returns an iterator over the non-finalized chains, with the best chain first.
264    //
265    // TODO: replace chain_set.iter().rev() with this method
266    pub fn chain_iter(&self) -> impl Iterator<Item = &Arc<Chain>> {
267        self.chain_set.iter().rev()
268    }
269
270    /// Insert `chain` into `self.chain_set`, apply `chain_filter` to the chains,
271    /// then limit the number of tracked chains.
272    fn insert_with<F>(&mut self, chain: Arc<Chain>, chain_filter: F)
273    where
274        F: FnOnce(&mut BTreeSet<Arc<Chain>>),
275    {
276        self.chain_set.insert(chain);
277
278        chain_filter(&mut self.chain_set);
279
280        while self.chain_set.len() > MAX_NON_FINALIZED_CHAIN_FORKS {
281            // The first chain is the chain with the lowest work.
282            self.chain_set.pop_first();
283        }
284
285        self.update_metrics_bars();
286    }
287
288    /// Insert `chain` into `self.chain_set`, then limit the number of tracked chains.
289    fn insert(&mut self, chain: Arc<Chain>) {
290        self.insert_with(chain, |_ignored_chain| { /* no filter */ })
291    }
292
293    /// Finalize the lowest height block in the non-finalized portion of the best
294    /// chain and update all side-chains to match.
295    pub fn finalize(&mut self) -> FinalizableBlock {
296        // Chain::cmp uses the partial cumulative work, and the hash of the tip block.
297        // Neither of these fields has interior mutability.
298        // (And when the tip block is dropped for a chain, the chain is also dropped.)
299        #[allow(clippy::mutable_key_type)]
300        let chains = mem::take(&mut self.chain_set);
301        let mut chains = chains.into_iter();
302
303        // extract best chain
304        let mut best_chain = chains.next_back().expect("there's at least one chain");
305
306        // clone if required
307        let mut_best_chain = Arc::make_mut(&mut best_chain);
308
309        // extract the rest into side_chains so they can be mutated
310        let side_chains = chains;
311
312        // Pop the lowest height block from the best chain to be finalized, and
313        // also obtain its associated treestate.
314        let (best_chain_root, root_treestate) = mut_best_chain.pop_root();
315
316        // add best_chain back to `self.chain_set`
317        if !best_chain.is_empty() {
318            self.insert(best_chain);
319        }
320
321        // for each remaining chain in side_chains
322        for mut side_chain in side_chains.rev() {
323            if side_chain.non_finalized_root_hash() != best_chain_root.hash {
324                // If we popped the root, the chain would be empty or orphaned,
325                // so just drop it now.
326                drop(side_chain);
327
328                continue;
329            }
330
331            // otherwise, the popped root block is the same as the finalizing block
332
333            // clone if required
334            let mut_side_chain = Arc::make_mut(&mut side_chain);
335
336            // remove the first block from `chain`
337            let (side_chain_root, _treestate) = mut_side_chain.pop_root();
338            assert_eq!(side_chain_root.hash, best_chain_root.hash);
339
340            // add the chain back to `self.chain_set`
341            if !side_chain.is_empty() {
342                self.insert(side_chain);
343            }
344        }
345
346        // Remove all invalidated_blocks at or below the finalized height
347        self.invalidated_blocks
348            .retain(|height, _blocks| *height >= best_chain_root.height);
349
350        self.update_metrics_for_chains();
351
352        // Add the treestate to the finalized block.
353        FinalizableBlock::new(best_chain_root, root_treestate)
354    }
355
356    /// Commit block to the non-finalized state, on top of:
357    /// - an existing chain's tip, or
358    /// - a newly forked chain.
359    #[tracing::instrument(level = "debug", skip(self, finalized_state, prepared))]
360    pub fn commit_block(
361        &mut self,
362        prepared: SemanticallyVerifiedBlock,
363        finalized_state: &ZebraDb,
364    ) -> Result<(), ValidateContextError> {
365        let parent_hash = prepared.block.header.previous_block_hash;
366        let (height, hash) = (prepared.height, prepared.hash);
367
368        let parent_chain = self.parent_chain(parent_hash)?;
369
370        // If the block is invalid, return the error,
371        // and drop the cloned parent Arc, or newly created chain fork.
372        let modified_chain = self.validate_and_commit(parent_chain, prepared, finalized_state)?;
373
374        // If the block is valid:
375        // - add the new chain fork or updated chain to the set of recent chains
376        // - remove the parent chain, if it was in the chain set
377        //   (if it was a newly created fork, it won't be in the chain set)
378        self.insert_with(modified_chain, |chain_set| {
379            chain_set.retain(|chain| chain.non_finalized_tip_hash() != parent_hash)
380        });
381
382        self.update_metrics_for_committed_block(height, hash);
383
384        Ok(())
385    }
386
387    /// Invalidate block with hash `block_hash` and all descendants from the non-finalized state. Insert
388    /// the new chain into the chain_set and discard the previous.
389    #[allow(clippy::unwrap_in_result)]
390    pub fn invalidate_block(&mut self, block_hash: Hash) -> Result<block::Hash, InvalidateError> {
391        let Some(chain) = self.find_chain(|chain| chain.contains_block_hash(block_hash)) else {
392            return Err(InvalidateError::BlockNotFound(block_hash));
393        };
394
395        let invalidated_blocks = if chain.non_finalized_root_hash() == block_hash {
396            self.chain_set.remove(&chain);
397            chain.blocks.values().cloned().collect()
398        } else {
399            let (new_chain, invalidated_blocks) = chain
400                .invalidate_block(block_hash)
401                .expect("already checked that chain contains hash");
402
403            // Add the new chain fork or updated chain to the set of recent chains, and
404            // remove the chain containing the hash of the block from chain set
405            self.insert_with(Arc::new(new_chain.clone()), |chain_set| {
406                chain_set.retain(|c| !c.contains_block_hash(block_hash))
407            });
408
409            invalidated_blocks
410        };
411
412        // TODO: Allow for invalidating multiple block hashes at a given height (#9552).
413        self.invalidated_blocks.insert(
414            invalidated_blocks
415                .first()
416                .expect("should not be empty")
417                .clone()
418                .height,
419            Arc::new(invalidated_blocks),
420        );
421
422        while self.invalidated_blocks.len() > MAX_INVALIDATED_BLOCKS {
423            self.invalidated_blocks.shift_remove_index(0);
424        }
425
426        self.update_metrics_for_chains();
427        self.update_metrics_bars();
428
429        Ok(block_hash)
430    }
431
432    /// Reconsiders a previously invalidated block and its descendants into the non-finalized state
433    /// based on a block_hash. Reconsidered blocks are inserted into the previous chain and re-inserted
434    /// into the chain_set.
435    #[allow(clippy::unwrap_in_result)]
436    pub fn reconsider_block(
437        &mut self,
438        block_hash: block::Hash,
439        finalized_state: &ZebraDb,
440    ) -> Result<Vec<block::Hash>, ReconsiderError> {
441        // Locate the record but keep it live until replay succeeds, so a
442        // recoverable error can't lose it; it is `shift_remove`d atomically with
443        // the insert below.
444        let (height, invalidated_blocks) = self
445            .invalidated_blocks
446            .iter()
447            .find_map(|(height, blocks)| {
448                if blocks.first()?.hash == block_hash {
449                    Some((*height, (**blocks).clone()))
450                } else {
451                    None
452                }
453            })
454            .ok_or(ReconsiderError::MissingInvalidatedBlock(block_hash))?;
455
456        let invalidated_block_hashes = invalidated_blocks
457            .iter()
458            .map(|block| block.hash)
459            .collect::<Vec<_>>();
460
461        // Find and fork the parent chain of the invalidated_root. Update the parent chain
462        // with the invalidated_descendants
463        let invalidated_root = invalidated_blocks
464            .first()
465            .ok_or(ReconsiderError::InvalidatedBlocksEmpty)?;
466
467        let root_parent_hash = invalidated_root.block.header.previous_block_hash;
468
469        // If the parent is the tip of the finalized_state we create a new chain and insert it
470        // into the non finalized state
471        let chain_result = if root_parent_hash == finalized_state.finalized_tip_hash() {
472            let chain = Chain::new(
473                &self.network,
474                finalized_state
475                    .finalized_tip_height()
476                    .ok_or(ReconsiderError::ParentChainNotFound(block_hash))?,
477                finalized_state.note_commitment_trees_for_tip(),
478                finalized_state.history_tree(),
479                finalized_state.finalized_value_pool(),
480            );
481            Arc::new(chain)
482        } else {
483            // The parent is not the finalized_tip and still exist in the NonFinalizedState
484            // or else we return an error due to the parent not existing in the NonFinalizedState
485            self.parent_chain(root_parent_hash)
486                .map_err(|_| ReconsiderError::ParentChainNotFound(block_hash))?
487        };
488
489        let mut modified_chain = Arc::unwrap_or_clone(chain_result);
490        for block in invalidated_blocks {
491            modified_chain = modified_chain
492                .push(block)
493                .map_err(ReconsiderError::ReplayFailed)?;
494        }
495
496        let (tip_height, tip_hash) = modified_chain.non_finalized_tip();
497
498        // All fallible steps have succeeded; remove the invalidation record
499        // atomically with installing the restored chain so a failed attempt
500        // does not destroy the record.
501        self.invalidated_blocks.shift_remove(&height);
502
503        // Only track invalidated_blocks that are not yet finalized. Once blocks are finalized (below the best_chain_root_height)
504        // we can discard the block.
505        if let Some(best_chain_root_height) = finalized_state.finalized_tip_height() {
506            self.invalidated_blocks
507                .retain(|height, _blocks| *height >= best_chain_root_height);
508        }
509
510        self.insert_with(Arc::new(modified_chain), |chain_set| {
511            chain_set.retain(|chain| chain.non_finalized_tip_hash() != root_parent_hash)
512        });
513
514        self.update_metrics_for_committed_block(tip_height, tip_hash);
515
516        Ok(invalidated_block_hashes)
517    }
518
519    /// Commit block to the non-finalized state as a new chain where its parent
520    /// is the finalized tip.
521    #[tracing::instrument(level = "debug", skip(self, finalized_state, prepared))]
522    #[allow(clippy::unwrap_in_result)]
523    pub fn commit_new_chain(
524        &mut self,
525        prepared: SemanticallyVerifiedBlock,
526        finalized_state: &ZebraDb,
527    ) -> Result<(), ValidateContextError> {
528        let finalized_tip_height = finalized_state.finalized_tip_height();
529
530        // TODO: fix tests that don't initialize the finalized state
531        #[cfg(not(test))]
532        let finalized_tip_height = finalized_tip_height.expect("finalized state contains blocks");
533        #[cfg(test)]
534        let finalized_tip_height = finalized_tip_height.unwrap_or(zebra_chain::block::Height(0));
535
536        let chain = Chain::new(
537            &self.network,
538            finalized_tip_height,
539            finalized_state.note_commitment_trees_for_tip(),
540            finalized_state.history_tree(),
541            finalized_state.finalized_value_pool(),
542        );
543
544        let (height, hash) = (prepared.height, prepared.hash);
545
546        // If the block is invalid, return the error, and drop the newly created chain fork
547        let chain = self.validate_and_commit(Arc::new(chain), prepared, finalized_state)?;
548
549        // If the block is valid, add the new chain fork to the set of recent chains.
550        self.insert(chain);
551        self.update_metrics_for_committed_block(height, hash);
552
553        Ok(())
554    }
555
556    /// Contextually validate `prepared` using `finalized_state`.
557    /// If validation succeeds, push `prepared` onto `new_chain`.
558    ///
559    /// `new_chain` should start as a clone of the parent chain fork,
560    /// or the finalized tip.
561    #[tracing::instrument(level = "debug", skip(self, finalized_state, new_chain))]
562    fn validate_and_commit(
563        &self,
564        new_chain: Arc<Chain>,
565        prepared: SemanticallyVerifiedBlock,
566        finalized_state: &ZebraDb,
567    ) -> Result<Arc<Chain>, ValidateContextError> {
568        if self
569            .invalidated_blocks
570            .values()
571            .any(|blocks| blocks.iter().any(|block| block.hash == prepared.hash))
572        {
573            return Err(ValidateContextError::BlockPreviouslyInvalidated {
574                block_hash: prepared.hash,
575            });
576        }
577
578        // Reads from disk
579        //
580        // TODO: if these disk reads show up in profiles, run them in parallel, using std::thread::spawn()
581        let spent_utxos = check::utxo::transparent_spend(
582            &prepared,
583            &new_chain.unspent_utxos(),
584            &new_chain.spent_utxos,
585            finalized_state,
586        )?;
587
588        // Reads from disk
589        check::anchors::block_sapling_orchard_anchors_refer_to_final_treestates(
590            finalized_state,
591            &new_chain,
592            &prepared,
593        )?;
594
595        // Reads from disk
596        let sprout_final_treestates = check::anchors::block_fetch_sprout_final_treestates(
597            finalized_state,
598            &new_chain,
599            &prepared,
600        );
601
602        // Quick check that doesn't read from disk
603        let contextual = ContextuallyVerifiedBlock::with_block_and_spent_utxos(
604            prepared.clone(),
605            spent_utxos.clone(),
606            calculate_deferred_pool_balance_change(prepared.height, &self.network),
607        )
608        .map_err(|value_balance_error| {
609            ValidateContextError::CalculateBlockChainValueChange {
610                value_balance_error,
611                height: prepared.height,
612                block_hash: prepared.hash,
613                transaction_count: prepared.block.transactions.len(),
614                spent_utxo_count: spent_utxos.len(),
615            }
616        })?;
617
618        Self::validate_and_update_parallel(new_chain, contextual, sprout_final_treestates)
619    }
620
621    /// Validate `contextual` and update `new_chain`, doing CPU-intensive work in parallel batches.
622    #[allow(clippy::unwrap_in_result)]
623    #[tracing::instrument(skip(new_chain, sprout_final_treestates))]
624    fn validate_and_update_parallel(
625        new_chain: Arc<Chain>,
626        contextual: ContextuallyVerifiedBlock,
627        sprout_final_treestates: HashMap<sprout::tree::Root, Arc<sprout::tree::NoteCommitmentTree>>,
628    ) -> Result<Arc<Chain>, ValidateContextError> {
629        let mut block_commitment_result = None;
630        let mut sprout_anchor_result = None;
631        let mut chain_push_result = None;
632
633        // Clone function arguments for different threads
634        let block = contextual.block.clone();
635        let network = new_chain.network();
636        let history_tree = new_chain.history_block_commitment_tree();
637
638        let block2 = contextual.block.clone();
639        let height = contextual.height;
640        let transaction_hashes = contextual.transaction_hashes.clone();
641
642        rayon::in_place_scope_fifo(|scope| {
643            scope.spawn_fifo(|_scope| {
644                block_commitment_result = Some(check::block_commitment_is_valid_for_chain_history(
645                    block,
646                    &network,
647                    &history_tree,
648                ));
649            });
650
651            scope.spawn_fifo(|_scope| {
652                sprout_anchor_result =
653                    Some(check::anchors::block_sprout_anchors_refer_to_treestates(
654                        sprout_final_treestates,
655                        block2,
656                        transaction_hashes,
657                        height,
658                    ));
659            });
660
661            // We're pretty sure the new block is valid,
662            // so clone the inner chain if needed, then add the new block.
663            //
664            // Pushing a block onto a Chain can launch additional parallel batches.
665            // TODO: should we pass _scope into Chain::push()?
666            scope.spawn_fifo(|_scope| {
667                // TODO: Replace with Arc::unwrap_or_clone() when it stabilises:
668                // https://github.com/rust-lang/rust/issues/93610
669                let new_chain = Arc::try_unwrap(new_chain)
670                    .unwrap_or_else(|shared_chain| (*shared_chain).clone());
671                chain_push_result = Some(new_chain.push(contextual).map(Arc::new));
672            });
673        });
674
675        // Don't return the updated Chain unless all the parallel results were Ok
676        block_commitment_result.expect("scope has finished")?;
677        sprout_anchor_result.expect("scope has finished")?;
678
679        chain_push_result.expect("scope has finished")
680    }
681
682    /// Returns the length of the non-finalized portion of the current best chain
683    /// or `None` if the best chain has no blocks.
684    pub fn best_chain_len(&self) -> Option<u32> {
685        // This `as` can't overflow because the number of blocks in the chain is limited to i32::MAX,
686        // and the non-finalized chain is further limited by the rollback window
687        // (`MAX_BLOCK_REORG_HEIGHT`, currently 1000 blocks).
688        Some(self.best_chain()?.blocks.len() as u32)
689    }
690
691    /// Returns the root height of the non-finalized state, if the non-finalized state is not empty.
692    pub fn root_height(&self) -> Option<block::Height> {
693        self.best_chain()
694            .map(|chain| chain.non_finalized_root_height())
695    }
696
697    /// Returns `true` if `hash` is contained in the non-finalized portion of any
698    /// known chain.
699    #[allow(dead_code)]
700    pub fn any_chain_contains(&self, hash: &block::Hash) -> bool {
701        self.chain_set
702            .iter()
703            .rev()
704            .any(|chain| chain.height_by_hash.contains_key(hash))
705    }
706
707    /// Returns the first chain satisfying the given predicate.
708    ///
709    /// If multiple chains satisfy the predicate, returns the chain with the highest difficulty.
710    /// (Using the tip block hash tie-breaker.)
711    pub fn find_chain<P>(&self, mut predicate: P) -> Option<Arc<Chain>>
712    where
713        P: FnMut(&Chain) -> bool,
714    {
715        // Reverse the iteration order, to find highest difficulty chains first.
716        self.chain_set
717            .iter()
718            .rev()
719            .find(|chain| predicate(chain))
720            .cloned()
721    }
722
723    /// Returns the [`transparent::Utxo`] pointed to by the given
724    /// [`transparent::OutPoint`] if it is present in any chain.
725    ///
726    /// UTXOs are returned regardless of whether they have been spent.
727    pub fn any_utxo(&self, outpoint: &transparent::OutPoint) -> Option<transparent::Utxo> {
728        self.chain_set
729            .iter()
730            .rev()
731            .find_map(|chain| chain.created_utxo(outpoint))
732    }
733
734    /// Returns the `block` with the given hash in any chain.
735    #[allow(dead_code)]
736    pub fn any_block_by_hash(&self, hash: block::Hash) -> Option<Arc<Block>> {
737        // This performs efficiently because the number of chains is limited to 10.
738        for chain in self.chain_set.iter().rev() {
739            if let Some(prepared) = chain
740                .height_by_hash
741                .get(&hash)
742                .and_then(|height| chain.blocks.get(height))
743            {
744                return Some(prepared.block.clone());
745            }
746        }
747
748        None
749    }
750
751    /// Returns the previous block hash for the given block hash in any chain.
752    #[allow(dead_code)]
753    pub fn any_prev_block_hash_for_hash(&self, hash: block::Hash) -> Option<block::Hash> {
754        // This performs efficiently because the blocks are in memory.
755        self.any_block_by_hash(hash)
756            .map(|block| block.header.previous_block_hash)
757    }
758
759    /// Returns the hash for a given `block::Height` if it is present in the best chain.
760    #[allow(dead_code)]
761    pub fn best_hash(&self, height: block::Height) -> Option<block::Hash> {
762        self.best_chain()?
763            .blocks
764            .get(&height)
765            .map(|prepared| prepared.hash)
766    }
767
768    /// Returns the tip of the best chain.
769    #[allow(dead_code)]
770    pub fn best_tip(&self) -> Option<(block::Height, block::Hash)> {
771        let best_chain = self.best_chain()?;
772        let height = best_chain.non_finalized_tip_height();
773        let hash = best_chain.non_finalized_tip_hash();
774
775        Some((height, hash))
776    }
777
778    /// Returns the block at the tip of the best chain.
779    #[allow(dead_code)]
780    pub fn best_tip_block(&self) -> Option<&ContextuallyVerifiedBlock> {
781        let best_chain = self.best_chain()?;
782
783        best_chain.tip_block()
784    }
785
786    /// Returns the height of `hash` in the best chain.
787    #[allow(dead_code)]
788    pub fn best_height_by_hash(&self, hash: block::Hash) -> Option<block::Height> {
789        let best_chain = self.best_chain()?;
790        let height = *best_chain.height_by_hash.get(&hash)?;
791        Some(height)
792    }
793
794    /// Returns the height of `hash` in any chain.
795    #[allow(dead_code)]
796    pub fn any_height_by_hash(&self, hash: block::Hash) -> Option<block::Height> {
797        for chain in self.chain_set.iter().rev() {
798            if let Some(height) = chain.height_by_hash.get(&hash) {
799                return Some(*height);
800            }
801        }
802
803        None
804    }
805
806    /// Returns `true` if the best chain contains `sprout_nullifier`.
807    #[cfg(any(test, feature = "proptest-impl"))]
808    #[allow(dead_code)]
809    pub fn best_contains_sprout_nullifier(&self, sprout_nullifier: &sprout::Nullifier) -> bool {
810        self.best_chain()
811            .map(|best_chain| best_chain.sprout_nullifiers.contains_key(sprout_nullifier))
812            .unwrap_or(false)
813    }
814
815    /// Returns `true` if the best chain contains `sapling_nullifier`.
816    #[cfg(any(test, feature = "proptest-impl"))]
817    #[allow(dead_code)]
818    pub fn best_contains_sapling_nullifier(
819        &self,
820        sapling_nullifier: &zebra_chain::sapling::Nullifier,
821    ) -> bool {
822        self.best_chain()
823            .map(|best_chain| {
824                best_chain
825                    .sapling_nullifiers
826                    .contains_key(sapling_nullifier)
827            })
828            .unwrap_or(false)
829    }
830
831    /// Returns `true` if the best chain contains `orchard_nullifier`.
832    #[cfg(any(test, feature = "proptest-impl"))]
833    #[allow(dead_code)]
834    pub fn best_contains_orchard_nullifier(
835        &self,
836        orchard_nullifier: &zebra_chain::orchard::Nullifier,
837    ) -> bool {
838        self.best_chain()
839            .map(|best_chain| {
840                best_chain
841                    .orchard_nullifiers
842                    .contains_key(orchard_nullifier)
843            })
844            .unwrap_or(false)
845    }
846
847    /// Return the non-finalized portion of the current best chain.
848    pub fn best_chain(&self) -> Option<&Arc<Chain>> {
849        self.chain_iter().next()
850    }
851
852    /// Return the number of chains.
853    pub fn chain_count(&self) -> usize {
854        self.chain_set.len()
855    }
856
857    /// Returns true if this [`NonFinalizedState`] contains no chains.
858    pub fn is_chain_set_empty(&self) -> bool {
859        self.chain_count() == 0
860    }
861
862    /// Return the invalidated blocks.
863    pub fn invalidated_blocks(&self) -> IndexMap<Height, Arc<Vec<ContextuallyVerifiedBlock>>> {
864        self.invalidated_blocks.clone()
865    }
866
867    /// Return the chain whose tip block hash is `parent_hash`.
868    ///
869    /// The chain can be an existing chain in the non-finalized state, or a freshly
870    /// created fork.
871    fn parent_chain(&self, parent_hash: block::Hash) -> Result<Arc<Chain>, ValidateContextError> {
872        match self.find_chain(|chain| chain.non_finalized_tip_hash() == parent_hash) {
873            // Clone the existing Arc<Chain> in the non-finalized state
874            Some(chain) => Ok(chain.clone()),
875            // Create a new fork
876            None => {
877                // Check the lowest difficulty chains first,
878                // because the fork could be closer to their tip.
879                let fork_chain = self
880                    .chain_set
881                    .iter()
882                    .rev()
883                    .find_map(|chain| chain.fork(parent_hash))
884                    .ok_or(ValidateContextError::NotReadyToBeCommitted)?;
885
886                Ok(Arc::new(fork_chain))
887            }
888        }
889    }
890
891    /// Should this `NonFinalizedState` instance track metrics and progress bars?
892    fn should_count_metrics(&self) -> bool {
893        self.should_count_metrics
894    }
895
896    /// Update the metrics after `block` is committed
897    fn update_metrics_for_committed_block(&self, height: block::Height, hash: block::Hash) {
898        if !self.should_count_metrics() {
899            return;
900        }
901
902        metrics::counter!("state.memory.committed.block.count").increment(1);
903        metrics::gauge!("state.memory.committed.block.height").set(height.0 as f64);
904
905        if self
906            .best_chain()
907            .expect("metrics are only updated after initialization")
908            .non_finalized_tip_hash()
909            == hash
910        {
911            metrics::counter!("state.memory.best.committed.block.count").increment(1);
912            metrics::gauge!("state.memory.best.committed.block.height").set(height.0 as f64);
913        }
914
915        self.update_metrics_for_chains();
916    }
917
918    /// Update the metrics after `self.chain_set` is modified
919    fn update_metrics_for_chains(&self) {
920        if !self.should_count_metrics() {
921            return;
922        }
923
924        metrics::gauge!("state.memory.chain.count").set(self.chain_set.len() as f64);
925        metrics::gauge!("state.memory.best.chain.length",)
926            .set(self.best_chain_len().unwrap_or_default() as f64);
927    }
928
929    /// Update the progress bars after any chain is modified.
930    /// This includes both chain forks and committed blocks.
931    fn update_metrics_bars(&mut self) {
932        // TODO: make chain_count_bar interior mutable, move to update_metrics_for_committed_block()
933
934        if !self.should_count_metrics() {
935            #[allow(clippy::needless_return)]
936            return;
937        }
938
939        #[cfg(feature = "progress-bar")]
940        {
941            use std::cmp::Ordering::*;
942
943            if matches!(howudoin::cancelled(), Some(true)) {
944                self.disable_metrics();
945                return;
946            }
947
948            // Update the chain count bar
949            if self.chain_count_bar.is_none() {
950                self.chain_count_bar = Some(howudoin::new_root().label("Chain Forks"));
951            }
952
953            let chain_count_bar = self
954                .chain_count_bar
955                .as_ref()
956                .expect("just initialized if missing");
957            let finalized_tip_height = self
958                .best_chain()
959                .map(|chain| chain.non_finalized_root_height().0 - 1);
960
961            chain_count_bar.set_pos(u64::try_from(self.chain_count()).expect("fits in u64"));
962            // .set_len(u64::try_from(MAX_NON_FINALIZED_CHAIN_FORKS).expect("fits in u64"));
963
964            if let Some(finalized_tip_height) = finalized_tip_height {
965                chain_count_bar.desc(format!("Finalized Root {finalized_tip_height}"));
966            }
967
968            // Update each chain length bar, creating or deleting bars as needed
969            let prev_length_bars = self.chain_fork_length_bars.len();
970
971            match self.chain_count().cmp(&prev_length_bars) {
972                Greater => self
973                    .chain_fork_length_bars
974                    .resize_with(self.chain_count(), || {
975                        howudoin::new_with_parent(chain_count_bar.id())
976                    }),
977                Less => {
978                    let redundant_bars = self.chain_fork_length_bars.split_off(self.chain_count());
979                    for bar in redundant_bars {
980                        bar.close();
981                    }
982                }
983                Equal => {}
984            }
985
986            // It doesn't matter what chain the bar was previously used for,
987            // because we update everything based on the latest chain in that position.
988            for (chain_length_bar, chain) in
989                std::iter::zip(self.chain_fork_length_bars.iter(), self.chain_iter())
990            {
991                let fork_height = chain
992                    .last_fork_height
993                    .unwrap_or_else(|| chain.non_finalized_tip_height())
994                    .0;
995
996                // We need to initialize and set all the values of the bar here, because:
997                // - the bar might have been newly created, or
998                // - the chain this bar was previously assigned to might have changed position.
999                chain_length_bar
1000                    .label(format!("Fork {fork_height}"))
1001                    .set_pos(u64::try_from(chain.len()).expect("fits in u64"));
1002                // TODO: should this be MAX_BLOCK_REORG_HEIGHT?
1003                // .set_len(u64::from(
1004                //     zebra_chain::transparent::MIN_TRANSPARENT_COINBASE_MATURITY,
1005                // ));
1006
1007                // TODO: store work in the finalized state for each height (#7109),
1008                //       and show the full chain work here, like `zcashd` (#7110)
1009                //
1010                // For now, we don't show any work here, see the deleted code in PR #7087.
1011                let mut desc = String::new();
1012
1013                if let Some(recent_fork_height) = chain.recent_fork_height() {
1014                    let recent_fork_length = chain
1015                        .recent_fork_length()
1016                        .expect("just checked recent fork height");
1017
1018                    let mut plural = "s";
1019                    if recent_fork_length == 1 {
1020                        plural = "";
1021                    }
1022
1023                    desc.push_str(&format!(
1024                        " at {recent_fork_height:?} + {recent_fork_length} block{plural}"
1025                    ));
1026                }
1027
1028                chain_length_bar.desc(desc);
1029            }
1030        }
1031    }
1032
1033    /// Stop tracking metrics for this non-finalized state and all its chains.
1034    pub fn disable_metrics(&mut self) {
1035        self.should_count_metrics = false;
1036
1037        #[cfg(feature = "progress-bar")]
1038        {
1039            let count_bar = self.chain_count_bar.take().into_iter();
1040            let fork_bars = self.chain_fork_length_bars.drain(..);
1041            count_bar.chain(fork_bars).for_each(howudoin::Tx::close);
1042        }
1043    }
1044}
1045
1046impl Drop for NonFinalizedState {
1047    fn drop(&mut self) {
1048        self.disable_metrics();
1049    }
1050}