fix(merkle-tree): Fix incoherent Merkle tree view (#2071)

## What ❔

Makes `MerkleTreeInfo` returned from `AsyncTreeReader` consistent.

## Why ❔

Right now, it's not consistent, which e.g., leads to sporadic CI
failures, and could lead to inconsistent data exposed via tree API or
tree health check.

## Checklist

- [x] PR title corresponds to the body of PR (we generate changelog
entries from PRs).
- [x] Tests for the changes have been added / updated.
- [x] Documentation comments have been added / updated.
- [x] Code has been formatted via `zk fmt` and `zk lint`.
- [x] Spellcheck has been run via `zk spellcheck`.

core/lib/merkle_tree/src/domain.rs

@@ -47,6 +47,11 @@ pub struct ZkSyncTree {
 }
 
 impl ZkSyncTree {
+    /// Returns a hash of an empty tree. This is a constant value.
+    pub fn empty_tree_hash() -> ValueHash {
+        Blake2Hasher.empty_tree_hash()
+    }
+
     fn create_thread_pool(thread_count: usize) -> ThreadPool {
         ThreadPoolBuilder::new()
             .thread_name(|idx| format!("new-merkle-tree-{idx}"))
@@ -375,9 +380,10 @@ impl ZkSyncTreeReader {
         &self.0.db
     }
 
-    /// Returns the current root hash of this tree.
-    pub fn root_hash(&self) -> ValueHash {
-        self.0.latest_root_hash()
+    /// Returns the root hash and leaf count at the specified L1 batch.
+    pub fn root_info(&self, l1_batch_number: L1BatchNumber) -> Option<(ValueHash, u64)> {
+        let root = self.0.root(l1_batch_number.0.into())?;
+        Some((root.hash(&Blake2Hasher), root.leaf_count()))
     }
 
     /// Returns the next L1 batch number that should be processed by the tree.
@@ -397,11 +403,6 @@ impl ZkSyncTreeReader {
         })
     }
 
-    /// Returns the number of leaves in the tree.
-    pub fn leaf_count(&self) -> u64 {
-        self.0.latest_root().leaf_count()
-    }
-
     /// Reads entries together with Merkle proofs with the specified keys from the tree. The entries are returned
     /// in the same order as requested.
     ///

core/lib/merkle_tree/src/hasher/nodes.rs

@@ -4,7 +4,8 @@ use std::slice;
 
 use crate::{
     hasher::HasherWithStats,
-    types::{ChildRef, InternalNode, LeafNode, Node, ValueHash, TREE_DEPTH},
+    types::{ChildRef, InternalNode, LeafNode, Node, Root, ValueHash, TREE_DEPTH},
+    HashTree,
 };
 
 impl LeafNode {
@@ -256,6 +257,15 @@ impl Node {
     }
 }
 
+impl Root {
+    pub(crate) fn hash(&self, hasher: &dyn HashTree) -> ValueHash {
+        let Self::Filled { node, .. } = self else {
+            return hasher.empty_tree_hash();
+        };
+        node.hash(&mut HasherWithStats::new(&hasher), 0)
+    }
+}
+
 #[cfg(test)]
 mod tests {
     use zksync_crypto::hasher::{blake2::Blake2Hasher, Hasher};

core/lib/merkle_tree/src/lib.rs

@@ -61,7 +61,7 @@ pub use crate::{
         TreeLogEntry, TreeLogEntryWithProof, ValueHash,
     },
 };
-use crate::{hasher::HasherWithStats, storage::Storage, types::Root};
+use crate::{storage::Storage, types::Root};
 
 mod consistency;
 pub mod domain;
@@ -166,10 +166,7 @@ impl<DB: Database, H: HashTree> MerkleTree<DB, H> {
     /// was not written yet.
     pub fn root_hash(&self, version: u64) -> Option<ValueHash> {
         let root = self.root(version)?;
-        let Root::Filled { node, .. } = root else {
-            return Some(self.hasher.empty_tree_hash());
-        };
-        Some(node.hash(&mut HasherWithStats::new(&self.hasher), 0))
+        Some(root.hash(&self.hasher))
     }
 
     pub(crate) fn root(&self, version: u64) -> Option<Root> {

core/node/metadata_calculator/src/helpers.rs

@@ -300,12 +300,35 @@ impl AsyncTreeReader {
     }
 
     pub async fn info(self) -> MerkleTreeInfo {
-        tokio::task::spawn_blocking(move || MerkleTreeInfo {
-            mode: self.mode,
-            root_hash: self.inner.root_hash(),
-            next_l1_batch_number: self.inner.next_l1_batch_number(),
-            min_l1_batch_number: self.inner.min_l1_batch_number(),
-            leaf_count: self.inner.leaf_count(),
+        tokio::task::spawn_blocking(move || {
+            loop {
+                let next_l1_batch_number = self.inner.next_l1_batch_number();
+                let latest_l1_batch_number = next_l1_batch_number.checked_sub(1);
+                let root_info = if let Some(number) = latest_l1_batch_number {
+                    self.inner.root_info(L1BatchNumber(number))
+                } else {
+                    // No L1 batches in the tree yet.
+                    Some((ZkSyncTree::empty_tree_hash(), 0))
+                };
+                let Some((root_hash, leaf_count)) = root_info else {
+                    // It is possible (although very unlikely) that the latest tree version was removed after requesting it,
+                    // hence the outer loop; RocksDB doesn't provide consistent data views by default.
+                    tracing::info!(
+                        "Tree version at L1 batch {latest_l1_batch_number:?} was removed after requesting the latest tree L1 batch; \
+                         re-requesting tree information"
+                    );
+                    continue;
+                };
+
+                // `min_l1_batch_number` is not necessarily consistent with other retrieved tree data, but this looks fine.
+                break MerkleTreeInfo {
+                    mode: self.mode,
+                    root_hash,
+                    next_l1_batch_number,
+                    min_l1_batch_number: self.inner.min_l1_batch_number(),
+                    leaf_count,
+                };
+            }
         })
         .await
         .unwrap()