[consensus/simplex+marshal] Auto-certify locally proposed views + Improve Durability Guarantees

consensus/fuzz/src/lib.rs

@@ -375,7 +375,7 @@ where
         propose_latency: (10.0, 5.0),
         verify_latency: (10.0, 5.0),
         certify_latency: (10.0, 5.0),
-        should_certify: application::Certifier::Sometimes,
+        should_certify: application::Certifier::Always,
     };
     let (actor, application) =
         application::Application::new(context.with_label("application"), app_cfg);
@@ -609,7 +609,7 @@ fn run_with_twin_mutator<P: simplex::Simplex>(input: FuzzInput) {
                 propose_latency: (10.0, 5.0),
                 verify_latency: (10.0, 5.0),
                 certify_latency: (10.0, 5.0),
-                should_certify: application::Certifier::Sometimes,
+                should_certify: application::Certifier::Always,
             };
             let (actor, application) =
                 application::Application::new(primary_context.with_label("application"), app_cfg);

consensus/src/lib.rs

@@ -99,6 +99,13 @@ stability_scope!(BETA, cfg(not(target_arch = "wasm32")) {
         /// If it is possible to generate a payload, the Digest should be returned over the provided
         /// channel. If it is not possible to generate a payload, the channel can be dropped. If construction
         /// takes too long, the consensus engine may drop the provided proposal.
+        ///
+        /// Returning a payload from `propose` commits the local proposer to verifying
+        /// the same `(context, payload)`.
+        ///
+        /// For [`CertifiableAutomaton`] implementations, returning a payload from
+        /// `propose` also commits the local proposer to certifying that same
+        /// `(round, payload)` if it later becomes notarized.
         fn propose(
             &mut self,
             context: Self::Context,
@@ -134,18 +141,12 @@ stability_scope!(BETA, cfg(not(target_arch = "wasm32")) {
         /// Determine whether a verified payload is safe to commit.
         ///
         /// The round parameter identifies which consensus round is being certified, allowing
-        /// applications to associate certification with the correct verification context.
-        ///
-        /// Note: In applications where payloads incorporate the round number (recommended),
-        /// each round will have a unique payload digest. However, the same payload may appear
-        /// in multiple rounds when re-proposing notarized blocks at epoch boundaries or in
-        /// integrations where payloads are round-agnostic.
-        ///
-        /// This is particularly useful for applications that employ erasure coding, which
-        /// can override this method to delay or prevent finalization until they have
-        /// reconstructed and validated the full block (e.g., after receiving enough shards).
+        /// applications to associate certification with the correct verification context. The
+        /// same payload may appear in multiple rounds, so implementations must key any state
+        /// on `(round, payload)` rather than `payload` alone.
         ///
-        /// Like [`Automaton::verify`], certification is single-shot for the given
+        /// Like [`Automaton::verify`], payloads produced by [`Automaton::propose`] are certifiable-by-construction.
+        /// Also like [`Automaton::verify`], certification is single-shot for the given
         /// `(round, payload)`. Once the returned channel resolves or closes, consensus treats
         /// certification as concluded and will not retry the same request.
         ///
@@ -193,7 +194,7 @@ stability_scope!(BETA, cfg(not(target_arch = "wasm32")) {
         /// treat every broadcast identically can set this to `()`.
         type Plan: Send;
 
-        /// Broadcast a payload to the given recipients.
+        /// Broadcast a payload according to the given plan.
         fn broadcast(
             &mut self,
             payload: Self::Digest,

consensus/src/marshal/application/validation.rs

@@ -3,13 +3,37 @@
 //! This module centralizes pure invariant checks shared across marshal verification
 //! and certification flows.
 
-use crate::types::{Epoch, Epocher, Height, Round};
-use commonware_utils::sync::Mutex;
-use std::sync::Arc;
+use crate::{
+    marshal::core::{Mailbox, Variant},
+    types::{Epoch, Epocher, Height, Round},
+};
+use commonware_cryptography::certificate::Scheme;
 
-/// Cache for the last block built during proposal, shared between the
-/// proposer task and the broadcast path.
-pub(crate) type LastBuilt<B> = Arc<Mutex<Option<(Round, B)>>>;
+/// Which stage of verification a block has reached.
+///
+/// This is used to determine which marshal cache a block should be stored in.
+#[derive(Clone, Copy, Debug)]
+pub(crate) enum Stage {
+    /// The block has been verified (store in `verified_blocks`).
+    Verified,
+    /// The block has been certified (store in `notarized_blocks`).
+    Certified,
+}
+
+impl Stage {
+    /// Store `block` in the marshal cache for the provided stage.
+    pub(crate) async fn store<S: Scheme, V: Variant>(
+        self,
+        marshal: &mut Mailbox<S, V>,
+        round: Round,
+        block: V::Block,
+    ) -> bool {
+        match self {
+            Self::Verified => marshal.verified(round, block).await,
+            Self::Certified => marshal.certified(round, block).await,
+        }
+    }
+}
 
 /// Returns true if the block is at an epoch boundary (last block in its epoch).
 #[inline]

consensus/src/marshal/application/verification_tasks.rs

@@ -59,3 +59,114 @@ where
             .retain(|(task_round, _), _| task_round > finalized_round);
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::types::{Epoch, View};
+    use commonware_cryptography::{sha256::Digest as Sha256Digest, Hasher, Sha256};
+
+    type D = Sha256Digest;
+
+    fn round(view: u64) -> Round {
+        Round::new(Epoch::zero(), View::new(view))
+    }
+
+    fn pending_task() -> oneshot::Receiver<bool> {
+        let (_tx, rx) = oneshot::channel();
+        rx
+    }
+
+    #[test]
+    fn test_insert_and_take_returns_task() {
+        let tasks = VerificationTasks::<D>::new();
+        let digest = Sha256::hash(b"block");
+        tasks.insert(round(1), digest, pending_task());
+
+        assert!(tasks.take(round(1), digest).is_some());
+        assert!(
+            tasks.take(round(1), digest).is_none(),
+            "taking twice should yield None"
+        );
+    }
+
+    #[test]
+    fn test_take_absent_key_is_none() {
+        let tasks = VerificationTasks::<D>::new();
+        assert!(tasks.take(round(1), Sha256::hash(b"missing")).is_none());
+    }
+
+    #[test]
+    fn test_take_distinguishes_rounds_and_digests() {
+        let tasks = VerificationTasks::<D>::new();
+        let digest_a = Sha256::hash(b"a");
+        let digest_b = Sha256::hash(b"b");
+        tasks.insert(round(1), digest_a, pending_task());
+        tasks.insert(round(2), digest_a, pending_task());
+        tasks.insert(round(1), digest_b, pending_task());
+
+        assert!(tasks.take(round(1), digest_a).is_some());
+        assert!(tasks.take(round(2), digest_a).is_some());
+        assert!(tasks.take(round(1), digest_b).is_some());
+    }
+
+    #[test]
+    fn test_retain_after_drops_at_and_below_boundary() {
+        let tasks = VerificationTasks::<D>::new();
+        let digest = Sha256::hash(b"block");
+        tasks.insert(round(1), digest, pending_task());
+        tasks.insert(round(2), digest, pending_task());
+        tasks.insert(round(3), digest, pending_task());
+
+        tasks.retain_after(&round(2));
+
+        assert!(
+            tasks.take(round(1), digest).is_none(),
+            "tasks strictly below boundary should be dropped"
+        );
+        assert!(
+            tasks.take(round(2), digest).is_none(),
+            "tasks at boundary should be dropped"
+        );
+        assert!(
+            tasks.take(round(3), digest).is_some(),
+            "tasks strictly above boundary should be retained"
+        );
+    }
+
+    #[test]
+    fn test_retain_after_spans_epochs() {
+        let tasks = VerificationTasks::<D>::new();
+        let digest = Sha256::hash(b"block");
+        let early = Round::new(Epoch::zero(), View::new(100));
+        let late = Round::new(Epoch::new(1), View::zero());
+        tasks.insert(early, digest, pending_task());
+        tasks.insert(late, digest, pending_task());
+
+        tasks.retain_after(&early);
+
+        assert!(
+            tasks.take(early, digest).is_none(),
+            "task at boundary must be dropped"
+        );
+        assert!(
+            tasks.take(late, digest).is_some(),
+            "task in later epoch must outlive an earlier boundary"
+        );
+    }
+
+    #[test]
+    fn test_retain_after_empty_map_is_noop() {
+        let tasks = VerificationTasks::<D>::new();
+        tasks.retain_after(&round(5));
+        assert!(tasks.take(round(5), Sha256::hash(b"x")).is_none());
+    }
+
+    #[test]
+    fn test_default_matches_new() {
+        let default = <VerificationTasks<D> as Default>::default();
+        let digest = Sha256::hash(b"block");
+        default.insert(round(1), digest, pending_task());
+        assert!(default.take(round(1), digest).is_some());
+    }
+}