add verified check on propose

Author: patrick-ogrady

consensus/src/marshal/coding/marshaled.rs

@@ -529,6 +529,28 @@ where
             .with_label("propose")
             .with_attribute("round", consensus_context.round)
             .spawn(move |runtime_context| async move {
+                // On leader recovery, marshal may already hold a verified
+                // block for this round (persisted before voting in consensus). Building
+                // a fresh block would land on the same view index in the
+                // prunable archive and be silently dropped, so reuse the
+                // stored block instead.
+                if let Some(block) = marshal.get_verified(consensus_context.round).await {
+                    let commitment = block.commitment();
+                    let round = consensus_context.round;
+                    {
+                        let mut lock = last_built.lock();
+                        *lock = Some((round, block));
+                    }
+                    let success = tx.send_lossy(commitment);
+                    debug!(
+                        ?round,
+                        ?commitment,
+                        success,
+                        "reused verified block from marshal on leader recovery"
+                    );
+                    return;
+                }
+
                 let (parent_view, parent_commitment) = consensus_context.parent;
                 let parent_request = fetch_parent(
                     parent_commitment,

consensus/src/marshal/coding/mod.rs

@@ -1982,4 +1982,99 @@ mod tests {
             );
         });
     }
+
+    /// Regression: if marshal already holds a verified block for a round
+    /// (say, persisted by a pre-crash propose whose notarize vote never
+    /// reached the journal), a restarted leader's `propose` must return
+    /// that block's commitment instead of rebuilding. Otherwise the
+    /// new block lands on the same view index in the prunable archive,
+    /// gets silently dropped (`skip_if_index_exists=true`), and the
+    /// leader's notarize targets a commitment no peer can serve.
+    #[test_traced("WARN")]
+    fn test_propose_reuses_verified_block_on_restart() {
+        let runner = deterministic::Runner::timed(Duration::from_secs(60));
+        runner.start(|mut context| async move {
+            let Fixture {
+                participants,
+                schemes,
+                ..
+            } = bls12381_threshold_vrf::fixture::<V, _>(&mut context, NAMESPACE, NUM_VALIDATORS);
+            let mut oracle =
+                setup_network_with_participants(context.clone(), NZUsize!(1), participants.clone())
+                    .await;
+
+            let me = participants[0].clone();
+            let coding_config = coding_config_for_participants(NUM_VALIDATORS as u16);
+
+            let setup = CodingHarness::setup_validator(
+                context.with_label("validator_0"),
+                &mut oracle,
+                me.clone(),
+                ConstantProvider::new(schemes[0].clone()),
+            )
+            .await;
+            let marshal = setup.mailbox;
+            let shards = setup.extra;
+
+            let genesis_ctx = CodingCtx {
+                round: Round::zero(),
+                leader: default_leader(),
+                parent: (View::zero(), genesis_commitment()),
+            };
+            let genesis = make_coding_block(genesis_ctx, Sha256::hash(b""), Height::zero(), 0);
+            let genesis_parent_commitment = genesis_coding_commitment::<Sha256, _>(&genesis);
+
+            let round = Round::new(Epoch::zero(), View::new(1));
+            let ctx = CodingCtx {
+                round,
+                leader: me.clone(),
+                parent: (View::zero(), genesis_parent_commitment),
+            };
+
+            // Seed block A in marshal's verified cache for `round`.
+            let block_a = make_coding_block(ctx.clone(), genesis.digest(), Height::new(1), 100);
+            let coded_a: CodedBlock<_, ReedSolomon<Sha256>, Sha256> =
+                CodedBlock::new(block_a.clone(), coding_config, &Sequential);
+            let commitment_a = coded_a.commitment();
+            assert!(marshal.proposed(round, coded_a).await);
+
+            // After restart, a fresh application would build a different
+            // block for the same round.
+            let block_b = make_coding_block(ctx.clone(), genesis.digest(), Height::new(1), 200);
+            let coded_b: CodedBlock<_, ReedSolomon<Sha256>, Sha256> =
+                CodedBlock::new(block_b.clone(), coding_config, &Sequential);
+            let commitment_b = coded_b.commitment();
+            assert_ne!(
+                commitment_a, commitment_b,
+                "test requires distinct commitments"
+            );
+
+            let mock_app: MockVerifyingApp<CodingB, S> =
+                MockVerifyingApp::new(genesis).with_propose_result(block_b);
+            let cfg = MarshaledConfig {
+                application: mock_app,
+                marshal: marshal.clone(),
+                shards: shards.clone(),
+                scheme_provider: ConstantProvider::new(schemes[0].clone()),
+                epocher: FixedEpocher::new(BLOCKS_PER_EPOCH),
+                strategy: Sequential,
+            };
+            let mut marshaled = Marshaled::new(context.clone(), cfg);
+
+            let commitment = marshaled
+                .propose(ctx)
+                .await
+                .await
+                .expect("propose must return a commitment");
+            assert_eq!(
+                commitment, commitment_a,
+                "propose must reuse the block marshal already persisted for this round"
+            );
+
+            assert!(
+                marshaled.broadcast(commitment_a, Plan::Propose).await,
+                "relay broadcast must succeed after re-propose"
+            );
+        });
+    }
 }

consensus/src/marshal/core/actor.rs

@@ -521,6 +521,14 @@ where
                         };
                         response.send_lossy(info);
                     }
+                    Message::GetVerified { round, response } => {
+                        let block = self
+                            .cache
+                            .get_verified(round)
+                            .await
+                            .map(Into::into);
+                        response.send_lossy(block);
+                    }
                     Message::Proposed { round, block, ack } => {
                         self.cache_verified(round, block.digest(), block.clone())
                             .await;

consensus/src/marshal/core/cache.rs

@@ -371,6 +371,16 @@ where
             .expect("failed to get notarization")
     }
 
+    /// Get the block previously persisted in the verified archive for `round`.
+    pub(crate) async fn get_verified(&self, round: Round) -> Option<V::StoredBlock> {
+        let cache = self.caches.get(&round.epoch())?;
+        cache
+            .verified_blocks
+            .get(Identifier::Index(round.view().get()))
+            .await
+            .expect("failed to get verified block")
+    }
+
     /// Get a finalization from the prunable archive by block digest.
     ///
     /// SAFETY: For blocks/certificates admitted by marshal verification, a block digest

consensus/src/marshal/core/mailbox.rs

@@ -85,6 +85,13 @@ pub(crate) enum Message<S: Scheme, V: Variant> {
         /// A channel to send the retrieved block.
         response: oneshot::Sender<V::Block>,
     },
+    /// A request to retrieve the verified block previously persisted for `round`.
+    GetVerified {
+        /// The round to query.
+        round: Round,
+        /// A channel to send the retrieved block, if any.
+        response: oneshot::Sender<Option<V::Block>>,
+    },
     /// A request to broadcast a proposed block to peers.
     Proposed {
         /// The round in which the block was proposed.
@@ -296,6 +303,14 @@ impl<S: Scheme, V: Variant> Mailbox<S, V> {
             .map(|block| AncestorStream::new(self.clone(), [V::into_inner(block)]))
     }
 
+    /// Returns the verified block previously persisted for `round`, if any.
+    pub async fn get_verified(&self, round: Round) -> Option<V::Block> {
+        self.sender
+            .request(|response| Message::GetVerified { round, response })
+            .await
+            .flatten()
+    }
+
     /// Requests that a proposed block is sent to peers, awaiting the actor's
     /// confirmation that the block has been durably persisted before returning.
     #[must_use = "callers must consider block durability before proceeding"]

consensus/src/marshal/standard/deferred.rs

@@ -312,6 +312,28 @@ where
             .with_label("propose")
             .with_attribute("round", consensus_context.round)
             .spawn(move |runtime_context| async move {
+                // On leader recovery, marshal may already hold a verified
+                // block for this round (persisted by a pre-crash propose
+                // whose notarize vote never reached the journal). Building
+                // a fresh block would land on the same view index in the
+                // prunable archive and be silently dropped, so reuse the
+                // stored block instead.
+                if let Some(block) = marshal.get_verified(consensus_context.round).await {
+                    let digest = block.digest();
+                    {
+                        let mut lock = last_built.lock();
+                        *lock = Some((consensus_context.round, block));
+                    }
+                    let success = tx.send_lossy(digest);
+                    debug!(
+                        round = ?consensus_context.round,
+                        ?digest,
+                        success,
+                        "reused verified block from marshal on leader recovery"
+                    );
+                    return;
+                }
+
                 let (parent_view, parent_digest) = consensus_context.parent;
                 let parent_request = fetch_parent(
                     parent_digest,
@@ -683,9 +705,9 @@ mod tests {
             },
             verifying::{GatedVerifyingApp, MockVerifyingApp},
         },
-        simplex::scheme::bls12381_threshold::vrf as bls12381_threshold_vrf,
+        simplex::{scheme::bls12381_threshold::vrf as bls12381_threshold_vrf, Plan},
         types::{Epoch, Epocher, FixedEpocher, Height, Round, View},
-        Automaton, CertifiableAutomaton,
+        Automaton, CertifiableAutomaton, Relay,
     };
     use commonware_broadcast::Broadcaster;
     use commonware_cryptography::{
@@ -1101,4 +1123,69 @@ mod tests {
             }
         });
     }
+
+    /// Regression: when marshal holds a verified block for a round from a
+    /// pre-crash propose, a restarted leader's `propose` must return that
+    /// block's digest instead of asking the application to build afresh.
+    /// See `standard::inline::tests::test_propose_reuses_verified_block_on_restart`.
+    #[test_traced("WARN")]
+    fn test_propose_reuses_verified_block_on_restart() {
+        let runner = deterministic::Runner::timed(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 =
+                setup_network_with_participants(context.clone(), NZUsize!(1), participants.clone())
+                    .await;
+
+            let me = participants[0].clone();
+            let setup = StandardHarness::setup_validator(
+                context.with_label("validator_0"),
+                &mut oracle,
+                me.clone(),
+                ConstantProvider::new(schemes[0].clone()),
+            )
+            .await;
+            let marshal = setup.mailbox;
+
+            let genesis = make_raw_block(Sha256::hash(b""), Height::zero(), 0);
+            let round = Round::new(Epoch::zero(), View::new(1));
+            let ctx = Ctx {
+                round,
+                leader: me.clone(),
+                parent: (View::zero(), genesis.digest()),
+            };
+            let block_a = B::new::<Sha256>(ctx.clone(), genesis.digest(), Height::new(1), 100);
+            let digest_a = block_a.digest();
+            assert!(marshal.proposed(round, block_a.clone()).await);
+
+            let block_b = B::new::<Sha256>(ctx.clone(), genesis.digest(), Height::new(1), 200);
+            let digest_b = block_b.digest();
+            assert_ne!(digest_a, digest_b, "test requires distinct digests");
+
+            let mock_app: MockVerifyingApp<B, S> =
+                MockVerifyingApp::new(genesis.clone()).with_propose_result(block_b);
+            let mut marshaled = Deferred::new(
+                context.clone(),
+                mock_app,
+                marshal.clone(),
+                FixedEpocher::new(BLOCKS_PER_EPOCH),
+            );
+
+            let digest_rx = marshaled.propose(ctx).await;
+            let digest = digest_rx.await.expect("propose must return a digest");
+            assert_eq!(
+                digest, digest_a,
+                "propose must reuse the block marshal already persisted for this round"
+            );
+
+            assert!(
+                marshaled.broadcast(digest_a, Plan::Propose).await,
+                "relay broadcast must succeed after re-propose"
+            );
+        });
+    }
 }