fix+test: close verify_grovedb cidx content-consistency gap

Was the highest-priority audit item from my earlier self-grade: the
verify_grovedb H1-A walk verifies *chain* integrity but not *content*
consistency between the primary's count_value field and what the
secondary actually contains. The nested-cidx bug found in a8bb34f
was exactly this class — stale secondary that internally hashes
correctly but reports wrong counts. H1-A passed; queries lied.

Three changes:

1. CONTENT-CONSISTENCY CHECK in verify_grovedb. After the H1-A check
   on each cidx primary, walk both Merks' raw storage and assert
   per-entry consistency. Mismatches are recorded in the existing
   VerificationIssues HashMap with sentinel path suffixes
   (__cidx_primary_orphan__, __cidx_secondary_orphan__,
   __cidx_count_mismatch__, __cidx_secondary_malformed_key__) so the
   public API stays unchanged.

2. db.insert() REJECTS cidx primary targets. Adding the check
   revealed a real direct-path bug: db.insert(cidx_primary, ...)
   wrote to the primary without mirroring to the secondary, leaving
   the same kind of drift the new check catches. Route users to
   insert_into_count_indexed_tree with a NotSupported error.

3. DELIBERATE-CORRUPTION TESTS. Three tests directly manipulate the
   secondary's storage via Element::insert/delete to introduce each
   drift class:
     - verify_grovedb_catches_secondary_missing_entry_for_primary
     - verify_grovedb_catches_orphan_in_secondary
     - verify_grovedb_catches_count_mismatch_between_primary_and_
       secondary
   Plus direct_db_insert_into_cidx_primary_is_rejected covering
   the rejection from item 2.

Without item 1, all three corruption tests would silently pass an
integrity check. With it, the class of bug that took an audit to
find is now CI-caught for any future regression.

All 1576 grovedb tests pass.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Author: QuantumExplorer

grovedb/src/lib.rs

@@ -1305,6 +1305,99 @@ impl GroveDb {
                         );
                     }
 
+                    // Cidx content-consistency check.
+                    //
+                    // The H1-A check above verifies *chain* integrity —
+                    // the cidx element's recorded value_hash matches
+                    // `combine_hash_three(value_hash(bytes),
+                    // primary_root, secondary_root)`. That binds the
+                    // two Merks' root hashes into the parent's hash,
+                    // but it does NOT verify the secondary's contents
+                    // match what the primary says they should be.
+                    //
+                    // A bug like nested-cidx batch bubble-up forgetting
+                    // to mirror the count change (caught at audit in
+                    // a8bb34fb) leaves the secondary internally
+                    // consistent — its root hash is the correct hash
+                    // of its on-disk content — but its content is
+                    // stale relative to the primary. H1-A passes;
+                    // queries return wrong results.
+                    //
+                    // Walk both Merks here and assert per-entry
+                    // consistency: every primary entry with
+                    // count_value=c at key=k must correspond to
+                    // exactly one secondary entry at
+                    // (c.to_be_bytes() ‖ k). Each mismatch is recorded
+                    // in `issues` with a sentinel path suffix so the
+                    // existing `VerificationIssues` type doesn't need
+                    // to change.
+                    let mut primary_entries: HashMap<Vec<u8>, u64> = HashMap::new();
+                    let mut content_iter =
+                        KVIterator::new(primary_merk.storage.raw_iter(), &all_query).unwrap();
+                    while let Some((p_key, p_value)) = content_iter.next_kv().unwrap() {
+                        let p_elem = Element::raw_decode(&p_value, grove_version)?;
+                        primary_entries.insert(p_key, p_elem.count_value_or_default());
+                    }
+                    drop(content_iter);
+
+                    let mut secondary_entries: HashMap<Vec<u8>, u64> = HashMap::new();
+                    let mut sec_iter =
+                        KVIterator::new(secondary_merk.storage.raw_iter(), &all_query).unwrap();
+                    while let Some((sec_key, _sec_value)) = sec_iter.next_kv().unwrap() {
+                        // Secondary keys are (count_be ‖ original_key);
+                        // a malformed key short of 8 bytes is itself a
+                        // drift indicator.
+                        if sec_key.len() < 8 {
+                            let mut p = new_path.to_vec();
+                            p.push(b"__cidx_secondary_malformed_key__".to_vec());
+                            p.push(sec_key.clone());
+                            issues.insert(p, ([0u8; 32], [0u8; 32], [0u8; 32]));
+                            continue;
+                        }
+                        let mut count_bytes = [0u8; 8];
+                        count_bytes.copy_from_slice(&sec_key[..8]);
+                        let sec_count = u64::from_be_bytes(count_bytes);
+                        let original_key = sec_key[8..].to_vec();
+                        secondary_entries.insert(original_key, sec_count);
+                    }
+                    drop(sec_iter);
+
+                    // For each primary entry, the secondary must have
+                    // a matching entry at the same count_value.
+                    for (p_key, p_count) in &primary_entries {
+                        match secondary_entries.get(p_key) {
+                            None => {
+                                let mut p = new_path.to_vec();
+                                p.push(b"__cidx_primary_orphan__".to_vec());
+                                p.push(p_key.clone());
+                                issues.insert(p, ([0u8; 32], [0u8; 32], [0u8; 32]));
+                            }
+                            Some(s_count) if s_count != p_count => {
+                                let mut p = new_path.to_vec();
+                                p.push(b"__cidx_count_mismatch__".to_vec());
+                                p.push(p_key.clone());
+                                let mut expected = [0u8; 32];
+                                expected[24..32].copy_from_slice(&p_count.to_be_bytes());
+                                let mut actual = [0u8; 32];
+                                actual[24..32].copy_from_slice(&s_count.to_be_bytes());
+                                issues.insert(p, ([0u8; 32], expected, actual));
+                            }
+                            Some(_) => { /* matches */ }
+                        }
+                    }
+                    // For each secondary entry, the primary must have
+                    // a matching entry at that exact key (count match
+                    // is already checked above; here we look for
+                    // orphans in the secondary).
+                    for s_key in secondary_entries.keys() {
+                        if !primary_entries.contains_key(s_key) {
+                            let mut p = new_path.to_vec();
+                            p.push(b"__cidx_secondary_orphan__".to_vec());
+                            p.push(s_key.clone());
+                            issues.insert(p, ([0u8; 32], [0u8; 32], [0u8; 32]));
+                        }
+                    }
+
                     issues.extend(self.verify_merk_and_submerks_in_transaction(
                         primary_merk,
                         &new_path_ref,

grovedb/src/operations/insert/mod.rs

@@ -201,6 +201,30 @@ impl GroveDb {
                 grove_version
             )
         );
+
+        // Inserting an item directly into a CountIndexedTree primary
+        // via `db.insert()` would update the primary's content but
+        // leave the secondary stale — `add_element_on_transaction`
+        // has no secondary-mirror hook. The dedicated
+        // `insert_into_count_indexed_tree` exists precisely for this
+        // case and handles the dual-Merk write. Reject early with a
+        // pointer to the right API to prevent silent data drift.
+        // (References — which carry no count change at the cidx
+        // level — are still allowed; the cidx primary handles them via
+        // `insert_into_count_indexed_tree`'s reference resolution
+        // path, but the path used here goes through generic merk
+        // insertion only and does not need the cidx mirror.)
+        if subtree_to_insert_into.tree_type.is_count_indexed_primary() {
+            return Err(Error::NotSupported(
+                "direct `db.insert` into a CountIndexedTree primary is not supported \
+                 (the secondary index would not be mirrored). Use \
+                 `db.insert_into_count_indexed_tree(...)` for element insertion or \
+                 `db.delete_from_count_indexed_tree(...)` for removal"
+                    .to_string(),
+            ))
+            .wrap_with_cost(cost);
+        }
+
         // if we don't allow a tree override then we should check
 
         if options.checks_for_override() {