Initial version of the marshal fuzzer

consensus/fuzz/Cargo.toml

@@ -20,7 +20,9 @@ cargo-fuzz = true
 [dependencies]
 arbitrary = { workspace = true, features = ["derive"] }
 bytes.workspace = true
+commonware-broadcast.workspace = true
 commonware-codec.workspace = true
+commonware-coding.workspace = true
 commonware-consensus = { workspace = true, features = ["mocks", "arbitrary"] }
 commonware-cryptography = { workspace = true, features = ["mocks"] }
 commonware-macros.workspace = true
@@ -362,3 +364,17 @@ path = "fuzz_targets/aggregation_cert_mock.rs"
 test = false
 doc = false
 bench = false
+
+[[bin]]
+name = "marshal_standard"
+path = "fuzz_targets/marshal_standard.rs"
+test = false
+doc = false
+bench = false
+
+[[bin]]
+name = "marshal_coding"
+path = "fuzz_targets/marshal_coding.rs"
+test = false
+doc = false
+bench = false

consensus/fuzz/fuzz_targets/marshal_coding.rs

@@ -0,0 +1,12 @@
+#![no_main]
+
+#[cfg(feature = "mocks")]
+mod fuzz {
+    use commonware_consensus::marshal::mocks::harness::CodingHarness;
+    use commonware_consensus_fuzz::marshal::{fuzz_marshal, MarshalFuzzInput};
+    use libfuzzer_sys::fuzz_target;
+
+    fuzz_target!(|input: MarshalFuzzInput| {
+        fuzz_marshal::<CodingHarness>(input);
+    });
+}

consensus/fuzz/fuzz_targets/marshal_standard.rs

@@ -0,0 +1,12 @@
+#![no_main]
+
+#[cfg(feature = "mocks")]
+mod fuzz {
+    use commonware_consensus::marshal::mocks::harness::StandardHarness;
+    use commonware_consensus_fuzz::marshal::{fuzz_marshal, MarshalFuzzInput};
+    use libfuzzer_sys::fuzz_target;
+
+    fuzz_target!(|input: MarshalFuzzInput| {
+        fuzz_marshal::<StandardHarness>(input);
+    });
+}

consensus/fuzz/src/lib.rs

@@ -7,6 +7,8 @@ pub mod byzzfuzz;
 pub mod disrupter;
 pub mod id_mock;
 pub mod invariants;
+#[cfg(feature = "mocks")]
+pub mod marshal;
 pub mod network;
 #[cfg(feature = "mocks")]
 pub mod ordered_broadcast;

consensus/fuzz/src/marshal/input.rs

@@ -0,0 +1,58 @@
+//! Libfuzzer-facing scenario: a byte tape (consumed by `FuzzRng`) plus a
+//! length-bounded list of events the driver replays against marshal.
+
+use arbitrary::Arbitrary;
+
+const MIN_EVENTS: usize = 1;
+const MAX_EVENTS: usize = 128;
+
+#[derive(Debug, Clone, Copy, Arbitrary)]
+pub enum MarshalEvent {
+    /// Notify marshal that a block was locally proposed.
+    Propose { block_idx: u8 },
+    /// Notify marshal that a block was verified.
+    Verify { block_idx: u8 },
+    /// Notify marshal that a block was certified.
+    Certify { block_idx: u8 },
+    /// Report a finalization for a block.
+    ReportFinalization { block_idx: u8 },
+    /// Report a notarization for a block.
+    ReportNotarization { block_idx: u8 },
+    /// Publish a block through the variant's local buffer (buffered
+    /// broadcast engine for Standard, shards engine for Coding) without
+    /// going through marshal's mailbox.
+    PublishViaVariant { block_idx: u8 },
+    /// Release one pending application ack, recording the popped height
+    /// as a delivery observation.
+    AckNext,
+    /// Abort the marshal actor and re-initialize from the same on-disk
+    /// state. Pending acks at the moment of restart are NOT signaled,
+    /// so marshal's persistent state retains them as un-processed and
+    /// the new instance must redeliver them (at-least-once).
+    Restart,
+    /// Yield without dispatching a marshal-facing event.
+    Idle,
+}
+
+#[derive(Debug, Clone)]
+pub struct MarshalFuzzInput {
+    pub raw_bytes: Vec<u8>,
+    pub events: Vec<MarshalEvent>,
+}
+
+impl Arbitrary<'_> for MarshalFuzzInput {
+    fn arbitrary(u: &mut arbitrary::Unstructured<'_>) -> arbitrary::Result<Self> {
+        let event_count = u.int_in_range(MIN_EVENTS..=MAX_EVENTS)?;
+        let mut events = Vec::with_capacity(event_count);
+        for _ in 0..event_count {
+            events.push(MarshalEvent::arbitrary(u)?);
+        }
+        let remaining = u.len().min(crate::MAX_RAW_BYTES);
+        let raw_bytes = if remaining == 0 {
+            vec![0]
+        } else {
+            u.bytes(remaining)?.to_vec()
+        };
+        Ok(Self { raw_bytes, events })
+    }
+}

consensus/fuzz/src/marshal/invariant.rs

@@ -0,0 +1,149 @@
+//! Marshal fuzz invariants.
+//!
+//! Each function asserts one property of the marshal-under-test against
+//! the driver's shadow state. The orchestrator [`check_all`] runs them
+//! in order; runner.rs only calls `check_all`.
+//!
+//! Conventions match the rest of the consensus fuzz crate: panics on
+//! violation, with a message that includes the relevant shadow state so
+//! libFuzzer's crash log is self-explanatory.
+
+use commonware_consensus::{marshal::mocks::harness::TestHarness, types::Height};
+use commonware_cryptography::Digestible;
+use std::collections::{BTreeMap, HashSet};
+
+/// Run every marshal invariant. Called from the driver at end of run.
+pub fn check_all<H: TestHarness>(
+    ready_prefix: u64,
+    delivery_log: &[Height],
+    segment_bounds: &[usize],
+    segment_starts: &[u64],
+    expected_redeliveries: &[Vec<Height>],
+    application_blocks: &BTreeMap<Height, H::ApplicationBlock>,
+    canonical: &[H::TestBlock],
+) {
+    check_ready_prefix_delivered(ready_prefix, delivery_log);
+    check_segment_ordering(segment_bounds, segment_starts, delivery_log);
+    check_redelivery_after_restart(expected_redeliveries, segment_bounds, delivery_log);
+    check_digest_fidelity::<H>(application_blocks, canonical);
+}
+
+/// Invariant: ready-prefix delivery.
+///
+/// Every height in `1..=ready_prefix` must appear at least once in
+/// `delivery_log`. The driver advances `ready_prefix` only when an
+/// above-floor `ReportFinalization` (or restart-triggered repair)
+/// observes a complete chain back to height 1, which is precisely when
+/// marshal is obliged to deliver the prefix.
+pub fn check_ready_prefix_delivered(ready_prefix: u64, delivery_log: &[Height]) {
+    let delivered_set: HashSet<u64> = delivery_log.iter().map(|h| h.get()).collect();
+    for h in 1..=ready_prefix {
+        assert!(
+            delivered_set.contains(&h),
+            "marshal violated at-least-once delivery: ready height {h} never reached \
+             the application (ready_prefix={ready_prefix}, delivered={delivered_set:?})",
+        );
+    }
+}
+
+/// Invariant: per-segment in-order delivery.
+///
+/// Within each actor instance (segment between restarts) marshal must
+/// deliver heights starting at `restored processed_height + 1` and
+/// advance strictly by one. The driver pre-populates `segment_starts`
+/// from each `setup.height` and `segment_bounds` from the delivery_log
+/// positions at restart boundaries.
+pub fn check_segment_ordering(
+    segment_bounds: &[usize],
+    segment_starts: &[u64],
+    delivery_log: &[Height],
+) {
+    assert_eq!(
+        segment_bounds.len(),
+        segment_starts.len() + 1,
+        "segment bookkeeping inconsistency",
+    );
+    for (segment_idx, window) in segment_bounds.windows(2).enumerate() {
+        let (start_idx, end_idx) = (window[0], window[1]);
+        if start_idx == end_idx {
+            continue;
+        }
+        let segment = &delivery_log[start_idx..end_idx];
+        let expected_start = segment_starts[segment_idx];
+        assert_eq!(
+            segment[0].get(),
+            expected_start,
+            "segment #{segment_idx} must start at restored processed height + 1 \
+             ({expected_start}), got {} (segment={:?})",
+            segment[0].get(),
+            segment,
+        );
+        for (offset, h) in segment.iter().enumerate() {
+            let expected = expected_start + offset as u64;
+            assert_eq!(
+                h.get(),
+                expected,
+                "marshal violated in-order delivery within segment #{segment_idx}: \
+                 expected height {expected}, observed {} (segment={:?})",
+                h.get(),
+                segment,
+            );
+        }
+    }
+}
+
+/// Invariant: at-least-once across restart.
+///
+/// Each height that was pending ack at the moment of restart `i` must
+/// reappear somewhere in `delivery_log[segment_bounds[i+1]..]`. Their
+/// ack handles were never signaled, so marshal's persistent state
+/// retains them as un-processed and the new instance is obliged to
+/// redeliver.
+pub fn check_redelivery_after_restart(
+    expected_redeliveries: &[Vec<Height>],
+    segment_bounds: &[usize],
+    delivery_log: &[Height],
+) {
+    for (restart_idx, expected) in expected_redeliveries.iter().enumerate() {
+        if expected.is_empty() {
+            continue;
+        }
+        let post_restart_start = segment_bounds[restart_idx + 1];
+        let post_restart: HashSet<u64> = delivery_log[post_restart_start..]
+            .iter()
+            .map(|h| h.get())
+            .collect();
+        for h in expected {
+            assert!(
+                post_restart.contains(&h.get()),
+                "marshal violated at-least-once across restart: height {} was \
+                 pending at restart #{} but was never redelivered \
+                 (post-restart deliveries={post_restart:?})",
+                h.get(),
+                restart_idx + 1,
+            );
+        }
+    }
+}
+
+/// Invariant: digest fidelity.
+///
+/// Every block surfaced in `application.blocks()` must match the
+/// canonical chain digest at its height. Re-emits after restart
+/// overwrite the prior `BTreeMap` entry, so the latest delivery at each
+/// height is the one we compare against canonical.
+pub fn check_digest_fidelity<H: TestHarness>(
+    application_blocks: &BTreeMap<Height, H::ApplicationBlock>,
+    canonical: &[H::TestBlock],
+) {
+    for (height, block) in application_blocks.iter() {
+        let canonical_block = &canonical[(height.get() - 1) as usize];
+        assert_eq!(
+            block.digest(),
+            H::digest(canonical_block),
+            "marshal delivered a block whose digest does not match the canonical \
+             chain at height {}",
+            height.get(),
+        );
+    }
+}

consensus/fuzz/src/marshal/mod.rs

@@ -0,0 +1,101 @@
+//! Fuzz driver for the marshal actor.
+//!
+//! Drives a single marshal actor under test by synthesizing every input
+//! marshal would normally receive from the consensus engine and from peers
+//! (blocks, notarizations, finalizations) and feeding them through the
+//! mailbox directly. Generic over `H: TestHarness` so the standard and
+//! coding variants share the same driver and corpora-per-binary discipline.
+//!
+//! # Invariants checked
+//!
+//! - **In-order delivery, no gaps within a marshal instance.** Within each
+//!   actor lifetime (segment between restarts), the first delivery is
+//!   `setup.height + 1` and subsequent deliveries advance strictly by one.
+//!   Marshal documents this guarantee on `Update::Block`.
+//! - **Ready-prefix delivery (anchor-based, chain-aware repair).**
+//!   When a `ReportFinalization` at height `h` arrives while block
+//!   `h` is locally available (durable or variant), marshal stores a
+//!   finalized anchor at `h` in its finalized archive. The driver
+//!   mirrors this with a persistent `finalized_anchors` set.
+//!
+//!   A `ReportFinalization` only triggers a repair wake when its
+//!   height is strictly above marshal's `processed_height` AND the
+//!   block is locally available. At-or-below-floor finalizations
+//!   are dropped by marshal's `store_finalization` (see
+//!   actor.rs:1732) and `try_repair_gaps` is gated on store success
+//!   (actor.rs:648). The driver mirrors this with a shadow
+//!   `processed_height`: initialized to `setup.height.get()`,
+//!   advanced on non-stale `AckNext`, and reset to
+//!   `setup.height.get()` after `Restart`.
+//!
+//!   On each repair wake (every above-floor `ReportFinalization`
+//!   that found its block, and every `Restart` after the variant
+//!   cache is cleared, since marshal's startup path runs
+//!   `try_repair_gaps` unconditionally) the driver finds the largest
+//!   anchor `a` for which every height `1..=a` is currently
+//!   available in (`durable_available` union `variant_available`).
+//!   If `a > ready_prefix`, the gap is repairable: marshal can walk
+//!   the chain from `a` back to 1 and deliver. The driver bumps
+//!   `ready_prefix = a` and promotes heights `prev_ready+1..=a` into
+//!   `durable_available` (marshal moves them to the finalized
+//!   archive, so they survive future restarts even if originally
+//!   sourced from the variant cache).
+//!
+//!   Availability state:
+//!     - `durable_available`: heights set by Propose / Verify /
+//!       Certify (marshal persists them), anchor blocks persisted by
+//!       `ReportFinalization` when the block was locally available
+//!       at that moment (marshal writes the block to
+//!       `finalized_blocks` alongside the finalization), plus
+//!       heights promoted by `ready_prefix` advances. Survives
+//!       restart.
+//!     - `variant_available`: heights set by `PublishViaVariant`
+//!       after confirmed local availability. Lives only in the
+//!       in-memory buffered / shards cache; cleared on `Restart`.
+//!     - `finalized_anchors`: heights at which a usable finalization
+//!       is stored. Survives restart.
+//! - **At-least-once across restart.** Heights pending ack at the moment
+//!   of restart are tracked. The new actor instance must redeliver each
+//!   of them at least once before the run ends.
+//! - **Digest fidelity.** Every block surfaced in `application.blocks()`
+//!   must match the canonical chain digest at its height.
+//! - **Durability acks.** `H::propose`/`H::verify`/`H::certify` return
+//!   `true` on durable persist; `false` surfaces an actor-died panic.
+//!
+//! # Variant buffer coverage
+//!
+//! `PublishViaVariant` exercises marshal's interaction with the local
+//! variant cache (buffered broadcast engine for Standard, shards engine
+//! for Coding). After publishing, the driver verifies the block actually
+//! landed in the local cache before counting it as `provided`; a publish
+//! that silently drops does not register.
+//!
+//! The marshal-mailbox path (`H::propose`/`H::verify`/`H::certify`) does
+//! NOT route through the shards mailbox for the coding harness: those
+//! wrappers call `handle.mailbox.proposed/verified/certified` directly.
+//! Shards-mailbox coverage is therefore exclusively via
+//! `PublishViaVariant`.
+//!
+//! # Known scope limitations
+//!
+//! - Single-validator only: peer-to-peer shard *dissemination* and
+//!   *reconstruction-from-peer-shards* are not exercised. Multi-validator
+//!   coding fuzz is a follow-up.
+//!
+//! # Layout
+//!
+//! - `input` defines the libFuzzer-facing scenario type.
+//! - `variant` adapts the standard / coding variant mailboxes to a
+//!   single publish trait the driver can call generically.
+//! - [`invariant`] holds the end-of-run assertions, one per property.
+//! - `runner` holds the deterministic-runtime driver and delegates
+//!   to [`invariant::check_all`] at the end.
+
+mod input;
+pub mod invariant;
+mod runner;
+mod variant;
+
+pub use input::{MarshalEvent, MarshalFuzzInput};
+pub use runner::fuzz_marshal;
+pub use variant::VariantPublish;