feat: replace indexed-tree placeholder rows with canonical references (#814)

docs/book/src/count-indexed-tree.md

@@ -152,19 +152,71 @@ The secondary Merk holds one entry per element in the primary, keyed by:
 
 ```text
 secondary_key = count_be_bytes(8) ‖ original_key
-secondary_val = ()        // empty; the original_key is encoded in the key
+secondary_val = ReferenceWithSumItem(
+                    SiblingReference(original_key),
+                    max_reference_hop = Some(1),
+                    sum = count_value,
+                )
 ```
 
 - **`count_be_bytes`** is the element's `count_value` encoded big-endian, 8
   bytes. Big-endian gives natural numeric order under lexicographic
   comparison, so right-to-left iteration yields highest-count-first.
 - **`original_key`** is appended to break ties among elements with equal
-  counts and to make each secondary key unique and reversible.
+  counts and to make each secondary key unique and reversible. It stays
+  in the key even though the row also names it: the key is what orders
+  and de-duplicates, and it has to be decodable on its own.
 
-The secondary Merk uses node feature type `ProvableCountedMerkNode(1)` —
-every entry contributes a count of `1`, so the aggregated count at the
-secondary's root equals the total number of indexed entries (which also
-equals the number of entries in the primary).
+<a id="secondary-rows"></a>
+
+### Secondary rows
+
+A row is a **canonical one-hop reference back to its primary entry**,
+written as a *combined* reference so the row's committed value hash is
+
+```text
+combine_hash(H(reference bytes), primary_node_committed_value_hash)
+```
+
+Three consequences worth stating plainly:
+
+- **Reads and proofs return the primary value.** Every non-aggregate
+  indexed read returns `IndexedAxisEntry { ordering_value, primary_key,
+  value }`, so a top-k result carries the values rather than pointers to
+  them — no follow-up `db.get` per row, and no extra inclusion proof per
+  row for a verified read. If the primary entry is itself a reference,
+  `value` is its TERMINAL, exactly as `db.get` on that key would give
+  you.
+
+  Callers that genuinely only rank (leaderboards, ranking views) can drop
+  the value with `IndexedAxisEntry::key_pair`.
+- **The binding is to the IMMEDIATE primary node**, not to a terminal
+  reached by following a chain. That keeps the invariant local: the only
+  thing that can staleness a row is a write to the primary entry itself,
+  which is exactly the event the mirror is driven by. This is dedicated
+  indexed-tree behaviour — ordinary GroveDB references keep their normal
+  terminal semantics, and an ordinary `max_hop = 1` reference pointing at
+  another reference remains ill-formed.
+- **Value-only updates now write.** Because the row binds a commitment,
+  an update that changes a primary entry's bytes without moving its
+  `count_value` still rewrites the row on every configured axis. So does
+  a deep mutation that only moves a child subtree's root. This write
+  amplification is intentional and is charged in the cost estimates.
+
+`SiblingReference` rather than an absolute path keeps a row's size
+independent of how deep the grove is. The reference is interpreted
+against the row's **logical origin** — the indexed primary's path — not
+against the derived storage prefix the secondary physically lives under.
+That prefix is `blake3(primary_prefix ‖ axis_tag)` and is not a GroveDB
+path at all, so resolution of a row's reference is purpose-built
+machinery rather than the ordinary path-keyed reference following.
+
+The secondary Merk uses node feature type
+`ProvableCountedAndProvableSummedMerkNode(1, count_value)` — every entry
+contributes a count of `1`, so the aggregated count at the secondary's
+root equals the total number of indexed entries (which also equals the
+number of entries in the primary), while the sum half makes a band TOTAL
+answerable as one committed scalar.
 
 The reason the secondary is a *provable* count tree (rather than the
 simpler `BasicMerkNode`) is that this lets the existing
@@ -427,7 +479,7 @@ Merk is not touched. The verifier receives the primary's root hash plus a
 
 ```rust
 // Shipped API on `GroveDb`:
-let entries: Vec<(u64, Vec<u8>)> = db
+let entries: Vec<IndexedAxisEntry<u64>> = db
     .indexed_count_top_k(path, k, /* descending: */ true, transaction, grove_version)?
     .expect("top-k");
 
@@ -436,14 +488,11 @@ let proof_bytes = db
     .prove_indexed_count_top_k(path, k, /* descending: */ true, transaction, grove_version)?
     .expect("prove");
 let result = GroveDb::verify_indexed_count_top_k(&proof_bytes, &path, k)?;
-// result.entries: Vec<(u64, Vec<u8>)>, result.root_hash: [u8; 32]
+// result.entries: Vec<IndexedAxisEntry<u64>>, result.root_hash: [u8; 32]
 ```
 
-The query returns `(count, key)` pairs. To resolve a primary value the
-caller follows up with `db.get(path, key, ...)`; the dedicated proof
-shape carries only the secondary range proof + a 32-byte attestation
-of the primary's root hash. Workloads that don't need values
-(leaderboards, ranking views) pay nothing for data they wouldn't read.
+The query returns `IndexedAxisEntry` rows — the count, the primary key,
+and the resolved primary value.
 
 Internally:
 
@@ -453,19 +502,32 @@ Internally:
 3. Run a **descending range query** with `limit = k` over the full
    secondary keyspace. This yields the k highest-count entries, with a
    standard Merk range proof.
-4. *(only if `resolve_values: true`)* For each `(c_be ‖ k)` in the
-   result, open the **primary** Merk and query for `k`. Each resolution
-   is one extra Merk read with one extra Merk inclusion proof.
-
-The default keeps the proof minimal: secondary range proof + a 32-byte
-attestation of the primary's root hash. Workloads that don't need the
-values (leaderboards, ranking views, "top N usernames") pay nothing for
-data they wouldn't read.
+4. Attach one **target chain** per returned row: the immediate primary
+   entry, then any ordinary reference hops through to the terminal. Each
+   chain entry carries its serialized bytes plus the rule that turns them
+   into a commitment (`Simple`, `Layered`, `IndexedSingle`,
+   `IndexedMulti`, `Reference`).
+
+A chain carries **no per-row path proofs**. It authenticates itself from
+the row's own committed hash: each entry's commitment is rebuilt from its
+bytes plus the next entry's, and the head's is what the row binds — and
+the row is bound into the secondary root, the indexed element, and the
+grove root. That is the same trust model shipped GroveDB reference proofs
+already use, so a chain is neither weaker nor stronger than reading the
+same reference through an ordinary proof. The practical effect is that a
+top-k result costs roughly one value plus one hash per row, instead of
+`k` inclusion proofs.
+
+The verifier also rebuilds the canonical row that the resolved primary
+value implies, and compares it against what the proof carried. That one
+comparison covers the ordering prefix, the primary-key suffix, the
+reference path, the hop budget and the carried sum — so a row filed under
+one key whose reference points at another cannot verify.
 
 ### Range by count
 
 ```rust
-let entries: Vec<(u64, Vec<u8>)> = db
+let entries: Vec<IndexedAxisEntry<u64>> = db
     .indexed_count_range(
         path,
         min,                       // u64, inclusive

grovedb/src/batch/indexed_tree/mirror.rs

@@ -22,9 +22,9 @@ use grovedb_merk::{
 use grovedb_storage::StorageContext;
 use grovedb_version::version::GroveVersion;
 
-use super::{read_entry_aggregates, AggregatePair, AggregateTransition};
+use super::{read_entry_aggregates, AggregateTransition, MaybeEntryState};
 use crate::{
-    operations::indexed_tree::{axis_row_payload, make_axis_secondary_key},
+    operations::indexed_tree::{axis_row_reference, make_axis_secondary_key},
     Element, Error,
 };
 
@@ -63,7 +63,7 @@ fn enforce_axis_item_key_bound<'a>(
 /// what makes each axis's assembled batch deterministic.
 pub(crate) fn read_post_apply_transitions<'db, S: StorageContext<'db>>(
     primary_merk: &Merk<S>,
-    pre: &BTreeMap<Vec<u8>, AggregatePair>,
+    pre: &BTreeMap<Vec<u8>, MaybeEntryState>,
     grove_version: &GroveVersion,
 ) -> CostResult<Vec<AggregateTransition>, Error> {
     let mut cost = OperationCost::default();
@@ -80,10 +80,18 @@ pub(crate) fn read_post_apply_transitions<'db, S: StorageContext<'db>>(
 
 /// Assemble one axis's secondary row moves into a sorted Merk batch.
 ///
-/// Each transition is compared as this axis's `(key, payload)` rather than
-/// the raw aggregates: on the avg axis two different `(count, sum)` pairs
-/// can share a sort key while carrying different payloads, and on the count
-/// axis a sum change moves nothing at all.
+/// Each transition is compared as this axis's `(key, row, target hash)`
+/// rather than the raw aggregates: on the avg axis two different
+/// `(count, sum)` pairs can share a sort key while carrying different
+/// payloads, on the count axis a sum change moves nothing at all, and a
+/// canonical reference row is stale whenever the primary node's committed
+/// value hash moves even if nothing about the sort position changed.
+///
+/// That last case is the intentional write amplification the reference
+/// representation buys: a value-only primary update — and equally a deep
+/// mutation that only changes a child subtree's root, or a
+/// `RefreshReference` on a reference-shaped primary — now rewrites every
+/// configured axis's row.
 fn build_axis_mirror_batch(
     transitions: &[AggregateTransition],
     axis: IndexAxis,
@@ -92,19 +100,20 @@ fn build_axis_mirror_batch(
     let mut cost = OperationCost::default();
     let secondary_tree_type = crate::operations::indexed_tree::axis_secondary_tree_type(axis);
     let mut secondary_batch: Vec<BatchEntry<Vec<u8>>> = Vec::with_capacity(transitions.len() * 2);
-    for (key, old_aggregates, new_aggregates) in transitions {
-        let entry_for = |aggregates: &AggregatePair| -> Result<_, Error> {
-            aggregates
-                .map(|(c, s)| {
+    for (key, old_state, new_state) in transitions {
+        let entry_for = |state: &MaybeEntryState| -> Result<_, Error> {
+            state
+                .map(|s| {
                     Ok((
-                        make_axis_secondary_key(axis, c, s, key),
-                        axis_row_payload(axis, c, s)?,
+                        make_axis_secondary_key(axis, s.count, s.sum, key),
+                        axis_row_reference(axis, key, s.count, s.sum)?,
+                        s.value_hash,
                     ))
                 })
                 .transpose()
         };
-        let old_entry = cost_return_on_error_no_add!(cost, entry_for(old_aggregates));
-        let new_entry = cost_return_on_error_no_add!(cost, entry_for(new_aggregates));
+        let old_entry = cost_return_on_error_no_add!(cost, entry_for(old_state));
+        let new_entry = cost_return_on_error_no_add!(cost, entry_for(new_state));
         if old_entry == new_entry {
             continue;
         }
@@ -115,8 +124,8 @@ fn build_axis_mirror_batch(
         // Merk batch is rejected outright ("Keys in batch must be unique"),
         // failing the whole GroveDB batch. Where the key is unchanged the put
         // alone overwrites the payload, which is what the row needs.
-        let new_secondary_key_ref = new_entry.as_ref().map(|(key, _)| key);
-        if let Some((old_secondary_key, _)) = &old_entry
+        let new_secondary_key_ref = new_entry.as_ref().map(|(key, ..)| key);
+        if let Some((old_secondary_key, ..)) = &old_entry
             && Some(old_secondary_key) != new_secondary_key_ref
         {
             cost_return_on_error!(
@@ -131,18 +140,26 @@ fn build_axis_mirror_batch(
                 .map_err(Error::MerkError)
             );
         }
-        if let Some((new_secondary_key, entry)) = new_entry {
+        if let Some((new_secondary_key, entry, target_value_hash)) = new_entry {
             let feature_type = cost_return_on_error_no_add!(
                 cost,
                 entry
                     .get_feature_type(secondary_tree_type)
                     .map_err(Error::MerkError)
             );
+            // `PutCombinedReference`, not a plain put: the row's committed
+            // value hash must be
+            // `combine_hash(H(reference bytes), target_value_hash)` so the
+            // secondary root binds the primary entry it points at. A plain
+            // put would commit only the reference bytes, leaving a row that
+            // authenticates its own path while saying nothing about the
+            // value at the other end of it.
             cost_return_on_error!(
                 &mut cost,
                 entry
-                    .insert_into_batch_operations(
+                    .insert_reference_into_batch_operations(
                         new_secondary_key,
+                        target_value_hash,
                         &mut secondary_batch,
                         feature_type,
                         grove_version,

grovedb/src/batch/indexed_tree/mod.rs

@@ -65,22 +65,30 @@ pub(crate) use preflight::reject_indexed_overwrite_with_descendants;
 
 use crate::{Element, Error};
 
-/// A primary entry's `(count, sum)` aggregate pair, `None` when the entry
-/// does not exist on that side of the transition.
-type AggregatePair = Option<(u64, i64)>;
+use crate::operations::indexed_tree::IndexedEntryState;
 
-/// One captured key's aggregate transition: `(item_key, old, new)`.
-type AggregateTransition = (Vec<u8>, AggregatePair, AggregatePair);
+/// A primary entry's state, `None` when the entry does not exist on that
+/// side of the transition.
+type MaybeEntryState = Option<IndexedEntryState>;
 
-/// Read one primary entry's current `(count, sum)` pair, `None` if the key
-/// does not exist. `phase` labels error messages ("pre" for the capture
-/// before the primary apply, "post" for the delta read after it).
+/// One captured key's transition: `(item_key, old, new)`.
+type AggregateTransition = (Vec<u8>, MaybeEntryState, MaybeEntryState);
+
+/// Read one primary entry's current state, `None` if the key does not
+/// exist. `phase` labels error messages ("pre" for the capture before the
+/// primary apply, "post" for the delta read after it).
+///
+/// Reads the element and the node's value hash. The value hash is taken
+/// from the Merk node rather than recomputed from the serialized bytes:
+/// for a tree-shaped or reference-shaped entry the stored hash is a
+/// combined hash that `value_hash(bytes)` cannot reproduce, and it is the
+/// stored one the row must bind.
 fn read_entry_aggregates<'db, S: StorageContext<'db>>(
     primary_merk: &Merk<S>,
     key: &[u8],
     phase: &str,
     grove_version: &GroveVersion,
-) -> CostResult<AggregatePair, Error> {
+) -> CostResult<MaybeEntryState, Error> {
     let mut cost = OperationCost::default();
     let maybe_bytes = cost_return_on_error!(
         &mut cost,
@@ -96,16 +104,40 @@ fn read_entry_aggregates<'db, S: StorageContext<'db>>(
                 hex::encode(key)
             )))
     );
-    let aggregates = if let Some(bytes) = maybe_bytes {
-        let elem = cost_return_on_error_no_add!(
-            cost,
-            Element::deserialize(bytes.as_slice(), grove_version).map_err(|e| {
-                Error::CorruptedData(format!("indexed {phase}-state deserialize: {e}"))
-            })
-        );
-        Some(elem.count_sum_value_or_default())
-    } else {
-        None
+    let Some(bytes) = maybe_bytes else {
+        return Ok(None).wrap_with_cost(cost);
     };
-    Ok(aggregates).wrap_with_cost(cost)
+    let elem = cost_return_on_error_no_add!(
+        cost,
+        Element::deserialize(bytes.as_slice(), grove_version)
+            .map_err(|e| Error::CorruptedData(format!("indexed {phase}-state deserialize: {e}")))
+    );
+    let value_hash = cost_return_on_error!(
+        &mut cost,
+        primary_merk
+            .get_value_hash(
+                key,
+                true,
+                Some(&Element::value_defined_cost_for_serialized_value),
+                grove_version,
+            )
+            .map_err(|e| Error::CorruptedData(format!(
+                "indexed {phase}-state value hash for key {}: {e}",
+                hex::encode(key)
+            )))
+    );
+    let value_hash = cost_return_on_error_no_add!(
+        cost,
+        value_hash.ok_or_else(|| Error::CorruptedData(format!(
+            "indexed {phase}-state: key {} has element bytes but no node value hash",
+            hex::encode(key)
+        )))
+    );
+    let (count, sum) = elem.count_sum_value_or_default();
+    Ok(Some(IndexedEntryState {
+        count,
+        sum,
+        value_hash,
+    }))
+    .wrap_with_cost(cost)
 }