fix: properly handle deletions within overlays

Author: gabriele-0201

nomt/src/merkle/seek.rs

@@ -192,6 +192,7 @@ impl SeekRequest {
 
     fn continue_leaf_fetch<H: HashAlgorithm>(&mut self, leaf: Option<LeafNodeRef>) {
         let RequestState::FetchingLeaf {
+            ref mut overlay_deletions,
             ref mut beatree_iterator,
             ..
         } = self.state
@@ -203,13 +204,24 @@ impl SeekRequest {
             beatree_iterator.provide_leaf(leaf);
         }
 
-        let (key, value_hash) = match beatree_iterator.next() {
-            None => panic!("leaf must exist position={}", self.position.path()),
-            Some(IterOutput::Blocked) => return,
-            Some(IterOutput::Item(key, value)) => {
-                (key, H::hash_value(&value)) // hash
+        let mut deletions_idx = 0;
+        let (key, value_hash) = loop {
+            match beatree_iterator.next() {
+                None => panic!("leaf must exist position={}", self.position.path()),
+                Some(IterOutput::Blocked) => {
+                    overlay_deletions.drain(..deletions_idx);
+                    return;
+                }
+                Some(IterOutput::Item(key, _value))
+                    if deletions_idx < overlay_deletions.len()
+                        && overlay_deletions[deletions_idx] == key =>
+                {
+                    deletions_idx += 1;
+                    continue;
+                }
+                Some(IterOutput::Item(key, value)) => break (key, H::hash_value(&value)),
+                Some(IterOutput::OverflowItem(key, value_hash, _)) => break (key, value_hash),
             }
-            Some(IterOutput::OverflowItem(key, value_hash, _)) => (key, value_hash),
         };
 
         self.state = RequestState::Completed(Some(trie::LeafData {
@@ -351,6 +363,7 @@ enum RequestState {
     Seeking,
     // Fetching one leaf
     FetchingLeaf {
+        overlay_deletions: Vec<KeyPath>,
         beatree_iterator: BeatreeIterator,
         needed_leaves: NeededLeavesIter,
     },
@@ -373,16 +386,18 @@ impl RequestState {
     ) -> Self {
         let (start, end) = range_bounds(pos.raw_path(), pos.depth() as usize);
 
-        // First see if the item is present within the overlay.
-        let overlay_item = overlay
-            .value_iter(start, end)
-            .filter(|(_, v)| v.as_option().is_some())
-            .next();
+        let overlay_items = overlay.value_iter(start, end);
+        let mut overlay_deletions = vec![];
 
-        if let Some((key_path, overlay_leaf)) = overlay_item {
-            let value_hash = match overlay_leaf {
-                // PANIC: we filtered out all deletions above.
-                ValueChange::Delete => panic!(),
+        for (key_path, overlay_change) in overlay_items {
+            let value_hash = match overlay_change {
+                ValueChange::Delete => {
+                    // All deletes must be collected to filter out from the beatree iterator.
+                    overlay_deletions.push(key_path);
+                    continue;
+                }
+                // If an insertion is found within the overlay, it is expected to be
+                // the item associated with the leaf that is being fetched.
                 ValueChange::Insert(value) => H::hash_value(value),
                 ValueChange::InsertOverflow(_, value_hash) => value_hash.clone(),
             };
@@ -397,6 +412,7 @@ impl RequestState {
         let beatree_iterator = read_transaction.iterator(start, end);
         let needed_leaves = beatree_iterator.needed_leaves();
         RequestState::FetchingLeaf {
+            overlay_deletions,
             beatree_iterator,
             needed_leaves,
         }

nomt/tests/overlay.rs

@@ -160,3 +160,95 @@ fn overlay_uncommitted_not_on_disk() {
     assert_eq!(test.read([2; 32]), None);
     assert_eq!(test.read([3; 32]), None);
 }
+
+#[test]
+fn overlay_deletions() {
+    let test_db = || -> Test {
+        let mut test = Test::new("overlay_deletions");
+        // subtree at 0000000_0/1
+        test.write([0; 32], Some(vec![1, 1]));
+        test.write([1; 32], Some(vec![2, 2]));
+
+        // subtree at 001000_00/01/10
+        test.write([32; 32], Some(vec![1, 1]));
+        test.write([33; 32], Some(vec![2, 2]));
+        test.write([34; 32], Some(vec![3, 3]));
+
+        // subtree at 100000_00/01/10/11
+        test.write([128; 32], Some(vec![4, 4]));
+        test.write([129; 32], Some(vec![5, 5]));
+        test.write([130; 32], Some(vec![6, 6]));
+        test.write([131; 32], Some(vec![7, 7]));
+
+        test.commit();
+        test
+    };
+
+    // Delete the first item for each subtree
+    let mut test = test_db();
+
+    test.write([0; 32], None);
+    test.write([32; 32], None);
+    test.write([128; 32], None);
+    let overlay_a = test.update().0;
+
+    test.start_overlay_session([&overlay_a]);
+    assert_eq!(test.read([0; 32]), None);
+    assert_eq!(test.read([1; 32]), Some(vec![2, 2]));
+
+    assert_eq!(test.read([32; 32]), None);
+    assert_eq!(test.read([33; 32]), Some(vec![2, 2]));
+    assert_eq!(test.read([34; 32]), Some(vec![3, 3]));
+
+    assert_eq!(test.read([128; 32]), None);
+    assert_eq!(test.read([129; 32]), Some(vec![5, 5]));
+    assert_eq!(test.read([130; 32]), Some(vec![6, 6]));
+    assert_eq!(test.read([131; 32]), Some(vec![7, 7]));
+
+    let _overlay_b = test.update().0;
+
+    // Delete the second item for each subtree
+    let mut test = test_db();
+
+    test.write([1; 32], None);
+    test.write([33; 32], None);
+    test.write([129; 32], None);
+    let overlay_a = test.update().0;
+
+    test.start_overlay_session([&overlay_a]);
+    assert_eq!(test.read([0; 32]), Some(vec![1, 1]));
+    assert_eq!(test.read([1; 32]), None);
+
+    assert_eq!(test.read([32; 32]), Some(vec![1, 1]));
+    assert_eq!(test.read([33; 32]), None);
+    assert_eq!(test.read([34; 32]), Some(vec![3, 3]));
+
+    assert_eq!(test.read([128; 32]), Some(vec![4, 4]));
+    assert_eq!(test.read([129; 32]), None);
+    assert_eq!(test.read([130; 32]), Some(vec![6, 6]));
+    assert_eq!(test.read([131; 32]), Some(vec![7, 7]));
+
+    let _overlay_b = test.update().0;
+
+    // Sequence of deletes
+    let mut test = test_db();
+
+    test.write([32; 32], None);
+    test.write([33; 32], None);
+    test.write([128; 32], None);
+    test.write([129; 32], None);
+    test.write([131; 32], None);
+    let overlay_a = test.update().0;
+
+    test.start_overlay_session([&overlay_a]);
+    assert_eq!(test.read([32; 32]), None);
+    assert_eq!(test.read([33; 32]), None);
+    assert_eq!(test.read([34; 32]), Some(vec![3, 3]));
+
+    assert_eq!(test.read([128; 32]), None);
+    assert_eq!(test.read([129; 32]), None);
+    assert_eq!(test.read([130; 32]), Some(vec![6, 6]));
+    assert_eq!(test.read([131; 32]), None);
+
+    let _overlay_b = test.update().0;
+}