fix(bulk): bill the committed-slot read, bound BulkAppend estimates, validate chunk header (#822 follow-up)

docs/crates/costs.md

@@ -144,13 +144,16 @@ over an epoch matches the database growth. `removed_bytes` stays
 `NoStorageRemoval` in both — a rolling buffer refunds nobody. Stored bytes,
 roots and proofs are identical under both versions.
 
-Mechanically: the dense tree's `SlotWriteAccounting::AgainstCommitted`
-reads the slot's committed value before the write (the read is billed) and
-attaches `KeyValueStorageCost::for_in_place_value_rewrite`; the MMR
+Mechanically: `BulkAppendTree` reads the committed value of a slot that
+already holds one (judged against the count at open; never this session's
+cache) before rewriting it — one billed seek plus the value's bytes — and
+asks the dense tree for `SlotWriteAccounting::Overwrite { previous_value_len }`,
+which attaches `KeyValueStorageCost::for_in_place_value_rewrite`; the MMR
 `MmrStore` takes a `LeafValueStorageCost::PartlyPrepaid` policy that the
-bulk tree feeds `chunk_blob_entry_bytes`; `BulkAppendTree` appends report
-`prepaid_chunk_bytes` for the caller to bill (already included in the cost
-of the `CostResult`-returning `append_deferred_roots`).
+bulk tree feeds `chunk_blob_entry_bytes`; the `Result`-returning appends
+report a `storage_accounting_cost` (the prepaid share plus the slot read)
+for the caller to bill, while the `CostResult`-returning
+`append_deferred_roots` already includes it in its cost.
 
 ## Cost Context
 

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

@@ -22,21 +22,31 @@
 mod v0;
 mod v1;
 
-#[cfg(feature = "storage")]
-use grovedb_dense_fixed_sized_merkle_tree::SlotWriteAccounting;
 #[cfg(feature = "storage")]
 use grovedb_merkle_mountain_range::LeafValueStorageCost;
 use grovedb_version::{error::GroveVersionError, version::GroveVersion};
 
 use crate::BulkAppendError;
 
+/// How the dense-buffer slot write is sized for the storage cost layer.
+#[cfg(feature = "storage")]
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+pub(crate) enum SlotRewriteAccounting {
+    /// Every slot write is new storage (no cost information on the put).
+    AsNew,
+    /// A slot that holds a committed value is read first (the read is
+    /// billed) and the write is reported as its replacement; a slot written
+    /// for the first time stays new storage.
+    AgainstCommitted,
+}
+
 /// How an append's data-storage writes are reported to the storage cost
 /// layer. Selected per grove version by [`append_storage_accounting`].
 #[cfg(feature = "storage")]
 #[derive(Clone, Copy)]
 pub(crate) struct AppendStorageAccounting {
-    /// How the dense-buffer slot write is reported.
-    pub slot_write: SlotWriteAccounting,
+    /// How the dense-buffer slot write is sized.
+    pub slot_rewrite: SlotRewriteAccounting,
     /// How the chunk-blob leaf is reported when the MMR overlay is flushed.
     pub chunk_leaf: LeafValueStorageCost,
     /// Whether the entry's chunk-blob share is charged as added storage at

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

@@ -7,14 +7,12 @@
 //! Locked: GROVE_V1..V3 are released and CommitmentTree has been billing this
 //! figure on mainnet.
 
-#[cfg(feature = "storage")]
-use grovedb_dense_fixed_sized_merkle_tree::SlotWriteAccounting;
 use grovedb_merkle_mountain_range::hash_count_for_push;
 #[cfg(feature = "storage")]
 use grovedb_merkle_mountain_range::LeafValueStorageCost;
 
 #[cfg(feature = "storage")]
-use super::AppendStorageAccounting;
+use super::{AppendStorageAccounting, SlotRewriteAccounting};
 
 pub(super) fn compaction_hash_count(leaf_count: u64) -> u32 {
     hash_count_for_push(leaf_count)
@@ -30,7 +28,7 @@ pub(super) fn compaction_hash_count(leaf_count: u64) -> u32 {
 #[cfg(feature = "storage")]
 pub(super) fn append_storage_accounting() -> AppendStorageAccounting {
     AppendStorageAccounting {
-        slot_write: SlotWriteAccounting::AsNew,
+        slot_rewrite: SlotRewriteAccounting::AsNew,
         chunk_leaf: LeafValueStorageCost::New,
         prepay_chunk_share: false,
     }

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

@@ -2,14 +2,12 @@
 //!
 //! Adds the peak-bagging merges [`v0`](super::v0) omitted. Used from GROVE_V4.
 
-#[cfg(feature = "storage")]
-use grovedb_dense_fixed_sized_merkle_tree::SlotWriteAccounting;
 #[cfg(feature = "storage")]
 use grovedb_merkle_mountain_range::LeafValueStorageCost;
 use grovedb_merkle_mountain_range::{hash_count_for_push, hash_count_for_root_bagging};
 
 #[cfg(feature = "storage")]
-use super::AppendStorageAccounting;
+use super::{AppendStorageAccounting, SlotRewriteAccounting};
 #[cfg(feature = "storage")]
 use crate::chunk::chunk_blob_entry_bytes;
 
@@ -26,16 +24,17 @@ pub(super) fn compaction_hash_count(leaf_count: u64, mmr_size_after_push: u64) -
 /// - The entry's chunk-blob share (its own bytes) is charged as added
 ///   storage at its append — these are the bytes that persist.
 /// - The buffer slot write is sized against the value the slot already holds
-///   in committed storage: a rewrite (epoch 2 onward) is replaced, growth is
-///   added, shrink is not credited; a slot written for the first time stays
-///   fully added.
+///   in committed storage, which is read first (and the read billed): a
+///   rewrite (epoch 2 onward) is replaced, growth is added, shrink is not
+///   credited; a slot written for the first time stays fully added and is
+///   not read.
 /// - The compaction blob is reported as a replacement of the entry bytes it
 ///   supersedes (all prepaid), leaving only its framing — and the MMR
 ///   internal nodes — as added storage.
 #[cfg(feature = "storage")]
 pub(super) fn append_storage_accounting() -> AppendStorageAccounting {
     AppendStorageAccounting {
-        slot_write: SlotWriteAccounting::AgainstCommitted,
+        slot_rewrite: SlotRewriteAccounting::AgainstCommitted,
         chunk_leaf: LeafValueStorageCost::PartlyPrepaid(chunk_blob_entry_bytes),
         prepay_chunk_share: true,
     }

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

@@ -1,6 +1,7 @@
 //! Append and compaction logic for BulkAppendTree.
 
 use grovedb_costs::{CostResult, CostsExt, OperationCost};
+use grovedb_dense_fixed_sized_merkle_tree::{position_key, SlotWriteAccounting};
 use grovedb_merkle_mountain_range::{mmr_size_to_leaf_count, MmrKeySize, MmrNode, MmrStore, MMR};
 use grovedb_storage::StorageContext;
 use grovedb_version::version::GroveVersion;
@@ -11,7 +12,10 @@ use super::{
 };
 use crate::{
     chunk::serialize_chunk_blob,
-    cost::{append_storage_accounting, compaction_hash_count},
+    cost::{
+        append_storage_accounting, compaction_hash_count, AppendStorageAccounting,
+        SlotRewriteAccounting,
+    },
     BulkAppendError,
 };
 
@@ -29,6 +33,7 @@ impl<'db, S: StorageContext<'db>> BulkAppendTree<S> {
             mmr_overlay: Vec::new(),
             // Empty tree → empty MMR → zero root.
             last_mmr_root: Some([0u8; 32]),
+            committed_total_count: 0,
         })
     }
 
@@ -56,9 +61,61 @@ impl<'db, S: StorageContext<'db>> BulkAppendTree<S> {
             // 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,
+            committed_total_count: total_count,
         })
     }
 
+    /// Whether buffer slot `position` holds a value in committed storage:
+    /// every slot once a chunk has ever been completed (the buffer was full
+    /// when it compacted), otherwise the slots below the committed buffer
+    /// count. Judged against the state at open — see `committed_total_count`.
+    pub(crate) fn slot_is_committed(&self, position: u16) -> bool {
+        let epoch_size = self.epoch_size();
+        self.committed_total_count / epoch_size > 0
+            || (position as u64) < self.committed_total_count % epoch_size
+    }
+
+    /// Decide how the next buffer-slot write is reported, reading the slot's
+    /// committed value when the accounting sizes rewrites against it.
+    ///
+    /// The read — one seek, the committed value's bytes — is billed into
+    /// `cost`. It goes to the underlying storage (committed state plus the
+    /// surrounding transaction), never to the session's write-through cache:
+    /// a `StorageBatch` keeps one put per key, so the put that is eventually
+    /// charged must describe the transition from the committed value. A
+    /// slot written for the first time, and a full buffer (the append
+    /// compacts and writes no slot), are not read. A committed slot that
+    /// storage does not hold — corruption, not a state this code produces —
+    /// is charged as new, the safe direction.
+    fn slot_write_accounting(
+        &self,
+        accounting: &AppendStorageAccounting,
+        cost: &mut OperationCost,
+    ) -> Result<SlotWriteAccounting, BulkAppendError> {
+        if accounting.slot_rewrite == SlotRewriteAccounting::AsNew {
+            return Ok(SlotWriteAccounting::AsNew);
+        }
+        let position = self.dense_tree.count();
+        if position >= self.dense_tree.capacity() || !self.slot_is_committed(position) {
+            return Ok(SlotWriteAccounting::AsNew);
+        }
+        match self
+            .dense_tree
+            .storage
+            .get(position_key(position))
+            .unwrap_add_cost(cost)
+        {
+            Ok(Some(previous)) => Ok(SlotWriteAccounting::Overwrite {
+                previous_value_len: previous.len() as u32,
+            }),
+            Ok(None) => Ok(SlotWriteAccounting::AsNew),
+            Err(e) => Err(BulkAppendError::StorageError(format!(
+                "committed slot {} read before rewrite failed: {}",
+                position, e
+            ))),
+        }
+    }
+
     /// Append a value to the tree.
     ///
     /// Handles dense tree insert, auto-compaction when the buffer fills, and
@@ -78,7 +135,7 @@ impl<'db, S: StorageContext<'db>> BulkAppendTree<S> {
             // +1 for the blake3 state-root computation we just did.
             hash_count: r.hash_count.saturating_add(1),
             compacted: r.compacted,
-            prepaid_chunk_bytes: r.prepaid_chunk_bytes,
+            storage_accounting_cost: r.storage_accounting_cost,
         })
     }
 
@@ -97,7 +154,7 @@ impl<'db, S: StorageContext<'db>> BulkAppendTree<S> {
     /// version; what the version selects is the reported `hash_count` (for an
     /// append that compacts) and the storage accounting of the data writes —
     /// the cost information attached to the slot put, and the
-    /// `prepaid_chunk_bytes` the caller bills as added storage.
+    /// `storage_accounting_cost` the caller bills.
     pub fn append_no_state_root(
         &mut self,
         value: &[u8],
@@ -106,11 +163,13 @@ impl<'db, S: StorageContext<'db>> BulkAppendTree<S> {
         let mut hash_count: u32 = 0;
         let global_position = self.total_count;
         let accounting = append_storage_accounting(grove_version)?;
+        let mut storage_accounting_cost = OperationCost::default();
+        let slot_write = self.slot_write_accounting(&accounting, &mut storage_accounting_cost)?;
 
         // 1. Try to insert into the dense tree buffer.
         let try_result = self
             .dense_tree
-            .try_insert_with_accounting(value, accounting.slot_write)
+            .try_insert_with_accounting(value, slot_write)
             .unwrap()
             .map_err(|e| {
                 BulkAppendError::StorageError(format!("dense tree insert failed: {}", e))
@@ -142,11 +201,16 @@ impl<'db, S: StorageContext<'db>> BulkAppendTree<S> {
 
         self.total_count += 1;
 
+        storage_accounting_cost.storage_cost.added_bytes = storage_accounting_cost
+            .storage_cost
+            .added_bytes
+            .saturating_add(accounting.prepaid_chunk_bytes(value.len()));
+
         Ok(AppendNoStateRootResult {
             global_position,
             hash_count,
             compacted,
-            prepaid_chunk_bytes: accounting.prepaid_chunk_bytes(value.len()),
+            storage_accounting_cost,
         })
     }
 
@@ -178,10 +242,19 @@ impl<'db, S: StorageContext<'db>> BulkAppendTree<S> {
             Ok(a) => a,
             Err(e) => return Err(e).wrap_with_cost(cost),
         };
+        // The committed-slot read is billed here, in the returned cost, and
+        // mirrored (with the prepaid share) in the result for information.
+        let mut storage_accounting_cost = OperationCost::default();
+        let slot_write = self.slot_write_accounting(&accounting, &mut storage_accounting_cost);
+        cost += storage_accounting_cost.clone();
+        let slot_write = match slot_write {
+            Ok(s) => s,
+            Err(e) => return Err(e).wrap_with_cost(cost),
+        };
 
         let try_result = match self
             .dense_tree
-            .try_insert_no_root_with_accounting(value, accounting.slot_write)
+            .try_insert_no_root_with_accounting(value, slot_write)
             .unwrap_add_cost(&mut cost)
         {
             Ok(r) => r,
@@ -238,12 +311,16 @@ impl<'db, S: StorageContext<'db>> BulkAppendTree<S> {
             .storage_cost
             .added_bytes
             .saturating_add(prepaid_chunk_bytes);
+        storage_accounting_cost.storage_cost.added_bytes = storage_accounting_cost
+            .storage_cost
+            .added_bytes
+            .saturating_add(prepaid_chunk_bytes);
 
         Ok(AppendNoStateRootResult {
             global_position,
             hash_count,
             compacted,
-            prepaid_chunk_bytes,
+            storage_accounting_cost,
         })
         .wrap_with_cost(cost)
     }

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

@@ -22,6 +22,8 @@ mod storage_accounting_tests;
 #[cfg(all(test, feature = "storage"))]
 mod tests;
 
+#[cfg(feature = "storage")]
+use grovedb_costs::OperationCost;
 use grovedb_dense_fixed_sized_merkle_tree::DenseFixedSizedMerkleTree;
 use grovedb_merkle_mountain_range::MmrNode;
 
@@ -40,10 +42,9 @@ pub struct AppendResult {
     pub hash_count: u32,
     /// Whether compaction (epoch flush) occurred.
     pub compacted: bool,
-    /// The entry's chunk-blob share, which the caller bills as
-    /// `storage_cost.added_bytes`. See
-    /// [`AppendNoStateRootResult::prepaid_chunk_bytes`].
-    pub prepaid_chunk_bytes: u32,
+    /// The storage-accounting cost the caller bills. See
+    /// [`AppendNoStateRootResult::storage_accounting_cost`].
+    pub storage_accounting_cost: OperationCost,
 }
 
 /// Result returned by [`BulkAppendTree::append_no_state_root`].
@@ -52,7 +53,7 @@ pub struct AppendResult {
 /// once at the end of a batch via
 /// [`BulkAppendTree::compute_current_state_root`].
 #[cfg(feature = "storage")]
-#[derive(Debug, Clone, Copy)]
+#[derive(Debug, Clone)]
 pub struct AppendNoStateRootResult {
     /// The 0-based global position of the appended value.
     pub global_position: u64,
@@ -61,22 +62,28 @@ pub struct AppendNoStateRootResult {
     pub hash_count: u32,
     /// Whether compaction (epoch flush) occurred.
     pub compacted: bool,
-    /// The entry's share of the chunk blob it will eventually be compacted
-    /// into — its own bytes — to be charged as `storage_cost.added_bytes` at
-    /// this append, so that the blob written at compaction can be reported as
-    /// a replacement of bytes already paid for (issue #822). `0` under the
-    /// shipped accounting (GROVE_V1..V3), where the blob is charged in full
-    /// at compaction instead.
+    /// The cost of this append's storage accounting, which the
+    /// `Result`-returning appends cannot otherwise surface:
+    ///
+    /// - `storage_cost.added_bytes`: the entry's share of the chunk blob it
+    ///   will eventually be compacted into — its own bytes — charged at this
+    ///   append so that the blob written at compaction can be reported as a
+    ///   replacement of bytes already paid for (issue #822);
+    /// - `seek_count` / `storage_loaded_bytes`: the read of the committed
+    ///   value a buffer slot already holds (epoch 2 onward), performed to
+    ///   size the rewrite.
+    ///
+    /// Zero under the shipped accounting (GROVE_V1..V3), where the blob is
+    /// charged in full at compaction and no slot is read.
     ///
     /// Like `hash_count`, this follows the "caller bills" convention of the
     /// `Result`-returning appends ([`append`](BulkAppendTree::append),
-    /// [`append_no_state_root`](BulkAppendTree::append_no_state_root)):
-    /// add it to the operation's `storage_cost.added_bytes`. The
-    /// `CostResult`-returning
+    /// [`append_no_state_root`](BulkAppendTree::append_no_state_root)): add it
+    /// to the operation's `OperationCost`. The `CostResult`-returning
     /// [`append_deferred_roots`](BulkAppendTree::append_deferred_roots)
-    /// already includes it in the returned cost; there the field is a
-    /// mirror for information only — do not bill it twice.
-    pub prepaid_chunk_bytes: u32,
+    /// already includes it in the returned cost; there the field is a mirror
+    /// for information only — do not bill it twice.
+    pub storage_accounting_cost: OperationCost,
 }
 
 /// Compute MMR size from leaf count: `2 * n - popcount(n)`.
@@ -129,6 +136,15 @@ pub struct BulkAppendTree<S> {
     ///
     /// [`from_state`]: BulkAppendTree::from_state
     pub(crate) last_mmr_root: Option<[u8; 32]>,
+    /// `total_count` as of the open ([`new`] → 0, [`from_state`] → the
+    /// persisted count): what committed storage holds, which this session's
+    /// appends have not changed. Used to tell a buffer slot that holds a
+    /// committed value — whose rewrite is read and reported as a replacement
+    /// — from one written for the first time.
+    ///
+    /// [`new`]: BulkAppendTree::new
+    /// [`from_state`]: BulkAppendTree::from_state
+    pub(crate) committed_total_count: u64,
 }
 
 impl<S> BulkAppendTree<S> {