Merge remote-tracking branch 'origin/main' into danlaine/contiguous-2

danlaine · danlaine · commit 17b68c31dfe5 · 2026-04-20T12:34:16.000-04:00
diff --git a/storage/fuzz/fuzz_targets/fixed_journal_operations.rs b/storage/fuzz/fuzz_targets/fixed_journal_operations.rs
@@ -3,7 +3,13 @@
 use arbitrary::{Arbitrary, Result, Unstructured};
 use commonware_cryptography::{Hasher as _, Sha256};
 use commonware_runtime::{buffer::paged::CacheRef, deterministic, Metrics, Runner};
-use commonware_storage::journal::contiguous::fixed::{Config as JournalConfig, Journal};
+use commonware_storage::journal::{
+    contiguous::{
+        fixed::{Config as JournalConfig, Journal},
+        Contiguous as _, Many, Mutable as _, Reader,
+    },
+    Error,
+};
 use commonware_utils::{NZUsize, NZU16, NZU64};
 use futures::{pin_mut, StreamExt};
 use libfuzzer_sys::fuzz_target;
@@ -12,12 +18,23 @@ use std::num::NonZeroU16;
 const MAX_REPLAY_BUF: usize = 2048;
 const MAX_WRITE_BUF: usize = 2048;
 const MAX_OPERATIONS: usize = 50;
+const MAX_APPEND_MANY: u8 = 20;
+const MAX_READ_MANY: usize = 16;
 
 fn bounded_non_zero(u: &mut Unstructured<'_>) -> Result<usize> {
     let v = u.int_in_range(1..=MAX_REPLAY_BUF)?;
     Ok(v)
 }
 
+fn bounded_append_count(u: &mut Unstructured<'_>) -> Result<u8> {
+    u.int_in_range(0..=MAX_APPEND_MANY)
+}
+
+fn bounded_positions(u: &mut Unstructured<'_>) -> Result<Vec<u64>> {
+    let len = u.int_in_range(0..=MAX_READ_MANY)?;
+    (0..len).map(|_| u64::arbitrary(u)).collect()
+}
+
 #[derive(Arbitrary, Debug, Clone)]
 enum JournalOperation {
     Append {
@@ -42,10 +59,31 @@ enum JournalOperation {
     },
     Restart,
     Destroy,
+    ReadMany {
+        #[arbitrary(with = bounded_positions)]
+        positions: Vec<u64>,
+    },
     AppendMany {
+        #[arbitrary(with = bounded_append_count)]
         count: u8,
     },
+    AppendNested {
+        #[arbitrary(with = bounded_append_count)]
+        count_a: u8,
+        #[arbitrary(with = bounded_append_count)]
+        count_b: u8,
+    },
+    RewindTo {
+        keep_value: u64,
+    },
     MultipleSync,
+    TryReadSync {
+        pos: u64,
+    },
+    PruningBoundary,
+    InitAtSize {
+        size: u64,
+    },
 }
 
 #[derive(Debug)]
@@ -99,6 +137,33 @@ fn fuzz(input: FuzzInput) {
                     }
                 }
 
+                JournalOperation::ReadMany { positions } => {
+                    let reader = journal.reader().await;
+                    let bounds = reader.bounds();
+                    // Map fuzz positions into valid, sorted, deduplicated positions
+                    let mut mapped: Vec<u64> = positions
+                        .iter()
+                        .filter_map(|p| {
+                            if bounds.is_empty() {
+                                return None;
+                            }
+                            let len = bounds.end - bounds.start;
+                            Some(bounds.start + (*p % len))
+                        })
+                        .collect();
+                    mapped.sort_unstable();
+                    mapped.dedup();
+                    if !mapped.is_empty() {
+                        let batch = reader.read_many(&mapped).await.unwrap();
+                        assert_eq!(batch.len(), mapped.len());
+                        // Cross-check against individual reads
+                        for (i, &pos) in mapped.iter().enumerate() {
+                            let single = reader.read(pos).await.unwrap();
+                            assert_eq!(batch[i], single);
+                        }
+                    }
+                }
+
                 JournalOperation::Size => {
                     let size = journal.size();
                     assert_eq!(journal_size, size, "unexpected size");
@@ -172,11 +237,20 @@ fn fuzz(input: FuzzInput) {
                 }
 
                 JournalOperation::AppendMany { count } => {
-                    for _ in 0..*count {
-                        let digest = Sha256::hash(&next_value.to_be_bytes());
-                        journal.append(&digest).await.unwrap();
-                        next_value += 1;
-                        journal_size += 1;
+                    if *count == 0 {
+                        // Exercise the EmptyAppend error path
+                        let err = journal.append_many(Many::Flat(&[])).await;
+                        assert!(matches!(err, Err(Error::EmptyAppend)));
+                    } else {
+                        let items: Vec<_> = (0..*count)
+                            .map(|_| {
+                                let d = Sha256::hash(&next_value.to_be_bytes());
+                                next_value += 1;
+                                d
+                            })
+                            .collect();
+                        journal.append_many(Many::Flat(&items)).await.unwrap();
+                        journal_size += *count as u64;
                     }
                 }
 
@@ -185,6 +259,76 @@ fn fuzz(input: FuzzInput) {
                     journal.sync().await.unwrap();
                     journal.sync().await.unwrap();
                 }
+
+                JournalOperation::AppendNested { count_a, count_b } => {
+                    if *count_a == 0 && *count_b == 0 {
+                        let err = journal.append_many(Many::Nested(&[&[], &[]])).await;
+                        assert!(matches!(err, Err(Error::EmptyAppend)));
+                    } else {
+                        let items_a: Vec<_> = (0..*count_a)
+                            .map(|_| {
+                                let d = Sha256::hash(&next_value.to_be_bytes());
+                                next_value += 1;
+                                d
+                            })
+                            .collect();
+                        let items_b: Vec<_> = (0..*count_b)
+                            .map(|_| {
+                                let d = Sha256::hash(&next_value.to_be_bytes());
+                                next_value += 1;
+                                d
+                            })
+                            .collect();
+                        let slices: &[&[_]] = &[&items_a, &items_b];
+                        journal.append_many(Many::Nested(slices)).await.unwrap();
+                        journal_size += *count_a as u64 + *count_b as u64;
+                    }
+                }
+
+                JournalOperation::RewindTo { keep_value } => {
+                    if journal_size > oldest_retained_pos {
+                        let target = Sha256::hash(&keep_value.to_be_bytes());
+                        let new_size = journal.rewind_to(|item| *item == target).await.unwrap();
+                        journal.sync().await.unwrap();
+                        journal_size = new_size;
+                        oldest_retained_pos = journal.reader().await.bounds().start;
+                    }
+                }
+
+                JournalOperation::TryReadSync { pos } => {
+                    let reader = journal.reader().await;
+                    let bounds = reader.bounds();
+                    if bounds.contains(pos) {
+                        // Cross-check: sync result must match async result
+                        if let Some(sync_val) = reader.try_read_sync(*pos) {
+                            let async_val = reader.read(*pos).await.unwrap();
+                            assert_eq!(sync_val, async_val);
+                        }
+                    }
+                }
+
+                JournalOperation::PruningBoundary => {
+                    let boundary = journal.pruning_boundary();
+                    assert_eq!(boundary, oldest_retained_pos);
+                }
+
+                JournalOperation::InitAtSize { size } => {
+                    // Cap to avoid excessive memory use
+                    let target_size = *size % 256;
+                    drop(journal);
+                    journal = Journal::init_at_size(
+                        context
+                            .with_label("journal")
+                            .with_attribute("instance", restarts),
+                        cfg.clone(),
+                        target_size,
+                    )
+                    .await
+                    .unwrap();
+                    restarts += 1;
+                    journal_size = journal.size();
+                    oldest_retained_pos = journal.reader().await.bounds().start;
+                }
             }
         }
     });
diff --git a/storage/src/qmdb/current/batch.rs b/storage/src/qmdb/current/batch.rs
@@ -25,11 +25,13 @@ use crate::{
     },
     Context,
 };
+use ahash::AHasher;
 use commonware_codec::Codec;
 use commonware_cryptography::{Digest, Hasher};
 use commonware_utils::bitmap::{Prunable as BitMap, Readable as BitmapReadable};
 use std::{
-    collections::{BTreeMap, BTreeSet},
+    collections::{BTreeSet, HashMap},
+    hash::BuildHasherDefault,
     sync::Arc,
 };
 
@@ -41,17 +43,20 @@ use std::{
 #[derive(Clone, Debug, Default)]
 pub(crate) struct ChunkOverlay<const N: usize> {
     /// Dirty chunks: chunk_idx -> materialized chunk bytes.
-    pub(crate) chunks: BTreeMap<usize, [u8; N]>,
+    ///
+    /// `ahash` (fast on integer keys) with `BuildHasherDefault` (no per-construction RNG
+    /// sampling). Iteration order is not observed by any consumer.
+    pub(crate) chunks: HashMap<usize, [u8; N], BuildHasherDefault<AHasher>>,
     /// Total number of bits (parent + new operations).
     pub(crate) len: u64,
 }
 
 impl<const N: usize> ChunkOverlay<N> {
     const CHUNK_BITS: u64 = BitMap::<N>::CHUNK_SIZE_BITS;
 
-    const fn new(len: u64) -> Self {
+    fn new(len: u64) -> Self {
         Self {
-            chunks: BTreeMap::new(),
+            chunks: HashMap::default(),
             len,
         }
     }
@@ -590,6 +595,7 @@ where
     // compute_db_root sees newly completed chunks. Using bitmap_parent alone would miss chunks
     // that transitioned from partial to complete in this batch.
     let bitmap_batch = BitmapBatch::Layer(Arc::new(BitmapBatchLayer {
+        pruned_chunks: bitmap_parent.pruned_chunks(),
         parent: bitmap_parent.clone(),
         overlay: Arc::new(overlay),
     }));
@@ -648,6 +654,10 @@ pub(crate) struct BitmapBatchLayer<const N: usize> {
     pub(crate) parent: BitmapBatch<N>,
     /// Chunk-level overlay: materialized bytes for every chunk that differs from parent.
     pub(crate) overlay: Arc<ChunkOverlay<N>>,
+    /// Pruned-chunk count, copied from `parent` at construction. Invariant across the whole
+    /// layer chain (pruning only happens on the committed base), so caching here lets
+    /// `BitmapBatch::pruned_chunks` return in O(1) instead of walking to the Base.
+    pub(crate) pruned_chunks: usize,
 }
 
 impl<const N: usize> BitmapBatch<N> {
@@ -660,14 +670,17 @@ impl<const N: usize> BitmapReadable<N> for BitmapBatch<N> {
     }
 
     fn get_chunk(&self, idx: usize) -> [u8; N] {
-        match self {
-            Self::Base(bm) => *bm.get_chunk(idx),
-            Self::Layer(layer) => {
-                // Check overlay first; fall through to parent if unmodified.
-                if let Some(&chunk) = layer.overlay.get(idx) {
-                    chunk
-                } else {
-                    layer.parent.get_chunk(idx)
+        // Walk the layer chain. Each layer's overlay either holds the chunk (return it) or
+        // doesn't (descend).
+        let mut current = self;
+        loop {
+            match current {
+                Self::Base(bm) => return *bm.get_chunk(idx),
+                Self::Layer(layer) => {
+                    if let Some(&chunk) = layer.overlay.get(idx) {
+                        return chunk;
+                    }
+                    current = &layer.parent;
                 }
             }
         }
@@ -691,7 +704,7 @@ impl<const N: usize> BitmapReadable<N> for BitmapBatch<N> {
     fn pruned_chunks(&self) -> usize {
         match self {
             Self::Base(bm) => bm.pruned_chunks(),
-            Self::Layer(layer) => layer.parent.pruned_chunks(),
+            Self::Layer(layer) => layer.pruned_chunks,
         }
     }
 
@@ -723,8 +736,13 @@ impl<const N: usize> BitmapBatch<N> {
         }
 
         // Slow path: create a new layer.
+        let pruned_chunks = self.pruned_chunks();
         let parent = self.clone();
-        *self = Self::Layer(Arc::new(BitmapBatchLayer { parent, overlay }));
+        *self = Self::Layer(Arc::new(BitmapBatchLayer {
+            parent,
+            overlay,
+            pruned_chunks,
+        }));
     }
 
     /// Flatten all layers back to a single `Base(Arc<BitMap<N>>)`.
diff --git a/utils/src/bitmap/mod.rs b/utils/src/bitmap/mod.rs
@@ -967,7 +967,13 @@ pub trait Readable<const N: usize> {
     where
         Self: Sized,
     {
-        OnesIter { bitmap: self, pos }
+        let len = self.len();
+        let pruned_start = (self.pruned_chunks() as u64) * BitMap::<N>::CHUNK_SIZE_BITS;
+        OnesIter {
+            bitmap: self,
+            pos: pos.max(pruned_start),
+            len,
+        }
     }
 }
 
@@ -1002,19 +1008,19 @@ impl<const N: usize> Readable<N> for BitMap<N> {
 pub struct OnesIter<'a, B, const N: usize> {
     bitmap: &'a B,
     pos: u64,
+    /// Cached `bitmap.len()` at iterator construction. The underlying bitmap is borrowed
+    /// immutably for the iterator's lifetime, so this can never change mid-iteration.
+    /// For layered bitmaps (e.g. `BitmapBatch`), `len()` walks the layer chain, so caching
+    /// this avoids that walk on every `next`.
+    len: u64,
 }
 
 impl<B: Readable<N>, const N: usize> iter::Iterator for OnesIter<'_, B, N> {
     type Item = u64;
 
     fn next(&mut self) -> Option<u64> {
-        let len = self.bitmap.len();
         let chunk_bits = BitMap::<N>::CHUNK_SIZE_BITS;
-        let pruned_start = self.bitmap.pruned_chunks() as u64 * chunk_bits;
-        if self.pos < pruned_start {
-            self.pos = pruned_start;
-        }
-        while self.pos < len {
+        while self.pos < self.len {
             let chunk_idx = BitMap::<N>::to_chunk_index(self.pos);
             let chunk = self.bitmap.get_chunk(chunk_idx);
             let chunk_start = chunk_idx as u64 * chunk_bits;
@@ -1027,7 +1033,7 @@ impl<B: Readable<N>, const N: usize> iter::Iterator for OnesIter<'_, B, N> {
                     let found = chunk_start
                         + (byte_idx * 8 + bit_in_byte) as u64
                         + masked.trailing_zeros() as u64;
-                    if found >= len {
+                    if found >= self.len {
                         return None;
                     }
                     self.pos = found + 1;
