Change Resolver from fn pointer to Arc<dyn Fn> for storage backends

src/batch_avl_prover.rs

@@ -470,7 +470,7 @@ mod tests {
     fn empty_digest_hash_is_correct() {
         let prover = BatchAVLProver::new(
             AVLTree::new(
-                |digest| Node::LabelOnly(NodeHeader::new(Some(*digest), None)),
+                alloc::sync::Arc::new(|digest: &Digest32| Node::LabelOnly(NodeHeader::new(Some(*digest), None))),
                 32,
                 None,
             ),

src/batch_node.rs

@@ -18,7 +18,7 @@ pub(crate) const END_OF_TREE_IN_PACKAGED_PROOF: u8 = 4;
 pub type Balance = i8;
 pub type SerializedAdProof = Bytes;
 pub type NodeId = Rc<RefCell<Node>>;
-pub type Resolver = fn(&Digest32) -> Node;
+pub type Resolver = alloc::sync::Arc<dyn Fn(&Digest32) -> Node + Send + Sync>;
 pub type Blake2b256 = Blake2b<blake2::digest::typenum::U32>;
 
 #[derive(Debug, Clone)]
@@ -426,30 +426,49 @@ impl AVLTree {
         key_found: bool,
     ) -> bool {
         if &self.label(node) == label {
-            true
-        } else {
-            if let Node::Internal(r) = &mut *node.borrow_mut() {
-                if key_found {
-                    self.contains_recursive(&self.resolve(&mut r.left), key, label, true)
-                } else {
-                    match (*key).cmp(r.hdr.key.as_ref().unwrap()) {
-                        Ordering::Equal =>
-                        // found in the tree -- go one step right, then left to the leaf
-                        {
-                            self.contains_recursive(&self.resolve(&mut r.right), key, label, true)
-                        }
-                        Ordering::Less =>
-                        // going left, not yet found
-                        {
-                            self.contains_recursive(&self.resolve(&mut r.left), key, label, false)
-                        }
-                        Ordering::Greater => {
-                            self.contains_recursive(&self.resolve(&mut r.right), key, label, false)
+            return true;
+        }
+
+        // Discriminate the node kind in a scoped immutable borrow so we can
+        // re-borrow mutably below for the Internal walk without a conflict.
+        enum Kind { Internal, Leaf, LabelOnly }
+        let kind = {
+            let n = node.borrow();
+            match &*n {
+                Node::Internal(_) => Kind::Internal,
+                Node::Leaf(_) => Kind::Leaf,
+                Node::LabelOnly(_) => Kind::LabelOnly,
+            }
+        };
+
+        match kind {
+            Kind::Leaf => false,
+            // Reached a LabelOnly that the resolver could not materialize
+            // (digest not in storage). We cannot conclude whether the target
+            // label is present in the unresolved subtree, so fail safe by
+            // returning true. Used by `removed_nodes()` to decide what to
+            // delete from storage; treating an unresolvable subtree as
+            // "definitely absent" silently deletes nodes that may still be
+            // referenced from this subtree, producing dangling parent->child
+            // references on disk and later walks bailing with
+            // "should never reach this point".
+            Kind::LabelOnly => true,
+            Kind::Internal => {
+                if let Node::Internal(r) = &mut *node.borrow_mut() {
+                    if key_found {
+                        self.contains_recursive(&self.resolve(&mut r.left), key, label, true)
+                    } else {
+                        match (*key).cmp(r.hdr.key.as_ref().unwrap()) {
+                            // found in the tree -- go one step right, then left to the leaf
+                            Ordering::Equal => self.contains_recursive(&self.resolve(&mut r.right), key, label, true),
+                            // going left, not yet found
+                            Ordering::Less => self.contains_recursive(&self.resolve(&mut r.left), key, label, false),
+                            Ordering::Greater => self.contains_recursive(&self.resolve(&mut r.right), key, label, false),
                         }
                     }
+                } else {
+                    unreachable!("kind already verified Internal")
                 }
-            } else {
-                false
             }
         }
     }

src/versioned_avl_storage.rs

@@ -53,4 +53,17 @@ pub trait VersionedAVLStorage {
     /// @return versions store keeps
     ///
     fn rollback_versions<'a>(&'a self) -> Box<dyn Iterator<Item = ADDigest> + 'a>;
+
+    ////
+    /// Force a durable commit of outstanding writes. Implementations that
+    /// use deferred/non-durable writes (e.g. `Durability::None` in redb)
+    /// should fsync the backing store here. Default is a no-op for
+    /// in-memory or always-durable storages.
+    ///
+    /// Callers should invoke this periodically during long-running write
+    /// sequences and on graceful shutdown to bound crash data loss.
+    ///
+    fn flush(&self) -> Result<()> {
+        Ok(())
+    }
 }

tests/common/mod.rs

@@ -10,6 +10,7 @@ use ergo_avltree_rust::versioned_avl_storage::*;
 use rand::prelude::*;
 use sha2::{Digest, Sha256};
 use std::collections::HashMap;
+use std::sync::Arc;
 
 pub const INITIAL_TREE_SIZE: usize = 1000;
 pub const KEY_LENGTH: usize = 32;
@@ -97,7 +98,7 @@ pub fn generate_verifier(
 }
 
 pub fn generate_tree(key_length: usize, value_length: Option<usize>) -> AVLTree {
-    AVLTree::new(dummy_resolver, key_length, value_length)
+    AVLTree::new(Arc::new(dummy_resolver), key_length, value_length)
 }
 
 pub fn generate_prover(key_length: usize, value_length: Option<usize>) -> BatchAVLProver {