[marshal] Start at Anchor (#3855)

Author: patrick-ogrady

consensus/conformance.toml

@@ -22,6 +22,14 @@ hash = "51b04637257c54cecdc4e5ded4568efccafbc08f20be3fe0951369cab43ba2b6"
 n_cases = 65536
 hash = "e53b4b684f0344590427aae976d298e3b6fd2b08ef46ff2cad5def04b4f7690b"
 
+["commonware_consensus::marshal::conformance::CodingStorageConformance"]
+n_cases = 32
+hash = "9234e6644b5aaf723be352e7e17b8613efafb7e80869e8dffe8b5527c04128ef"
+
+["commonware_consensus::marshal::conformance::StandardStorageConformance"]
+n_cases = 32
+hash = "05033526f41131ceb367df9177c4090d9670d6ea0119d8478211f894e2d557d7"
+
 ["commonware_consensus::marshal::resolver::handler::tests::conformance::CodecConformance<Key<D>>"]
 n_cases = 65536
 hash = "481cd68f2e6452c0dda512c75d74ddd2e19ac6c9fb19c642c1a152a0d830c1b2"

consensus/src/marshal/coding/mod.rs

@@ -779,6 +779,11 @@ mod tests {
         harness::ack_pipeline_backlog_persists_on_restart::<CodingHarness>();
     }
 
+    #[test_traced("WARN")]
+    fn test_coding_genesis_emitted_once() {
+        harness::genesis_emitted_once::<CodingHarness>();
+    }
+
     #[test_traced("WARN")]
     fn test_coding_proposed_success_implies_recoverable_after_restart() {
         harness::proposed_success_implies_recoverable_after_restart::<CodingHarness>(0..16);

consensus/src/marshal/config.rs

@@ -18,7 +18,7 @@ use std::num::{NonZeroU64, NonZeroUsize};
 pub enum Start<S: Scheme, C: Digest, B> {
     /// Start from the height-zero genesis block.
     Genesis(B),
-    /// Start from an already-processed finalized commitment.
+    /// Start from a finalized commitment.
     Floor(Finalization<S, C>),
 }
 

consensus/src/marshal/conformance.rs

@@ -0,0 +1,127 @@
+//! Marshal storage conformance tests.
+
+use super::mocks::{
+    application::Application,
+    harness::{
+        self, CodingHarness, StandardHarness, TestHarness, ValidatorHandle, ValidatorSetup,
+        BLOCKS_PER_EPOCH, NAMESPACE, NUM_VALIDATORS, QUORUM, V,
+    },
+};
+use crate::{
+    simplex::{scheme::bls12381_threshold::vrf as bls12381_threshold_vrf, types::Proposal},
+    types::{Epoch, Height, Round, View},
+};
+use commonware_conformance::{conformance_tests, Conformance};
+use commonware_cryptography::certificate::{mocks::Fixture, ConstantProvider};
+use commonware_runtime::{deterministic, Clock, Runner, Supervisor as _};
+use commonware_utils::NZUsize;
+use rand::Rng;
+use std::time::Duration;
+
+const CASES: usize = 32;
+
+struct StandardStorageConformance;
+struct CodingStorageConformance;
+
+impl Conformance for StandardStorageConformance {
+    async fn commit(seed: u64) -> Vec<u8> {
+        marshal_commit::<StandardHarness>(seed)
+    }
+}
+
+impl Conformance for CodingStorageConformance {
+    async fn commit(seed: u64) -> Vec<u8> {
+        marshal_commit::<CodingHarness>(seed)
+    }
+}
+
+fn marshal_commit<H: TestHarness>(seed: u64) -> Vec<u8> {
+    let runner = deterministic::Runner::new(
+        deterministic::Config::default()
+            .with_seed(seed)
+            .with_timeout(Some(Duration::from_secs(30))),
+    );
+    runner.start(|mut context| async move {
+        let Fixture {
+            participants,
+            schemes,
+            ..
+        } = bls12381_threshold_vrf::fixture::<V, _>(&mut context, NAMESPACE, NUM_VALIDATORS);
+        let mut oracle = harness::setup_network_with_participants(
+            context.child("network"),
+            NZUsize!(1),
+            participants.clone(),
+        )
+        .await;
+
+        let validator = participants[0].clone();
+        let provider = ConstantProvider::new(schemes[0].clone());
+        let application = Application::<H::ApplicationBlock>::manual_ack();
+        let setup = H::setup_validator_with(
+            context.child("validator"),
+            &mut oracle,
+            validator,
+            provider,
+            NZUsize!(1),
+            application,
+        )
+        .await;
+
+        assert_eq!(setup.application.acknowledged().await, Height::zero());
+        wait_processed(&mut context, &setup, Height::zero()).await;
+
+        let mut handle = ValidatorHandle::<H> {
+            mailbox: setup.mailbox.clone(),
+            extra: setup.extra.clone(),
+        };
+        let mut peers = Vec::<ValidatorHandle<H>>::new();
+        let mut parent = H::genesis_block(NUM_VALIDATORS as u16);
+        let count = context.gen_range(1..=BLOCKS_PER_EPOCH.get().min(4));
+        for height in 1..=count {
+            let height = Height::new(height);
+            let round = Round::new(Epoch::zero(), View::new(height.get()));
+            let parent_view = height
+                .previous()
+                .map_or(View::zero(), |h| View::new(h.get()));
+            let block = H::make_test_block(
+                H::digest(&parent),
+                H::commitment(&parent),
+                height,
+                context.gen(),
+                NUM_VALIDATORS as u16,
+            );
+            H::verify(&mut handle, round, &block, &mut peers).await;
+
+            let proposal = Proposal::new(round, parent_view, H::commitment(&block));
+            let finalization = H::make_finalization(proposal, &schemes, QUORUM);
+            let mut mailbox = setup.mailbox.clone();
+            H::report_finalization(&mut mailbox, finalization).await;
+
+            assert_eq!(setup.application.acknowledged().await, height);
+            wait_processed(&mut context, &setup, height).await;
+            parent = block;
+        }
+
+        setup.actor_handle.abort();
+        let _ = setup.actor_handle.await;
+        context.storage_audit().to_vec()
+    })
+}
+
+async fn wait_processed<H: TestHarness>(
+    context: &mut deterministic::Context,
+    setup: &ValidatorSetup<H>,
+    height: Height,
+) {
+    loop {
+        if setup.mailbox.get_processed_height().await == Some(height) {
+            break;
+        }
+        context.sleep(Duration::from_millis(1)).await;
+    }
+}
+
+conformance_tests! {
+    StandardStorageConformance => CASES,
+    CodingStorageConformance => CASES,
+}

consensus/src/marshal/core/acks.rs

@@ -62,12 +62,12 @@ impl<V: Variant, A: Acknowledgement> PendingAcks<V, A> {
     }
 
     /// Returns the next height to dispatch while preserving sequential order.
-    pub(super) fn next_dispatch_height(&self, last_processed_height: Height) -> Height {
+    pub(super) fn next_dispatch_height(&self, start_height: Height) -> Height {
         self.queue
             .back()
             .map(|ack| ack.height.next())
             .or_else(|| self.current.as_ref().map(|ack| ack.height.next()))
-            .unwrap_or_else(|| last_processed_height.next())
+            .unwrap_or(start_height)
     }
 
     /// Enqueues a newly dispatched ack, arming it immediately when idle.
@@ -136,18 +136,18 @@ mod tests {
     fn enqueue_tracks_capacity_and_fifo_ready_order() {
         let mut pending = PendingAcks::<TestVariant, Exact>::new(2);
         assert!(pending.has_capacity());
-        assert_eq!(pending.next_dispatch_height(Height::new(7)), Height::new(8));
+        assert_eq!(pending.next_dispatch_height(Height::new(8)), Height::new(8));
 
         let (first, first_ack) = pending_ack(8, 1);
         pending.enqueue(first);
         assert!(pending.has_capacity());
-        assert_eq!(pending.next_dispatch_height(Height::new(7)), Height::new(9));
+        assert_eq!(pending.next_dispatch_height(Height::new(8)), Height::new(9));
 
         let (second, second_ack) = pending_ack(9, 2);
         pending.enqueue(second);
         assert!(!pending.has_capacity());
         assert_eq!(
-            pending.next_dispatch_height(Height::new(7)),
+            pending.next_dispatch_height(Height::new(8)),
             Height::new(10)
         );
 
@@ -185,7 +185,7 @@ mod tests {
         assert!(pending.pop_ready().is_none());
         assert!(pending.has_capacity());
         assert_eq!(
-            pending.next_dispatch_height(Height::new(9)),
+            pending.next_dispatch_height(Height::new(10)),
             Height::new(10)
         );
     }