fix(state): Avoid temporary failures verifying the first non-finalized block or attempting to fork the chain before the final checkpoint (#6810)

* Fix #6388, rename sent_hashes field

* Removes prune_by_height, uses new SentHashes instead

* update queue_and_commit_to_non_finalized_state to start with children of non-finalized tip when dropping the finalized block write sender

* revert rename for now

* removes outdated TODO

---------

Co-authored-by: mergify[bot] <37929162+mergify[bot]@users.noreply.github.com>

zebra-state/src/service.rs

@@ -707,23 +707,21 @@ impl StateService {
             // Tell the block write task to stop committing checkpoint verified blocks to the finalized state,
             // and move on to committing semantically verified blocks to the non-finalized state.
             std::mem::drop(self.finalized_block_write_sender.take());
-
+            // Remove any checkpoint-verified block hashes from `non_finalized_block_write_sent_hashes`.
+            self.non_finalized_block_write_sent_hashes = SentHashes::default();
+            // Mark `SentHashes` as usable by the `can_fork_chain_at()` method.
+            self.non_finalized_block_write_sent_hashes
+                .can_fork_chain_at_hashes = true;
+            // Send blocks from non-finalized queue
+            self.send_ready_non_finalized_queued(self.finalized_block_write_last_sent_hash);
             // We've finished committing checkpoint verified blocks to finalized state, so drop any repeated queued blocks.
             self.clear_finalized_block_queue(
                 "already finished committing checkpoint verified blocks: dropped duplicate block, \
                  block is already committed to the state",
             );
-        }
+        } else if !self.can_fork_chain_at(&parent_hash) {
-
-        // TODO: avoid a temporary verification failure that can happen
-        //       if the first non-finalized block arrives before the last finalized block is committed
-        //       (#5125)
-        if !self.can_fork_chain_at(&parent_hash) {
             tracing::trace!("unready to verify, returning early");
-            return rsp_rx;
+        } else if self.finalized_block_write_sender.is_none() {
-        }
-
-        if self.finalized_block_write_sender.is_none() {
             // Wait until block commit task is ready to write non-finalized blocks before dequeuing them
             self.send_ready_non_finalized_queued(parent_hash);
 
@@ -743,7 +741,8 @@ impl StateService {
 
     /// Returns `true` if `hash` is a valid previous block hash for new non-finalized blocks.
     fn can_fork_chain_at(&self, hash: &block::Hash) -> bool {
-        self.non_finalized_block_write_sent_hashes.contains(hash)
+        self.non_finalized_block_write_sent_hashes
+            .can_fork_chain_at(hash)
             || &self.read_service.db.finalized_tip_hash() == hash
     }
 

zebra-state/src/service/queued_blocks.rs

@@ -237,6 +237,10 @@ pub(crate) struct SentHashes {
 
     /// Known UTXOs.
     known_utxos: HashMap<transparent::OutPoint, transparent::Utxo>,
+
+    /// Whether the hashes in this struct can be used check if the chain can be forked.
+    /// This is set to false until all checkpoint-verified block hashes have been pruned.
+    pub(crate) can_fork_chain_at_hashes: bool,
 }
 
 impl SentHashes {
@@ -335,6 +339,8 @@ impl SentHashes {
         });
 
         self.sent.shrink_to_fit();
+        self.known_utxos.shrink_to_fit();
+        self.bufs.shrink_to_fit();
 
         self.update_metrics_for_cache();
     }
@@ -344,6 +350,11 @@ impl SentHashes {
         self.sent.contains_key(hash)
     }
 
+    /// Returns true if the chain can be forked at the provided hash
+    pub fn can_fork_chain_at(&self, hash: &block::Hash) -> bool {
+        self.can_fork_chain_at_hashes && self.contains(hash)
+    }
+
     /// Update sent block metrics after a block is sent.
     fn update_metrics_for_block(&self, height: block::Height) {
         metrics::counter!("state.memory.sent.block.count", 1);