Implement scenario-based fuzzing with state prefixes

consensus/fuzz/Cargo.toml

@@ -50,6 +50,20 @@ tracing-subscriber.workspace = true
 [dev-dependencies]
 proptest.workspace = true
 
+[[bin]]
+name = "marshal_scenario_standard_deferred_cert_mock"
+path = "fuzz_targets/marshal_scenario_standard_deferred_cert_mock.rs"
+test = false
+doc = false
+bench = false
+
+[[bin]]
+name = "marshal_scenario_standard_inline_cert_mock"
+path = "fuzz_targets/marshal_scenario_standard_inline_cert_mock.rs"
+test = false
+doc = false
+bench = false
+
 [[bin]]
 name = "simplex_elector"
 path = "fuzz_targets/simplex_elector.rs"

consensus/fuzz/fuzz_targets/marshal_scenario_standard_deferred_cert_mock.rs

@@ -0,0 +1,14 @@
+#![no_main]
+
+#[cfg(feature = "mocks")]
+mod fuzz {
+    use commonware_consensus_fuzz::{
+        SimplexCertificateMock,
+        scenarios::{MarshalScenarioPrefixInput, fuzz_marshal_scenario_prefix_deferred},
+    };
+    use libfuzzer_sys::fuzz_target;
+
+    fuzz_target!(|input: MarshalScenarioPrefixInput| {
+        fuzz_marshal_scenario_prefix_deferred::<SimplexCertificateMock>(input);
+    });
+}

consensus/fuzz/fuzz_targets/marshal_scenario_standard_inline_cert_mock.rs

@@ -0,0 +1,14 @@
+#![no_main]
+
+#[cfg(feature = "mocks")]
+mod fuzz {
+    use commonware_consensus_fuzz::{
+        SimplexCertificateMock,
+        scenarios::{MarshalScenarioPrefixInput, fuzz_marshal_scenario_prefix_inline},
+    };
+    use libfuzzer_sys::fuzz_target;
+
+    fuzz_target!(|input: MarshalScenarioPrefixInput| {
+        fuzz_marshal_scenario_prefix_inline::<SimplexCertificateMock>(input);
+    });
+}

consensus/fuzz/src/disrupter.rs

@@ -398,7 +398,7 @@ where
         }
 
         // Optionally send mutated vote
-        if self.context.random_bool(0.5) {
+        if self.strategy.emit_byte_corruption() && self.context.random_bool(0.5) {
             let mutated = self.mutate_bytes(&msg);
             let _ = sender.send(Recipients::All, mutated, true);
         }
@@ -479,7 +479,7 @@ where
         }
 
         // Optionally send mutated certificate
-        if self.context.random_bool(0.5) {
+        if self.strategy.emit_byte_corruption() && self.context.random_bool(0.5) {
             let cert = self
                 .strategy
                 .mutate_certificate_bytes(self.context.as_mut(), &msg);
@@ -492,7 +492,7 @@ where
             return;
         }
         // Optionally send malformed resolver data
-        if self.context.random_bool(0.5) {
+        if self.strategy.emit_byte_corruption() && self.context.random_bool(0.5) {
             let mutated = self
                 .strategy
                 .mutate_resolver_bytes(self.context.as_mut(), &msg);

consensus/fuzz/src/lib.rs

@@ -20,6 +20,8 @@ pub mod network;
 pub mod ordered_broadcast;
 #[cfg(feature = "mocks")]
 pub mod ordered_broadcast_certificate_mock;
+#[cfg(feature = "mocks")]
+pub mod scenarios;
 pub mod simplex;
 pub(crate) mod simplex_audit;
 #[cfg(feature = "mocks")]

consensus/fuzz/src/marshal/end_to_end/app.rs

@@ -31,13 +31,69 @@ use commonware_cryptography::{
     Digestible, Sha256, certificate::Scheme, sha256::Digest as Sha256Digest,
 };
 use commonware_runtime::{Clock as _, deterministic};
-use commonware_utils::{FuzzRng, sync::Mutex};
+use commonware_utils::{FuzzRng, channel::mpsc, sync::Mutex};
 use futures::StreamExt;
 use rand_core::Rng as _;
 use std::{
     collections::HashMap, fmt, marker::PhantomData, num::NonZeroUsize, sync::Arc, time::Duration,
 };
 
+/// Buffer for delivery-height updates; far above the single-epoch block count so
+/// a subscriber never misses one.
+const PROGRESS_CHANNEL_CAPACITY: usize = 256;
+
+/// Shared delivery-height progress for one node, so a liveness watcher can await a
+/// target height on a channel instead of polling (and cloning) the application's
+/// block map. All clones share one subscriber set.
+#[derive(Clone)]
+pub(crate) struct ProgressHandle {
+    inner: Arc<Mutex<Progress>>,
+}
+
+struct Progress {
+    latest: u64,
+    subscribers: Vec<mpsc::Sender<u64>>,
+}
+
+impl ProgressHandle {
+    pub(crate) fn new() -> Self {
+        Self {
+            inner: Arc::new(Mutex::new(Progress {
+                latest: 0,
+                subscribers: Vec::new(),
+            })),
+        }
+    }
+
+    /// Highest delivered height recorded so far.
+    pub(crate) fn latest(&self) -> u64 {
+        self.inner.lock().latest
+    }
+
+    /// Subscribe to delivery-height updates: returns the current height and a
+    /// receiver yielding each subsequent height.
+    pub(crate) fn subscribe(&self) -> (u64, mpsc::Receiver<u64>) {
+        let mut progress = self.inner.lock();
+        let (tx, rx) = mpsc::channel(PROGRESS_CHANNEL_CAPACITY);
+        progress.subscribers.push(tx);
+        (progress.latest, rx)
+    }
+
+    fn record(&self, height: u64) {
+        let mut progress = self.inner.lock();
+        if height <= progress.latest {
+            return;
+        }
+        progress.latest = height;
+        progress.subscribers.retain(|tx| {
+            !matches!(
+                tx.try_send(height),
+                Err(mpsc::error::TrySendError::Closed(_))
+            )
+        });
+    }
+}
+
 #[derive(Clone)]
 pub(crate) struct DeliveryReporter<C>
 where
@@ -48,6 +104,7 @@ where
     tips: Arc<Mutex<Vec<(Round, Height, Sha256Digest)>>>,
     max_pending_acks: Option<NonZeroUsize>,
     stack: Arc<str>,
+    progress: Option<ProgressHandle>,
 }
 
 impl<C> DeliveryReporter<C>
@@ -66,8 +123,16 @@ where
             tips: Arc::new(Mutex::new(Vec::new())),
             max_pending_acks,
             stack,
+            progress: None,
         }
     }
+
+    /// Publish delivery-height updates to `progress` so a liveness watcher can
+    /// await progress on a channel instead of polling.
+    pub(crate) fn with_progress(mut self, progress: ProgressHandle) -> Self {
+        self.progress = Some(progress);
+        self
+    }
 }
 
 impl<C> Reporter for DeliveryReporter<C>
@@ -142,6 +207,9 @@ where
                         );
                     }
                 }
+                if let Some(progress) = &self.progress {
+                    progress.record(block.height().get());
+                }
             }
         }
         self.application.report(activity)

consensus/fuzz/src/marshal/end_to_end/invariants.rs

@@ -346,34 +346,41 @@ pub(super) fn check_certificate_backfill_retry<P: commonware_cryptography::Publi
 }
 
 /// Run block-ordering and agreement invariants.
-pub(super) fn check_all_blocks<D: ConsensusParentDigest, P: PublicKey>(
+///
+/// `floor` is the height every honest node started from (0 for genesis-started
+/// clusters); it anchors [`check_in_order`]'s first-delivery check.
+pub(crate) fn check_all_blocks<D: ConsensusParentDigest, P: PublicKey>(
     honest_apps: &[(usize, AuditedApplication<D, P>)],
     genesis: Sha256Digest,
+    floor: Height,
     stack: Option<&str>,
 ) {
     let stack = stack.unwrap_or("unspecified");
     for (idx, app) in honest_apps {
-        check_local_blocks(*idx, app, genesis, stack);
+        check_local_blocks(*idx, app, genesis, floor, stack);
     }
     agreement(honest_apps, stack);
 }
 
 /// Run block-ordering and parent-linkage invariants for one node.
-pub(super) fn check_local_blocks<D: ConsensusParentDigest, P: PublicKey>(
+///
+/// `floor` is the height the node started from (0 for a genesis-started node).
+pub(crate) fn check_local_blocks<D: ConsensusParentDigest, P: PublicKey>(
     idx: usize,
     app: &AuditedApplication<D, P>,
     genesis: Sha256Digest,
+    floor: Height,
     stack: &str,
 ) {
-    check_in_order(idx, &app.delivered(), stack);
+    check_in_order(idx, &app.delivered(), floor, stack);
     check_parent_linkage(idx, &app.blocks(), genesis, stack);
 }
 
 /// Invariant: every pair of consecutively delivered blocks is parent-linked.
 ///
 /// [`check_in_order`] runs first and guarantees that after exact duplicate
 /// deliveries are collapsed, the by-height snapshot is one contiguous chain.
-fn check_parent_linkage<D: ConsensusParentDigest, P: PublicKey>(
+pub(crate) fn check_parent_linkage<D: ConsensusParentDigest, P: PublicKey>(
     idx: usize,
     blocks: &AuditedBlocks<D, P>,
     genesis: Sha256Digest,
@@ -473,18 +480,31 @@ pub(super) fn check_pending_acks<B: Block>(
 
 /// Invariant: per-node in-order, gap-free delivery.
 ///
-/// Walks the arrival-ordered delivery log. Delivery starts either at the
-/// genesis floor block (height 0, surfaced on a fresh start) or at the first
-/// finalized container (height 1), then every subsequent delivery must advance
-/// by exactly one or repeat the identical `(height, digest)`. Because delivery
-/// is at-least-once, any number of exact duplicates is allowed; an out-of-order
-/// delivery, gap, or same-height fork fails the check.
-fn check_in_order<D: Debug + PartialEq>(idx: usize, delivered: &[(Height, D)], stack: &str) {
+/// Walks the arrival-ordered delivery log. The first delivery must be either
+/// height 0 (the anchor block surfaced on a fresh start) or the node's floor
+/// height: a genesis-started node has floor 0, whose first finalized container
+/// is height 1, while a node started from a floor at height `H` (via
+/// `Start::Floor` or `set_floor`) delivers that anchor block at `H` first. Every
+/// subsequent delivery must advance by exactly one or repeat the identical
+/// `(height, digest)`. Because delivery is at-least-once, any number of exact
+/// duplicates is allowed; an out-of-order delivery, gap, or same-height fork
+/// fails the check.
+fn check_in_order<D: Debug + PartialEq>(
+    idx: usize,
+    delivered: &[(Height, D)],
+    floor: Height,
+    stack: &str,
+) {
+    // A genesis-started node (floor 0) accepts a first delivery at height 0 or
+    // 1; a node started above genesis at floor `H` accepts 0 or `H`. `max(1)`
+    // keeps the genesis floor's anchor at 1, so genesis nodes behave exactly as
+    // before.
+    let anchor = floor.get().max(1);
     let first = delivered.first().map_or(0, |(height, _)| height.get());
     assert!(
-        first <= 1,
-        "node{idx} first delivery at height {first} is above the genesis floor + 1; \
-         sequence={delivered:?}; stack={stack}",
+        first == 0 || first == anchor,
+        "node{idx} first delivery at height {first} is neither genesis (0) nor the floor \
+         anchor {anchor}; sequence={delivered:?}; stack={stack}",
     );
     for window in delivered.windows(2) {
         let (height_0, digest_0) = &window[0];
@@ -519,7 +539,7 @@ fn check_in_order<D: Debug + PartialEq>(idx: usize, delivered: &[(Height, D)], s
 ///
 /// This detects conflicting finalization, fork divergence, and recovery that
 /// delivers a different block at an already observed height.
-fn agreement<B: Block<Digest = Sha256Digest>>(
+pub(crate) fn agreement<B: Block<Digest = Sha256Digest>>(
     honest_apps: &[(usize, Application<B>)],
     stack: &str,
 ) {
@@ -703,6 +723,7 @@ mod tests {
                 (Height::new(1), digest(0xA)),
                 (Height::new(2), digest(0xB)),
             ],
+            Height::zero(),
             "test",
         );
     }
@@ -716,6 +737,7 @@ mod tests {
                 (Height::new(1), digest(0xA)),
                 (Height::new(1), digest(0xA)),
             ],
+            Height::zero(),
             "test",
         );
     }
@@ -726,6 +748,7 @@ mod tests {
         check_in_order(
             0,
             &[(Height::new(1), digest(0xA)), (Height::new(1), digest(0xB))],
+            Height::zero(),
             "test",
         );
     }
@@ -736,6 +759,65 @@ mod tests {
         check_in_order(
             0,
             &[(Height::new(1), digest(0xA)), (Height::new(3), digest(0xB))],
+            Height::zero(),
+            "test",
+        );
+    }
+
+    #[test]
+    fn floor_started_delivery_from_floor_is_allowed() {
+        // A node started from a floor at height 5 delivers its anchor block at
+        // height 5 first, then advances contiguously.
+        check_in_order(
+            0,
+            &[
+                (Height::new(5), digest(0xA)),
+                (Height::new(6), digest(0xB)),
+                (Height::new(7), digest(0xC)),
+            ],
+            Height::new(5),
+            "test",
+        );
+    }
+
+    #[test]
+    #[should_panic(expected = "in-order delivery")]
+    fn floor_started_gap_above_floor_is_rejected() {
+        // First delivery at the floor is accepted; the jump to height 7 skips 6.
+        check_in_order(
+            0,
+            &[(Height::new(5), digest(0xA)), (Height::new(7), digest(0xB))],
+            Height::new(5),
+            "test",
+        );
+    }
+
+    #[test]
+    #[should_panic(expected = "floor anchor")]
+    fn floor_started_delivery_below_floor_is_rejected() {
+        // A node with floor 5 must not deliver first below its floor.
+        check_in_order(
+            0,
+            &[(Height::new(3), digest(0xA)), (Height::new(4), digest(0xB))],
+            Height::new(5),
+            "test",
+        );
+    }
+
+    #[test]
+    fn genesis_started_delivery_is_unchanged() {
+        // Floor 0 accepts a first delivery at genesis (0) or the first finalized
+        // container (1), exactly as before the floor generalization.
+        check_in_order(
+            0,
+            &[(Height::zero(), digest(0xA)), (Height::new(1), digest(0xB))],
+            Height::zero(),
+            "test",
+        );
+        check_in_order(
+            0,
+            &[(Height::new(1), digest(0xA)), (Height::new(2), digest(0xB))],
+            Height::zero(),
             "test",
         );
     }

consensus/fuzz/src/marshal/end_to_end/mod.rs

@@ -32,12 +32,12 @@
 
 use commonware_consensus::marshal::mocks::harness::BLOCKS_PER_EPOCH;
 
-mod app;
+pub(crate) mod app;
 mod block_disrupter;
 mod coding_disrupter;
 pub(crate) mod coding_stack;
 mod input;
-pub(super) mod invariants;
+pub(crate) mod invariants;
 mod runner;
 mod scenario;
 pub(crate) mod twins;