wip

grovedb/src/merk_cache.rs

@@ -18,7 +18,7 @@ use crate::{Element, Error, GroveDb, Transaction};
 type TxMerk<'db> = Merk<PrefixedRocksDbTransactionContext<'db>>;
 
 struct CachedMerk<'db> {
-    updated: bool,
+    to_propagate: bool,
     merk: TxMerk<'db>,
 }
 
@@ -29,6 +29,7 @@ struct CachedMerk<'db> {
 /// those Merks, so it's better to have them cached and proceed through the same
 /// structure. Eventually we'll have enough info at the same place to perform
 /// necessary propagations as well.
+// SAFETY: please consult with other safety docs here before doing any changes
 pub(crate) struct MerkCache<'db, 'b, B> {
     merks: BTreeMap<SubtreePath<'b, B>, CachedMerk<'db>>,
     db: &'db GroveDb,
@@ -38,14 +39,14 @@ pub(crate) struct MerkCache<'db, 'b, B> {
 }
 
 impl<'db, 'b, B: AsRef<[u8]>> MerkCache<'db, 'b, B> {
-    pub(crate) fn new(
+    pub(crate) fn new<'tx>(
         db: &'db GroveDb,
         tx: &'db Transaction<'db>,
         version: &'db GroveVersion,
     ) -> Self {
         MerkCache {
             db,
-            tx,
+            tx: &tx,
             version,
             batch: Box::leak(Box::new(StorageBatch::default())),
             merks: Default::default(),
@@ -55,10 +56,10 @@ impl<'db, 'b, B: AsRef<[u8]>> MerkCache<'db, 'b, B> {
     /// Get a mutable Merk reference from the cache.
     /// If it doesn't present then it will be opened.
     /// Returns `None` if  there is no Merk under this path.
-    fn get_merk_mut_internal<'s>(
-        &'s mut self,
+    fn get_merk_mut_internal<'c>(
+        &'c mut self,
         path: SubtreePath<'b, B>,
-    ) -> CostResult<&'s mut CachedMerk<'db>, Error> {
+    ) -> CostResult<&'c mut CachedMerk<'db>, Error> {
         let mut cost = Default::default();
 
         match self.merks.entry(path) {
@@ -75,7 +76,7 @@ impl<'db, 'b, B: AsRef<[u8]>> MerkCache<'db, 'b, B> {
                 );
                 Ok(e.insert(CachedMerk {
                     merk,
-                    updated: false,
+                    to_propagate: false,
                 }))
                 .wrap_with_cost(cost)
             }
@@ -89,11 +90,11 @@ impl<'db, 'b, B: AsRef<[u8]>> MerkCache<'db, 'b, B> {
     /// # Panics
     /// All input paths *must* be unique, otherwise it could provide multiple
     /// mutable references to the same memory which is strictly prohibited.
-    pub(crate) fn get_multi_mut<'s, const N: usize>(
-        &'s mut self,
+    pub(crate) fn get_multi_mut<'c, const N: usize>(
+        &'c mut self,
         paths: [SubtreePath<'b, B>; N],
-    ) -> CostResult<[MerkHandle<'db, 's, 'b, B>; N], Error> {
-        let mut result_uninit = [const { MaybeUninit::<MerkHandle<'db, 's, 'b, B>>::uninit() }; N];
+    ) -> CostResult<[MerkHandle<'db, 'c>; N], Error> {
+        let mut result_uninit = [const { MaybeUninit::<MerkHandle<'db, 'c>>::uninit() }; N];
         let mut cost = Default::default();
 
         let unique_args: HashSet<_> = paths.iter().collect();
@@ -109,10 +110,9 @@ impl<'db, 'b, B: AsRef<[u8]>> MerkCache<'db, 'b, B> {
             // sure no overlapping memory will be referenced.
             let merk_ref = unsafe {
                 MerkHandle::new(
-                    path.clone(),
                     (cost_return_on_error!(&mut cost, self.get_merk_mut_internal(path))
                         as *mut CachedMerk<'db>)
-                        .as_mut::<'s>()
+                        .as_mut::<'c>()
                         .expect("not a null pointer"),
                     &self.version,
                 )
@@ -125,45 +125,92 @@ impl<'db, 'b, B: AsRef<[u8]>> MerkCache<'db, 'b, B> {
         // N in our case. `mem::transmute` would represent it better, however,
         // due to poor support of const generics in stable Rust we bypass
         // compile-time size checks with pointer casts.
-        let result = unsafe { (&result_uninit as *const _ as *const [MerkHandle<B>; N]).read() };
+        let result = unsafe { (&result_uninit as *const _ as *const [MerkHandle; N]).read() };
         mem::forget(result_uninit);
 
         Ok(result).wrap_with_cost(cost)
     }
 
     /// Summarizes all performed operations on this `MerkCache` with necessary
     /// propagations into a `Storagebatch`.
-    pub(crate) fn finalize(self) -> Box<StorageBatch> {
+    pub(crate) fn finalize(mut self) -> CostResult<Box<StorageBatch>, Error> {
         let batch_ptr = self.batch as *const _;
-        // Remove all possible usages of the `StorageBatch`:
-        drop(self);
+
+        // Propagate updated subtrees' hashes up to the root and dropping all possible
+        // batch users:
+        let propagation_result = self.propagate_updated_merks();
 
         // SAFETY: The batch reference was created by `Box::leak` and restored into a
         // `Box` form. No shared usage of that memory exists because the
         // `MerkCache` structure was dropped above.
-        unsafe { Box::from_raw(batch_ptr as *mut _) }
+        let result_batch = unsafe { Box::from_raw(batch_ptr as *mut _) };
+
+        // The batch's unsafety cleanup happens regardless of propagation results, so no
+        // early termination allowed. We do the mapping afterwards:
+        propagation_result.map_ok(|_| result_batch)
+    }
+
+    /// Finalizes each Merk starting from the deepest subtrees, updating hashes
+    /// up to the root.
+    fn propagate_updated_merks(mut self) -> CostResult<(), Error> {
+        let mut cost = Default::default();
+
+        let version: &'db GroveVersion = self.version;
+
+        // Picking Merks one by one as long as they have a parent
+        while let Some((parent_path, parent_key, subtree)) =
+            self.merks.pop_first().and_then(|(path, subtree)| {
+                path.derive_parent()
+                    .map(|(parent_path, parent_key)| (parent_path, parent_key, subtree))
+            })
+        {
+            // If a cached Merk wasn't changed, we don't need to propagate it:
+            if !subtree.to_propagate {
+                continue;
+            }
+
+            let parent_subtree =
+                cost_return_on_error!(&mut cost, self.get_merk_mut_internal(parent_path));
+            parent_subtree.to_propagate = true;
+            let (root_hash, root_key, root_sum) = cost_return_on_error!(
+                &mut cost,
+                subtree.merk.root_hash_key_and_sum().map_err(|e| e.into())
+            );
+            cost_return_on_error!(
+                &mut cost,
+                GroveDb::update_tree_item_preserve_flag(
+                    &mut parent_subtree.merk,
+                    parent_key,
+                    root_key,
+                    root_hash,
+                    root_sum,
+                    version
+                )
+            );
+        }
+
+        Ok(()).wrap_with_cost(cost)
     }
 }
 
 /// Handle to a cached Merk.
-pub(crate) struct MerkHandle<'db, 'c, 'b, B> {
-    path: SubtreePath<'b, B>,
+pub(crate) struct MerkHandle<'db, 'c> {
     merk: &'c mut TxMerk<'db>,
     version: &'db GroveVersion,
-    updated: &'c mut bool,
+    to_propagate: &'c mut bool,
 }
 
 /// It is allowed to dereference `MerkHandle` to regular Merks but in a
 /// non-mutable way since we want to track what have been done to those Merks.
-impl<'db, 'c, 'b, B> Deref for MerkHandle<'db, 'c, 'b, B> {
+impl<'db, 'c> Deref for MerkHandle<'db, 'c> {
     type Target = TxMerk<'db>;
 
     fn deref(&self) -> &Self::Target {
         &self.merk
     }
 }
 
-impl<'db, 'c, 'b, B> MerkHandle<'db, 'c, 'b, B> {
+impl<'db, 'c> MerkHandle<'db, 'c> {
     pub(crate) fn insert(
         &mut self,
         key: impl AsRef<[u8]>,
@@ -172,19 +219,14 @@ impl<'db, 'c, 'b, B> MerkHandle<'db, 'c, 'b, B> {
     ) -> CostResult<(), Error> {
         element
             .insert(self.merk, key, options, self.version)
-            .for_ok(|_| *self.updated = true)
+            .for_ok(|_| *self.to_propagate = true)
     }
 
-    fn new(
-        path: SubtreePath<'b, B>,
-        cached_merk: &'c mut CachedMerk<'db>,
-        version: &'db GroveVersion,
-    ) -> Self {
+    fn new(cached_merk: &'c mut CachedMerk<'db>, version: &'db GroveVersion) -> Self {
         Self {
-            path,
             merk: &mut cached_merk.merk,
             version,
-            updated: &mut cached_merk.updated,
+            to_propagate: &mut cached_merk.to_propagate,
         }
     }
 }
@@ -197,14 +239,16 @@ mod tests {
     use grovedb_version::version::GroveVersion;
 
     use super::MerkCache;
-    use crate::tests::{make_deep_tree, ANOTHER_TEST_LEAF, TEST_LEAF};
+    use crate::{
+        tests::{make_deep_tree, ANOTHER_TEST_LEAF, TEST_LEAF},
+        Element,
+    };
 
     #[test]
     fn cached_subtrees_are_free() {
         let version = GroveVersion::latest();
         let db = make_deep_tree(&version);
         let tx = db.start_transaction();
-        let batch = StorageBatch::new();
         let mut cache = MerkCache::new(&db, &tx, version);
 
         let mut cost: OperationCost = Default::default();
@@ -242,12 +286,40 @@ mod tests {
         let version = GroveVersion::latest();
         let db = make_deep_tree(&version);
         let tx = db.start_transaction();
-        let batch = StorageBatch::new();
         let mut cache = MerkCache::new(&db, &tx, version);
 
         let _ = cache.get_multi_mut([
             SubtreePath::from(&[TEST_LEAF]),
             SubtreePath::from(&[TEST_LEAF]),
         ]);
     }
+
+    #[test]
+    fn changes_to_merk_cache_provide_non_empty_batch() {
+        let version = GroveVersion::latest();
+        let db = make_deep_tree(&version);
+        let tx = db.start_transaction();
+
+        let mut cache = MerkCache::new(&db, &tx, version);
+        let [_subtree] = cache
+            .get_multi_mut([SubtreePath::from(&[TEST_LEAF])])
+            .unwrap()
+            .unwrap();
+
+        // Do nothing and finalize a batch:
+        assert!(cache.finalize().unwrap().unwrap().len() == 0);
+
+        let mut cache = MerkCache::new(&db, &tx, version);
+        let [mut subtree] = cache
+            .get_multi_mut([SubtreePath::from(&[TEST_LEAF])])
+            .unwrap()
+            .unwrap();
+        subtree
+            .insert(b"ayy", Element::new_item(b"lmao".to_vec()), None)
+            .unwrap()
+            .unwrap();
+
+        // Do something and finalize another batch:
+        assert!(cache.finalize().unwrap().unwrap().len() > 0);
+    }
 }

path/src/subtree_path.rs

@@ -99,6 +99,7 @@ where
             iter_a
                 .len()
                 .cmp(&iter_b.len())
+                .reverse()
                 .then_with(|| iter_a.cmp(iter_b)),
         )
     }
@@ -340,8 +341,8 @@ mod tests {
             cmp::Ordering::Equal
         ));
 
-        // Longer paths are "greater than":
-        assert!(path_a < path_b);
-        assert!(path_a < path_c);
+        // Longer paths come first
+        assert!(path_a > path_b);
+        assert!(path_a > path_c);
     }
 }

storage/src/storage.rs

@@ -305,8 +305,8 @@ impl StorageBatch {
         }
     }
 
-    #[cfg(test)]
-    pub(crate) fn len(&self) -> usize {
+    /// Returns a number of operations in the batch.
+    pub fn len(&self) -> usize {
         let operations = self.operations.borrow();
         operations.data.len()
             + operations.roots.len()