From d308d7879617bc9340bd4781906f788b3a9502ae Mon Sep 17 00:00:00 2001 From: Quantum Explorer Date: Sun, 8 Mar 2026 17:34:13 +0700 Subject: [PATCH 1/2] fix: enforce subtree emptiness check and storage cleanup in batch DeleteTree Batch DeleteTree had two issues: H1: When DeleteTree ran in batch mode for standard Merk trees, only the parent key was removed. Child subtree storage and all nested data remained as orphans, causing corruption when re-inserting at the same path. Now after apply_body, we call find_subtrees to recursively discover all nested subtrees and clear their storage, matching the non-batch delete behavior. H2: The batch path unconditionally deleted tree elements without checking is_empty_tree(). The allow_deleting_non_empty_trees option in BatchApplyOptions was never consulted. Now before apply_body, we check subtree emptiness (using is_empty_tree_except for Merk trees and non_merk_entry_count for non-Merk trees) and respect the batch options. Co-Authored-By: Claude Opus 4.6 --- grovedb/src/batch/mod.rs | 398 +++++++++++++++++-- grovedb/src/tests/batch_delete_tree_tests.rs | 394 ++++++++++++++++++ grovedb/src/tests/mod.rs | 1 + 3 files changed, 760 insertions(+), 33 deletions(-) create mode 100644 grovedb/src/tests/batch_delete_tree_tests.rs diff --git a/grovedb/src/batch/mod.rs b/grovedb/src/batch/mod.rs index 1b06788fd..c58534775 100644 --- a/grovedb/src/batch/mod.rs +++ b/grovedb/src/batch/mod.rs @@ -3303,6 +3303,8 @@ impl GroveDb { self.preprocess_dense_tree_ops(ops, tx.as_ref(), grove_version) ); + // Collect paths of subtrees being deleted, separated by type. + // // Non-Merk trees (MmrTree, BulkAppendTree, DenseTree, CommitmentTree) // store their data in the data storage namespace of their subtree path // (e.g. MMR nodes, buffer entries, dense tree values). When apply_body @@ -3310,21 +3312,155 @@ impl GroveDb { // and clears the subtree's Merk metadata, but does NOT clear the raw // data these tree types wrote. Without explicit cleanup, deleting a // non-Merk tree would leave orphaned data that could corrupt a new tree - // inserted at the same path. Standard Merk trees don't need this because - // their data IS their Merk — deleting the subtree clears everything. - let non_merk_delete_paths: Vec>> = ops - .iter() - .filter_map(|op| { - if let GroveOp::DeleteTree(tree_type) = &op.op - && tree_type.uses_non_merk_data_storage() - { - let mut child_path = op.path.to_path(); - child_path.push(op.key.as_ref()?.as_slice().to_vec()); - return Some(child_path); + // inserted at the same path. + // + // Standard Merk trees also need cleanup: when a Merk subtree has child + // subtrees, deleting the parent key in the parent Merk does NOT clear + // the child subtree's storage. We must recursively find and clear all + // nested subtrees. + let mut non_merk_delete_paths: Vec>> = Vec::new(); + let mut merk_delete_paths: Vec>> = Vec::new(); + + let batch_apply_options_ref = batch_apply_options.as_ref().cloned().unwrap_or_default(); + + for op in ops.iter() { + if let GroveOp::DeleteTree(tree_type) = &op.op + && let Some(key) = op.key.as_ref() + { + let mut child_path = op.path.to_path(); + child_path.push(key.as_slice().to_vec()); + + // H2 fix: check emptiness of the subtree before allowing + // deletion, respecting batch_apply_options just like the + // non-batch path does. + if !batch_apply_options_ref.allow_deleting_non_empty_trees { + let is_empty = if tree_type.uses_non_merk_data_storage() { + // Non-Merk trees: check element-level entry count. + let parent_path_vec = op.path.to_path(); + let parent_path: SubtreePath> = parent_path_vec.as_slice().into(); + let parent_storage = self + .db + .get_transactional_storage_context( + parent_path, + Some(&storage_batch), + tx.as_ref(), + ) + .unwrap_add_cost(&mut cost); + let element = cost_return_on_error!( + &mut cost, + Element::get_from_storage( + &parent_storage, + key.as_slice(), + grove_version, + ) + .map_err(|e| { + Error::CorruptedData(format!( + "unable to get element for delete tree emptiness check: {e}" + )) + }) + ); + element.non_merk_entry_count().unwrap_or(0) == 0 + } else { + // Standard Merk trees: use is_empty_tree_except to + // account for other delete ops in the same batch that + // target this subtree. + let batch_deleted_keys = ops + .iter() + .filter_map(|other_op| match &other_op.op { + GroveOp::Delete | GroveOp::DeleteTree(_) => { + if other_op.path.to_path() == child_path { + Some(other_op.key.as_ref()?.as_slice().to_vec()) + } else { + None + } + } + _ => None, + }) + .collect::>>(); + let batch_deleted_keys_refs: std::collections::BTreeSet<&[u8]> = + batch_deleted_keys.iter().map(|k| k.as_slice()).collect(); + + let child_subtree_path: SubtreePath> = child_path.as_slice().into(); + let child_storage = self + .db + .get_transactional_storage_context( + child_subtree_path, + Some(&storage_batch), + tx.as_ref(), + ) + .unwrap_add_cost(&mut cost); + + let child_merk = cost_return_on_error!( + &mut cost, + Merk::open_layered_with_root_key( + child_storage, + None, + *tree_type, + Some(&Element::value_defined_cost_for_serialized_value), + grove_version, + ) + .map_err(|e| { + Error::CorruptedData(format!( + "unable to open subtree for emptiness check: {e}" + )) + }) + ); + + child_merk + .is_empty_tree_except(batch_deleted_keys_refs) + .unwrap_add_cost(&mut cost) + }; + + if !is_empty { + if batch_apply_options_ref.deleting_non_empty_trees_returns_error { + return Err(Error::DeletingNonEmptyTree( + "trying to do a batch delete operation for a non empty tree, \ + but options not allowing this", + )) + .wrap_with_cost(cost); + } else { + // Skip this DeleteTree op — don't add to cleanup + // paths and the op will still be in the batch but + // we filter it out below. + continue; + } + } } - None - }) - .collect(); + + if tree_type.uses_non_merk_data_storage() { + non_merk_delete_paths.push(child_path); + } else { + merk_delete_paths.push(child_path); + } + } + } + + // When allow_deleting_non_empty_trees is false and + // deleting_non_empty_trees_returns_error is false, we need to filter + // out DeleteTree ops for non-empty trees (they were skipped above). + let ops = if !batch_apply_options_ref.allow_deleting_non_empty_trees + && !batch_apply_options_ref.deleting_non_empty_trees_returns_error + { + let all_delete_paths: std::collections::HashSet>> = non_merk_delete_paths + .iter() + .chain(merk_delete_paths.iter()) + .cloned() + .collect(); + ops.into_iter() + .filter(|op| { + if let GroveOp::DeleteTree(_) = &op.op + && let Some(key) = op.key.as_ref() + { + let mut child_path = op.path.to_path(); + child_path.push(key.as_slice().to_vec()); + return all_delete_paths.contains(&child_path); + } + true + }) + .collect() + } else { + ops + }; // With the only one difference (if there is a transaction) do the following: // 2. If nothing left to do and we were on a non-leaf subtree or we're done with @@ -3378,6 +3514,34 @@ impl GroveDb { ); } + // Clean up storage for deleted standard Merk subtrees. + // The parent key has been removed from the parent Merk by apply_body, + // but the child subtree's storage (and any nested subtrees) remains. + // We use find_subtrees to recursively discover all nested subtrees + // and clear their storage, matching the non-batch delete behavior. + for child_path in &merk_delete_paths { + let child_subtree_path: SubtreePath> = child_path.as_slice().into(); + let subtrees_paths = cost_return_on_error!( + &mut cost, + self.find_subtrees(&child_subtree_path, Some(tx.as_ref()), grove_version) + ); + for subtree_path in subtrees_paths { + let p: SubtreePath<_> = subtree_path.as_slice().into(); + let mut storage = self + .db + .get_transactional_storage_context(p, Some(&storage_batch), tx.as_ref()) + .unwrap_add_cost(&mut cost); + cost_return_on_error!( + &mut cost, + storage.clear().map_err(|e| { + Error::CorruptedData(format!( + "unable to clean up merk subtree storage in batch delete: {e}", + )) + }) + ); + } + } + // TODO: compute batch costs cost_return_on_error!( &mut cost, @@ -3492,23 +3656,142 @@ impl GroveDb { self.preprocess_dense_tree_ops(ops, tx.as_ref(), grove_version) ); - // See comment above in apply_batch_with_element_flags_update for why - // non-Merk tree deletions need explicit data storage cleanup. - let non_merk_delete_paths: Vec>> = ops - .iter() - .filter_map(|op| { - if let GroveOp::DeleteTree(tree_type) = &op.op - && tree_type.uses_non_merk_data_storage() - { - let mut child_path = op.path.to_path(); - child_path.push(op.key.as_ref()?.as_slice().to_vec()); - return Some(child_path); - } - None - }) - .collect(); + // See comment in apply_batch_with_element_flags_update for why + // deleted tree subtrees need explicit storage cleanup, and why + // emptiness checks are needed (H2). + let mut non_merk_delete_paths: Vec>> = Vec::new(); + let mut merk_delete_paths: Vec>> = Vec::new(); let mut batch_apply_options = batch_apply_options.unwrap_or_default(); + + for op in ops.iter() { + if let GroveOp::DeleteTree(tree_type) = &op.op + && let Some(key) = op.key.as_ref() + { + let mut child_path = op.path.to_path(); + child_path.push(key.as_slice().to_vec()); + + // H2 fix: check emptiness of the subtree before allowing + // deletion (same logic as apply_batch_with_element_flags_update). + if !batch_apply_options.allow_deleting_non_empty_trees { + let is_empty = if tree_type.uses_non_merk_data_storage() { + let parent_path_vec = op.path.to_path(); + let parent_path: SubtreePath> = parent_path_vec.as_slice().into(); + let parent_storage = self + .db + .get_transactional_storage_context( + parent_path, + Some(&storage_batch), + tx.as_ref(), + ) + .unwrap_add_cost(&mut cost); + let element = cost_return_on_error!( + &mut cost, + Element::get_from_storage( + &parent_storage, + key.as_slice(), + grove_version, + ) + .map_err(|e| { + Error::CorruptedData(format!( + "unable to get element for delete tree emptiness check: {e}" + )) + }) + ); + element.non_merk_entry_count().unwrap_or(0) == 0 + } else { + let batch_deleted_keys = ops + .iter() + .filter_map(|other_op| match &other_op.op { + GroveOp::Delete | GroveOp::DeleteTree(_) => { + if other_op.path.to_path() == child_path { + Some(other_op.key.as_ref()?.as_slice().to_vec()) + } else { + None + } + } + _ => None, + }) + .collect::>>(); + let batch_deleted_keys_refs: std::collections::BTreeSet<&[u8]> = + batch_deleted_keys.iter().map(|k| k.as_slice()).collect(); + + let child_subtree_path: SubtreePath> = child_path.as_slice().into(); + let child_storage = self + .db + .get_transactional_storage_context( + child_subtree_path, + Some(&storage_batch), + tx.as_ref(), + ) + .unwrap_add_cost(&mut cost); + + let child_merk = cost_return_on_error!( + &mut cost, + Merk::open_layered_with_root_key( + child_storage, + None, + *tree_type, + Some(&Element::value_defined_cost_for_serialized_value), + grove_version, + ) + .map_err(|e| { + Error::CorruptedData(format!( + "unable to open subtree for emptiness check: {e}" + )) + }) + ); + + child_merk + .is_empty_tree_except(batch_deleted_keys_refs) + .unwrap_add_cost(&mut cost) + }; + + if !is_empty { + if batch_apply_options.deleting_non_empty_trees_returns_error { + return Err(Error::DeletingNonEmptyTree( + "trying to do a batch delete operation for a non empty tree, \ + but options not allowing this", + )) + .wrap_with_cost(cost); + } else { + continue; + } + } + } + + if tree_type.uses_non_merk_data_storage() { + non_merk_delete_paths.push(child_path); + } else { + merk_delete_paths.push(child_path); + } + } + } + + // Filter out skipped DeleteTree ops when non-error mode is active. + let ops = if !batch_apply_options.allow_deleting_non_empty_trees + && !batch_apply_options.deleting_non_empty_trees_returns_error + { + let all_delete_paths: std::collections::HashSet>> = non_merk_delete_paths + .iter() + .chain(merk_delete_paths.iter()) + .cloned() + .collect(); + ops.into_iter() + .filter(|op| { + if let GroveOp::DeleteTree(_) = &op.op + && let Some(key) = op.key.as_ref() + { + let mut child_path = op.path.to_path(); + child_path.push(key.as_slice().to_vec()); + return all_delete_paths.contains(&child_path); + } + true + }) + .collect() + } else { + ops + }; if batch_apply_options.batch_pause_height.is_none() { // we default to pausing at the root tree, which is the most common case batch_apply_options.batch_pause_height = Some(1); @@ -3613,6 +3896,35 @@ impl GroveDb { ); } + // Clean up storage for deleted standard Merk subtrees (same as + // apply_batch_with_element_flags_update). + for child_path in &merk_delete_paths { + let child_subtree_path: SubtreePath> = child_path.as_slice().into(); + let subtrees_paths = cost_return_on_error!( + &mut cost, + self.find_subtrees(&child_subtree_path, Some(tx.as_ref()), grove_version) + ); + for subtree_path in subtrees_paths { + let p: SubtreePath<_> = subtree_path.as_slice().into(); + let mut storage = self + .db + .get_transactional_storage_context( + p, + Some(&continue_storage_batch), + tx.as_ref(), + ) + .unwrap_add_cost(&mut cost); + cost_return_on_error!( + &mut cost, + storage.clear().map_err(|e| { + Error::CorruptedData(format!( + "unable to clean up merk subtree storage in batch delete: {e}", + )) + }) + ); + } + } + // let's build the write batch let continued_pending_costs = cost_return_on_error!( &mut cost, @@ -5059,14 +5371,20 @@ mod tests { .unwrap() .expect("insert commitment tree data"); - // Delete it via batch. + // Delete it via batch. The tree is non-empty (has one entry), + // so we must set allow_deleting_non_empty_trees. let ops = vec![QualifiedGroveDbOp::delete_tree_op( vec![], b"ct".to_vec(), grovedb_merk::tree_type::TreeType::CommitmentTree(4), )]; - db.apply_batch(ops, None, Some(&tx), grove_version) + let batch_options = Some(BatchApplyOptions { + allow_deleting_non_empty_trees: true, + ..Default::default() + }); + + db.apply_batch(ops, batch_options, Some(&tx), grove_version) .unwrap() .expect("batch delete non-merk tree"); @@ -5141,13 +5459,20 @@ mod tests { .expect("append mmr value"); } + // The tree is non-empty (has 3 entries), so we must set + // allow_deleting_non_empty_trees. let ops = vec![QualifiedGroveDbOp::delete_tree_op( vec![], b"mmr".to_vec(), grovedb_merk::tree_type::TreeType::MmrTree, )]; - db.apply_batch(ops, None, Some(&tx), grove_version) + let batch_options = Some(BatchApplyOptions { + allow_deleting_non_empty_trees: true, + ..Default::default() + }); + + db.apply_batch(ops, batch_options, Some(&tx), grove_version) .unwrap() .expect("batch delete mmr tree"); @@ -5213,15 +5538,22 @@ mod tests { .expect("insert dense tree value"); } + // The tree is non-empty (has 3 entries), so we must set + // allow_deleting_non_empty_trees. let ops = vec![QualifiedGroveDbOp::delete_tree_op( vec![], b"dense".to_vec(), grovedb_merk::tree_type::TreeType::DenseAppendOnlyFixedSizeTree(3), )]; + let batch_options = Some(BatchApplyOptions { + allow_deleting_non_empty_trees: true, + ..Default::default() + }); + db.apply_partial_batch( ops, - None, + batch_options, |_cost, _left_over_ops| Ok(vec![]), Some(&tx), grove_version, diff --git a/grovedb/src/tests/batch_delete_tree_tests.rs b/grovedb/src/tests/batch_delete_tree_tests.rs new file mode 100644 index 000000000..38ccf7180 --- /dev/null +++ b/grovedb/src/tests/batch_delete_tree_tests.rs @@ -0,0 +1,394 @@ +//! Tests for batch DeleteTree cleanup (H1) and emptiness check (H2). +//! +//! H1: Batch DeleteTree for standard Merk trees must clean up child subtree +//! storage, not just remove the parent key. +//! +//! H2: Batch DeleteTree must consult `allow_deleting_non_empty_trees` before +//! deleting a non-empty tree. + +#[cfg(feature = "minimal")] +mod tests { + use grovedb_merk::tree_type::TreeType; + use grovedb_version::version::GroveVersion; + + use crate::{ + batch::{BatchApplyOptions, QualifiedGroveDbOp}, + tests::{common::EMPTY_PATH, make_empty_grovedb}, + Element, Error, + }; + + // =================================================================== + // H2: Batch DeleteTree should respect allow_deleting_non_empty_trees + // =================================================================== + + #[test] + fn test_batch_delete_tree_non_empty_should_fail_when_not_allowed() { + // When `allow_deleting_non_empty_trees` is false (default) and + // `deleting_non_empty_trees_returns_error` is true (default), + // deleting a non-empty tree via batch should return an error. + let grove_version = GroveVersion::latest(); + let db = make_empty_grovedb(); + + // Insert a tree at root level + db.insert( + EMPTY_PATH, + b"parent_tree", + Element::empty_tree(), + None, + None, + grove_version, + ) + .unwrap() + .expect("insert parent tree"); + + // Insert an item inside the tree to make it non-empty + db.insert( + [b"parent_tree".as_ref()].as_ref(), + b"child_item", + Element::new_item(b"value".to_vec()), + None, + None, + grove_version, + ) + .unwrap() + .expect("insert child item"); + + // Try to delete the non-empty tree via batch with default options + // (allow_deleting_non_empty_trees: false, deleting_non_empty_trees_returns_error: true) + let ops = vec![QualifiedGroveDbOp::delete_tree_op( + vec![], + b"parent_tree".to_vec(), + TreeType::NormalTree, + )]; + + let batch_options = Some(BatchApplyOptions { + allow_deleting_non_empty_trees: false, + deleting_non_empty_trees_returns_error: true, + ..Default::default() + }); + + let result = db + .apply_batch(ops, batch_options, None, grove_version) + .unwrap(); + + assert!( + result.is_err(), + "batch DeleteTree on a non-empty tree with allow_deleting_non_empty_trees=false \ + should fail, but got: {:?}", + result, + ); + match result { + Err(Error::DeletingNonEmptyTree(_)) => { /* expected */ } + Err(e) => panic!("expected DeletingNonEmptyTree error, got: {:?}", e), + Ok(()) => panic!("expected error but got Ok"), + } + } + + #[test] + fn test_batch_delete_tree_non_empty_succeeds_when_allowed() { + // When `allow_deleting_non_empty_trees` is true, + // deleting a non-empty tree via batch should succeed. + let grove_version = GroveVersion::latest(); + let db = make_empty_grovedb(); + + // Insert a tree at root level + db.insert( + EMPTY_PATH, + b"parent_tree", + Element::empty_tree(), + None, + None, + grove_version, + ) + .unwrap() + .expect("insert parent tree"); + + // Insert an item inside the tree to make it non-empty + db.insert( + [b"parent_tree".as_ref()].as_ref(), + b"child_item", + Element::new_item(b"value".to_vec()), + None, + None, + grove_version, + ) + .unwrap() + .expect("insert child item"); + + let ops = vec![QualifiedGroveDbOp::delete_tree_op( + vec![], + b"parent_tree".to_vec(), + TreeType::NormalTree, + )]; + + let batch_options = Some(BatchApplyOptions { + allow_deleting_non_empty_trees: true, + deleting_non_empty_trees_returns_error: true, + ..Default::default() + }); + + db.apply_batch(ops, batch_options, None, grove_version) + .unwrap() + .expect("batch delete of non-empty tree should succeed when allowed"); + + // Verify tree is gone + let result = db + .get(EMPTY_PATH, b"parent_tree", None, grove_version) + .unwrap(); + assert!(result.is_err(), "parent tree should have been deleted"); + } + + #[test] + fn test_batch_delete_empty_tree_succeeds() { + // Deleting an empty tree should always succeed regardless of options. + let grove_version = GroveVersion::latest(); + let db = make_empty_grovedb(); + + // Insert an empty tree at root level + db.insert( + EMPTY_PATH, + b"empty_tree", + Element::empty_tree(), + None, + None, + grove_version, + ) + .unwrap() + .expect("insert empty tree"); + + let ops = vec![QualifiedGroveDbOp::delete_tree_op( + vec![], + b"empty_tree".to_vec(), + TreeType::NormalTree, + )]; + + let batch_options = Some(BatchApplyOptions { + allow_deleting_non_empty_trees: false, + deleting_non_empty_trees_returns_error: true, + ..Default::default() + }); + + db.apply_batch(ops, batch_options, None, grove_version) + .unwrap() + .expect("batch delete of empty tree should succeed"); + + // Verify tree is gone + let result = db + .get(EMPTY_PATH, b"empty_tree", None, grove_version) + .unwrap(); + assert!(result.is_err(), "empty tree should have been deleted"); + } + + // =================================================================== + // H1: Batch DeleteTree must clean up child subtree storage + // =================================================================== + + #[test] + fn test_batch_delete_tree_cleans_up_subtree_storage() { + // When we delete a tree that has nested subtrees via batch, + // the storage for those subtrees should be cleaned up. + // Then inserting a new tree at the same path should work without + // corruption from stale data. + let grove_version = GroveVersion::latest(); + let db = make_empty_grovedb(); + + // Step 1: Create a tree with nested subtrees + db.insert( + EMPTY_PATH, + b"outer", + Element::empty_tree(), + None, + None, + grove_version, + ) + .unwrap() + .expect("insert outer tree"); + + db.insert( + [b"outer".as_ref()].as_ref(), + b"inner", + Element::empty_tree(), + None, + None, + grove_version, + ) + .unwrap() + .expect("insert inner tree"); + + db.insert( + [b"outer".as_ref(), b"inner".as_ref()].as_ref(), + b"item1", + Element::new_item(b"old_value".to_vec()), + None, + None, + grove_version, + ) + .unwrap() + .expect("insert item into inner tree"); + + // Step 2: Delete the outer tree via batch (with allow_deleting_non_empty_trees) + let ops = vec![QualifiedGroveDbOp::delete_tree_op( + vec![], + b"outer".to_vec(), + TreeType::NormalTree, + )]; + + let batch_options = Some(BatchApplyOptions { + allow_deleting_non_empty_trees: true, + deleting_non_empty_trees_returns_error: true, + ..Default::default() + }); + + db.apply_batch(ops, batch_options.clone(), None, grove_version) + .unwrap() + .expect("batch delete tree should succeed"); + + // Verify that outer tree is gone + let outer_get = db.get(EMPTY_PATH, b"outer", None, grove_version).unwrap(); + assert!(outer_get.is_err(), "outer tree should have been deleted"); + + // Step 3: Insert a new tree at the same path + db.insert( + EMPTY_PATH, + b"outer", + Element::empty_tree(), + None, + None, + grove_version, + ) + .unwrap() + .expect("insert new outer tree at same path"); + + // Step 4: The new tree should be empty (no stale data from old inner tree) + db.insert( + [b"outer".as_ref()].as_ref(), + b"new_inner", + Element::empty_tree(), + None, + None, + grove_version, + ) + .unwrap() + .expect("insert new inner tree"); + + db.insert( + [b"outer".as_ref(), b"new_inner".as_ref()].as_ref(), + b"new_item", + Element::new_item(b"new_value".to_vec()), + None, + None, + grove_version, + ) + .unwrap() + .expect("insert item into new inner tree"); + + // Verify the new data is correct + let result = db + .get( + [b"outer".as_ref(), b"new_inner".as_ref()].as_ref(), + b"new_item", + None, + grove_version, + ) + .unwrap() + .expect("get new item"); + assert_eq!(result, Element::new_item(b"new_value".to_vec())); + + // The old item should NOT be accessible at the old path + // (the inner tree no longer exists, so this path is invalid) + let old_result = db + .get( + [b"outer".as_ref(), b"inner".as_ref()].as_ref(), + b"item1", + None, + grove_version, + ) + .unwrap(); + assert!( + old_result.is_err(), + "old inner tree data should not be accessible after delete and re-insert" + ); + } + + #[test] + fn test_batch_delete_tree_then_reinsert_produces_clean_tree() { + // This test verifies that after batch-deleting a tree with children, + // re-inserting at the same path produces a genuinely empty tree + // with no leftover data from previous children. + let grove_version = GroveVersion::latest(); + let db = make_empty_grovedb(); + + // Create structure: root -> parent -> child (with item) + let ops = vec![ + QualifiedGroveDbOp::insert_or_replace_op( + vec![], + b"parent".to_vec(), + Element::empty_tree(), + ), + QualifiedGroveDbOp::insert_or_replace_op( + vec![b"parent".to_vec()], + b"child".to_vec(), + Element::empty_tree(), + ), + QualifiedGroveDbOp::insert_or_replace_op( + vec![b"parent".to_vec(), b"child".to_vec()], + b"key1".to_vec(), + Element::new_item(b"data".to_vec()), + ), + ]; + + db.apply_batch(ops, None, None, grove_version) + .unwrap() + .expect("batch insert tree structure"); + + // Verify structure was created + let val = db + .get( + [b"parent".as_ref(), b"child".as_ref()].as_ref(), + b"key1", + None, + grove_version, + ) + .unwrap() + .expect("verify item exists"); + assert_eq!(val, Element::new_item(b"data".to_vec())); + + // Delete the parent tree via batch + let delete_ops = vec![QualifiedGroveDbOp::delete_tree_op( + vec![], + b"parent".to_vec(), + TreeType::NormalTree, + )]; + + let batch_options = Some(BatchApplyOptions { + allow_deleting_non_empty_trees: true, + ..Default::default() + }); + + db.apply_batch(delete_ops, batch_options, None, grove_version) + .unwrap() + .expect("batch delete parent tree"); + + // Re-insert parent as empty tree + db.insert( + EMPTY_PATH, + b"parent", + Element::empty_tree(), + None, + None, + grove_version, + ) + .unwrap() + .expect("re-insert parent tree"); + + // Verify the re-inserted tree is truly empty by checking + // that the old child subtree does not leak through + let child_get = db + .get([b"parent".as_ref()].as_ref(), b"child", None, grove_version) + .unwrap(); + assert!( + child_get.is_err(), + "re-inserted tree should not contain old child subtree data" + ); + } +} diff --git a/grovedb/src/tests/mod.rs b/grovedb/src/tests/mod.rs index 95ba8979f..28a2893b4 100644 --- a/grovedb/src/tests/mod.rs +++ b/grovedb/src/tests/mod.rs @@ -7,6 +7,7 @@ mod query_tests; mod sum_tree_tests; mod batch_coverage_tests; +mod batch_delete_tree_tests; mod batch_rejection_tests; mod batch_unit_tests; mod bulk_append_tree_tests; From e581ce6cdf6d3e33cfe4d463edc7e64dfb959caf Mon Sep 17 00:00:00 2001 From: Quantum Explorer Date: Sun, 8 Mar 2026 19:55:44 +0700 Subject: [PATCH 2/2] test: add coverage for batch DeleteTree cleanup, emptiness checks, and InsertIfNotExists Add 19 tests covering previously uncovered code paths in the batch processing changes: - Batch DeleteTree skip mode (allow=false, error=false): verify non-empty tree deletion is silently skipped while the tree remains intact - Batch DeleteTree with simultaneous child deletion: test that is_empty_tree_except correctly accounts for children being deleted in the same batch - Batch DeleteTree with partial child deletion: verify tree is still considered non-empty when only some children are deleted - Partial batch (apply_partial_batch) emptiness checks: error mode, skip mode, and allow mode with recursive Merk cleanup - Partial batch with simultaneous child deletion - Transactional batch emptiness check - Deeply nested subtree cleanup (3+ levels) - Skip mode mixed batch (skipped DeleteTree + successful DeleteTree + insert) - InsertIfNotExists: success for new key, error when key exists - InsertIfNotExists or skip: silently skip existing key - InsertWithKnownToNotAlreadyExist: success for new key - InsertIfNotExists through apply_batch_with_element_flags_update path - Debug format coverage for new InsertIfNotExists/InsertWithKnownToNotAlreadyExist variants Co-Authored-By: Claude Opus 4.6 --- grovedb/src/batch/mod.rs | 329 ++++++++ grovedb/src/tests/batch_delete_tree_tests.rs | 759 +++++++++++++++++++ 2 files changed, 1088 insertions(+) diff --git a/grovedb/src/batch/mod.rs b/grovedb/src/batch/mod.rs index c58534775..dcad3bbcd 100644 --- a/grovedb/src/batch/mod.rs +++ b/grovedb/src/batch/mod.rs @@ -5601,6 +5601,335 @@ mod tests { } } + // =================================================================== + // InsertIfNotExists and InsertWithKnownToNotAlreadyExist tests + // =================================================================== + + #[test] + fn test_batch_insert_if_not_exists_succeeds_for_new_key() { + // InsertIfNotExists should succeed when the key does not exist. + let grove_version = GroveVersion::latest(); + let db = make_test_grovedb(grove_version); + + let ops = vec![QualifiedGroveDbOp::insert_if_not_exists_op( + vec![TEST_LEAF.to_vec()], + b"new_key".to_vec(), + Element::new_item(b"value".to_vec()), + )]; + + db.apply_batch(ops, None, None, grove_version) + .unwrap() + .expect("insert_if_not_exists should succeed for new key"); + + let result = db + .get([TEST_LEAF].as_ref(), b"new_key", None, grove_version) + .unwrap() + .expect("get inserted item"); + assert_eq!(result, Element::new_item(b"value".to_vec())); + } + + #[test] + fn test_batch_insert_if_not_exists_errors_when_key_exists() { + // InsertIfNotExists (with error_if_exists=true) should fail when key exists. + let grove_version = GroveVersion::latest(); + let db = make_test_grovedb(grove_version); + + // Insert an item first + db.insert( + [TEST_LEAF].as_ref(), + b"existing", + Element::new_item(b"original".to_vec()), + None, + None, + grove_version, + ) + .unwrap() + .expect("insert original"); + + // Try to insert at the same key with insert_if_not_exists + let ops = vec![QualifiedGroveDbOp::insert_if_not_exists_op( + vec![TEST_LEAF.to_vec()], + b"existing".to_vec(), + Element::new_item(b"new_value".to_vec()), + )]; + + let result = db.apply_batch(ops, None, None, grove_version).unwrap(); + + assert!( + result.is_err(), + "insert_if_not_exists should fail when key exists, got: {:?}", + result, + ); + + // Original value should be preserved + let val = db + .get([TEST_LEAF].as_ref(), b"existing", None, grove_version) + .unwrap() + .expect("get existing"); + assert_eq!(val, Element::new_item(b"original".to_vec())); + } + + #[test] + fn test_batch_insert_if_not_exists_or_skip_silently_skips() { + // InsertIfNotExists with error_if_exists=false should silently skip + // when the key already exists. + let grove_version = GroveVersion::latest(); + let db = make_test_grovedb(grove_version); + + // Insert an item first + db.insert( + [TEST_LEAF].as_ref(), + b"existing", + Element::new_item(b"original".to_vec()), + None, + None, + grove_version, + ) + .unwrap() + .expect("insert original"); + + // Use insert_if_not_exists_or_skip (error_if_exists=false) + let ops = vec![QualifiedGroveDbOp::insert_if_not_exists_or_skip_op( + vec![TEST_LEAF.to_vec()], + b"existing".to_vec(), + Element::new_item(b"new_value".to_vec()), + )]; + + db.apply_batch(ops, None, None, grove_version) + .unwrap() + .expect("insert_if_not_exists_or_skip should succeed (skip)"); + + // Original value should be preserved + let val = db + .get([TEST_LEAF].as_ref(), b"existing", None, grove_version) + .unwrap() + .expect("get existing"); + assert_eq!(val, Element::new_item(b"original".to_vec())); + } + + #[test] + fn test_batch_insert_only_known_to_not_already_exist() { + // InsertWithKnownToNotAlreadyExist should succeed for a new key. + let grove_version = GroveVersion::latest(); + let db = make_test_grovedb(grove_version); + + let ops = vec![ + QualifiedGroveDbOp::insert_only_known_to_not_already_exist_op( + vec![TEST_LEAF.to_vec()], + b"brand_new".to_vec(), + Element::new_item(b"data".to_vec()), + ), + ]; + + db.apply_batch(ops, None, None, grove_version) + .unwrap() + .expect("insert_only_known_to_not_already_exist should succeed"); + + let result = db + .get([TEST_LEAF].as_ref(), b"brand_new", None, grove_version) + .unwrap() + .expect("get inserted item"); + assert_eq!(result, Element::new_item(b"data".to_vec())); + } + + #[test] + fn test_batch_insert_if_not_exists_with_flags_update() { + // Test InsertIfNotExists through the apply_batch_with_element_flags_update + // code path (with element flags function). + let grove_version = GroveVersion::latest(); + let db = make_test_grovedb(grove_version); + let tx = db.start_transaction(); + + // Insert an item first + db.insert( + [TEST_LEAF].as_ref(), + b"flagged", + Element::new_item(b"original".to_vec()), + None, + Some(&tx), + grove_version, + ) + .unwrap() + .expect("insert original"); + + // Try insert_if_not_exists with element flags (uses the flags update path) + let ops = vec![QualifiedGroveDbOp::insert_if_not_exists_op( + vec![TEST_LEAF.to_vec()], + b"flagged".to_vec(), + Element::new_item(b"new_value".to_vec()), + )]; + + let batch_options = Some(BatchApplyOptions { + validate_insertion_does_not_override: false, + ..Default::default() + }); + + let result = db + .apply_batch_with_element_flags_update( + ops, + batch_options, + |_cost, _old_flags, _new_flags| Ok(false), + |_flags, _removed_key_bytes, _removed_value_bytes| { + Ok((NoStorageRemoval, NoStorageRemoval)) + }, + Some(&tx), + grove_version, + ) + .unwrap(); + + assert!( + result.is_err(), + "insert_if_not_exists via flags update should fail when key exists: {:?}", + result, + ); + + // Original should be preserved + let val = db + .get([TEST_LEAF].as_ref(), b"flagged", Some(&tx), grove_version) + .unwrap() + .expect("get existing"); + assert_eq!(val, Element::new_item(b"original".to_vec())); + } + + #[test] + fn test_batch_insert_if_not_exists_or_skip_with_flags_update() { + // Test InsertIfNotExists (skip mode) through the flags update path. + let grove_version = GroveVersion::latest(); + let db = make_test_grovedb(grove_version); + let tx = db.start_transaction(); + + // Insert an item first + db.insert( + [TEST_LEAF].as_ref(), + b"flagged2", + Element::new_item(b"original".to_vec()), + None, + Some(&tx), + grove_version, + ) + .unwrap() + .expect("insert original"); + + // Use insert_if_not_exists_or_skip with element flags + let ops = vec![QualifiedGroveDbOp::insert_if_not_exists_or_skip_op( + vec![TEST_LEAF.to_vec()], + b"flagged2".to_vec(), + Element::new_item(b"new_value".to_vec()), + )]; + + let batch_options = Some(BatchApplyOptions { + validate_insertion_does_not_override: false, + ..Default::default() + }); + + db.apply_batch_with_element_flags_update( + ops, + batch_options, + |_cost, _old_flags, _new_flags| Ok(false), + |_flags, _removed_key_bytes, _removed_value_bytes| { + Ok((NoStorageRemoval, NoStorageRemoval)) + }, + Some(&tx), + grove_version, + ) + .unwrap() + .expect("insert_if_not_exists_or_skip via flags update should succeed (skip)"); + + // Original should be preserved + let val = db + .get([TEST_LEAF].as_ref(), b"flagged2", Some(&tx), grove_version) + .unwrap() + .expect("get existing"); + assert_eq!(val, Element::new_item(b"original".to_vec())); + } + + #[test] + fn test_batch_insert_if_not_exists_new_key_with_flags_update() { + // Test InsertIfNotExists for a new key through the flags update path. + let grove_version = GroveVersion::latest(); + let db = make_test_grovedb(grove_version); + let tx = db.start_transaction(); + + let ops = vec![QualifiedGroveDbOp::insert_if_not_exists_op( + vec![TEST_LEAF.to_vec()], + b"new_flagged".to_vec(), + Element::new_item(b"fresh".to_vec()), + )]; + + let batch_options = Some(BatchApplyOptions { + validate_insertion_does_not_override: false, + ..Default::default() + }); + + db.apply_batch_with_element_flags_update( + ops, + batch_options, + |_cost, _old_flags, _new_flags| Ok(false), + |_flags, _removed_key_bytes, _removed_value_bytes| { + Ok((NoStorageRemoval, NoStorageRemoval)) + }, + Some(&tx), + grove_version, + ) + .unwrap() + .expect("insert_if_not_exists should succeed for new key via flags update"); + + let val = db + .get( + [TEST_LEAF].as_ref(), + b"new_flagged", + Some(&tx), + grove_version, + ) + .unwrap() + .expect("get new item"); + assert_eq!(val, Element::new_item(b"fresh".to_vec())); + } + + // =================================================================== + // Debug formatting tests for new op variants + // =================================================================== + + #[test] + fn test_debug_format_insert_if_not_exists_ops() { + // Verify Debug formatting covers the new InsertIfNotExists variants + let op_error = QualifiedGroveDbOp::insert_if_not_exists_op( + vec![b"path".to_vec()], + b"key".to_vec(), + Element::new_item(b"val".to_vec()), + ); + let debug_str = format!("{:?}", op_error); + assert!( + debug_str.contains("Insert If Not Exists (error on existing)"), + "unexpected debug format: {}", + debug_str, + ); + + let op_skip = QualifiedGroveDbOp::insert_if_not_exists_or_skip_op( + vec![b"path".to_vec()], + b"key".to_vec(), + Element::new_item(b"val".to_vec()), + ); + let debug_str = format!("{:?}", op_skip); + assert!( + debug_str.contains("Insert If Not Exists (skip on existing)"), + "unexpected debug format: {}", + debug_str, + ); + + let op_known = QualifiedGroveDbOp::insert_only_known_to_not_already_exist_op( + vec![b"path".to_vec()], + b"key".to_vec(), + Element::new_item(b"val".to_vec()), + ); + let debug_str = format!("{:?}", op_known); + assert!( + debug_str.contains("Insert With Known To Not Already Exist"), + "unexpected debug format: {}", + debug_str, + ); + } + #[test] fn test_batch_rejects_key_longer_than_255_bytes() { let grove_version = GroveVersion::latest(); diff --git a/grovedb/src/tests/batch_delete_tree_tests.rs b/grovedb/src/tests/batch_delete_tree_tests.rs index 38ccf7180..b8e3a7204 100644 --- a/grovedb/src/tests/batch_delete_tree_tests.rs +++ b/grovedb/src/tests/batch_delete_tree_tests.rs @@ -391,4 +391,763 @@ mod tests { "re-inserted tree should not contain old child subtree data" ); } + + // =================================================================== + // H2: Skip mode — deleting_non_empty_trees_returns_error = false + // =================================================================== + + #[test] + fn test_batch_delete_tree_non_empty_skip_mode_silently_skips() { + // When `allow_deleting_non_empty_trees` is false and + // `deleting_non_empty_trees_returns_error` is also false, + // the DeleteTree op for a non-empty tree should be silently skipped. + let grove_version = GroveVersion::latest(); + let db = make_empty_grovedb(); + + // Insert a tree with a child item to make it non-empty + db.insert( + EMPTY_PATH, + b"skip_tree", + Element::empty_tree(), + None, + None, + grove_version, + ) + .unwrap() + .expect("insert tree"); + + db.insert( + [b"skip_tree".as_ref()].as_ref(), + b"child", + Element::new_item(b"data".to_vec()), + None, + None, + grove_version, + ) + .unwrap() + .expect("insert child"); + + // Attempt to delete the non-empty tree in skip mode + let ops = vec![QualifiedGroveDbOp::delete_tree_op( + vec![], + b"skip_tree".to_vec(), + TreeType::NormalTree, + )]; + + let batch_options = Some(BatchApplyOptions { + allow_deleting_non_empty_trees: false, + deleting_non_empty_trees_returns_error: false, + ..Default::default() + }); + + // Should succeed (no error) but the tree should still exist + db.apply_batch(ops, batch_options, None, grove_version) + .unwrap() + .expect("skip mode should not return error"); + + // The tree should still exist because the delete was skipped + let result = db + .get(EMPTY_PATH, b"skip_tree", None, grove_version) + .unwrap(); + assert!( + result.is_ok(), + "tree should still exist after skip-mode DeleteTree on non-empty tree" + ); + + // Child should still be accessible + let child_result = db + .get( + [b"skip_tree".as_ref()].as_ref(), + b"child", + None, + grove_version, + ) + .unwrap(); + assert!( + child_result.is_ok(), + "child should still be accessible after skipped delete" + ); + } + + #[test] + fn test_batch_delete_tree_skip_mode_allows_empty_tree_deletion() { + // Even in skip mode, an empty tree should still be deleted. + let grove_version = GroveVersion::latest(); + let db = make_empty_grovedb(); + + db.insert( + EMPTY_PATH, + b"empty_tree", + Element::empty_tree(), + None, + None, + grove_version, + ) + .unwrap() + .expect("insert empty tree"); + + let ops = vec![QualifiedGroveDbOp::delete_tree_op( + vec![], + b"empty_tree".to_vec(), + TreeType::NormalTree, + )]; + + let batch_options = Some(BatchApplyOptions { + allow_deleting_non_empty_trees: false, + deleting_non_empty_trees_returns_error: false, + ..Default::default() + }); + + db.apply_batch(ops, batch_options, None, grove_version) + .unwrap() + .expect("empty tree deletion should succeed in skip mode"); + + let result = db + .get(EMPTY_PATH, b"empty_tree", None, grove_version) + .unwrap(); + assert!( + result.is_err(), + "empty tree should have been deleted even in skip mode" + ); + } + + // =================================================================== + // Batch DeleteTree with simultaneous child deletion (is_empty_tree_except) + // =================================================================== + + #[test] + fn test_batch_delete_tree_with_simultaneous_child_delete() { + // When the batch contains both a Delete of the child and a DeleteTree + // of the parent, the emptiness check should account for the child + // deletion (is_empty_tree_except). The tree should be considered + // empty because the child is also being deleted in the same batch. + let grove_version = GroveVersion::latest(); + let db = make_empty_grovedb(); + + // Create a tree with one child + db.insert( + EMPTY_PATH, + b"parent", + Element::empty_tree(), + None, + None, + grove_version, + ) + .unwrap() + .expect("insert parent"); + + db.insert( + [b"parent".as_ref()].as_ref(), + b"child", + Element::new_item(b"value".to_vec()), + None, + None, + grove_version, + ) + .unwrap() + .expect("insert child"); + + // Batch: delete the child item AND delete the parent tree + // The emptiness check should see the child as "being deleted" and + // consider the tree empty. + let ops = vec![ + QualifiedGroveDbOp::delete_op(vec![b"parent".to_vec()], b"child".to_vec()), + QualifiedGroveDbOp::delete_tree_op(vec![], b"parent".to_vec(), TreeType::NormalTree), + ]; + + let batch_options = Some(BatchApplyOptions { + allow_deleting_non_empty_trees: false, + deleting_non_empty_trees_returns_error: true, + ..Default::default() + }); + + // Should succeed: the child is being deleted in the same batch, + // so the tree is effectively empty. + db.apply_batch(ops, batch_options, None, grove_version) + .unwrap() + .expect("batch delete with simultaneous child deletion should succeed"); + + // Verify parent is gone + let result = db.get(EMPTY_PATH, b"parent", None, grove_version).unwrap(); + assert!(result.is_err(), "parent tree should have been deleted"); + } + + #[test] + fn test_batch_delete_tree_with_partial_child_delete_still_non_empty() { + // When only some children are deleted in the same batch, the tree + // should still be considered non-empty and the delete should fail. + let grove_version = GroveVersion::latest(); + let db = make_empty_grovedb(); + + // Create a tree with two children + db.insert( + EMPTY_PATH, + b"parent", + Element::empty_tree(), + None, + None, + grove_version, + ) + .unwrap() + .expect("insert parent"); + + db.insert( + [b"parent".as_ref()].as_ref(), + b"child1", + Element::new_item(b"value1".to_vec()), + None, + None, + grove_version, + ) + .unwrap() + .expect("insert child1"); + + db.insert( + [b"parent".as_ref()].as_ref(), + b"child2", + Element::new_item(b"value2".to_vec()), + None, + None, + grove_version, + ) + .unwrap() + .expect("insert child2"); + + // Only delete one child in the batch, then try to delete the parent tree. + // The tree still has child2, so it should fail. + let ops = vec![ + QualifiedGroveDbOp::delete_op(vec![b"parent".to_vec()], b"child1".to_vec()), + QualifiedGroveDbOp::delete_tree_op(vec![], b"parent".to_vec(), TreeType::NormalTree), + ]; + + let batch_options = Some(BatchApplyOptions { + allow_deleting_non_empty_trees: false, + deleting_non_empty_trees_returns_error: true, + ..Default::default() + }); + + let result = db + .apply_batch(ops, batch_options, None, grove_version) + .unwrap(); + + assert!( + result.is_err(), + "should fail when only some children are deleted: {:?}", + result, + ); + match result { + Err(Error::DeletingNonEmptyTree(_)) => { /* expected */ } + Err(e) => panic!("expected DeletingNonEmptyTree, got: {:?}", e), + Ok(()) => panic!("expected error but got Ok"), + } + } + + // =================================================================== + // Partial batch: emptiness checks and cleanup + // =================================================================== + + #[test] + fn test_partial_batch_delete_tree_non_empty_should_fail_when_not_allowed() { + // The partial batch path has its own copy of the emptiness check logic. + // Verify it also enforces non-empty tree deletion restrictions. + let grove_version = GroveVersion::latest(); + let db = make_empty_grovedb(); + let tx = db.start_transaction(); + + db.insert( + EMPTY_PATH, + b"tree_a", + Element::empty_tree(), + None, + Some(&tx), + grove_version, + ) + .unwrap() + .expect("insert tree"); + + db.insert( + [b"tree_a".as_ref()].as_ref(), + b"item", + Element::new_item(b"val".to_vec()), + None, + Some(&tx), + grove_version, + ) + .unwrap() + .expect("insert item"); + + let ops = vec![QualifiedGroveDbOp::delete_tree_op( + vec![], + b"tree_a".to_vec(), + TreeType::NormalTree, + )]; + + let batch_options = Some(BatchApplyOptions { + allow_deleting_non_empty_trees: false, + deleting_non_empty_trees_returns_error: true, + ..Default::default() + }); + + let result = db + .apply_partial_batch( + ops, + batch_options, + |_cost, _left_over_ops| Ok(vec![]), + Some(&tx), + grove_version, + ) + .unwrap(); + + assert!( + result.is_err(), + "partial batch should fail for non-empty tree: {:?}", + result, + ); + match result { + Err(Error::DeletingNonEmptyTree(_)) => { /* expected */ } + Err(e) => panic!("expected DeletingNonEmptyTree, got: {:?}", e), + Ok(()) => panic!("expected error but got Ok"), + } + } + + #[test] + fn test_partial_batch_delete_tree_non_empty_skip_mode() { + // Partial batch skip mode: non-empty tree should be skipped, not errored. + let grove_version = GroveVersion::latest(); + let db = make_empty_grovedb(); + let tx = db.start_transaction(); + + db.insert( + EMPTY_PATH, + b"tree_b", + Element::empty_tree(), + None, + Some(&tx), + grove_version, + ) + .unwrap() + .expect("insert tree"); + + db.insert( + [b"tree_b".as_ref()].as_ref(), + b"item", + Element::new_item(b"val".to_vec()), + None, + Some(&tx), + grove_version, + ) + .unwrap() + .expect("insert item"); + + let ops = vec![QualifiedGroveDbOp::delete_tree_op( + vec![], + b"tree_b".to_vec(), + TreeType::NormalTree, + )]; + + let batch_options = Some(BatchApplyOptions { + allow_deleting_non_empty_trees: false, + deleting_non_empty_trees_returns_error: false, + ..Default::default() + }); + + db.apply_partial_batch( + ops, + batch_options, + |_cost, _left_over_ops| Ok(vec![]), + Some(&tx), + grove_version, + ) + .unwrap() + .expect("partial batch skip mode should succeed"); + + // Tree should still exist + let result = db + .get(EMPTY_PATH, b"tree_b", Some(&tx), grove_version) + .unwrap(); + assert!( + result.is_ok(), + "tree should still exist after skipped partial batch delete" + ); + } + + #[test] + fn test_partial_batch_delete_tree_non_empty_succeeds_when_allowed() { + // Partial batch: deleting a non-empty tree should succeed when allowed, + // exercising the merk cleanup path in apply_partial_batch. + let grove_version = GroveVersion::latest(); + let db = make_empty_grovedb(); + let tx = db.start_transaction(); + + // Create a tree with a nested subtree + db.insert( + EMPTY_PATH, + b"outer", + Element::empty_tree(), + None, + Some(&tx), + grove_version, + ) + .unwrap() + .expect("insert outer"); + + db.insert( + [b"outer".as_ref()].as_ref(), + b"inner", + Element::empty_tree(), + None, + Some(&tx), + grove_version, + ) + .unwrap() + .expect("insert inner"); + + db.insert( + [b"outer".as_ref(), b"inner".as_ref()].as_ref(), + b"data", + Element::new_item(b"some_value".to_vec()), + None, + Some(&tx), + grove_version, + ) + .unwrap() + .expect("insert data"); + + let ops = vec![QualifiedGroveDbOp::delete_tree_op( + vec![], + b"outer".to_vec(), + TreeType::NormalTree, + )]; + + let batch_options = Some(BatchApplyOptions { + allow_deleting_non_empty_trees: true, + ..Default::default() + }); + + db.apply_partial_batch( + ops, + batch_options, + |_cost, _left_over_ops| Ok(vec![]), + Some(&tx), + grove_version, + ) + .unwrap() + .expect("partial batch delete non-empty tree"); + + // Tree should be gone + let result = db + .get(EMPTY_PATH, b"outer", Some(&tx), grove_version) + .unwrap(); + assert!(result.is_err(), "tree should have been deleted"); + + // Re-insert and verify clean state (no stale data) + db.insert( + EMPTY_PATH, + b"outer", + Element::empty_tree(), + None, + Some(&tx), + grove_version, + ) + .unwrap() + .expect("re-insert outer"); + + let inner_get = db + .get( + [b"outer".as_ref()].as_ref(), + b"inner", + Some(&tx), + grove_version, + ) + .unwrap(); + assert!( + inner_get.is_err(), + "re-inserted tree should be clean with no stale inner subtree" + ); + } + + #[test] + fn test_partial_batch_delete_tree_with_simultaneous_child_delete() { + // Partial batch version of the is_empty_tree_except test. + let grove_version = GroveVersion::latest(); + let db = make_empty_grovedb(); + let tx = db.start_transaction(); + + db.insert( + EMPTY_PATH, + b"parent", + Element::empty_tree(), + None, + Some(&tx), + grove_version, + ) + .unwrap() + .expect("insert parent"); + + db.insert( + [b"parent".as_ref()].as_ref(), + b"only_child", + Element::new_item(b"data".to_vec()), + None, + Some(&tx), + grove_version, + ) + .unwrap() + .expect("insert only child"); + + // Delete the only child and the parent tree in the same batch + let ops = vec![ + QualifiedGroveDbOp::delete_op(vec![b"parent".to_vec()], b"only_child".to_vec()), + QualifiedGroveDbOp::delete_tree_op(vec![], b"parent".to_vec(), TreeType::NormalTree), + ]; + + let batch_options = Some(BatchApplyOptions { + allow_deleting_non_empty_trees: false, + deleting_non_empty_trees_returns_error: true, + ..Default::default() + }); + + db.apply_partial_batch( + ops, + batch_options, + |_cost, _left_over_ops| Ok(vec![]), + Some(&tx), + grove_version, + ) + .unwrap() + .expect("partial batch: should succeed since child is also being deleted"); + + let result = db + .get(EMPTY_PATH, b"parent", Some(&tx), grove_version) + .unwrap(); + assert!(result.is_err(), "parent should have been deleted"); + } + + // =================================================================== + // Transactional batch: emptiness check + Merk cleanup + // =================================================================== + + #[test] + fn test_batch_delete_tree_non_empty_error_with_transaction() { + // Same as the non-transactional test but uses a transaction, + // exercising the apply_batch_with_element_flags_update code path + // which is slightly different (uses storage_batch). + let grove_version = GroveVersion::latest(); + let db = make_empty_grovedb(); + let tx = db.start_transaction(); + + db.insert( + EMPTY_PATH, + b"tree_tx", + Element::empty_tree(), + None, + Some(&tx), + grove_version, + ) + .unwrap() + .expect("insert tree"); + + db.insert( + [b"tree_tx".as_ref()].as_ref(), + b"item", + Element::new_item(b"val".to_vec()), + None, + Some(&tx), + grove_version, + ) + .unwrap() + .expect("insert item"); + + let ops = vec![QualifiedGroveDbOp::delete_tree_op( + vec![], + b"tree_tx".to_vec(), + TreeType::NormalTree, + )]; + + let batch_options = Some(BatchApplyOptions { + allow_deleting_non_empty_trees: false, + deleting_non_empty_trees_returns_error: true, + ..Default::default() + }); + + let result = db + .apply_batch(ops, batch_options, Some(&tx), grove_version) + .unwrap(); + + match result { + Err(Error::DeletingNonEmptyTree(_)) => { /* expected */ } + Err(e) => panic!("expected DeletingNonEmptyTree, got: {:?}", e), + Ok(()) => panic!("expected error but got Ok"), + } + } + + #[test] + fn test_batch_delete_tree_cleans_up_deeply_nested_subtrees() { + // Verify that recursive cleanup works for 3+ levels of nesting. + let grove_version = GroveVersion::latest(); + let db = make_empty_grovedb(); + + // Create: root -> level1 -> level2 -> level3 -> item + db.insert( + EMPTY_PATH, + b"l1", + Element::empty_tree(), + None, + None, + grove_version, + ) + .unwrap() + .expect("insert l1"); + db.insert( + [b"l1".as_ref()].as_ref(), + b"l2", + Element::empty_tree(), + None, + None, + grove_version, + ) + .unwrap() + .expect("insert l2"); + db.insert( + [b"l1".as_ref(), b"l2".as_ref()].as_ref(), + b"l3", + Element::empty_tree(), + None, + None, + grove_version, + ) + .unwrap() + .expect("insert l3"); + db.insert( + [b"l1".as_ref(), b"l2".as_ref(), b"l3".as_ref()].as_ref(), + b"item", + Element::new_item(b"deep_value".to_vec()), + None, + None, + grove_version, + ) + .unwrap() + .expect("insert deep item"); + + // Delete the top-level tree + let ops = vec![QualifiedGroveDbOp::delete_tree_op( + vec![], + b"l1".to_vec(), + TreeType::NormalTree, + )]; + + let batch_options = Some(BatchApplyOptions { + allow_deleting_non_empty_trees: true, + ..Default::default() + }); + + db.apply_batch(ops, batch_options, None, grove_version) + .unwrap() + .expect("delete deeply nested tree"); + + // Re-insert and verify clean state at all levels + db.insert( + EMPTY_PATH, + b"l1", + Element::empty_tree(), + None, + None, + grove_version, + ) + .unwrap() + .expect("re-insert l1"); + + // l2 should not exist in the re-inserted tree + let l2_get = db + .get([b"l1".as_ref()].as_ref(), b"l2", None, grove_version) + .unwrap(); + assert!(l2_get.is_err(), "l2 should not exist in re-inserted tree"); + } + + #[test] + fn test_batch_delete_tree_skip_mode_mixed_batch() { + // Batch with a DeleteTree of a non-empty tree (skipped) mixed with + // other operations that should still succeed. + let grove_version = GroveVersion::latest(); + let db = make_empty_grovedb(); + + // Create two trees: one non-empty, one empty + db.insert( + EMPTY_PATH, + b"non_empty", + Element::empty_tree(), + None, + None, + grove_version, + ) + .unwrap() + .expect("insert non_empty"); + db.insert( + [b"non_empty".as_ref()].as_ref(), + b"child", + Element::new_item(b"data".to_vec()), + None, + None, + grove_version, + ) + .unwrap() + .expect("insert child"); + + db.insert( + EMPTY_PATH, + b"empty_one", + Element::empty_tree(), + None, + None, + grove_version, + ) + .unwrap() + .expect("insert empty_one"); + + // Batch: delete non-empty tree (should be skipped) and delete empty tree + // (should succeed), plus insert an item + let ops = vec![ + QualifiedGroveDbOp::delete_tree_op(vec![], b"non_empty".to_vec(), TreeType::NormalTree), + QualifiedGroveDbOp::delete_tree_op(vec![], b"empty_one".to_vec(), TreeType::NormalTree), + QualifiedGroveDbOp::insert_or_replace_op( + vec![], + b"new_item".to_vec(), + Element::new_item(b"hello".to_vec()), + ), + ]; + + let batch_options = Some(BatchApplyOptions { + allow_deleting_non_empty_trees: false, + deleting_non_empty_trees_returns_error: false, + ..Default::default() + }); + + db.apply_batch(ops, batch_options, None, grove_version) + .unwrap() + .expect("mixed batch should succeed"); + + // non_empty tree should still exist (skipped) + assert!( + db.get(EMPTY_PATH, b"non_empty", None, grove_version) + .unwrap() + .is_ok(), + "non-empty tree should still exist" + ); + + // empty_one should be gone + assert!( + db.get(EMPTY_PATH, b"empty_one", None, grove_version) + .unwrap() + .is_err(), + "empty tree should have been deleted" + ); + + // new_item should exist + assert_eq!( + db.get(EMPTY_PATH, b"new_item", None, grove_version) + .unwrap() + .expect("get new_item"), + Element::new_item(b"hello".to_vec()), + ); + } }