fix: report the append-only family's write churn as replaced bytes, not new storage

grovedb-bulk-append-tree/Cargo.toml

@@ -27,6 +27,7 @@ grovedb-merkle-mountain-range = { version = "5.0.1", path = "../grovedb-merkle-m
 grovedb-dense-fixed-sized-merkle-tree = { version = "5.0.1", path = "../grovedb-dense-fixed-sized-merkle-tree", default-features = false }
 grovedb-query = { version = "5.0.1", path = "../grovedb-query" }
 grovedb-costs = { version = "5.0.1", path = "../costs" }
+integer-encoding = { workspace = true }
 grovedb-version = { version = "5.0.1", path = "../grovedb-version" }
 grovedb-storage = { version = "5.0.1", path = "../storage", optional = true }
 blake3 = { workspace = true }

grovedb-bulk-append-tree/src/cost/mod.rs

@@ -9,10 +9,107 @@
 mod v0;
 mod v1;
 
+use grovedb_costs::storage_cost::{key_value_cost::KeyValueStorageCost, StorageCost};
 use grovedb_version::{error::GroveVersionError, version::GroveVersion};
+use integer_encoding::VarInt;
 
 use crate::BulkAppendError;
 
+/// Per-entry framing the chunk blob adds over the raw entry bytes (length
+/// prefix and per-entry overhead), charged to each append as its amortized
+/// share of the blob under storage accounting v1 — so that a compacting
+/// append's blob put can be reported as replacement of bytes that were
+/// already paid for. Kept equal to the estimator's per-entry chunk overhead
+/// so the bound and the actual agree on what is pre-paid.
+pub const CHUNK_ENTRY_AMORTIZED_BYTES: u32 = 16;
+
+/// Which storage-accounting report the append-only writes use; see
+/// `BulkAppendTreeCostVersions::storage_accounting`.
+pub(crate) fn storage_accounting_version(
+    grove_version: &GroveVersion,
+) -> Result<u16, BulkAppendError> {
+    match grove_version
+        .bulk_append_tree_versions
+        .cost
+        .storage_accounting
+    {
+        v @ (0 | 1) => Ok(v),
+        version => Err(BulkAppendError::VersionError(
+            GroveVersionError::UnknownVersionMismatch {
+                method: "BulkAppendTree storage accounting".to_string(),
+                known_versions: vec![0, 1],
+                received: version,
+            }
+            .to_string(),
+        )),
+    }
+}
+
+/// Cost info for writing one entry into the dense buffer under storage
+/// accounting v1: the entry's bytes (plus the value-length varint the
+/// storage layer would have counted) and its amortized share of the chunk
+/// blob's framing are charged as **added** — each entry's permanent bytes,
+/// paid once, by the append that creates it. `new_key` is whether this
+/// buffer position has never been written (first epoch); from the second
+/// epoch on the position key already exists and only the value is new.
+pub(crate) fn buffer_entry_cost_info(value_len: u32, new_key: bool) -> KeyValueStorageCost {
+    let added = entry_charge_bytes(value_len);
+    KeyValueStorageCost {
+        // The key's own bytes are appended by the storage context when
+        // `new_node` (it alone knows the prefix).
+        key_storage_cost: StorageCost::default(),
+        value_storage_cost: StorageCost {
+            added_bytes: added,
+            replaced_bytes: 0,
+            removed_bytes: Default::default(),
+        },
+        new_node: new_key,
+        needs_value_verification: false,
+    }
+}
+
+/// Cost info for writing the chunk blob a compaction produces, under
+/// storage accounting v1.
+///
+/// The blob supersedes the buffer entries it was built from; their bytes
+/// (entry + varint + amortized framing share) were already paid as added
+/// by the appends that wrote them — `pre_paid_bytes` — so they are
+/// reported as **replaced**. The value that triggered the compaction never
+/// entered the buffer (it goes straight into the blob), so this append
+/// pays for it here, as added, exactly as a buffered append would have
+/// (`compacting_entry_bytes`, the same entry + varint + framing share).
+/// Any residual the blob carries beyond the pre-paid bytes (its own
+/// header; nothing, when the compact fixed-size format undercuts the
+/// per-entry framing pre-payment) is added too. The blob's key is a new
+/// MMR position.
+pub(crate) fn chunk_blob_cost_info(
+    blob_len: u32,
+    pre_paid_bytes: u32,
+    compacting_entry_bytes: u32,
+) -> KeyValueStorageCost {
+    let total = blob_len.saturating_add(blob_len.required_space() as u32);
+    let replaced = total.min(pre_paid_bytes);
+    let residual = total.saturating_sub(pre_paid_bytes);
+    KeyValueStorageCost {
+        key_storage_cost: StorageCost::default(),
+        value_storage_cost: StorageCost {
+            added_bytes: compacting_entry_bytes.saturating_add(residual),
+            replaced_bytes: replaced,
+            removed_bytes: Default::default(),
+        },
+        new_node: true,
+        needs_value_verification: false,
+    }
+}
+
+/// The bytes an append is charged (as added) for one buffered entry:
+/// entry + value-length varint + its amortized chunk-framing share.
+pub(crate) fn entry_charge_bytes(value_len: u32) -> u32 {
+    value_len
+        .saturating_add(value_len.required_space() as u32)
+        .saturating_add(CHUNK_ENTRY_AMORTIZED_BYTES)
+}
+
 /// Hashes to report for a compacting append.
 ///
 /// `leaf_count` is the MMR leaf count BEFORE the push (what

grovedb-bulk-append-tree/src/tree/append.rs

@@ -1,5 +1,6 @@
 //! Append and compaction logic for BulkAppendTree.
 
+use grovedb_costs::storage_cost::key_value_cost::KeyValueStorageCost;
 use grovedb_costs::{CostResult, CostsExt, OperationCost};
 use grovedb_merkle_mountain_range::{mmr_size_to_leaf_count, MmrKeySize, MmrNode, MmrStore, MMR};
 use grovedb_storage::StorageContext;
@@ -9,7 +10,14 @@ use super::{
     capacity_for_height, hash::compute_state_root, AppendNoStateRootResult, AppendResult,
     BulkAppendTree,
 };
-use crate::{chunk::serialize_chunk_blob, cost::compaction_hash_count, BulkAppendError};
+use crate::{
+    chunk::serialize_chunk_blob,
+    cost::{
+        buffer_entry_cost_info, chunk_blob_cost_info, compaction_hash_count, entry_charge_bytes,
+        storage_accounting_version,
+    },
+    BulkAppendError,
+};
 
 impl<'db, S: StorageContext<'db>> BulkAppendTree<S> {
     /// Create a new empty tree.
@@ -23,6 +31,7 @@ impl<'db, S: StorageContext<'db>> BulkAppendTree<S> {
             total_count: 0,
             dense_tree,
             mmr_overlay: Vec::new(),
+            pending_blob_cost_infos: std::collections::BTreeMap::new(),
             // Empty tree → empty MMR → zero root.
             last_mmr_root: Some([0u8; 32]),
         })
@@ -49,6 +58,7 @@ impl<'db, S: StorageContext<'db>> BulkAppendTree<S> {
             total_count,
             dense_tree,
             mmr_overlay: Vec::new(),
+            pending_blob_cost_infos: std::collections::BTreeMap::new(),
             // Lazy: the restored MMR may not be readable until an append occurs,
             // so don't compute the root here. The first append fills the cache.
             last_mmr_root: None,
@@ -100,9 +110,14 @@ impl<'db, S: StorageContext<'db>> BulkAppendTree<S> {
         let global_position = self.total_count;
 
         // 1. Try to insert into the dense tree buffer.
-        let try_result = self.dense_tree.try_insert(value).unwrap().map_err(|e| {
-            BulkAppendError::StorageError(format!("dense tree insert failed: {}", e))
-        })?;
+        let cost_info = self.buffer_entry_cost_info(value, grove_version)?;
+        let try_result = self
+            .dense_tree
+            .try_insert_with_cost_info(value, cost_info)
+            .unwrap()
+            .map_err(|e| {
+                BulkAppendError::StorageError(format!("dense tree insert failed: {}", e))
+            })?;
 
         let compacted = match try_result {
             Some((_dense_root, _position)) => {
@@ -162,9 +177,13 @@ impl<'db, S: StorageContext<'db>> BulkAppendTree<S> {
         let mut cost = OperationCost::default();
         let global_position = self.total_count;
 
+        let cost_info = match self.buffer_entry_cost_info(value, grove_version) {
+            Ok(c) => c,
+            Err(e) => return Err(e).wrap_with_cost(cost),
+        };
         let try_result = match self
             .dense_tree
-            .try_insert_no_root(value)
+            .try_insert_no_root_with_cost_info(value, cost_info)
             .unwrap_add_cost(&mut cost)
         {
             Ok(r) => r,
@@ -282,6 +301,29 @@ impl<'db, S: StorageContext<'db>> BulkAppendTree<S> {
         Ok(compute_state_root(&mmr_root, &dense_root)).wrap_with_cost(cost)
     }
 
+    /// Storage cost info for the next buffer entry write, per the
+    /// version's storage accounting: `None` (v0) lets the storage layer
+    /// report key + value as added; v1 reports the entry's permanent bytes
+    /// plus its amortized blob-framing share as added, marking the
+    /// position key new only in the first epoch (later epochs overwrite a
+    /// stale slot at an existing key).
+    fn buffer_entry_cost_info(
+        &self,
+        value: &[u8],
+        grove_version: &GroveVersion,
+    ) -> Result<Option<KeyValueStorageCost>, BulkAppendError> {
+        match storage_accounting_version(grove_version)? {
+            0 => Ok(None),
+            _ => {
+                let first_epoch = self.total_count < self.epoch_size();
+                Ok(Some(buffer_entry_cost_info(
+                    value.len() as u32,
+                    first_epoch,
+                )))
+            }
+        }
+    }
+
     /// Compact all dense tree entries plus a new value into a chunk blob
     /// and append to the chunk MMR. Resets the dense tree.
     /// Returns `(hash_count, mmr_root)`.
@@ -341,6 +383,19 @@ impl<'db, S: StorageContext<'db>> BulkAppendTree<S> {
             entries.push(value);
         }
 
+        // Under storage accounting v1 the blob is reported as replacement of
+        // the buffer bytes it supersedes: every BUFFERED entry's bytes plus
+        // its amortized framing share were charged as added by the append
+        // that wrote it. The value that triggers the compaction never enters
+        // the buffer — it goes straight into the blob — so it is not
+        // pre-paid and stays in the blob's added residual, which is how this
+        // append pays for its own entry.
+        let pre_paid_bytes: u32 = entries
+            .iter()
+            .map(|e| entry_charge_bytes(e.len() as u32))
+            .fold(0u32, u32::saturating_add);
+        let compacting_entry_bytes = entry_charge_bytes(new_value.len() as u32);
+
         // Add the new value that didn't fit
         entries.push(new_value.to_vec());
 
@@ -349,6 +404,15 @@ impl<'db, S: StorageContext<'db>> BulkAppendTree<S> {
             Ok(b) => b,
             Err(e) => return Err(e).wrap_with_cost(cost),
         };
+        let blob_cost_info = match storage_accounting_version(grove_version) {
+            Ok(0) => None,
+            Ok(_) => Some(chunk_blob_cost_info(
+                blob.len() as u32,
+                pre_paid_bytes,
+                compacting_entry_bytes,
+            )),
+            Err(e) => return Err(e).wrap_with_cost(cost),
+        };
         // `MmrNode::leaf` hashes the blob eagerly.
         cost.hash_node_calls = cost.hash_node_calls.saturating_add(1);
         let leaf = MmrNode::leaf(blob);
@@ -370,11 +434,19 @@ impl<'db, S: StorageContext<'db>> BulkAppendTree<S> {
                 MMR::new_with_overlay(mmr_size, &mmr_store, std::mem::take(&mut self.mmr_overlay));
 
             let push_result = mmr.push(leaf, grove_version).unwrap_add_cost(&mut cost);
-            if let Err(e) = push_result {
-                // Restore overlay before returning error
-                self.mmr_overlay = mmr.batch.take_overlay();
-                return Err(BulkAppendError::MmrError(format!("MMR push failed: {}", e)))
-                    .wrap_with_cost(cost);
+            let leaf_pos = match push_result {
+                Ok(pos) => pos,
+                Err(e) => {
+                    // Restore overlay before returning error
+                    self.mmr_overlay = mmr.batch.take_overlay();
+                    return Err(BulkAppendError::MmrError(format!("MMR push failed: {}", e)))
+                        .wrap_with_cost(cost);
+                }
+            };
+            // The leaf is only staged here; it is written at `commit_mmr`,
+            // so its storage report travels with it until then.
+            if let Some(cost_info) = blob_cost_info {
+                self.pending_blob_cost_infos.insert(leaf_pos, cost_info);
             }
 
             let root_result = mmr.get_root(grove_version).unwrap_add_cost(&mut cost);
@@ -459,16 +531,19 @@ impl<'db, S: StorageContext<'db>> BulkAppendTree<S> {
         if self.mmr_overlay.is_empty() {
             return Ok(());
         }
-        let mmr_store = MmrStore::with_key_size(&self.dense_tree.storage, MmrKeySize::U32);
+        let mmr_store = MmrStore::with_key_size(&self.dense_tree.storage, MmrKeySize::U32)
+            .with_put_cost_infos(std::mem::take(&mut self.pending_blob_cost_infos));
         let mut mmr = MMR::new_with_overlay(
             self.mmr_size(),
             &mmr_store,
             std::mem::take(&mut self.mmr_overlay),
         );
         if let Err(e) = mmr.commit().unwrap() {
-            // Restore overlay before returning error so retries/get_mmr_root
-            // still see the staged nodes.
+            // Restore overlay (and the unconsumed storage reports) before
+            // returning error so retries/get_mmr_root still see the staged
+            // nodes.
             self.mmr_overlay = mmr.batch.take_overlay();
+            self.pending_blob_cost_infos = mmr_store.take_put_cost_infos();
             return Err(BulkAppendError::MmrError(format!(
                 "MMR commit failed: {}",
                 e

grovedb-bulk-append-tree/src/tree/mod.rs

@@ -20,6 +20,7 @@ pub use fetch::{BufferQueryResult, ChunkQueryResult};
 #[cfg(all(test, feature = "storage"))]
 mod tests;
 
+use grovedb_costs::storage_cost::key_value_cost::KeyValueStorageCost;
 use grovedb_dense_fixed_sized_merkle_tree::DenseFixedSizedMerkleTree;
 use grovedb_merkle_mountain_range::MmrNode;
 
@@ -92,6 +93,12 @@ pub struct BulkAppendTree<S> {
     /// lifetimes (compaction cycles) so that reads can find recently-pushed
     /// nodes without a storage round-trip.
     pub(crate) mmr_overlay: Vec<(u64, Vec<MmrNode>)>,
+    /// Storage cost info for chunk-blob leaves staged in `mmr_overlay` but
+    /// not yet committed, keyed by MMR position. Under storage accounting
+    /// v1 a compaction's blob is reported as replacement of the buffer
+    /// bytes it supersedes; since MMR nodes are written at `commit_mmr`,
+    /// the report travels with the staged node until then.
+    pub(crate) pending_blob_cost_infos: std::collections::BTreeMap<u64, KeyValueStorageCost>,
     /// Cached MMR root, refreshed only when a compaction mutates the MMR.
     ///
     /// The MMR is only touched on compaction (every `epoch_size` appends), so

grovedb-commitment-tree/Cargo.toml

@@ -22,6 +22,7 @@ incrementalmerkletree = "0.8"
 shardtree = { version = "0.6", optional = true }
 rusqlite = { version = "0.38", features = ["bundled"], optional = true }
 grovedb-costs = { version = "5.0.1", path = "../costs" }
+integer-encoding = { workspace = true }
 grovedb-version = { version = "5.0.1", path = "../grovedb-version" }
 grovedb-storage = { version = "5.0.1", path = "../storage", optional = true }
 grovedb-bulk-append-tree = { version = "5.0.1", path = "../grovedb-bulk-append-tree", default-features = false }