diff --git a/grovedb/src/lib.rs b/grovedb/src/lib.rs index a3f509a8a..484631764 100644 --- a/grovedb/src/lib.rs +++ b/grovedb/src/lib.rs @@ -311,12 +311,18 @@ type VerificationIssues = HashMap>, (CryptoHash, CryptoHash, CryptoH /// It represents a tuple containing: /// - A `Merk` instance with a prefixed RocksDB immediate storage context. /// - An optional `root_key`, represented as a vector of bytes. -/// - A boolean indicating whether the Merk is a sum tree. +/// - The `TreeType` of the subtree (with its parameters, e.g. chunk power, +/// preserved from the parent element). +/// - The parent-declared `Element` for this subtree, or `None` when opening +/// the root subtree (which has no parent element). Replication needs the +/// full element for non-Merk append-only trees, whose entry counts and +/// parameters drive the raw-replay restore path. #[cfg(feature = "minimal")] type OpenedMerkForReplication<'tx> = ( Merk>, Option>, TreeType, + Option, ); /// Verify that an indexed tree's secondary projection is exactly derivable @@ -549,7 +555,7 @@ impl GroveDb { )) }) .unwrap()?; - if let Some((root_key, tree_type)) = element.root_key_and_tree_type_owned() { + if let Some((root_key, tree_type)) = element.clone().root_key_and_tree_type_owned() { Ok(( Merk::open_layered_with_root_key( storage, @@ -566,6 +572,7 @@ impl GroveDb { .unwrap()?, root_key, tree_type, + Some(element), )) } else { Err(Error::CorruptedPath( @@ -584,6 +591,7 @@ impl GroveDb { .unwrap()?, None, TreeType::NormalTree, + None, )) } } @@ -2455,6 +2463,131 @@ impl GroveDb { _ => merk_root_hash, } } + + /// Strict variant of [`Self::compute_non_merk_child_hash`] for callers + /// that must not silently fall back when the payload is unreadable — + /// notably state-sync restore, where a missing or corrupt payload has to + /// reject the subtree rather than slip through as a hash that may + /// coincidentally match. + /// + /// Recomputes the tree-type-specific state root from the payload stored + /// in the subtree's data namespace: + /// - `CommitmentTree`: `blake3("ct_state" || sinsemilla_root || + /// bulk_state_root)` (reads the frontier and the bulk store) + /// - `BulkAppendTree`: `blake3("bulk_state" || mmr_root || dense_root)` + /// - `MmrTree`: the MMR root hash + /// - `DenseAppendOnlyFixedSizeTree`: the dense tree root hash + /// + /// For empty trees this returns the same conventions the insert path + /// binds into the parent: `EMPTY_COMMITMENT_TREE_STATE_ROOT` for an + /// empty commitment tree, `NULL_HASH` (the empty Merk root) for the + /// other three types. + /// + /// Returns an error if `element` is not a non-Merk data tree, or if the + /// payload cannot be read back as a consistent tree of the declared + /// size. + pub(crate) fn compute_non_merk_state_root<'b, B: AsRef<[u8]>>( + &self, + element: &Element, + subtree_path: SubtreePath<'b, B>, + transaction: &Transaction, + ) -> Result { + use grovedb_merk::tree::hash::NULL_HASH; + match element.underlying() { + Element::CommitmentTree(total_count, chunk_power, _) => { + if *total_count == 0 { + return Ok(grovedb_commitment_tree::EMPTY_COMMITMENT_TREE_STATE_ROOT); + } + let storage_ctx = self + .db + .get_transactional_storage_context(subtree_path, None, transaction) + .unwrap(); + let ct = grovedb_commitment_tree::CommitmentTree::<_>::open( + *total_count, + *chunk_power, + storage_ctx, + ) + .value + .map_err(|e| { + Error::CorruptedData(format!( + "cannot open commitment tree of {total_count} entries from payload: {e}" + )) + })?; + ct.compute_current_state_root().map_err(|e| { + Error::CorruptedData(format!( + "cannot compute commitment tree state root from payload: {e}" + )) + }) + } + Element::BulkAppendTree(total_count, chunk_power, _) => { + if *total_count == 0 { + return Ok(NULL_HASH); + } + let storage_ctx = self + .db + .get_transactional_storage_context(subtree_path, None, transaction) + .unwrap(); + let tree = grovedb_bulk_append_tree::BulkAppendTree::from_state( + *total_count, + *chunk_power, + storage_ctx, + ) + .map_err(|e| { + Error::CorruptedData(format!( + "cannot open bulk append tree of {total_count} entries from payload: {e}" + )) + })?; + tree.compute_current_state_root().map_err(|e| { + Error::CorruptedData(format!( + "cannot compute bulk append tree state root from payload: {e}" + )) + }) + } + Element::MmrTree(mmr_size, _) => { + if *mmr_size == 0 { + return Ok(NULL_HASH); + } + let storage_ctx = self + .db + .get_transactional_storage_context(subtree_path, None, transaction) + .unwrap(); + let store = grovedb_merkle_mountain_range::MmrStore::new(&storage_ctx); + let mmr = grovedb_merkle_mountain_range::MMR::new(*mmr_size, &store); + mmr.get_root().value.map(|root| root.hash()).map_err(|e| { + Error::CorruptedData(format!( + "cannot compute MMR root of size {mmr_size} from payload: {e}" + )) + }) + } + Element::DenseAppendOnlyFixedSizeTree(count, height, _) => { + if *count == 0 { + return Ok(NULL_HASH); + } + let storage_ctx = self + .db + .get_transactional_storage_context(subtree_path, None, transaction) + .unwrap(); + use grovedb_dense_fixed_sized_merkle_tree::DenseFixedSizedMerkleTree; + DenseFixedSizedMerkleTree::from_state(*height, *count, storage_ctx) + .map_err(|e| { + Error::CorruptedData(format!( + "cannot open dense tree of {count} entries from payload: {e}" + )) + })? + .root_hash() + .unwrap() + .map_err(|e| { + Error::CorruptedData(format!( + "cannot compute dense tree root from payload: {e}" + )) + }) + } + _ => Err(Error::InternalError(format!( + "compute_non_merk_state_root called on a non append-only element: {}", + element.type_str() + ))), + } + } } /// Inspect a tree-bearing Element together with the actual aggregate data of diff --git a/grovedb/src/replication.rs b/grovedb/src/replication.rs index cb6d104c0..b11cfa55f 100644 --- a/grovedb/src/replication.rs +++ b/grovedb/src/replication.rs @@ -1,3 +1,4 @@ +pub(crate) mod non_merk_sync; mod state_sync_session; use std::pin::Pin; @@ -93,6 +94,12 @@ impl GroveDb { /// - The function opens a `Merk` tree for each chunk and retrieves the /// associated data. /// - Empty trees return an empty byte vector. + /// - Non-Merk append-only subtrees (`CommitmentTree`, `MmrTree`, + /// `BulkAppendTree`, `DenseAppendOnlyFixedSizeTree`) are served as + /// cursor-based entry pages instead of Merk chunks. A request for one + /// of these subtrees without a page cursor returns + /// `Error::NotSupported`. + /// - Indexed-tree requests return `Error::NotSupported`. pub fn fetch_chunk( &self, packed_global_chunk_id: &[u8], @@ -143,6 +150,35 @@ impl GroveDb { )); } + // Non-Merk append-only trees (CommitmentTree / MmrTree / + // BulkAppendTree / DenseAppendOnlyFixedSizeTree) have no Merk + // nodes to chunk — their payload is served as target-driven + // entry pages instead. The target encodes a page cursor into + // every local chunk id; a request without one comes from a + // peer speaking the pre-#785 protocol, which cannot sync + // these subtrees. + if tree_type.uses_non_merk_data_storage() { + if nested_chunk_ids.is_empty() { + return Err(Error::NotSupported( + "append-only subtree chunk request is missing its page \ + cursor — the requesting peer does not support state \ + sync of append-only trees (see issue #785)" + .to_string(), + )); + } + let mut local_chunk_bytes: Vec> = vec![]; + for chunk_id in &nested_chunk_ids { + local_chunk_bytes.push(self.fetch_non_merk_page( + chunk_prefix, + tree_type, + chunk_id, + tx.as_ref(), + )?); + } + global_chunk_bytes.push(pack_nested_bytes(local_chunk_bytes)?); + continue; + } + let mut local_chunk_bytes: Vec> = vec![]; let merk = self diff --git a/grovedb/src/replication/non_merk_sync.rs b/grovedb/src/replication/non_merk_sync.rs new file mode 100644 index 000000000..4d511d075 --- /dev/null +++ b/grovedb/src/replication/non_merk_sync.rs @@ -0,0 +1,675 @@ +//! State-sync support for the non-Merk append-only tree family +//! (`CommitmentTree`, `MmrTree`, `BulkAppendTree`, +//! `DenseAppendOnlyFixedSizeTree`) — see +//! . +//! +//! These tree types keep an always-empty Merk (`root_key = None`) and store +//! their payload as raw non-Element entries in the subtree's data namespace, +//! so the Merk chunk protocol cannot transfer them. Instead, the transfer is +//! **target-driven entry replay**: +//! +//! - The target already holds the subtree's element (entry counts and +//! parameters) from the hash-verified parent Merk, so it encodes a +//! [`NonMerkChunkId`] — `(start_position, state, param)` — into every local +//! chunk id it requests. +//! - The source serves pages of **leaf entries only** (plus the serialized +//! Sinsemilla frontier for commitment trees, which is an accumulator and +//! cannot be recomputed from entries), read through the same public +//! accessors normal reads use. +//! - The target replays each entry through the real append primitives +//! (`BulkAppendTree::append`, `MMR::push`, `DenseFixedSizedMerkleTree:: +//! insert`), so every internal node, chunk blob, and cached hash on the +//! target is **locally derived** from the wire entries. +//! - At subtree completion the target recomputes the type-specific state +//! root from its own storage +//! ([`GroveDb::compute_non_merk_state_root`]) and requires +//! `combine_hash(value_hash(element_bytes), state_root)` to equal the +//! element value hash bound into the (already restored, hash-verified) +//! parent Merk. Any tampering with wire bytes — entries, frontier, counts +//! — changes the recomputed root and rejects the subtree. + +use grovedb_bulk_append_tree::{deserialize_chunk_blob, BulkAppendTree}; +use grovedb_commitment_tree::COMMITMENT_TREE_DATA_KEY; +use grovedb_dense_fixed_sized_merkle_tree::DenseFixedSizedMerkleTree; +use grovedb_merk::{ + tree::{combine_hash, hash::CryptoHash}, + tree_type::TreeType, +}; +use grovedb_merkle_mountain_range::{ + leaf_to_pos, mmr_size_to_leaf_count, MMRStoreReadOps, MmrNode, MmrStore, MMR, +}; +use grovedb_path::SubtreePath; +use grovedb_storage::{Storage, StorageContext}; + +use crate::{ + replication::utils::{pack_nested_bytes, unpack_nested_bytes}, + Element, Error, GroveDb, Transaction, +}; + +/// Soft byte budget for a single page of entries. A page always carries at +/// least one entry even if that entry alone exceeds the budget. +pub(crate) const MAX_PAGE_BYTES: usize = 1 << 20; // 1 MiB + +/// Hard cap on the number of entries in a single page, so pages of tiny +/// entries stay bounded in element count as well as bytes. +const MAX_PAGE_ENTRIES: usize = 8192; + +/// Encoded length of a [`NonMerkChunkId`]: start (8) + state (8) + param (1). +const NON_MERK_CHUNK_ID_LEN: usize = 17; + +/// Local chunk id for a non-Merk subtree page request. The target — which +/// holds the hash-verified element — tells the source everything it needs to +/// serve the page, so the source never has to reconstruct tree geometry from +/// its raw namespace. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub(crate) struct NonMerkChunkId { + /// First entry position (0-based) this page should start at. + pub start: u64, + /// Type-specific size state from the element: `total_count` for + /// commitment/bulk trees, `mmr_size` for MMR trees, entry `count` for + /// dense trees. + pub state: u64, + /// Type-specific parameter from the element: `chunk_power` for + /// commitment/bulk trees, `height` for dense trees, 0 for MMR trees. + pub param: u8, +} + +impl NonMerkChunkId { + pub(crate) fn encode(&self) -> Vec { + let mut out = Vec::with_capacity(NON_MERK_CHUNK_ID_LEN); + out.extend_from_slice(&self.start.to_be_bytes()); + out.extend_from_slice(&self.state.to_be_bytes()); + out.push(self.param); + out + } + + pub(crate) fn decode(bytes: &[u8]) -> Result { + if bytes.len() != NON_MERK_CHUNK_ID_LEN { + return Err(Error::CorruptedData(format!( + "non-merk chunk id must be {NON_MERK_CHUNK_ID_LEN} bytes, got {}", + bytes.len() + ))); + } + let start = u64::from_be_bytes(bytes[0..8].try_into().expect("checked length")); + let state = u64::from_be_bytes(bytes[8..16].try_into().expect("checked length")); + let param = bytes[16]; + Ok(NonMerkChunkId { + start, + state, + param, + }) + } +} + +/// Encode a page of entries. +/// +/// Layout: `[more_flag: u8] ++ pack_nested_bytes([aux, entry_1, ..., +/// entry_n])`. `more_flag` is 1 when further pages follow, 0 on the final +/// page. `aux` carries type-specific side data — the serialized Sinsemilla +/// frontier on a commitment tree's first page — and is empty everywhere +/// else. +pub(crate) fn encode_non_merk_page( + more: bool, + aux: Vec, + entries: Vec>, +) -> Result, Error> { + let mut sections = Vec::with_capacity(entries.len() + 1); + sections.push(aux); + sections.extend(entries); + let mut out = vec![u8::from(more)]; + out.extend(pack_nested_bytes(sections)?); + Ok(out) +} + +/// Decode a page of entries. Returns `(more, aux, entries)`. +pub(crate) fn decode_non_merk_page(data: &[u8]) -> Result<(bool, Vec, Vec>), Error> { + let (&flag, packed) = data.split_first().ok_or_else(|| { + Error::CorruptedData("non-merk page is empty (missing more-flag)".to_string()) + })?; + let more = match flag { + 0 => false, + 1 => true, + other => { + return Err(Error::CorruptedData(format!( + "non-merk page has invalid more-flag {other}" + ))); + } + }; + let mut sections = unpack_nested_bytes(packed)?; + if sections.is_empty() { + return Err(Error::CorruptedData( + "non-merk page is missing its aux section".to_string(), + )); + } + let entries = sections.split_off(1); + let aux = sections.pop().expect("checked non-empty"); + Ok((more, aux, entries)) +} + +// ── Source side ───────────────────────────────────────────────────────── + +impl GroveDb { + /// Serve one page of entries for a non-Merk append-only subtree, + /// identified by its 32-byte prefix. `chunk_id_bytes` is the + /// target-encoded [`NonMerkChunkId`] cursor. + /// + /// The `state`/`param` fields in the cursor come from the *target's* + /// element. An honest pair always agrees with the source's own data; a + /// mismatching cursor from a byzantine peer only produces read errors or + /// short pages here (bounded by the page budget) — target-side + /// verification is what protects the *syncing* node. + pub(crate) fn fetch_non_merk_page( + &self, + chunk_prefix: crate::SubtreePrefix, + tree_type: TreeType, + chunk_id_bytes: &[u8], + transaction: &Transaction, + ) -> Result, Error> { + let id = NonMerkChunkId::decode(chunk_id_bytes)?; + + match tree_type { + TreeType::CommitmentTree(_) | TreeType::BulkAppendTree(_) => { + // For a commitment tree, the first page also carries the + // serialized Sinsemilla frontier: it is an accumulator over + // the whole append history and cannot be replayed from + // entries without redoing every Sinsemilla hash. + let aux = if matches!(tree_type, TreeType::CommitmentTree(_)) && id.start == 0 { + let ctx = self + .db + .get_transactional_storage_context_by_subtree_prefix( + chunk_prefix, + None, + transaction, + ) + .unwrap(); + ctx.get(COMMITMENT_TREE_DATA_KEY) + .unwrap() + .map_err(|e| { + Error::CorruptedData(format!( + "cannot read commitment tree frontier: {e}" + )) + })? + .unwrap_or_default() + } else { + Vec::new() + }; + + let ctx = self + .db + .get_transactional_storage_context_by_subtree_prefix( + chunk_prefix, + None, + transaction, + ) + .unwrap(); + let tree = BulkAppendTree::from_state(id.state, id.param, ctx).map_err(|e| { + Error::CorruptedData(format!( + "cannot open bulk store of {} entries for page serving: {e}", + id.state + )) + })?; + + let epoch_size = tree.epoch_size(); + let chunk_count = tree.chunk_count(); + let buffer_start = chunk_count * epoch_size; + let total = id.state; + + let mut entries = Vec::new(); + let mut bytes = 0usize; + let mut pos = id.start; + // Cache the deserialized blob of the chunk currently being + // walked so a page spanning a chunk deserializes it once. + let mut cached_chunk: Option<(u64, Vec>)> = None; + while pos < total && entries.len() < MAX_PAGE_ENTRIES && bytes < MAX_PAGE_BYTES { + let value = if pos >= buffer_start { + tree.get_buffer_value((pos - buffer_start) as u16) + .map_err(|e| { + Error::CorruptedData(format!("cannot read buffer entry {pos}: {e}")) + })? + .ok_or_else(|| { + Error::CorruptedData(format!("missing buffer entry {pos}")) + })? + } else { + let chunk_idx = pos / epoch_size; + if cached_chunk.as_ref().map(|(idx, _)| *idx) != Some(chunk_idx) { + let blob = tree + .get_chunk_value(chunk_idx) + .map_err(|e| { + Error::CorruptedData(format!( + "cannot read chunk blob {chunk_idx}: {e}" + )) + })? + .ok_or_else(|| { + Error::CorruptedData(format!("missing chunk blob {chunk_idx}")) + })?; + let blob_entries = deserialize_chunk_blob(&blob).map_err(|e| { + Error::CorruptedData(format!( + "cannot deserialize chunk blob {chunk_idx}: {e}" + )) + })?; + cached_chunk = Some((chunk_idx, blob_entries)); + } + let (_, blob_entries) = cached_chunk.as_ref().expect("just cached"); + blob_entries + .get((pos % epoch_size) as usize) + .cloned() + .ok_or_else(|| { + Error::CorruptedData(format!( + "chunk blob {chunk_idx} has no entry at position {pos}" + )) + })? + }; + bytes += value.len(); + entries.push(value); + pos += 1; + } + + encode_non_merk_page(pos < total, aux, entries) + } + TreeType::MmrTree => { + let mmr_size = id.state; + let leaf_count = mmr_size_to_leaf_count(mmr_size); + let ctx = self + .db + .get_transactional_storage_context_by_subtree_prefix( + chunk_prefix, + None, + transaction, + ) + .unwrap(); + let store = MmrStore::new(&ctx); + + let store_ref: &MmrStore<_> = &store; + + let mut entries = Vec::new(); + let mut bytes = 0usize; + let mut leaf = id.start; + while leaf < leaf_count + && entries.len() < MAX_PAGE_ENTRIES + && bytes < MAX_PAGE_BYTES + { + let node = store_ref + .element_at_position(leaf_to_pos(leaf)) + .value + .map_err(|e| { + Error::CorruptedData(format!("cannot read MMR leaf {leaf}: {e}")) + })? + .ok_or_else(|| Error::CorruptedData(format!("missing MMR leaf {leaf}")))?; + let value = node.into_value().ok_or_else(|| { + Error::CorruptedData(format!("MMR leaf {leaf} carries no value")) + })?; + bytes += value.len(); + entries.push(value); + leaf += 1; + } + + encode_non_merk_page(leaf < leaf_count, Vec::new(), entries) + } + TreeType::DenseAppendOnlyFixedSizeTree(_) => { + let count = u16::try_from(id.state).map_err(|_| { + Error::CorruptedData(format!( + "dense tree count {} exceeds u16 in page cursor", + id.state + )) + })?; + let ctx = self + .db + .get_transactional_storage_context_by_subtree_prefix( + chunk_prefix, + None, + transaction, + ) + .unwrap(); + let tree = + DenseFixedSizedMerkleTree::from_state(id.param, count, ctx).map_err(|e| { + Error::CorruptedData(format!( + "cannot open dense tree of {count} entries for page serving: {e}" + )) + })?; + + let mut entries = Vec::new(); + let mut bytes = 0usize; + let mut pos = id.start; + while pos < count as u64 + && entries.len() < MAX_PAGE_ENTRIES + && bytes < MAX_PAGE_BYTES + { + let value = tree + .get(pos as u16) + .unwrap() + .map_err(|e| { + Error::CorruptedData(format!("cannot read dense entry {pos}: {e}")) + })? + .ok_or_else(|| { + Error::CorruptedData(format!("missing dense entry {pos}")) + })?; + bytes += value.len(); + entries.push(value); + pos += 1; + } + + encode_non_merk_page(pos < count as u64, Vec::new(), entries) + } + _ => Err(Error::InternalError(format!( + "fetch_non_merk_page called for non append-only tree type {tree_type}" + ))), + } + } +} + +// ── Target side ───────────────────────────────────────────────────────── + +/// Restorer for a non-Merk append-only subtree. Holds only plain data — the +/// per-page storage context is created (and dropped) inside each call, so +/// this type has no lifetime entanglement with the sync transaction. +#[derive(Debug)] +pub(crate) struct NonMerkRestorer { + /// The subtree's element as declared by the (hash-verified) parent. + element: Element, + /// The element value hash bound into the parent leaf: + /// `combine_hash(value_hash(element_bytes), state_root)`. + expected_elem_value_hash: CryptoHash, + /// `value_hash(element_bytes)` for the element as stored in the parent. + actual_value_hash: CryptoHash, + /// Total number of leaf entries the element declares. + expected_entries: u64, + /// `state` field for outgoing page cursors (see [`NonMerkChunkId`]). + state_for_source: u64, + /// `param` field for outgoing page cursors (see [`NonMerkChunkId`]). + param: u8, + /// Number of entries replayed so far. + replayed: u64, + /// Current MMR size of the partially replayed MMR (MmrTree only). + mmr_size_so_far: u64, + /// Whether the final page has been received. + finished: bool, +} + +impl NonMerkRestorer { + pub(crate) fn new( + element: Element, + expected_elem_value_hash: CryptoHash, + actual_value_hash: CryptoHash, + ) -> Result { + let (expected_entries, state_for_source, param) = match element.underlying() { + Element::CommitmentTree(total_count, chunk_power, _) => { + (*total_count, *total_count, *chunk_power) + } + Element::BulkAppendTree(total_count, chunk_power, _) => { + (*total_count, *total_count, *chunk_power) + } + Element::MmrTree(mmr_size, _) => (mmr_size_to_leaf_count(*mmr_size), *mmr_size, 0), + Element::DenseAppendOnlyFixedSizeTree(count, height, _) => { + (*count as u64, *count as u64, *height) + } + other => { + return Err(Error::InternalError(format!( + "NonMerkRestorer::new called on a non append-only element: {}", + other.type_str() + ))); + } + }; + Ok(NonMerkRestorer { + element, + expected_elem_value_hash, + actual_value_hash, + expected_entries, + state_for_source, + param, + replayed: 0, + mmr_size_so_far: 0, + finished: false, + }) + } + + /// The local chunk id for the first page of this subtree. + pub(crate) fn initial_chunk_id(&self) -> Vec { + NonMerkChunkId { + start: 0, + state: self.state_for_source, + param: self.param, + } + .encode() + } + + /// Apply one received page: replay its entries through the real append + /// primitives and return the next page cursor(s), empty when the source + /// declared this the final page. + pub(crate) fn apply_page( + &mut self, + db: &GroveDb, + tx: &Transaction, + path: &[Vec], + chunk_id: &[u8], + data: &[u8], + ) -> Result>, Error> { + let id = NonMerkChunkId::decode(chunk_id)?; + if id.start != self.replayed || id.state != self.state_for_source || id.param != self.param + { + return Err(Error::InternalError(format!( + "non-merk page cursor out of order: got start {}, expected {}", + id.start, self.replayed + ))); + } + if self.finished { + return Err(Error::InternalError( + "non-merk page received after the final page".to_string(), + )); + } + + let (more, aux, entries) = decode_non_merk_page(data)?; + + if self.replayed + entries.len() as u64 > self.expected_entries { + return Err(Error::CorruptedData(format!( + "non-merk page overflows declared entry count: {} + {} > {}", + self.replayed, + entries.len(), + self.expected_entries + ))); + } + if more && entries.is_empty() { + return Err(Error::CorruptedData( + "non-merk page declares more data but carries no entries".to_string(), + )); + } + + let path_refs: Vec<&[u8]> = path.iter().map(|v| v.as_slice()).collect(); + let subtree_path: SubtreePath<&[u8]> = SubtreePath::from(path_refs.as_slice()); + + // Aux section: only a commitment tree's first page may carry data — + // the serialized frontier, copied verbatim (it is authenticated at + // finalize time through the sinsemilla root inside the state root). + if self.element.is_commitment_tree() && self.replayed == 0 { + if !aux.is_empty() { + let ctx = db + .db + .get_immediate_storage_context(subtree_path.clone(), tx) + .unwrap(); + ctx.put(COMMITMENT_TREE_DATA_KEY, &aux, None, None) + .unwrap() + .map_err(|e| { + Error::CorruptedData(format!("cannot write commitment tree frontier: {e}")) + })?; + } else if self.expected_entries > 0 { + return Err(Error::CorruptedData( + "populated commitment tree page 0 is missing the frontier".to_string(), + )); + } + } else if !aux.is_empty() { + return Err(Error::CorruptedData( + "unexpected aux data in non-merk page".to_string(), + )); + } + + match self.element.underlying() { + Element::CommitmentTree(..) | Element::BulkAppendTree(..) => { + let ctx = db + .db + .get_immediate_storage_context(subtree_path, tx) + .unwrap(); + let mut tree = + BulkAppendTree::from_state(self.replayed, self.param, ctx).map_err(|e| { + Error::CorruptedData(format!( + "cannot open partially replayed bulk store ({} entries): {e}", + self.replayed + )) + })?; + for entry in &entries { + tree.append(entry).map_err(|e| { + Error::CorruptedData(format!("cannot replay bulk entry: {e}")) + })?; + } + tree.commit_mmr().map_err(|e| { + Error::CorruptedData(format!("cannot flush replayed chunk MMR: {e}")) + })?; + } + Element::MmrTree(..) => { + let ctx = db + .db + .get_immediate_storage_context(subtree_path, tx) + .unwrap(); + let store = MmrStore::new(&ctx); + let mut mmr = MMR::new(self.mmr_size_so_far, &store); + for entry in entries.iter().cloned() { + mmr.push(MmrNode::leaf(entry)).unwrap().map_err(|e| { + Error::CorruptedData(format!("cannot replay MMR leaf: {e}")) + })?; + } + mmr.commit() + .unwrap() + .map_err(|e| Error::CorruptedData(format!("cannot flush replayed MMR: {e}")))?; + self.mmr_size_so_far = mmr.mmr_size; + } + Element::DenseAppendOnlyFixedSizeTree(..) => { + let ctx = db + .db + .get_immediate_storage_context(subtree_path, tx) + .unwrap(); + let mut tree = + DenseFixedSizedMerkleTree::from_state(self.param, self.replayed as u16, ctx) + .map_err(|e| { + Error::CorruptedData(format!( + "cannot open partially replayed dense tree ({} entries): {e}", + self.replayed + )) + })?; + for entry in &entries { + tree.insert(entry).unwrap().map_err(|e| { + Error::CorruptedData(format!("cannot replay dense entry: {e}")) + })?; + } + } + _ => unreachable!("NonMerkRestorer::new only accepts append-only elements"), + } + + self.replayed += entries.len() as u64; + + if more { + Ok(vec![NonMerkChunkId { + start: self.replayed, + state: self.state_for_source, + param: self.param, + } + .encode()]) + } else { + self.finished = true; + Ok(vec![]) + } + } + + /// Verify the fully replayed subtree against the parent binding: + /// `combine_hash(value_hash(element_bytes), recomputed_state_root)` must + /// equal the element value hash the parent Merk committed to. + pub(crate) fn finalize( + &self, + db: &GroveDb, + tx: &Transaction, + path: &[Vec], + ) -> Result<(), Error> { + if self.replayed != self.expected_entries { + return Err(Error::CorruptedData(format!( + "non-merk subtree replay incomplete: got {} entries, element declares {}", + self.replayed, self.expected_entries + ))); + } + if let Element::MmrTree(mmr_size, _) = self.element.underlying() + && self.mmr_size_so_far != *mmr_size + { + return Err(Error::CorruptedData(format!( + "replayed MMR size {} does not match element mmr_size {}", + self.mmr_size_so_far, mmr_size + ))); + } + + let path_refs: Vec<&[u8]> = path.iter().map(|v| v.as_slice()).collect(); + let subtree_path: SubtreePath<&[u8]> = SubtreePath::from(path_refs.as_slice()); + + let state_root = db.compute_non_merk_state_root(&self.element, subtree_path, tx)?; + let combined = combine_hash(&self.actual_value_hash, &state_root).unwrap(); + if combined != self.expected_elem_value_hash { + return Err(Error::CorruptedData(format!( + "non-merk subtree state root mismatch after replay: combined hash {} does not \ + match the parent binding {}", + hex::encode(combined), + hex::encode(self.expected_elem_value_hash), + ))); + } + Ok(()) + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn non_merk_chunk_id_roundtrip() { + let id = NonMerkChunkId { + start: 123456789, + state: 987654321, + param: 7, + }; + let encoded = id.encode(); + assert_eq!(encoded.len(), NON_MERK_CHUNK_ID_LEN); + assert_eq!(NonMerkChunkId::decode(&encoded).unwrap(), id); + } + + #[test] + fn non_merk_chunk_id_rejects_wrong_length() { + assert!(NonMerkChunkId::decode(&[]).is_err()); + assert!(NonMerkChunkId::decode(&[0u8; 16]).is_err()); + assert!(NonMerkChunkId::decode(&[0u8; 18]).is_err()); + } + + #[test] + fn non_merk_page_roundtrip() { + let entries = vec![b"one".to_vec(), b"two".to_vec(), Vec::new()]; + let aux = b"frontier".to_vec(); + let encoded = encode_non_merk_page(true, aux.clone(), entries.clone()).unwrap(); + let (more, got_aux, got_entries) = decode_non_merk_page(&encoded).unwrap(); + assert!(more); + assert_eq!(got_aux, aux); + assert_eq!(got_entries, entries); + + let encoded = encode_non_merk_page(false, Vec::new(), Vec::new()).unwrap(); + let (more, got_aux, got_entries) = decode_non_merk_page(&encoded).unwrap(); + assert!(!more); + assert!(got_aux.is_empty()); + assert!(got_entries.is_empty()); + } + + #[test] + fn non_merk_page_rejects_malformed() { + // Empty data: missing flag. + assert!(decode_non_merk_page(&[]).is_err()); + // Invalid flag. + assert!(decode_non_merk_page(&[2u8, 0, 0, 0, 0]).is_err()); + // Valid flag but no sections at all (count = 0): aux is mandatory. + let no_sections = { + let mut d = vec![0u8]; + d.extend(pack_nested_bytes(vec![]).unwrap()); + d + }; + assert!(decode_non_merk_page(&no_sections).is_err()); + } +} diff --git a/grovedb/src/replication/state_sync_session.rs b/grovedb/src/replication/state_sync_session.rs index 599b1394f..3da53fa15 100644 --- a/grovedb/src/replication/state_sync_session.rs +++ b/grovedb/src/replication/state_sync_session.rs @@ -19,6 +19,7 @@ use grovedb_storage::{ use grovedb_version::version::GroveVersion; use super::{ + non_merk_sync::NonMerkRestorer, utils::{decode_vec_ops, encode_global_chunk_id, path_to_string}, CURRENT_STATE_SYNC_VERSION, }; @@ -34,10 +35,19 @@ pub const CONST_GROUP_PACKING_SIZE: usize = 32; pub(crate) type SubtreePrefix = [u8; 32]; +/// The restore backend for one subtree: Merk chunk restore for ordinary +/// subtrees, entry replay for the non-Merk append-only tree family +/// (CommitmentTree / MmrTree / BulkAppendTree / +/// DenseAppendOnlyFixedSizeTree) — see issue #785. +enum SubtreeRestorer<'db> { + Merk(Restorer>), + NonMerk(NonMerkRestorer), +} + /// Struct governing the state synchronization of one subtree. struct SubtreeStateSyncInfo<'db> { - /// Current Chunk restorer - restorer: Restorer>, + /// Current chunk restorer (Merk chunks or non-Merk entry replay) + restorer: SubtreeRestorer<'db>, /// Set of global chunk ids requested to be fetched and pending for /// processing. For the description of global chunk id check @@ -91,8 +101,10 @@ impl SubtreeStateSyncInfo<'_> { /// expected format. /// - The function modifies the state of the synchronization process, so it /// must be used carefully to maintain correctness. - fn apply_inner_chunk( + fn apply_inner_chunk<'tx, 'db: 'tx>( &mut self, + db: &'db GroveDb, + tx: &'tx Transaction<'db>, chunk_id: &[u8], chunk_data: &[u8], grove_version: &GroveVersion, @@ -105,28 +117,46 @@ impl SubtreeStateSyncInfo<'_> { )); } self.pending_chunks.remove(chunk_id); - if !chunk_data.is_empty() { - match decode_vec_ops(chunk_data) { - Ok(ops) => { - match self.restorer.process_chunk(chunk_id, ops, grove_version) { - Ok(next_chunk_ids) => { - self.num_processed_chunks += 1; - for next_chunk_id in next_chunk_ids { - self.pending_chunks.insert(next_chunk_id.clone()); - res.push(next_chunk_id); - } + match &mut self.restorer { + SubtreeRestorer::Merk(restorer) => { + if !chunk_data.is_empty() { + match decode_vec_ops(chunk_data) { + Ok(ops) => { + match restorer.process_chunk(chunk_id, ops, grove_version) { + Ok(next_chunk_ids) => { + self.num_processed_chunks += 1; + for next_chunk_id in next_chunk_ids { + self.pending_chunks.insert(next_chunk_id.clone()); + res.push(next_chunk_id); + } + } + Err(e) => { + return Err(Error::InternalError(format!( + "Unable to process incoming chunk: {e}" + ))); + } + }; } Err(e) => { - return Err(Error::InternalError(format!( - "Unable to process incoming chunk: {e}" + return Err(Error::CorruptedData(format!( + "Unable to decode incoming chunk: {e}" ))); } - }; + } } - Err(e) => { - return Err(Error::CorruptedData(format!( - "Unable to decode incoming chunk: {e}" - ))); + } + SubtreeRestorer::NonMerk(non_merk_restorer) => { + let next_chunk_ids = non_merk_restorer.apply_page( + db, + tx, + &self.current_path, + chunk_id, + chunk_data, + )?; + self.num_processed_chunks += 1; + for next_chunk_id in next_chunk_ids { + self.pending_chunks.insert(next_chunk_id.clone()); + res.push(next_chunk_id); } } } @@ -138,7 +168,7 @@ impl SubtreeStateSyncInfo<'_> { impl<'tx> SubtreeStateSyncInfo<'tx> { pub fn new(restorer: Restorer>) -> Self { SubtreeStateSyncInfo { - restorer, + restorer: SubtreeRestorer::Merk(restorer), root_key: None, tree_type: TreeType::NormalTree, pending_chunks: Default::default(), @@ -344,10 +374,48 @@ impl<'db> MultiStateSyncSession<'db> { &*(tx as *const _) }; - if let Ok((merk, root_key, tree_type)) = + if let Ok((merk, root_key, tree_type, element)) = self.db .open_merk_for_replication(path.clone(), transaction_ref, grove_version) { + if tree_type.uses_non_merk_data_storage() { + // Non-Merk append-only subtree: restored by replaying leaf + // entries rather than Merk chunks (see issue #785). The + // Merk opened above is structurally empty for these types + // and is not needed. + drop(merk); + let element = element.ok_or_else(|| { + Error::InternalError( + "append-only subtree must have a parent element".to_string(), + ) + })?; + let actual_hash = actual_hash.ok_or_else(|| { + Error::InternalError( + "append-only subtree sync requires the parent element value hash" + .to_string(), + ) + })?; + let non_merk_restorer = NonMerkRestorer::new(element, hash, actual_hash)?; + let first_chunk_id = non_merk_restorer.initial_chunk_id(); + let mut sync_info = SubtreeStateSyncInfo { + restorer: SubtreeRestorer::NonMerk(non_merk_restorer), + root_key: root_key.clone(), + tree_type, + pending_chunks: Default::default(), + current_path: path.to_vec(), + num_processed_chunks: 0, + }; + sync_info.pending_chunks.insert(first_chunk_id.clone()); + self.as_mut() + .current_prefixes() + .insert(chunk_prefix, sync_info); + return encode_global_chunk_id( + chunk_prefix, + root_key, + tree_type, + vec![first_chunk_id], + ); + } let restorer = Restorer::new(merk, hash, actual_hash); let mut sync_info = SubtreeStateSyncInfo::new(restorer); sync_info.pending_chunks.insert(vec![]); @@ -470,6 +538,22 @@ impl<'db> MultiStateSyncSession<'db> { )); } + let db = self.db; + // SAFETY: the transaction lives as long as the pinned session and is + // dropped last; the reference is only used within this call while + // the session is alive. This mirrors the pattern used by + // `add_subtree_sync_info` and `discover_new_subtrees_metadata`. + // + // ADDITIONAL INVARIANT for this call site: `set_new_transaction()` + // below replaces and commits `self.transaction`, which invalidates + // `transaction_ref`. Every use of `transaction_ref` MUST stay inside + // the per-chunk loop, above the `set_new_transaction()` call. Do not + // use `transaction_ref` after that point. + let transaction_ref: &'db Transaction<'db> = unsafe { + let tx: &Transaction<'db> = &self.as_ref().transaction; + &*(tx as *const _) + }; + let mut next_global_chunk_ids: Vec> = vec![]; for (iter_global_chunk_id, iter_packed_chunks) in nested_global_chunk_ids @@ -510,6 +594,8 @@ impl<'db> MultiStateSyncSession<'db> { it_chunk_ids.iter().zip(current_nested_chunk_data.iter()) { next_local_chunk_ids.extend(subtree_state_sync.apply_inner_chunk( + db, + transaction_ref, current_local_chunk_id.as_slice(), current_local_chunks.as_slice(), grove_version, @@ -531,24 +617,38 @@ impl<'db> MultiStateSyncSession<'db> { // Subtree is finished. We can save it. let is_subtree_empty = subtree_state_sync.num_processed_chunks == 0; + let mut is_non_merk_subtree = false; if let Some(prefix_data) = current_prefixes.remove(&chunk_prefix) { - if is_subtree_empty { - // For empty subtrees, verify the restorer's underlying merk has a - // NULL root hash. A malicious peer that sends empty data for a - // non-empty subtree will be caught here (and also at commit time - // via H3 root hash verification). - let merk = prefix_data.restorer.into_merk(); - let merk_root = merk.root_hash().unwrap(); - if merk_root != grovedb_merk::tree::hash::NULL_HASH { - return Err(Error::InternalError( - "empty subtree has non-null root hash".to_string(), - )); + match prefix_data.restorer { + SubtreeRestorer::Merk(restorer) => { + if is_subtree_empty { + // For empty subtrees, verify the restorer's underlying merk + // has a NULL root hash. A malicious peer that sends empty + // data for a non-empty subtree will be caught here (and + // also at commit time via H3 root hash verification). + let merk = restorer.into_merk(); + let merk_root = merk.root_hash().unwrap(); + if merk_root != grovedb_merk::tree::hash::NULL_HASH { + return Err(Error::InternalError( + "empty subtree has non-null root hash".to_string(), + )); + } + } else if let Err(err) = restorer.finalize(grove_version) { + return Err(Error::InternalError(format!( + "Unable to finalize Merk: {:?}", + err + ))); + } + } + SubtreeRestorer::NonMerk(non_merk_restorer) => { + // Entry replay finished: recompute the state + // root from the replayed payload and verify it + // against the parent binding. A byzantine + // source that tampered with any wire byte is + // rejected here. + is_non_merk_subtree = true; + non_merk_restorer.finalize(db, transaction_ref, &completed_path)?; } - } else if let Err(err) = prefix_data.restorer.finalize(grove_version) { - return Err(Error::InternalError(format!( - "Unable to finalize Merk: {:?}", - err - ))); } } else { return Err(Error::InternalError(format!( @@ -561,8 +661,15 @@ impl<'db> MultiStateSyncSession<'db> { *self.as_mut().num_processed_subtrees_in_batch() += 1; - let new_subtrees_metadata = - self.discover_new_subtrees_metadata(&completed_path, grove_version)?; + // Non-Merk append-only subtrees never contain child + // subtrees, and their data namespace holds raw payload + // entries (not Elements) — running element discovery over + // it would fail. Skip it. + let new_subtrees_metadata = if is_non_merk_subtree { + SubtreesMetadata::default() + } else { + self.discover_new_subtrees_metadata(&completed_path, grove_version)? + }; if self.num_processed_subtrees_in_batch >= self.subtrees_batch_size { match self.as_mut().pending_discovered_subtrees() { @@ -696,6 +803,11 @@ impl<'db> MultiStateSyncSession<'db> { .to_string(), )); } + // Non-Merk append-only trees (CommitmentTree / MmrTree / + // BulkAppendTree / DenseAppendOnlyFixedSizeTree) are + // discovered like any subtree; `add_subtree_sync_info` + // routes them to the entry-replay restore path instead of + // Merk chunk restore (see issue #785). subtree_keys.insert(key.to_vec()); } } diff --git a/grovedb/src/tests/replication_session_tests.rs b/grovedb/src/tests/replication_session_tests.rs index 8d29a3e5a..863eaefa9 100644 --- a/grovedb/src/tests/replication_session_tests.rs +++ b/grovedb/src/tests/replication_session_tests.rs @@ -13,16 +13,27 @@ mod tests { Element, GroveDb, }; - /// Helper: perform a full state sync from source to destination using - /// a checkpoint of the source (mirrors the tutorial/production pattern). - /// - /// Returns the destination TempGroveDb after committing the session. - fn sync_source_to_destination( + /// Optional in-flight mutation of a commitment tree page: + /// `(more, aux, entries) -> (more, aux, entries)`. Used by tamper tests. + type CtPageMutator<'a> = + &'a dyn Fn(bool, Vec, Vec>) -> (bool, Vec, Vec>); + + /// The single sync driver behind every test in this file: checkpoint the + /// source (the standard replication pattern — the tutorial does the + /// same), run the fetch/apply loop with the given subtree batch size, + /// optionally mutating commitment tree pages in flight, verify + /// completion, and commit the session. + fn run_sync( source: &TempGroveDb, grove_version: &GroveVersion, - ) -> TempGroveDb { - // Create a checkpoint from the source -- this is the standard pattern - // for replication (the tutorial does the same). + subtrees_batch_size: usize, + mutate_ct_page: Option, + ) -> Result { + use crate::replication::{ + non_merk_sync::{decode_non_merk_page, encode_non_merk_page}, + utils::{decode_global_chunk_id, pack_nested_bytes, unpack_nested_bytes}, + }; + let checkpoint_dir = TempDir::new().expect("should create temp dir for checkpoint"); let checkpoint_path = checkpoint_dir.path().join("checkpoint"); source @@ -37,45 +48,86 @@ mod tests { let dest = make_empty_grovedb(); - let mut session = dest - .start_snapshot_syncing(app_hash, 64, CURRENT_STATE_SYNC_VERSION, grove_version) - .expect("should start snapshot syncing"); + let mut session = dest.start_snapshot_syncing( + app_hash, + subtrees_batch_size, + CURRENT_STATE_SYNC_VERSION, + grove_version, + )?; // Use a queue-based approach as shown in the tutorial let mut chunk_queue: VecDeque> = VecDeque::new(); chunk_queue.push_back(app_hash.to_vec()); while let Some(chunk_id) = chunk_queue.pop_front() { - let chunk_data = checkpoint_db - .fetch_chunk( - chunk_id.as_slice(), - None, - CURRENT_STATE_SYNC_VERSION, - grove_version, - ) - .expect("should fetch chunk from checkpoint"); + let mut chunk_data = checkpoint_db.fetch_chunk( + chunk_id.as_slice(), + None, + CURRENT_STATE_SYNC_VERSION, + grove_version, + )?; - let more_ids = session - .apply_chunk( - chunk_id.as_slice(), - &chunk_data, - CURRENT_STATE_SYNC_VERSION, - grove_version, - ) - .expect("should apply chunk to destination"); + if let Some(mutate) = mutate_ct_page { + // Mirror apply_chunk's unpacking to find commitment tree + // pages and run them through the mutator. + let global_ids: Vec> = if chunk_id.as_slice() == app_hash.as_slice() { + vec![chunk_id.clone()] + } else { + unpack_nested_bytes(&chunk_id)? + }; + let global_data = unpack_nested_bytes(&chunk_data)?; + assert_eq!(global_ids.len(), global_data.len()); + let mut mutated_globals = Vec::with_capacity(global_data.len()); + for (gid, gdata) in global_ids.iter().zip(global_data) { + let (_, _, tree_type, _) = decode_global_chunk_id(gid, &app_hash)?; + if matches!( + tree_type, + grovedb_merk::tree_type::TreeType::CommitmentTree(_) + ) { + let pages = unpack_nested_bytes(&gdata)?; + let mut mutated_pages = Vec::with_capacity(pages.len()); + for page in pages { + let (more, aux, entries) = decode_non_merk_page(&page)?; + let (more, aux, entries) = mutate(more, aux, entries); + mutated_pages.push(encode_non_merk_page(more, aux, entries)?); + } + mutated_globals.push(pack_nested_bytes(mutated_pages)?); + } else { + mutated_globals.push(gdata); + } + } + chunk_data = pack_nested_bytes(mutated_globals)?; + } + + let more_ids = session.apply_chunk( + chunk_id.as_slice(), + &chunk_data, + CURRENT_STATE_SYNC_VERSION, + grove_version, + )?; chunk_queue.extend(more_ids); } - assert!( - session.is_sync_completed(), - "sync should be completed after all chunks are applied" - ); + if !session.is_sync_completed() { + return Err(crate::Error::InternalError( + "sync did not complete".to_string(), + )); + } - dest.commit_session(session, grove_version) - .expect("should commit sync session"); + dest.commit_session(session, grove_version)?; + Ok(dest) + } - dest + /// Helper: perform a full state sync from source to destination, + /// panicking on any error. + /// + /// Returns the destination TempGroveDb after committing the session. + fn sync_source_to_destination( + source: &TempGroveDb, + grove_version: &GroveVersion, + ) -> TempGroveDb { + run_sync(source, grove_version, 64, None).expect("state sync should succeed") } #[test] @@ -720,41 +772,7 @@ mod tests { source: &TempGroveDb, grove_version: &GroveVersion, ) -> Result<(), crate::Error> { - let checkpoint_dir = TempDir::new().expect("should create temp dir for checkpoint"); - let checkpoint_path = checkpoint_dir.path().join("checkpoint"); - source - .create_checkpoint(&checkpoint_path) - .expect("should create checkpoint"); - let checkpoint_db = GroveDb::open(&checkpoint_path).expect("should open checkpoint db"); - - let app_hash = checkpoint_db - .root_hash(None, grove_version) - .unwrap() - .expect("checkpoint root hash should be available"); - - let dest = make_empty_grovedb(); - let mut session = - dest.start_snapshot_syncing(app_hash, 64, CURRENT_STATE_SYNC_VERSION, grove_version)?; - - let mut chunk_queue: VecDeque> = VecDeque::new(); - chunk_queue.push_back(app_hash.to_vec()); - - while let Some(chunk_id) = chunk_queue.pop_front() { - let chunk_data = checkpoint_db.fetch_chunk( - chunk_id.as_slice(), - None, - CURRENT_STATE_SYNC_VERSION, - grove_version, - )?; - let more_ids = session.apply_chunk( - chunk_id.as_slice(), - &chunk_data, - CURRENT_STATE_SYNC_VERSION, - grove_version, - )?; - chunk_queue.extend(more_ids); - } - Ok(()) + run_sync(source, grove_version, 64, None).map(|_| ()) } #[test] @@ -889,7 +907,7 @@ mod tests { // Read the PCIT element to get its root key and confirm tree type. let tx = source.start_transaction(); - let (merk, root_key, tree_type) = source + let (merk, root_key, tree_type, _element) = source .open_merk_for_replication([TEST_LEAF, b"pcit"].as_ref().into(), &tx, grove_version) .expect("open pcit merk for replication"); drop(merk); @@ -921,4 +939,861 @@ mod tests { .expect_err("source-side fetch_chunk of an indexed tree must be rejected"); assert_not_supported_indexed(&err, "source-side fetch_chunk"); } + + fn assert_not_supported_append_only(err: &crate::Error, context: &str) { + let msg = format!("{err:?}"); + assert!( + matches!(err, crate::Error::NotSupported(_)), + "{context}: expected Error::NotSupported, got: {msg}" + ); + assert!( + msg.contains("append-only"), + "{context}: error should mention append-only trees, got: {msg}" + ); + } + + /// Full state-sync round trip for a POPULATED CommitmentTree (issue + /// #785, Phase 1). Uses chunk_power 2 (epoch of 4) with 6 notes so the + /// payload spans a compacted chunk blob AND the current buffer, plus + /// the Sinsemilla frontier. + #[test] + fn state_sync_populated_commitment_tree_round_trip() { + let grove_version = GroveVersion::latest(); + let source = make_test_grovedb(grove_version); + + source + .insert( + [TEST_LEAF].as_ref(), + b"ct", + Element::empty_commitment_tree(2).expect("valid chunk power"), + None, + None, + grove_version, + ) + .unwrap() + .expect("insert commitment tree"); + // A sibling item so the parent subtree holds mixed content. + source + .insert( + [TEST_LEAF].as_ref(), + b"sibling", + Element::new_item(b"item next to the ct".to_vec()), + None, + None, + grove_version, + ) + .unwrap() + .expect("insert sibling item"); + + for i in 1u8..=6 { + source + .commitment_tree_insert_raw( + [TEST_LEAF].as_ref(), + b"ct", + [i; 32], + [i.wrapping_add(100); 32], + [i.wrapping_add(200); 32], + vec![i; 216], + None, + grove_version, + ) + .unwrap() + .expect("insert commitment tree note"); + } + + let source_root_hash = source + .root_hash(None, grove_version) + .unwrap() + .expect("source root hash"); + let source_anchor = source + .commitment_tree_anchor([TEST_LEAF].as_ref(), b"ct", None, grove_version) + .unwrap() + .expect("source anchor"); + + let dest = sync_source_to_destination(&source, grove_version); + + let dest_root_hash = dest + .root_hash(None, grove_version) + .unwrap() + .expect("dest root hash"); + assert_eq!(source_root_hash, dest_root_hash, "app hash must match"); + + // The anchor (recomputed from the transferred frontier) matches. + let dest_anchor = dest + .commitment_tree_anchor([TEST_LEAF].as_ref(), b"ct", None, grove_version) + .unwrap() + .expect("dest anchor must be readable"); + assert_eq!(source_anchor, dest_anchor, "anchor must match"); + + // Every note value survives, both in the compacted chunk (positions + // 0..4) and in the buffer (positions 4..6). + for pos in 0u64..6 { + let source_value = source + .commitment_tree_get_value([TEST_LEAF].as_ref(), b"ct", pos, None, grove_version) + .unwrap() + .expect("source note value") + .expect("source note value present"); + let dest_value = dest + .commitment_tree_get_value([TEST_LEAF].as_ref(), b"ct", pos, None, grove_version) + .unwrap() + .expect("dest note value") + .expect("dest note value present"); + assert_eq!(source_value, dest_value, "note {pos} must match"); + } + + // The destination passes a full integrity check. + let dest_issues = dest + .verify_grovedb(None, true, false, grove_version) + .expect("dest verify_grovedb should run"); + assert!( + dest_issues.is_empty(), + "destination must verify clean, got: {:?}", + dest_issues + ); + + // The restored subtree is fully usable for future writes: appending + // the same note on both sides keeps the states identical. + for db in [&source, &dest] { + db.commitment_tree_insert_raw( + [TEST_LEAF].as_ref(), + b"ct", + // Small repeated bytes stay below the Pallas field modulus. + [7u8; 32], + [8u8; 32], + [9u8; 32], + vec![77u8; 216], + None, + grove_version, + ) + .unwrap() + .expect("post-sync append"); + } + assert_eq!( + source.root_hash(None, grove_version).unwrap().unwrap(), + dest.root_hash(None, grove_version).unwrap().unwrap(), + "post-sync appends must produce identical states" + ); + } + + /// A peer speaking the pre-#785 protocol requests an append-only + /// subtree the old way — with no page cursor in the global chunk id. + /// The source must reject that request descriptively instead of trying + /// (and opaquely failing) to build a Merk chunk producer. + #[test] + fn fetch_chunk_rejects_append_only_request_without_page_cursor() { + use crate::replication::utils::{encode_global_chunk_id, pack_nested_bytes}; + + let grove_version = GroveVersion::latest(); + let source = make_test_grovedb(grove_version); + + source + .insert( + [TEST_LEAF].as_ref(), + b"ct", + Element::empty_commitment_tree(4).expect("valid chunk power"), + None, + None, + grove_version, + ) + .unwrap() + .expect("insert commitment tree"); + source + .commitment_tree_insert_raw( + [TEST_LEAF].as_ref(), + b"ct", + [1u8; 32], + [2u8; 32], + [3u8; 32], + vec![0u8; 216], + None, + grove_version, + ) + .unwrap() + .expect("insert commitment tree note"); + + let tx = source.start_transaction(); + let (merk, root_key, tree_type, _element) = source + .open_merk_for_replication([TEST_LEAF, b"ct"].as_ref().into(), &tx, grove_version) + .expect("open ct merk for replication"); + drop(merk); + assert!( + tree_type.uses_non_merk_data_storage(), + "sanity: opened tree must be a non-Merk data tree, got {tree_type:?}" + ); + + let ct_path: &[&[u8]] = &[TEST_LEAF, b"ct"]; + let prefix = + grovedb_storage::rocksdb_storage::RocksDbStorage::build_prefix(ct_path.as_ref().into()) + .unwrap(); + // No nested chunk ids — the shape an old peer would send. + let global_chunk_id = + encode_global_chunk_id(prefix, root_key, tree_type, vec![]).expect("encode chunk id"); + let packed = pack_nested_bytes(vec![global_chunk_id]).expect("pack chunk id"); + + let err = source + .fetch_chunk( + packed.as_slice(), + Some(&tx), + CURRENT_STATE_SYNC_VERSION, + grove_version, + ) + .expect_err("cursor-less append-only chunk request must be rejected"); + assert_not_supported_append_only(&err, "source-side fetch_chunk"); + } + + /// An EMPTY CommitmentTree state-syncs cleanly: it has no payload + /// entries, so the entry-replay path transfers a single empty page and + /// verification reduces to the empty-tree state-root convention. + #[test] + fn state_sync_empty_commitment_tree_succeeds() { + let grove_version = GroveVersion::latest(); + let source = make_test_grovedb(grove_version); + + source + .insert( + [TEST_LEAF].as_ref(), + b"ct_empty", + Element::empty_commitment_tree(4).expect("valid chunk power"), + None, + None, + grove_version, + ) + .unwrap() + .expect("insert commitment tree"); + + let source_root_hash = source + .root_hash(None, grove_version) + .unwrap() + .expect("source root hash"); + + let dest = sync_source_to_destination(&source, grove_version); + + let dest_root_hash = dest + .root_hash(None, grove_version) + .unwrap() + .expect("dest root hash"); + assert_eq!(source_root_hash, dest_root_hash); + + // The empty CT element survives and is intact on the destination. + let elem = dest + .get([TEST_LEAF].as_ref(), b"ct_empty", None, grove_version) + .unwrap() + .expect("CT element must exist on destination"); + match elem.underlying() { + Element::CommitmentTree(total_count, _, _) => { + assert_eq!(*total_count, 0); + } + other => panic!("expected CommitmentTree element, got {:?}", other), + } + + let dest_issues = dest + .verify_grovedb(None, true, false, grove_version) + .expect("dest verify_grovedb should run"); + assert!( + dest_issues.is_empty(), + "empty-CT destination must verify clean, got: {:?}", + dest_issues + ); + } + + /// Round trip for a populated MmrTree. + #[test] + fn state_sync_populated_mmr_tree_round_trip() { + let grove_version = GroveVersion::latest(); + let source = make_test_grovedb(grove_version); + + source + .insert( + [TEST_LEAF].as_ref(), + b"mmr", + Element::empty_mmr_tree(), + None, + None, + grove_version, + ) + .unwrap() + .expect("insert mmr tree"); + for i in 0u8..3 { + source + .mmr_tree_append( + [TEST_LEAF].as_ref(), + b"mmr", + format!("leaf-{i}").into_bytes(), + None, + grove_version, + ) + .unwrap() + .expect("append mmr leaf"); + } + + let dest = sync_source_to_destination(&source, grove_version); + + assert_eq!( + source.root_hash(None, grove_version).unwrap().unwrap(), + dest.root_hash(None, grove_version).unwrap().unwrap(), + ); + for i in 0u64..3 { + assert_eq!( + dest.mmr_tree_get_value([TEST_LEAF].as_ref(), b"mmr", i, None, grove_version) + .unwrap() + .expect("dest mmr leaf"), + Some(format!("leaf-{i}").into_bytes()), + "mmr leaf {i} must survive the sync" + ); + } + let dest_issues = dest + .verify_grovedb(None, true, false, grove_version) + .expect("dest verify_grovedb should run"); + assert!(dest_issues.is_empty(), "got: {:?}", dest_issues); + } + + /// Round trip for a populated BulkAppendTree spanning multiple + /// compacted chunks plus the buffer (chunk_power 2 → epoch of 4; + /// 10 values → 2 chunk blobs + 2 buffer entries). + #[test] + fn state_sync_populated_bulk_append_tree_round_trip() { + let grove_version = GroveVersion::latest(); + let source = make_test_grovedb(grove_version); + + source + .insert( + [TEST_LEAF].as_ref(), + b"bulk", + Element::empty_bulk_append_tree(2).expect("valid chunk power"), + None, + None, + grove_version, + ) + .unwrap() + .expect("insert bulk append tree"); + for i in 0u8..10 { + source + .bulk_append( + [TEST_LEAF].as_ref(), + b"bulk", + format!("value-{i}").into_bytes(), + None, + grove_version, + ) + .unwrap() + .expect("bulk append"); + } + + let dest = sync_source_to_destination(&source, grove_version); + + assert_eq!( + source.root_hash(None, grove_version).unwrap().unwrap(), + dest.root_hash(None, grove_version).unwrap().unwrap(), + ); + for i in 0u64..10 { + assert_eq!( + dest.bulk_get_value([TEST_LEAF].as_ref(), b"bulk", i, None, grove_version) + .unwrap() + .expect("dest bulk value"), + Some(format!("value-{i}").into_bytes()), + "bulk value {i} must survive the sync" + ); + } + let dest_issues = dest + .verify_grovedb(None, true, false, grove_version) + .expect("dest verify_grovedb should run"); + assert!(dest_issues.is_empty(), "got: {:?}", dest_issues); + } + + /// Round trip for a populated DenseAppendOnlyFixedSizeTree. + #[test] + fn state_sync_populated_dense_tree_round_trip() { + let grove_version = GroveVersion::latest(); + let source = make_test_grovedb(grove_version); + + source + .insert( + [TEST_LEAF].as_ref(), + b"dense", + Element::empty_dense_tree(4), + None, + None, + grove_version, + ) + .unwrap() + .expect("insert dense tree"); + for i in 0u8..3 { + source + .dense_tree_insert( + [TEST_LEAF].as_ref(), + b"dense", + vec![i; 32], + None, + grove_version, + ) + .unwrap() + .expect("dense insert"); + } + + let dest = sync_source_to_destination(&source, grove_version); + + assert_eq!( + source.root_hash(None, grove_version).unwrap().unwrap(), + dest.root_hash(None, grove_version).unwrap().unwrap(), + ); + for i in 0u16..3 { + assert_eq!( + dest.dense_tree_get([TEST_LEAF].as_ref(), b"dense", i, None, grove_version) + .unwrap() + .expect("dest dense value"), + Some(vec![i as u8; 32]), + "dense value {i} must survive the sync" + ); + } + let dest_issues = dest + .verify_grovedb(None, true, false, grove_version) + .expect("dest verify_grovedb should run"); + assert!(dest_issues.is_empty(), "got: {:?}", dest_issues); + } + + /// Entry payloads larger than the page byte budget force the transfer + /// across multiple pages; the multi-page path must round-trip too. + #[test] + fn state_sync_mmr_tree_multi_page_round_trip() { + let grove_version = GroveVersion::latest(); + let source = make_test_grovedb(grove_version); + + source + .insert( + [TEST_LEAF].as_ref(), + b"mmr_big", + Element::empty_mmr_tree(), + None, + None, + grove_version, + ) + .unwrap() + .expect("insert mmr tree"); + // Size each leaf from the page budget so any two leaves exceed one + // page: the transfer is guaranteed to split into at least two pages + // even if MAX_PAGE_BYTES is raised later. + let leaf_size = crate::replication::non_merk_sync::MAX_PAGE_BYTES / 2 + 1; + for i in 0u8..4 { + source + .mmr_tree_append( + [TEST_LEAF].as_ref(), + b"mmr_big", + vec![i; leaf_size], + None, + grove_version, + ) + .unwrap() + .expect("append big mmr leaf"); + } + + let dest = sync_source_to_destination(&source, grove_version); + + assert_eq!( + source.root_hash(None, grove_version).unwrap().unwrap(), + dest.root_hash(None, grove_version).unwrap().unwrap(), + ); + for i in 0u64..4 { + assert_eq!( + dest.mmr_tree_get_value([TEST_LEAF].as_ref(), b"mmr_big", i, None, grove_version) + .unwrap() + .expect("dest mmr leaf"), + Some(vec![i as u8; leaf_size]), + "big mmr leaf {i} must survive the sync" + ); + } + let dest_issues = dest + .verify_grovedb(None, true, false, grove_version) + .expect("dest verify_grovedb should run"); + assert!(dest_issues.is_empty(), "got: {:?}", dest_issues); + } + + /// Drive the full sync loop while mutating the wire bytes of commitment + /// tree pages. Every mutation must be rejected before the session can + /// complete — the target recomputes the state root from the replayed + /// payload and checks it against the parent binding. + fn try_sync_with_ct_page_mutation( + source: &TempGroveDb, + grove_version: &GroveVersion, + mutate_page: CtPageMutator, + ) -> Result<(), crate::Error> { + run_sync(source, grove_version, 64, Some(mutate_page)).map(|_| ()) + } + + /// Byzantine-source coverage: any tampering with commitment tree wire + /// bytes — a flipped entry byte, a stripped frontier, a dropped entry — + /// must fail the sync instead of committing corrupt state. + #[test] + fn state_sync_commitment_tree_tampered_pages_rejected() { + let grove_version = GroveVersion::latest(); + let source = make_test_grovedb(grove_version); + + source + .insert( + [TEST_LEAF].as_ref(), + b"ct", + Element::empty_commitment_tree(2).expect("valid chunk power"), + None, + None, + grove_version, + ) + .unwrap() + .expect("insert commitment tree"); + for i in 1u8..=6 { + source + .commitment_tree_insert_raw( + [TEST_LEAF].as_ref(), + b"ct", + [i; 32], + [i.wrapping_add(100); 32], + [i.wrapping_add(200); 32], + vec![i; 216], + None, + grove_version, + ) + .unwrap() + .expect("insert commitment tree note"); + } + + // Sanity: with the identity mutation the sync completes. + try_sync_with_ct_page_mutation(&source, grove_version, &|more, aux, entries| { + (more, aux, entries) + }) + .expect("un-tampered sync must succeed"); + + // 1. Flip one byte of one entry: the replayed payload no longer + // hashes to the bound state root. + let err = + try_sync_with_ct_page_mutation(&source, grove_version, &|more, aux, mut entries| { + if let Some(first) = entries.first_mut() { + first[0] ^= 0x01; + } + (more, aux, entries) + }) + .expect_err("flipped entry byte must be rejected"); + assert!( + format!("{err:?}").contains("state root mismatch after replay"), + "expected state-root rejection, got: {err:?}" + ); + + // 2. Strip the frontier from the first page. + let err = try_sync_with_ct_page_mutation(&source, grove_version, &|more, _aux, entries| { + (more, Vec::new(), entries) + }) + .expect_err("stripped frontier must be rejected"); + assert!( + format!("{err:?}").contains("missing the frontier"), + "expected missing-frontier rejection, got: {err:?}" + ); + + // 3. Tamper with the frontier bytes: the recomputed sinsemilla root + // diverges from the one bound into ct_state. + let err = + try_sync_with_ct_page_mutation(&source, grove_version, &|more, mut aux, entries| { + if !aux.is_empty() { + let last = aux.len() - 1; + aux[last] ^= 0x01; + } + (more, aux, entries) + }) + .expect_err("tampered frontier must be rejected"); + let msg = format!("{err:?}"); + assert!( + msg.contains("state root mismatch after replay") + || msg.contains("cannot open commitment tree") + || msg.contains("cannot compute commitment tree state root"), + "expected frontier-integrity rejection, got: {msg}" + ); + + // 4. Drop the last entry while still claiming the page is final. + let err = + try_sync_with_ct_page_mutation(&source, grove_version, &|more, aux, mut entries| { + if !more { + entries.pop(); + } + (more, aux, entries) + }) + .expect_err("dropped entry must be rejected"); + assert!( + format!("{err:?}").contains("replay incomplete"), + "expected incomplete-replay rejection, got: {err:?}" + ); + } + + /// Non-Merk subtrees interleaved with the subtree-batch boundary: + /// a batch size of 1 forces a transaction swap after every completed + /// subtree, including append-only ones. + #[test] + fn state_sync_non_merk_trees_with_batch_size_one() { + let grove_version = GroveVersion::latest(); + let source = make_test_grovedb(grove_version); + + source + .insert( + [TEST_LEAF].as_ref(), + b"ct", + Element::empty_commitment_tree(2).expect("valid chunk power"), + None, + None, + grove_version, + ) + .unwrap() + .expect("insert commitment tree"); + source + .commitment_tree_insert_raw( + [TEST_LEAF].as_ref(), + b"ct", + [1u8; 32], + [2u8; 32], + [3u8; 32], + vec![9u8; 216], + None, + grove_version, + ) + .unwrap() + .expect("insert commitment tree note"); + source + .insert( + [TEST_LEAF].as_ref(), + b"mmr", + Element::empty_mmr_tree(), + None, + None, + grove_version, + ) + .unwrap() + .expect("insert mmr tree"); + source + .mmr_tree_append( + [TEST_LEAF].as_ref(), + b"mmr", + b"leaf".to_vec(), + None, + grove_version, + ) + .unwrap() + .expect("append mmr leaf"); + source + .insert( + [ANOTHER_TEST_LEAF].as_ref(), + b"dense", + Element::empty_dense_tree(4), + None, + None, + grove_version, + ) + .unwrap() + .expect("insert dense tree"); + source + .dense_tree_insert( + [ANOTHER_TEST_LEAF].as_ref(), + b"dense", + vec![5u8; 32], + None, + grove_version, + ) + .unwrap() + .expect("dense insert"); + + // Same sync loop as the shared driver but with subtrees_batch_size + // of 1, exercising set_new_transaction between subtrees. + let dest = run_sync(&source, grove_version, 1, None) + .expect("state sync with batch size 1 should succeed"); + + assert_eq!( + source.root_hash(None, grove_version).unwrap().unwrap(), + dest.root_hash(None, grove_version).unwrap().unwrap(), + ); + let dest_issues = dest + .verify_grovedb(None, true, false, grove_version) + .expect("dest verify_grovedb should run"); + assert!(dest_issues.is_empty(), "got: {:?}", dest_issues); + } + + /// Empty append-only trees of every type survive state sync: nothing + /// to transfer, and verification reduces to the empty-tree state-root + /// conventions (NULL_HASH for MMR / bulk / dense). + #[test] + fn state_sync_empty_non_merk_trees_round_trip() { + let grove_version = GroveVersion::latest(); + let source = make_test_grovedb(grove_version); + + source + .insert( + [TEST_LEAF].as_ref(), + b"mmr_empty", + Element::empty_mmr_tree(), + None, + None, + grove_version, + ) + .unwrap() + .expect("insert empty mmr tree"); + source + .insert( + [TEST_LEAF].as_ref(), + b"bulk_empty", + Element::empty_bulk_append_tree(2).expect("valid chunk power"), + None, + None, + grove_version, + ) + .unwrap() + .expect("insert empty bulk tree"); + source + .insert( + [TEST_LEAF].as_ref(), + b"dense_empty", + Element::empty_dense_tree(4), + None, + None, + grove_version, + ) + .unwrap() + .expect("insert empty dense tree"); + + let dest = sync_source_to_destination(&source, grove_version); + + assert_eq!( + source.root_hash(None, grove_version).unwrap().unwrap(), + dest.root_hash(None, grove_version).unwrap().unwrap(), + ); + let dest_issues = dest + .verify_grovedb(None, true, false, grove_version) + .expect("dest verify_grovedb should run"); + assert!(dest_issues.is_empty(), "got: {:?}", dest_issues); + } + + /// Direct misuse/malformed-input coverage for the non-Merk restorer: + /// every wire-level validation must reject before touching storage. + #[test] + fn non_merk_restorer_rejects_malformed_input() { + use crate::replication::non_merk_sync::{ + encode_non_merk_page, NonMerkChunkId, NonMerkRestorer, + }; + + let grove_version = GroveVersion::latest(); + let db = make_test_grovedb(grove_version); + let tx = db.start_transaction(); + let path: Vec> = vec![TEST_LEAF.to_vec(), b"ct".to_vec()]; + + // Non append-only elements are rejected outright. + let err = NonMerkRestorer::new(Element::new_item(b"nope".to_vec()), [0u8; 32], [1u8; 32]) + .expect_err("item element must be rejected"); + assert!( + format!("{err:?}").contains("non append-only"), + "got: {err:?}" + ); + + // A commitment tree declaring 3 entries with chunk_power 2. + let mut restorer = + NonMerkRestorer::new(Element::CommitmentTree(3, 2, None), [0u8; 32], [1u8; 32]) + .expect("valid CT element"); + let frontier = b"opaque frontier bytes".to_vec(); + + // Malformed cursor: wrong length. + let page = encode_non_merk_page(false, frontier.clone(), vec![b"e".to_vec()]) + .expect("encode page"); + let err = restorer + .apply_page(&db, &tx, &path, &[0u8; 3], &page) + .expect_err("short chunk id must be rejected"); + assert!(format!("{err:?}").contains("17 bytes"), "got: {err:?}"); + + // Out-of-order cursor: wrong start position. + let bad_id = NonMerkChunkId { + start: 1, + state: 3, + param: 2, + } + .encode(); + let err = restorer + .apply_page(&db, &tx, &path, &bad_id, &page) + .expect_err("out-of-order cursor must be rejected"); + assert!(format!("{err:?}").contains("out of order"), "got: {err:?}"); + + let good_id = restorer.initial_chunk_id(); + + // Empty page data cannot even be decoded. + let err = restorer + .apply_page(&db, &tx, &path, &good_id, &[]) + .expect_err("empty page must be rejected"); + assert!( + format!("{err:?}").contains("missing more-flag"), + "got: {err:?}" + ); + + // A page claiming more data but carrying no entries would loop + // forever; it must be rejected. + let page = encode_non_merk_page(true, frontier.clone(), vec![]).expect("encode page"); + let err = restorer + .apply_page(&db, &tx, &path, &good_id, &page) + .expect_err("more-without-entries must be rejected"); + assert!( + format!("{err:?}").contains("carries no entries"), + "got: {err:?}" + ); + + // More entries than the element declares. + let too_many: Vec> = (0u8..4).map(|i| vec![i; 8]).collect(); + let page = encode_non_merk_page(false, frontier.clone(), too_many).expect("encode page"); + let err = restorer + .apply_page(&db, &tx, &path, &good_id, &page) + .expect_err("entry overflow must be rejected"); + assert!(format!("{err:?}").contains("overflows"), "got: {err:?}"); + + // A populated commitment tree page 0 without the frontier. + let page = + encode_non_merk_page(false, Vec::new(), vec![b"e".to_vec()]).expect("encode page"); + let err = restorer + .apply_page(&db, &tx, &path, &good_id, &page) + .expect_err("missing frontier must be rejected"); + assert!( + format!("{err:?}").contains("missing the frontier"), + "got: {err:?}" + ); + + // Finalizing before all entries arrived is rejected. + let err = restorer + .finalize(&db, &tx, &path) + .expect_err("incomplete replay must be rejected"); + assert!( + format!("{err:?}").contains("replay incomplete"), + "got: {err:?}" + ); + + // Aux data is only valid on a commitment tree's first page; any + // other tree type must reject it. + let mmr_path: Vec> = vec![TEST_LEAF.to_vec(), b"mmr".to_vec()]; + let mut mmr_restorer = + NonMerkRestorer::new(Element::MmrTree(0, None), [0u8; 32], [1u8; 32]) + .expect("valid MMR element"); + let page = encode_non_merk_page(false, b"bogus aux".to_vec(), vec![]).expect("encode page"); + let err = mmr_restorer + .apply_page(&db, &tx, &mmr_path, &mmr_restorer.initial_chunk_id(), &page) + .expect_err("aux on a non-CT page must be rejected"); + assert!( + format!("{err:?}").contains("unexpected aux"), + "got: {err:?}" + ); + + // A page arriving after the final page is rejected. + let dense_path: Vec> = vec![TEST_LEAF.to_vec(), b"dense".to_vec()]; + let mut dense_restorer = NonMerkRestorer::new( + Element::DenseAppendOnlyFixedSizeTree(0, 4, None), + [0u8; 32], + [1u8; 32], + ) + .expect("valid dense element"); + let final_page = encode_non_merk_page(false, Vec::new(), vec![]).expect("encode page"); + let dense_id = dense_restorer.initial_chunk_id(); + dense_restorer + .apply_page(&db, &tx, &dense_path, &dense_id, &final_page) + .expect("final page applies"); + let err = dense_restorer + .apply_page(&db, &tx, &dense_path, &dense_id, &final_page) + .expect_err("page after final must be rejected"); + assert!( + format!("{err:?}").contains("after the final page"), + "got: {err:?}" + ); + } }