From 13343b2450f892ef234f11f51eaf8d140c71d9e4 Mon Sep 17 00:00:00 2001 From: Patrick O'Grady Date: Sun, 20 Sep 2026 21:26:27 -0700 Subject: [PATCH] consensus/simplex: supply selected commitments to applications --- consensus/src/aggregation/engine.rs | 2 +- .../src/aggregation/mocks/application.rs | 8 +- consensus/src/lib.rs | 18 +- consensus/src/marshal/coding/marshaled.rs | 9 +- consensus/src/marshal/coding/mod.rs | 372 ++++++-- consensus/src/marshal/standard/deferred.rs | 109 ++- consensus/src/marshal/standard/inline.rs | 79 +- consensus/src/marshal/standard/mod.rs | 199 +++- consensus/src/simplex/actors/voter/actor.rs | 155 +++- consensus/src/simplex/actors/voter/mod.rs | 140 ++- consensus/src/simplex/actors/voter/state.rs | 858 +++++++++++++++++- consensus/src/simplex/mocks/application.rs | 25 +- examples/bridge/src/application/ingress.rs | 4 +- examples/log/src/application/ingress.rs | 4 +- glue/src/dkg/tests/mocks.rs | 7 +- glue/src/stateful/tests/mod.rs | 49 +- 16 files changed, 1820 insertions(+), 218 deletions(-) diff --git a/consensus/src/aggregation/engine.rs b/consensus/src/aggregation/engine.rs index 08206db7b22..f844f83caa8 100644 --- a/consensus/src/aggregation/engine.rs +++ b/consensus/src/aggregation/engine.rs @@ -719,7 +719,7 @@ impl< let mut automaton = self.automaton.clone(); let timer = self.metrics.digest_duration.timer(self.context.as_ref()); self.digest_requests.push(async move { - let receiver = automaton.propose(height).await; + let receiver = automaton.propose(height, Arc::from([])).await; let result = receiver.await.map_err(Error::AppProposeCanceled); DigestRequest { height, diff --git a/consensus/src/aggregation/mocks/application.rs b/consensus/src/aggregation/mocks/application.rs index 34c3aa2e594..4aa985e734f 100644 --- a/consensus/src/aggregation/mocks/application.rs +++ b/consensus/src/aggregation/mocks/application.rs @@ -1,6 +1,7 @@ use crate::{Automaton as A, types::Height}; use commonware_cryptography::{Hasher, Sha256}; use commonware_utils::channel::oneshot; +use std::sync::Arc; use tracing::trace; #[derive(Clone, Debug)] @@ -30,7 +31,11 @@ impl A for Application { type Context = Height; type Digest = ::Digest; - async fn propose(&mut self, context: Self::Context) -> oneshot::Receiver { + async fn propose( + &mut self, + context: Self::Context, + _ancestry: Arc<[Self::Digest]>, + ) -> oneshot::Receiver { let (sender, receiver) = oneshot::channel(); let digest = match &self.strategy { @@ -56,6 +61,7 @@ impl A for Application { &mut self, context: Self::Context, payload: Self::Digest, + _ancestry: Arc<[Self::Digest]>, ) -> oneshot::Receiver { trace!(%context, ?payload, "verify"); let (sender, receiver) = oneshot::channel(); diff --git a/consensus/src/lib.rs b/consensus/src/lib.rs index 59bbe3fc498..1d72979de74 100644 --- a/consensus/src/lib.rs +++ b/consensus/src/lib.rs @@ -140,12 +140,19 @@ stability_scope!(BETA, cfg(not(target_arch = "wasm32")) { /// rather than rebuilding them from current local state. If consensus /// later abandons a dependency, it also abandons the proposal. /// + /// `ancestry` contains commitments on the selected parent branch in + /// forward order, ending at the parent and excluding the new payload. + /// For parent-linked consensus it begins at a finalized or genesis + /// anchor; applications may resolve older canonical history as needed. + /// Consecutive re-proposals of one payload contribute one commitment. + /// /// Closing the response declines this request, which consensus may /// treat as final for the context. Keep the response pending when /// temporary unavailability should not abandon the context. fn propose( &mut self, context: Self::Context, + ancestry: Arc<[Self::Digest]>, ) -> impl Future> + Send; /// Verify the payload is valid. @@ -167,10 +174,13 @@ stability_scope!(BETA, cfg(not(target_arch = "wasm32")) { /// /// The future-context requirement on [`Self::propose`] applies here /// too: the context's dependencies may not be resolvable locally yet. + /// `ancestry` has the same ordering and availability contract as + /// [`Self::propose`] and excludes `payload`. fn verify( &mut self, context: Self::Context, payload: Self::Digest, + ancestry: Arc<[Self::Digest]>, ) -> impl Future> + Send; } @@ -180,13 +190,18 @@ stability_scope!(BETA, cfg(not(target_arch = "wasm32")) { /// phase between notarization and finalization. Applications that do not need custom certification /// logic can use the default implementation which always certifies. pub trait CertifiableAutomaton: Automaton { - /// Determine whether a verified payload is safe to commit. + /// Determine whether a 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. The /// same payload may appear in multiple rounds, so implementations must key any state /// on `(round, payload)` rather than `payload` alone. /// + /// Certification may be requested without prior local verification. + /// `ancestry` supplies the selected parent branch independently of + /// verification, with the same ordering and availability contract as + /// [`Automaton::propose`], and excludes `payload`. + /// /// 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 @@ -211,6 +226,7 @@ stability_scope!(BETA, cfg(not(target_arch = "wasm32")) { &mut self, _round: Round, _payload: Self::Digest, + _ancestry: Arc<[Self::Digest]>, ) -> impl Future> + Send { #[allow(clippy::async_yields_async)] async move { diff --git a/consensus/src/marshal/coding/marshaled.rs b/consensus/src/marshal/coding/marshaled.rs index 56a66625fb0..ad0fad933a2 100644 --- a/consensus/src/marshal/coding/marshaled.rs +++ b/consensus/src/marshal/coding/marshaled.rs @@ -667,6 +667,7 @@ where async fn propose( &mut self, consensus_context: Context, ::PublicKey>, + _ancestry: Arc<[Self::Digest]>, ) -> oneshot::Receiver { let marshal = self.marshal.clone(); let mut application = self.application.clone(); @@ -884,6 +885,7 @@ where &mut self, consensus_context: Context::PublicKey>, payload: Self::Digest, + _ancestry: Arc<[Self::Digest]>, ) -> oneshot::Receiver { // If there's no scheme for the current epoch, we cannot vote on the proposal. // Send back a receiver with a dropped sender. @@ -1110,7 +1112,12 @@ where { #[allow(clippy::async_yields_async)] #[tracing::instrument(name = "marshal.coding.certify", level = "info", skip_all, fields(round = %round, commitment = %payload))] - async fn certify(&mut self, round: Round, payload: Self::Digest) -> oneshot::Receiver { + async fn certify( + &mut self, + round: Round, + payload: Self::Digest, + _ancestry: Arc<[Self::Digest]>, + ) -> oneshot::Receiver { self.gates.flush_unrelayed(&self.marshal, round, payload); // First, check for an in-progress certification gate task. diff --git a/consensus/src/marshal/coding/mod.rs b/consensus/src/marshal/coding/mod.rs index 5ba918c50be..84878ac0c81 100644 --- a/consensus/src/marshal/coding/mod.rs +++ b/consensus/src/marshal/coding/mod.rs @@ -938,7 +938,13 @@ mod tests { }; let mut marshaled = Marshaled::new(context.child("marshaled"), cfg); - let verify_rx = marshaled.verify(candidate_ctx, commitment).await; + let verify_rx = marshaled + .verify( + candidate_ctx.clone(), + commitment, + Arc::from([candidate_ctx.parent.1]), + ) + .await; context.sleep(Duration::from_millis(100)).await; assert!( @@ -1001,7 +1007,13 @@ mod tests { let proposal = Proposal::new(round, View::zero(), commitment); let notarization = CodingHarness::make_notarization(proposal, &schemes, QUORUM); resolver.respond_to_next_fetch((notarization, candidate).encode()); - let certify_rx = marshaled.certify(round, commitment).await; + let certify_rx = marshaled + .certify( + round, + commitment, + Arc::from([genesis_coding_commitment(&genesis_block())]), + ) + .await; let result = certify_rx.await.expect("certify result missing"); assert!(result, "fetched notarized candidate should certify"); @@ -1067,12 +1079,20 @@ mod tests { let (candidate_ctx, candidate) = missing_candidate(me); let commitment = candidate.commitment(); let round = candidate_ctx.round; - let _verify_rx = marshaled.verify(candidate_ctx, commitment).await; + let _verify_rx = marshaled + .verify( + candidate_ctx.clone(), + commitment, + Arc::from([candidate_ctx.parent.1]), + ) + .await; let proposal = Proposal::new(round, View::zero(), commitment); let notarization = CodingHarness::make_notarization(proposal, &schemes, QUORUM); resolver.respond_to_next_fetch((notarization, candidate).encode()); - let certify_rx = marshaled.certify(round, commitment).await; + let certify_rx = marshaled + .certify(round, commitment, Arc::from([candidate_ctx.parent.1])) + .await; let result = certify_rx.await.expect("certify result missing"); assert!( @@ -1438,14 +1458,18 @@ mod tests { let mut marshaled = Marshaled::new(context.child("marshaled"), cfg); let shard_validity = marshaled - .verify(block_ctx, commitment) + .verify( + block_ctx.clone(), + commitment, + Arc::from([block_ctx.parent.1]), + ) .await .await .expect("verify result missing"); assert!(shard_validity, "shard validity should pass"); let certify_result = marshaled - .certify(round, commitment) + .certify(round, commitment, Arc::from([block_ctx.parent.1])) .await .await .expect("certify result missing"); @@ -1654,20 +1678,58 @@ mod tests { context.sleep(Duration::from_millis(10)).await; // Step 1: Verify block A at view 5 - let _ = marshaled.verify(context_a, commitment_a).await.await; + let _ = marshaled + .verify( + context_a.clone(), + commitment_a, + Arc::from([ + genesis_coding_commitment(&genesis_block()), + context_a.parent.1, + ]), + ) + .await + .await; // Step 2: Verify block B at view 10 - let _ = marshaled.verify(context_b, commitment_b).await.await; + let _ = marshaled + .verify( + context_b.clone(), + commitment_b, + Arc::from([ + genesis_coding_commitment(&genesis_block()), + context_b.parent.1, + ]), + ) + .await + .await; // Step 3: Certify block B at view 10 FIRST - let certify_b = marshaled.certify(round_b, commitment_b).await; + let certify_b = marshaled + .certify( + round_b, + commitment_b, + Arc::from([ + genesis_coding_commitment(&genesis_block()), + context_b.parent.1, + ]), + ) + .await; assert!( certify_b.await.unwrap(), "Block B certification should succeed" ); // Step 4: Certify block A at view 5 - should succeed - let certify_a = marshaled.certify(round_a, commitment_a).await; + let certify_a = marshaled + .certify( + round_a, + commitment_a, + Arc::from([ + genesis_coding_commitment(&genesis_block()), + context_a.parent.1, + ]), + ) + .await; // Use select with timeout to detect never-resolving receiver select! { @@ -1816,7 +1878,11 @@ mod tests { // We must await the verify result to ensure the certification gate task is // registered before calling certify. let shard_validity = marshaled - .verify(reproposal_context.clone(), boundary_commitment) + .verify( + reproposal_context.clone(), + boundary_commitment, + Arc::from([reproposal_context.parent.1]), + ) .await .await; assert!( @@ -1839,7 +1905,11 @@ mod tests { // Use certify to get the actual deferred_verify result let certify_result = marshaled - .certify(reproposal_round, boundary_commitment) + .certify( + reproposal_round, + boundary_commitment, + Arc::from([reproposal_context.parent.1]), + ) .await .await; assert!( @@ -1860,7 +1930,11 @@ mod tests { parent: (reproposal_round.view(), boundary_commitment), }; let repeated_verify = marshaled - .verify(repeated_reproposal_context, boundary_commitment) + .verify( + repeated_reproposal_context.clone(), + boundary_commitment, + Arc::from([repeated_reproposal_context.parent.1]), + ) .await .await; assert!( @@ -1868,7 +1942,11 @@ mod tests { "Repeated re-proposal should remain valid as the parent view advances" ); let repeated_certify = marshaled - .certify(repeated_reproposal_round, boundary_commitment) + .certify( + repeated_reproposal_round, + boundary_commitment, + Arc::from([repeated_reproposal_context.parent.1]), + ) .await .await; assert!( @@ -1892,7 +1970,11 @@ mod tests { // We must await the verify result to ensure the certification gate task is // registered before calling certify. let shard_validity = marshaled - .verify(invalid_reproposal_context, non_boundary_commitment) + .verify( + invalid_reproposal_context.clone(), + non_boundary_commitment, + Arc::from([invalid_reproposal_context.parent.1]), + ) .await .await; assert!( @@ -1902,7 +1984,11 @@ mod tests { // Use certify to get the actual deferred_verify result let certify_result = marshaled - .certify(invalid_reproposal_round, non_boundary_commitment) + .certify( + invalid_reproposal_round, + non_boundary_commitment, + Arc::from([invalid_reproposal_context.parent.1]), + ) .await .await; assert!( @@ -1923,7 +2009,11 @@ mod tests { // We must await the verify result to ensure the certification gate task is // registered before calling certify. let shard_validity = marshaled - .verify(cross_epoch_reproposal_context.clone(), boundary_commitment) + .verify( + cross_epoch_reproposal_context.clone(), + boundary_commitment, + Arc::from([cross_epoch_reproposal_context.parent.1]), + ) .await .await; assert!( @@ -1933,7 +2023,11 @@ mod tests { // Use certify to get the actual deferred_verify result let certify_result = marshaled - .certify(cross_epoch_reproposal_round, boundary_commitment) + .certify( + cross_epoch_reproposal_round, + boundary_commitment, + Arc::from([cross_epoch_reproposal_context.parent.1]), + ) .await .await; assert!( @@ -2026,7 +2120,14 @@ mod tests { parent: (View::zero(), commitment), }; let assigned = shards.subscribe_assigned_shard_verified(commitment); - let verdict = marshaled.verify(reproposal_context, commitment).await.await; + let verdict = marshaled + .verify( + reproposal_context.clone(), + commitment, + Arc::from([reproposal_context.parent.1]), + ) + .await + .await; assert!(!verdict.expect("re-proposal verdict missing")); // The re-proposer delivers this node's assigned shard. Without reconstruction @@ -2140,7 +2241,11 @@ mod tests { parent: (View::new(boundary_height.get()), boundary_commitment), }; let verdict = marshaled - .verify(reproposal_context, boundary_commitment) + .verify( + reproposal_context.clone(), + boundary_commitment, + Arc::from([reproposal_context.parent.1]), + ) .await .await; assert!( @@ -2181,7 +2286,11 @@ mod tests { assert_eq!(block.commitment(), boundary_commitment); let certify = marshaled - .certify(reproposal_round, boundary_commitment) + .certify( + reproposal_round, + boundary_commitment, + Arc::from([reproposal_context.parent.1]), + ) .await .await; assert!( @@ -2287,7 +2396,14 @@ mod tests { leader: participants[1].clone(), parent: (View::new(5), commitment), }; - let verdict = marshaled.verify(reproposal_context, commitment).await.await; + let verdict = marshaled + .verify( + reproposal_context.clone(), + commitment, + Arc::from([reproposal_context.parent.1]), + ) + .await + .await; assert!( verdict.expect("re-proposal verdict missing"), "re-proposal should verify after fetching the block by parent round" @@ -2373,7 +2489,13 @@ mod tests { parent: (View::new(1), parent_commitment), }; - let verify_rx = marshaled.verify(context_b, commitment_a).await; + let verify_rx = marshaled + .verify( + context_b.clone(), + commitment_a, + Arc::from([context_b.parent.1]), + ) + .await; select! { result = verify_rx => { assert!( @@ -2461,7 +2583,11 @@ mod tests { // Start verify, then drop the receiver before the block is available. let verify_rx = marshaled - .verify(reproposal_context, boundary_commitment) + .verify( + reproposal_context.clone(), + boundary_commitment, + Arc::from([reproposal_context.parent.1]), + ) .await; drop(verify_rx); context.sleep(Duration::from_millis(10)).await; @@ -2472,7 +2598,11 @@ mod tests { // Certify should not return the stale closed certification gate task; it // should recover through the embedded-context certification path. let certify_rx = marshaled - .certify(reproposal_round, boundary_commitment) + .certify( + reproposal_round, + boundary_commitment, + Arc::from([reproposal_context.parent.1]), + ) .await; select! { result = certify_rx => { @@ -2543,7 +2673,7 @@ mod tests { }; // Verify must not synthesize `false` when the block cannot be fetched. - let verify_rx = marshaled.verify(reproposal_context, missing_payload).await; + let verify_rx = marshaled.verify(reproposal_context.clone(), missing_payload, Arc::from([reproposal_context.parent.1])).await; // Ensure the certification gate task has registered its subscription, then // force cancellation by pruning the missing commitment. @@ -2568,7 +2698,7 @@ mod tests { // Certify should not surface the closed certification gate task as the final result. // With no block available, it remains pending on the recovery path until the // certifier's caller times out or data arrives. - let mut certify_rx = marshaled.certify(round, missing_payload).await; + let mut certify_rx = marshaled.certify(round, missing_payload, Arc::from([reproposal_context.parent.1])).await; context.sleep(Duration::from_millis(100)).await; assert!( matches!( @@ -2848,12 +2978,20 @@ mod tests { // Call verify to kick off deferred verification let _shard_validity = marshaled - .verify(unsupported_context, block_commitment) + .verify( + unsupported_context.clone(), + block_commitment, + Arc::from([unsupported_context.parent.1]), + ) .await; // Use certify to get the actual deferred_verify result let certify_result = marshaled - .certify(unsupported_round, block_commitment) + .certify( + unsupported_round, + block_commitment, + Arc::from([unsupported_context.parent.1]), + ) .await .await; @@ -2965,12 +3103,20 @@ mod tests { // 3. Validate height is contiguous (fail) // 4. Return false let _shard_validity = marshaled - .verify(byzantine_context, malicious_commitment) + .verify( + byzantine_context.clone(), + malicious_commitment, + Arc::from([byzantine_context.parent.1]), + ) .await; // Use certify to get the actual deferred_verify result let certify_result = marshaled - .certify(byzantine_round, malicious_commitment) + .certify( + byzantine_round, + malicious_commitment, + Arc::from([byzantine_context.parent.1]), + ) .await .await; @@ -3011,12 +3157,20 @@ mod tests { // 4. Validate parent commitment matches (fail) // 5. Return false let _shard_validity = marshaled - .verify(byzantine_context2, malicious_commitment2) + .verify( + byzantine_context2.clone(), + malicious_commitment2, + Arc::from([byzantine_context2.parent.1]), + ) .await; // Use certify to get the actual deferred_verify result let certify_result = marshaled - .certify(byzantine_round2, malicious_commitment2) + .certify( + byzantine_round2, + malicious_commitment2, + Arc::from([byzantine_context2.parent.1]), + ) .await .await; @@ -3107,7 +3261,16 @@ mod tests { context.sleep(Duration::from_millis(10)).await; // Call certify directly without any prior verify (simulating crash recovery). - let certify_rx = marshaled.certify(child_round, child_commitment).await; + let certify_rx = marshaled + .certify( + child_round, + child_commitment, + Arc::from([ + genesis_coding_commitment(&genesis_block()), + parent_commitment, + ]), + ) + .await; select! { result = certify_rx => { assert!( @@ -3200,7 +3363,16 @@ mod tests { // No prior verify, so no gate exists and certify falls through to // the embedded-context path. - let certify_rx = marshaled.certify(child_round, child_commitment).await; + let certify_rx = marshaled + .certify( + child_round, + child_commitment, + Arc::from([ + genesis_coding_commitment(&genesis_block()), + parent_commitment, + ]), + ) + .await; select! { result = certify_rx => { assert!( @@ -3318,7 +3490,16 @@ mod tests { // Certify must register reconstruction interest with the shard // engine, drain the buffered shards, and verify the reconstructed // block through its embedded context. - let certify_rx = marshaled.certify(child_round, child_commitment).await; + let certify_rx = marshaled + .certify( + child_round, + child_commitment, + Arc::from([ + genesis_coding_commitment(&genesis_block()), + parent_commitment, + ]), + ) + .await; select! { result = certify_rx => { assert!( @@ -3442,14 +3623,32 @@ mod tests { context.sleep(Duration::from_millis(10)).await; - let optimistic = marshaled.verify(verify_context, commitment).await; + let optimistic = marshaled + .verify( + verify_context.clone(), + commitment, + Arc::from([ + genesis_coding_commitment(&genesis_block()), + verify_context.parent.1, + ]), + ) + .await; assert!( optimistic.await.expect("verify result missing"), "optimistic verify should pass pre-checks and schedule deferred verification" ); // 4) Certify must observe the deferred application failure and return false. - let certify = marshaled.certify(round, commitment).await; + let certify = marshaled + .certify( + round, + commitment, + Arc::from([ + genesis_coding_commitment(&genesis_block()), + verify_context.parent.1, + ]), + ) + .await; assert!( !certify.await.expect("certify result missing"), "certify should propagate deferred application verify failure" @@ -3560,7 +3759,16 @@ mod tests { leader: me.clone(), parent: (View::new(1), certified_commitment), }; - let verify_rx = marshaled.verify(equivocating_ctx, commitment).await; + let verify_rx = marshaled + .verify( + equivocating_ctx.clone(), + commitment, + Arc::from([ + genesis_coding_commitment(&genesis_block()), + certified_commitment, + ]), + ) + .await; select! { result = verify_rx => { assert!( @@ -3575,7 +3783,17 @@ mod tests { // The honest notarization for the same `(round, commitment)` // arrives. Certification recovers through the embedded context. - let certify_rx = marshaled.certify(round, commitment).await; + let certify_rx = marshaled + .certify( + round, + commitment, + Arc::from([ + genesis_coding_commitment(&genesis_block()), + certified_commitment, + notarized_commitment, + ]), + ) + .await; select! { result = certify_rx => { assert!( @@ -4049,7 +4267,11 @@ mod tests { }, ); let _shard_verdict = marshaled - .verify(candidate.context(), candidate.commitment()) + .verify( + candidate.context().clone(), + candidate.commitment(), + Arc::from([candidate.context().parent.1]), + ) .await; // Neither the certified fork nor a cached untrusted child authenticates these fetches @@ -4099,7 +4321,11 @@ mod tests { } assert!( marshaled - .certify(candidate.context().round, candidate.commitment()) + .certify( + candidate.context().round, + candidate.commitment(), + Arc::from([candidate.context().parent.1]) + ) .await .await .unwrap() @@ -4589,11 +4815,15 @@ mod tests { }; // propose with a missing scheme returns a dropped sender - let rx = marshaled.propose(ctx.clone()).await; + let rx = marshaled + .propose(ctx.clone(), Arc::from([ctx.parent.1])) + .await; assert!(rx.await.is_err()); // verify with a missing scheme returns a dropped sender - let rx = marshaled.verify(ctx, genesis_commitment()).await; + let rx = marshaled + .verify(ctx.clone(), genesis_commitment(), Arc::from([ctx.parent.1])) + .await; assert!(rx.await.is_err()); }); } @@ -4689,7 +4919,14 @@ mod tests { // Optimistic verify - returns shard validity (true). let shard_validity = marshaled - .verify(child_ctx, child_commitment) + .verify( + child_ctx.clone(), + child_commitment, + Arc::from([ + genesis_coding_commitment(&genesis_block()), + child_ctx.parent.1, + ]), + ) .await .await .expect("verify result missing"); @@ -4697,7 +4934,14 @@ mod tests { // Certify - this is the safety gate before finalize voting. let certify_result = marshaled - .certify(child_round, child_commitment) + .certify( + child_round, + child_commitment, + Arc::from([ + genesis_coding_commitment(&genesis_block()), + child_ctx.parent.1, + ]), + ) .await .await .expect("certify result missing"); @@ -4813,7 +5057,10 @@ mod tests { // and returns the commitment. Durability is established by the // certify flush below. let commitment = marshaled - .propose(propose_context) + .propose( + propose_context.clone(), + Arc::from([propose_context.parent.1]), + ) .await .await .expect("propose should produce a commitment"); @@ -4823,7 +5070,11 @@ mod tests { // sync handle and establishes durability before the finalize vote. assert!( marshaled - .certify(propose_round, commitment) + .certify( + propose_round, + commitment, + Arc::from([propose_context.parent.1]) + ) .await .await .expect("certify result missing"), @@ -4932,7 +5183,10 @@ mod tests { let mut marshaled = Marshaled::new(context.child("marshaled"), cfg); let commitment = marshaled - .propose(propose_context) + .propose( + propose_context.clone(), + Arc::from([propose_context.parent.1]), + ) .await .await .expect("propose should produce a commitment"); @@ -4956,7 +5210,11 @@ mod tests { // certification resolves durably without a flush. assert!( marshaled - .certify(propose_round, commitment) + .certify( + propose_round, + commitment, + Arc::from([propose_context.parent.1]) + ) .await .await .expect("certify result missing"), @@ -5043,7 +5301,7 @@ mod tests { let mut marshaled = Marshaled::new(context.child("marshaled"), cfg); let commitment = marshaled - .propose(ctx) + .propose(ctx.clone(), Arc::from([ctx.parent.1])) .await .await .expect("propose must return a commitment"); @@ -5057,7 +5315,9 @@ mod tests { // write), resolving the certification gate registered by the // recovery path. let _ = marshaled.broadcast(commitment, Plan::Propose { round }); - let certify_rx = marshaled.certify(round, commitment).await; + let certify_rx = marshaled + .certify(round, commitment, Arc::from([ctx.parent.1])) + .await; select! { result = certify_rx => { assert!( @@ -5154,7 +5414,7 @@ mod tests { let mut marshaled = Marshaled::new(context.child("marshaled"), cfg); let commitment = marshaled - .propose(ctx) + .propose(ctx.clone(), Arc::from([ctx.parent.1])) .await .await .expect("propose must return a commitment"); @@ -5164,7 +5424,7 @@ mod tests { ); let _ = marshaled.broadcast(commitment, Plan::Propose { round }); - let certify_rx = marshaled.certify(round, commitment).await; + let certify_rx = marshaled.certify(round, commitment, Arc::from([ctx.parent.1])).await; select! { result = certify_rx => { assert!( @@ -5260,7 +5520,9 @@ mod tests { }; let mut marshaled = Marshaled::new(context.child("marshaled"), cfg); - let commitment_rx = marshaled.propose(new_ctx).await; + let commitment_rx = marshaled + .propose(new_ctx.clone(), Arc::from([new_ctx.parent.1])) + .await; assert!( commitment_rx.await.is_err(), "propose must drop the receiver when the cached block's context no longer matches" diff --git a/consensus/src/marshal/standard/deferred.rs b/consensus/src/marshal/standard/deferred.rs index 10eedb51d9c..061fd5d6a11 100644 --- a/consensus/src/marshal/standard/deferred.rs +++ b/consensus/src/marshal/standard/deferred.rs @@ -493,6 +493,7 @@ where async fn propose( &mut self, consensus_context: Context, + _ancestry: Arc<[Self::Digest]>, ) -> oneshot::Receiver { let marshal = self.marshal.clone(); let mut application = self.application.clone(); @@ -687,6 +688,7 @@ where &mut self, context: Context, digest: Self::Digest, + _ancestry: Arc<[Self::Digest]>, ) -> oneshot::Receiver { let marshal = self.marshal.clone(); let mut marshaled = self.clone(); @@ -847,7 +849,12 @@ where { #[allow(clippy::async_yields_async)] #[tracing::instrument(name = "marshal.deferred.certify", level = "info", skip_all, fields(round = %round, digest = %digest))] - async fn certify(&mut self, round: Round, digest: Self::Digest) -> oneshot::Receiver { + async fn certify( + &mut self, + round: Round, + digest: Self::Digest, + _ancestry: Arc<[Self::Digest]>, + ) -> oneshot::Receiver { self.gates.flush_unrelayed(&self.marshal, round, digest); // Attempt to retrieve the existing certification gate task for this round/digest. @@ -922,7 +929,7 @@ mod tests { use commonware_macros::{select, test_traced}; use commonware_runtime::{Clock, Runner, Supervisor as _, deterministic}; use commonware_utils::{NZUsize, channel::fallible::OneshotExt}; - use std::time::Duration; + use std::{sync::Arc, time::Duration}; #[test_traced("INFO")] fn test_certify_lower_view_after_higher_view() { @@ -996,20 +1003,38 @@ mod tests { context.sleep(Duration::from_millis(10)).await; // Step 1: Verify block A at view 5 - let _ = marshaled.verify(context_a, commitment_a).await.await; + let _ = marshaled + .verify( + context_a.clone(), + commitment_a, + Arc::from([context_a.parent.1]), + ) + .await + .await; // Step 2: Verify block B at view 10 - let _ = marshaled.verify(context_b, commitment_b).await.await; + let _ = marshaled + .verify( + context_b.clone(), + commitment_b, + Arc::from([context_b.parent.1]), + ) + .await + .await; // Step 3: Certify block B at view 10 FIRST - let certify_b = marshaled.certify(round_b, commitment_b).await; + let certify_b = marshaled + .certify(round_b, commitment_b, Arc::from([context_b.parent.1])) + .await; assert!( certify_b.await.unwrap(), "Block B certification should succeed" ); // Step 4: Certify block A at view 5 - should succeed - let certify_a = marshaled.certify(round_a, commitment_a).await; + let certify_a = marshaled + .certify(round_a, commitment_a, Arc::from([context_a.parent.1])) + .await; select! { result = certify_a => { @@ -1140,7 +1165,11 @@ mod tests { // Call verify and wait for the result (verify returns optimistic result, // but also spawns deferred verification) let verify_result = marshaled - .verify(unsupported_context, block_commitment) + .verify( + unsupported_context.clone(), + block_commitment, + Arc::from([unsupported_context.parent.1]), + ) .await; // Wait for optimistic verify to complete so the certification gate task is registered @@ -1233,7 +1262,13 @@ mod tests { parent: (View::new(1), parent_commitment), }; - let verify_rx = marshaled.verify(context_b, commitment_a).await; + let verify_rx = marshaled + .verify( + context_b.clone(), + commitment_a, + Arc::from([context_b.parent.1]), + ) + .await; select! { result = verify_rx => { assert!( @@ -1296,7 +1331,13 @@ mod tests { B::new::(block_context.clone(), genesis.digest(), Height::new(1), 100); let digest = block.digest(); - let verify_rx = marshaled.verify(block_context, digest).await; + let verify_rx = marshaled + .verify( + block_context.clone(), + digest, + Arc::from([block_context.parent.1]), + ) + .await; drop(verify_rx); // Give the optimistic task a chance to observe the dropped receiver while its @@ -1304,7 +1345,9 @@ mod tests { context.sleep(Duration::from_millis(10)).await; assert!(marshal.verified(round, block).await); - let certify_rx = marshaled.certify(round, digest).await; + let certify_rx = marshaled + .certify(round, digest, Arc::from([block_context.parent.1])) + .await; select! { result = certify_rx => { assert!( @@ -1393,7 +1436,7 @@ mod tests { // Kick off the optimistic verify, which spawns `deferred_verify`. Its gated // `app.verify` blocks until we release it. - let optimistic_rx = marshaled.verify(child_ctx, child_digest).await; + let optimistic_rx = marshaled.verify(child_ctx.clone(), child_digest, Arc::from([child_ctx.parent.1])).await; assert!( optimistic_rx .await @@ -1414,7 +1457,7 @@ mod tests { // Releasing verification lets certification succeed (valid and durable). release_verify.send_lossy(()); - let certify_rx = marshaled.certify(child_round, child_digest).await; + let certify_rx = marshaled.certify(child_round, child_digest, Arc::from([child_ctx.parent.1])).await; select! { result = certify_rx => { assert!( @@ -1487,7 +1530,9 @@ mod tests { FixedEpocher::new(BLOCKS_PER_EPOCH), ); - let digest_rx = marshaled.propose(ctx).await; + let digest_rx = marshaled + .propose(ctx.clone(), Arc::from([ctx.parent.1])) + .await; let digest = digest_rx.await.expect("propose must return a digest"); assert_eq!( digest, digest_a, @@ -1499,7 +1544,9 @@ mod tests { // write), resolving the certification gate registered by the // recovery path. let _ = marshaled.broadcast(digest, Plan::Propose { round }); - let certify_rx = marshaled.certify(round, digest).await; + let certify_rx = marshaled + .certify(round, digest, Arc::from([ctx.parent.1])) + .await; select! { result = certify_rx => { assert!( @@ -1584,7 +1631,7 @@ mod tests { FixedEpocher::new(BLOCKS_PER_EPOCH), ); - let digest_rx = marshaled.propose(ctx).await; + let digest_rx = marshaled.propose(ctx.clone(), Arc::from([ctx.parent.1])).await; let digest = digest_rx.await.expect("propose must return a digest"); assert_eq!( digest, boundary_digest, @@ -1592,7 +1639,7 @@ mod tests { ); let _ = marshaled.broadcast(digest, Plan::Propose { round }); - let certify_rx = marshaled.certify(round, digest).await; + let certify_rx = marshaled.certify(round, digest, Arc::from([ctx.parent.1])).await; select! { result = certify_rx => { assert!( @@ -1670,7 +1717,9 @@ mod tests { FixedEpocher::new(BLOCKS_PER_EPOCH), ); - let digest_rx = marshaled.propose(new_ctx).await; + let digest_rx = marshaled + .propose(new_ctx.clone(), Arc::from([new_ctx.parent.1])) + .await; assert!( digest_rx.await.is_err(), "propose must drop the receiver when the cached block's context no longer matches" @@ -1741,7 +1790,7 @@ mod tests { ); let digest = marshaled - .propose(ctx) + .propose(ctx.clone(), Arc::from([ctx.parent.1])) .await .await .expect("propose must return a digest"); @@ -1753,7 +1802,7 @@ mod tests { // The leader certifies its own proposal; this awaits the deferred propose sync handle. assert!( marshaled - .certify(round, child_digest) + .certify(round, child_digest, Arc::from([ctx.parent.1])) .await .await .expect("certify result missing"), @@ -1917,7 +1966,7 @@ mod tests { let verify_rx = fixture .marshaled - .verify(fixture.equivocating_ctx.clone(), fixture.digest) + .verify(fixture.equivocating_ctx.clone(), fixture.digest, Arc::from([fixture.equivocating_ctx.parent.1])) .await; assert!( !verify_rx.await.expect("verify result missing"), @@ -1926,7 +1975,7 @@ mod tests { let certify_rx = fixture .marshaled - .certify(fixture.round, fixture.digest) + .certify(fixture.round, fixture.digest, Arc::from([fixture.equivocating_ctx.parent.1])) .await; select! { result = certify_rx => { @@ -1956,7 +2005,11 @@ mod tests { let verify_rx = fixture .marshaled - .verify(fixture.embedded_ctx.clone(), fixture.digest) + .verify( + fixture.embedded_ctx.clone(), + fixture.digest, + Arc::from([fixture.embedded_ctx.parent.1]), + ) .await; assert!( verify_rx.await.expect("verify result missing"), @@ -1965,7 +2018,11 @@ mod tests { let certify_rx = fixture .marshaled - .certify(fixture.round, fixture.digest) + .certify( + fixture.round, + fixture.digest, + Arc::from([fixture.embedded_ctx.parent.1]), + ) .await; select! { result = certify_rx => { @@ -1996,7 +2053,11 @@ mod tests { // the embedded-context path. let certify_rx = fixture .marshaled - .certify(fixture.round, fixture.digest) + .certify( + fixture.round, + fixture.digest, + Arc::from([fixture.embedded_ctx.parent.1]), + ) .await; select! { result = certify_rx => { diff --git a/consensus/src/marshal/standard/inline.rs b/consensus/src/marshal/standard/inline.rs index b3bf678d536..5bc53f2f1c9 100644 --- a/consensus/src/marshal/standard/inline.rs +++ b/consensus/src/marshal/standard/inline.rs @@ -248,6 +248,7 @@ where async fn propose( &mut self, consensus_context: Context, + _ancestry: Arc<[Self::Digest]>, ) -> oneshot::Receiver { let marshal = self.marshal.clone(); let mut application = self.application.clone(); @@ -420,6 +421,7 @@ where &mut self, context: Context, digest: Self::Digest, + _ancestry: Arc<[Self::Digest]>, ) -> oneshot::Receiver { let round = context.round; @@ -591,7 +593,12 @@ where { #[allow(clippy::async_yields_async)] #[tracing::instrument(name = "marshal.inline.certify", level = "info", skip_all, fields(round = %round, digest = %digest))] - async fn certify(&mut self, round: Round, digest: Self::Digest) -> oneshot::Receiver { + async fn certify( + &mut self, + round: Round, + digest: Self::Digest, + _ancestry: Arc<[Self::Digest]>, + ) -> oneshot::Receiver { self.gates.flush_unrelayed(&self.marshal, round, digest); // `propose`/`verify` register an in-flight certification gate whose result resolves @@ -727,7 +734,7 @@ mod tests { use commonware_runtime::{Clock, Metrics, Runner, Spawner, Supervisor as _, deterministic}; use commonware_utils::{NZUsize, channel::fallible::OneshotExt}; use rand::Rng; - use std::time::Duration; + use std::{sync::Arc, time::Duration}; // Compile-time assertion only: inline standard wrapper must not require `CertifiableBlock`. #[allow(dead_code)] @@ -812,14 +819,22 @@ mod tests { assert!(marshal.verified(round, block).await); // Complete verify first so the block is already available locally. - let verify_rx = inline.verify(verify_context, digest).await; + let verify_rx = inline + .verify( + verify_context.clone(), + digest, + Arc::from([verify_context.parent.1]), + ) + .await; assert!( verify_rx.await.unwrap(), "verify should complete successfully before certify" ); // Certify should return immediately instead of waiting on marshal. - let certify_rx = inline.certify(round, digest).await; + let certify_rx = inline + .certify(round, digest, Arc::from([verify_context.parent.1])) + .await; select! { result = certify_rx => { @@ -893,7 +908,9 @@ mod tests { assert!(marshal.verified(round, block).await); // Certify should still resolve by waiting on marshal block availability directly. - let certify_rx = inline.certify(round, digest).await; + let certify_rx = inline + .certify(round, digest, Arc::from([verify_context.parent.1])) + .await; select! { result = certify_rx => { @@ -965,7 +982,7 @@ mod tests { parent: (View::new(boundary_height.get()), boundary_digest), }; - let verify_rx = inline.verify(reproposal_context, boundary_digest).await; + let verify_rx = inline.verify(reproposal_context.clone(), boundary_digest, Arc::from([reproposal_context.parent.1])).await; assert!( verify_rx.await.unwrap(), "verify should accept a valid boundary re-proposal" @@ -975,7 +992,7 @@ mod tests { drop(marshal); context.sleep(Duration::from_millis(1)).await; - let certify_rx = inline.certify(reproposal_round, boundary_digest).await; + let certify_rx = inline.certify(reproposal_round, boundary_digest, Arc::from([reproposal_context.parent.1])).await; select! { result = certify_rx => { assert!( @@ -1053,7 +1070,13 @@ mod tests { leader: me, parent: (View::new(2), digest), }; - let verify_rx = inline.verify(reproposal_context, digest).await; + let verify_rx = inline + .verify( + reproposal_context.clone(), + digest, + Arc::from([reproposal_context.parent.1]), + ) + .await; assert!( !verify_rx.await.expect("verify result missing"), "a non-boundary re-proposal must be rejected" @@ -1061,7 +1084,9 @@ mod tests { // The header-scoped rejection must not become the certification // verdict for the notarized digest. - let certify_rx = inline.certify(round, digest).await; + let certify_rx = inline + .certify(round, digest, Arc::from([reproposal_context.parent.1])) + .await; select! { result = certify_rx => { assert!( @@ -1152,9 +1177,15 @@ mod tests { "buffer broadcast for child should be accepted" ); - let verify_rx = inline.verify(child_ctx, child_digest).await; + let verify_rx = inline + .verify( + child_ctx.clone(), + child_digest, + Arc::from([child_ctx.parent.1]), + ) + .await; let certify_result = inline - .certify(child_round, child_digest) + .certify(child_round, child_digest, Arc::from([child_ctx.parent.1])) .await .await .expect("certify result missing"); @@ -1249,7 +1280,7 @@ mod tests { ); let digest = inline - .propose(ctx) + .propose(ctx.clone(), Arc::from([ctx.parent.1])) .await .await .expect("propose must return a digest"); @@ -1261,7 +1292,7 @@ mod tests { // The leader certifies its own proposal, which awaits the deferred sync handle. assert!( inline - .certify(round, child_digest) + .certify(round, child_digest, Arc::from([ctx.parent.1])) .await .await .expect("certify result missing"), @@ -1339,7 +1370,13 @@ mod tests { B::new::(block_context.clone(), genesis.digest(), Height::new(1), 100); let digest = block.digest(); - let verify_rx = inline.verify(block_context, digest).await; + let verify_rx = inline + .verify( + block_context.clone(), + digest, + Arc::from([block_context.parent.1]), + ) + .await; drop(verify_rx); // Give the verify task a chance to observe the dropped receiver while its @@ -1347,7 +1384,9 @@ mod tests { context.sleep(Duration::from_millis(10)).await; assert!(marshal.verified(round, block).await); - let certify_rx = inline.certify(round, digest).await; + let certify_rx = inline + .certify(round, digest, Arc::from([block_context.parent.1])) + .await; select! { result = certify_rx => { assert!( @@ -1431,7 +1470,7 @@ mod tests { "buffer broadcast for child should be accepted" ); - let verify_rx = inline.verify(child_ctx, child_digest).await; + let verify_rx = inline.verify(child_ctx.clone(), child_digest, Arc::from([child_ctx.parent.1])).await; // Application verification is now blocked. The store request runs concurrently // with it, so the block is locally queryable even though the notarize vote has @@ -1451,7 +1490,7 @@ mod tests { verify_rx.await.expect("verify result missing"), "inline verify should pass once verification is released" ); - let certify_rx = inline.certify(child_round, child_digest).await; + let certify_rx = inline.certify(child_round, child_digest, Arc::from([child_ctx.parent.1])).await; select! { result = certify_rx => { assert!( @@ -1550,7 +1589,7 @@ mod tests { FixedEpocher::new(BLOCKS_PER_EPOCH), ); - let digest_rx = inline.propose(ctx).await; + let digest_rx = inline.propose(ctx.clone(), Arc::from([ctx.parent.1])).await; assert!( digest_rx.await.is_err(), "propose must drop the receiver so the voter nullifies the round via timeout" @@ -1653,14 +1692,14 @@ mod tests { leader, parent: (View::new(1), certified_digest), }; - let verify_rx = inline.verify(equivocating_ctx, digest).await; + let verify_rx = inline.verify(equivocating_ctx.clone(), digest, Arc::from([equivocating_ctx.parent.1])).await; assert!( !verify_rx.await.expect("verify result missing"), "the equivocating proposal must not be notarized" ); // The honest notarization for the same `(round, digest)` arrives. - let certify_rx = inline.certify(round, digest).await; + let certify_rx = inline.certify(round, digest, Arc::from([equivocating_ctx.parent.1])).await; select! { result = certify_rx => { assert!( diff --git a/consensus/src/marshal/standard/mod.rs b/consensus/src/marshal/standard/mod.rs index 650387df7ac..4415a4821b0 100644 --- a/consensus/src/marshal/standard/mod.rs +++ b/consensus/src/marshal/standard/mod.rs @@ -1943,24 +1943,34 @@ mod tests { } } - async fn propose(&mut self, context: Ctx) -> oneshot::Receiver { + async fn propose(&mut self, context: Ctx, ancestry: Arc<[D]>) -> oneshot::Receiver { match self { - Self::Inline(inline) => inline.propose(context).await, - Self::Deferred(deferred) => deferred.propose(context).await, + Self::Inline(inline) => inline.propose(context, ancestry).await, + Self::Deferred(deferred) => deferred.propose(context, ancestry).await, } } - async fn verify(&mut self, context: Ctx, digest: D) -> oneshot::Receiver { + async fn verify( + &mut self, + context: Ctx, + digest: D, + ancestry: Arc<[D]>, + ) -> oneshot::Receiver { match self { - Self::Inline(inline) => inline.verify(context, digest).await, - Self::Deferred(deferred) => deferred.verify(context, digest).await, + Self::Inline(inline) => inline.verify(context, digest, ancestry).await, + Self::Deferred(deferred) => deferred.verify(context, digest, ancestry).await, } } - async fn certify(&mut self, round: Round, digest: D) -> oneshot::Receiver { + async fn certify( + &mut self, + round: Round, + digest: D, + ancestry: Arc<[D]>, + ) -> oneshot::Receiver { match self { - Self::Inline(inline) => inline.certify(round, digest).await, - Self::Deferred(deferred) => deferred.certify(round, digest).await, + Self::Inline(inline) => inline.certify(round, digest, ancestry).await, + Self::Deferred(deferred) => deferred.certify(round, digest, ancestry).await, } } } @@ -1969,22 +1979,32 @@ mod tests { type Context = Ctx; type Digest = D; - async fn propose(&mut self, context: Self::Context) -> oneshot::Receiver { - Self::propose(self, context).await + async fn propose( + &mut self, + context: Self::Context, + ancestry: Arc<[Self::Digest]>, + ) -> oneshot::Receiver { + Self::propose(self, context, ancestry).await } async fn verify( &mut self, context: Self::Context, digest: Self::Digest, + ancestry: Arc<[Self::Digest]>, ) -> oneshot::Receiver { - Self::verify(self, context, digest).await + Self::verify(self, context, digest, ancestry).await } } impl CertifiableAutomaton for Wrapper { - async fn certify(&mut self, round: Round, digest: Self::Digest) -> oneshot::Receiver { - Self::certify(self, round, digest).await + async fn certify( + &mut self, + round: Round, + digest: Self::Digest, + ancestry: Arc<[Self::Digest]>, + ) -> oneshot::Receiver { + Self::certify(self, round, digest, ancestry).await } } @@ -2086,8 +2106,8 @@ mod tests { // until the leader's block arrives. A notarization can arrive in // this window, causing Simplex's sole `certify` request to consume // and await that still-pending gate. - let verify_rx = wrapper.verify(conflicting_context, digest).await; - let certify_rx = wrapper.certify(round, digest).await; + let verify_rx = wrapper.verify(conflicting_context, digest, Arc::from([genesis.digest(), certified_digest])).await; + let certify_rx = wrapper.certify(round, digest, Arc::from([genesis.digest(), certified_digest, skipped_digest])).await; assert!( buffer .broadcast(Recipients::Some(vec![]), block) @@ -2335,6 +2355,19 @@ mod tests { } context.sleep(Duration::from_millis(250)).await; + // The selected parent certificate reaches Simplex after rejection of + // the conflicting header. Its body is already available in Marshal. + let parent_notarization = StandardHarness::make_notarization( + Proposal::new(skipped_round, View::new(1), skipped_digest), + &schemes, + QUORUM, + ); + byzantine_certificate_sender.send( + Recipients::One(victim.clone()), + Certificate::::Notarization(parent_notarization).encode(), + true, + ); + // The Byzantine validator and the other two honest validators // notarize the header naming view 2 without the victim's vote. let good_votes: Vec<_> = [0usize, 2, 3] @@ -2431,7 +2464,7 @@ mod tests { context.sleep(Duration::from_millis(10)).await; let verify_result = wrapper - .verify(block_context, digest) + .verify(block_context, digest, Arc::from([genesis.digest()])) .await .await .expect("verify result missing"); @@ -2441,7 +2474,7 @@ mod tests { ); let certify_result = wrapper - .certify(round, digest) + .certify(round, digest, Arc::from([genesis.digest()])) .await .await .expect("certify result missing"); @@ -2491,7 +2524,13 @@ mod tests { parent: (View::zero(), genesis.digest()), }; let missing = Sha256::hash(&[b"missing candidate"]); - let mut verify = wrapper.verify(consensus_context, missing).await; + let mut verify = wrapper + .verify( + consensus_context.clone(), + missing, + Arc::from([consensus_context.parent.1]), + ) + .await; context.sleep(Duration::from_millis(50)).await; assert!( @@ -2571,7 +2610,9 @@ mod tests { let proposal = Proposal::new(round, View::zero(), digest); let notarization = StandardHarness::make_notarization(proposal, &schemes, QUORUM); resolver.respond_to_next_fetch((notarization, block).encode()); - let certify = wrapper.certify(round, digest).await; + let certify = wrapper + .certify(round, digest, Arc::from([genesis.digest()])) + .await; let result = certify.await.expect("certify result missing"); assert!( @@ -2688,7 +2729,7 @@ mod tests { // block subscription cannot pull from peers, so it stays parked // until something delivers the block locally. let block_context = case.block_context.clone(); - let verify_rx = case.wrapper.verify(block_context, digest).await; + let verify_rx = case.wrapper.verify(block_context, digest, Arc::from([case.block_context.parent.1])).await; // Stage the notarized response so the bump's fetch can resolve. let proposal = Proposal::new(round, View::zero(), digest); @@ -2702,7 +2743,7 @@ mod tests { // resolver delivers, and the marshal stores the block and wakes // verify's digest subscription, letting the pending verify task // resolve the gate that certify awaits. - let certify_rx = case.wrapper.certify(round, digest).await; + let certify_rx = case.wrapper.certify(round, digest, Arc::from([case.block_context.parent.1])).await; select! { result = verify_rx => { @@ -2765,8 +2806,18 @@ mod tests { // The next lookup returns ownership while removing the buffer entry, modeling // same-peer cache pressure. let block_context = case.block_context.clone(); - let verify_rx = case.wrapper.verify(block_context, digest).await; - let certify_rx = case.wrapper.certify(round, digest).await; + let verify_rx = case + .wrapper + .verify( + block_context, + digest, + Arc::from([case.block_context.parent.1]), + ) + .await; + let certify_rx = case + .wrapper + .certify(round, digest, Arc::from([case.block_context.parent.1])) + .await; // This request is ordered after the verification subscription and // certification hint in the marshal mailbox. Once it returns, the @@ -2848,14 +2899,18 @@ mod tests { B::new::(block_context.clone(), genesis.digest(), Height::new(1), 100); let digest = block.digest(); - let verify_rx = wrapper.verify(block_context, digest).await; + let verify_rx = wrapper + .verify(block_context, digest, Arc::from([genesis.digest()])) + .await; drop(verify_rx); context.sleep(Duration::from_millis(10)).await; let proposal = Proposal::new(round, View::zero(), digest); let notarization = StandardHarness::make_notarization(proposal, &schemes, QUORUM); resolver.respond_to_next_fetch((notarization, block).encode()); - let certify_rx = wrapper.certify(round, digest).await; + let certify_rx = wrapper + .certify(round, digest, Arc::from([genesis.digest()])) + .await; select! { result = certify_rx => { @@ -2936,7 +2991,13 @@ mod tests { let child_digest = child.digest(); assert!(marshal.verified(child_round, child).await); - let verify = wrapper.verify(child_context, child_digest).await; + let verify = wrapper + .verify( + child_context, + child_digest, + Arc::from([genesis.digest(), parent_digest]), + ) + .await; wait_until( &context, Duration::from_secs(5), @@ -2982,7 +3043,13 @@ mod tests { verify_result, "deferred verify should optimistically pass pre-checks" ); - let certify = wrapper.certify(child_round, child_digest).await; + let certify = wrapper + .certify( + child_round, + child_digest, + Arc::from([genesis.digest(), parent_digest]), + ) + .await; assert!( !certify.await.expect("certify result missing"), "deferred certify should reject non-contiguous ancestry" @@ -3105,11 +3172,11 @@ mod tests { mock_app, victim_mailbox.clone(), ); - let verify = wrapper.verify(child_context, child_digest).await; + let verify = wrapper.verify(child_context, child_digest, Arc::from([genesis.digest(), parent_digest])).await; let verify_or_certify = if kind == WrapperKind::Deferred { let optimistic = verify.await.expect("verify result missing"); assert!(optimistic, "deferred verify should optimistically succeed"); - wrapper.certify(child_round, child_digest).await + wrapper.certify(child_round, child_digest, Arc::from([genesis.digest(), parent_digest])).await } else { verify }; @@ -3211,7 +3278,9 @@ mod tests { leader: me.clone(), parent: (View::zero(), genesis.digest()), }; - let proposal_rx = wrapper.propose(non_boundary_context).await; + let proposal_rx = wrapper + .propose(non_boundary_context, Arc::from([genesis.digest()])) + .await; assert!( proposal_rx.await.is_err(), "{kind:?}: proposal should be dropped when application returns no block" @@ -3254,7 +3323,9 @@ mod tests { leader: me, parent: (View::new(boundary_height.get()), boundary_digest), }; - let reproposal_rx = wrapper.propose(reproposal_context).await; + let reproposal_rx = wrapper + .propose(reproposal_context, Arc::from([boundary_digest])) + .await; assert_eq!( reproposal_rx.await.expect("reproposal result missing"), boundary_digest, @@ -3263,7 +3334,13 @@ mod tests { // The re-proposal registers a certification gate whose durability // certify awaits before the finalize vote. - let certify_rx = wrapper.certify(reproposal_round, boundary_digest).await; + let certify_rx = wrapper + .certify( + reproposal_round, + boundary_digest, + Arc::from([boundary_digest]), + ) + .await; assert!( certify_rx.await.expect("certify result missing"), "{kind:?}: certify must succeed for the re-proposed boundary block" @@ -3343,7 +3420,11 @@ mod tests { }; assert!( wrapper - .verify(valid_reproposal_context, boundary_digest) + .verify( + valid_reproposal_context, + boundary_digest, + Arc::from([boundary_digest]) + ) .await .await .expect("verify result missing"), @@ -3383,7 +3464,11 @@ mod tests { }; assert!( !wrapper - .verify(invalid_reproposal_context, non_boundary_digest) + .verify( + invalid_reproposal_context, + non_boundary_digest, + Arc::from([non_boundary_digest]) + ) .await .await .expect("verify result missing"), @@ -3398,7 +3483,11 @@ mod tests { }; assert!( !wrapper - .verify(cross_epoch_context, boundary_digest) + .verify( + cross_epoch_context, + boundary_digest, + Arc::from([boundary_digest]) + ) .await .await .expect("verify result missing"), @@ -3409,7 +3498,11 @@ mod tests { // Deferred-only crash-recovery path: certify without prior verify. let certify_only_round = Round::new(Epoch::zero(), View::new(21)); let certify_result = wrapper - .certify(certify_only_round, boundary_digest) + .certify( + certify_only_round, + boundary_digest, + Arc::from([boundary_digest]), + ) .await .await; assert!( @@ -3483,7 +3576,11 @@ mod tests { context.sleep(Duration::from_millis(10)).await; let malformed_verify = wrapper - .verify(malformed_context.clone(), malformed_digest) + .verify( + malformed_context.clone(), + malformed_digest, + Arc::from([genesis.digest()]), + ) .await .await .expect("verify result missing"); @@ -3499,7 +3596,13 @@ mod tests { malformed_verify, "deferred verify should optimistically pass pre-checks" ); - let certify = wrapper.certify(malformed_round, malformed_digest).await; + let certify = wrapper + .certify( + malformed_round, + malformed_digest, + Arc::from([genesis.digest()]), + ) + .await; assert!( !certify.await.expect("certify result missing"), "deferred certify should reject non-contiguous ancestry" @@ -3542,7 +3645,11 @@ mod tests { context.sleep(Duration::from_millis(10)).await; let mismatch_verify = wrapper - .verify(mismatched_context, mismatched_digest) + .verify( + mismatched_context, + mismatched_digest, + Arc::from([genesis.digest(), parent_digest]), + ) .await .await .expect("verify result missing"); @@ -3558,7 +3665,13 @@ mod tests { mismatch_verify, "deferred verify should optimistically pass pre-checks" ); - let certify = wrapper.certify(mismatch_round, mismatched_digest).await; + let certify = wrapper + .certify( + mismatch_round, + mismatched_digest, + Arc::from([genesis.digest(), parent_digest]), + ) + .await; assert!( !certify.await.expect("certify result missing"), "deferred certify should reject mismatched parent digest" @@ -3628,7 +3741,7 @@ mod tests { // - Inline fails in `verify`. // - Deferred returns optimistic success and fails in `certify`. let verify_result = wrapper - .verify(verify_context, digest) + .verify(verify_context, digest, Arc::from([genesis.digest(), parent_digest])) .await .await .expect("verify result missing"); @@ -3642,7 +3755,7 @@ mod tests { verify_result, "deferred verify should pass pre-checks and schedule deferred verification" ); - let certify = wrapper.certify(round, digest).await; + let certify = wrapper.certify(round, digest, Arc::from([genesis.digest(), parent_digest])).await; assert!( !certify.await.expect("certify result missing"), "deferred certify should propagate deferred application verification failure" diff --git a/consensus/src/simplex/actors/voter/actor.rs b/consensus/src/simplex/actors/voter/actor.rs index c5d946e71f9..993808d57e2 100644 --- a/consensus/src/simplex/actors/voter/actor.rs +++ b/consensus/src/simplex/actors/voter/actor.rs @@ -358,9 +358,19 @@ impl< /// Attempt to propose a new block. #[allow(clippy::async_yields_async)] - async fn try_propose(&mut self) -> Option, D>> { - // Check if we are ready to propose - let context = self.state.try_propose()?; + async fn try_propose( + &mut self, + resolver: &mut resolver::Mailbox, + ) -> Option, D>> { + let (context, fetches) = self.state.try_propose(); + for CertificateFetch { proposal, view } in fetches { + resolver.resolve(proposal, view, Kind::Notarization, None); + } + let context = context?; + let ancestry = self + .state + .ancestry(context.parent.0) + .expect("proposal ancestry ready"); // Request proposal from application let span = info_span!( @@ -371,7 +381,7 @@ impl< ); let receiver = async { debug!(round = ?context.round, "requested proposal from automaton"); - self.automaton.propose(context.clone()).await + self.automaton.propose(context.clone(), ancestry).await } .instrument(span.clone()) .await; @@ -398,6 +408,10 @@ impl< } Verify::Wait => return None, }; + let ancestry = self + .state + .ancestry(context.parent.0) + .expect("verification ancestry ready"); // Request verification let span = info_span!( @@ -409,7 +423,7 @@ impl< let receiver = async { debug!(?proposal, "requested proposal verification"); self.automaton - .verify(context.clone(), proposal.payload) + .verify(context.clone(), proposal.payload, ancestry) .await } .instrument(span.clone()) @@ -445,7 +459,7 @@ impl< // State and Round prevent duplicate requests when both checkpoints // observe the same view. if pending_propose.is_none() { - *pending_propose = self.try_propose().await; + *pending_propose = self.try_propose(resolver).await; } if pending_verify.is_none() { *pending_verify = self.try_verify(resolver).await; @@ -1125,6 +1139,7 @@ impl< resolver.resolve(proposal, view, Kind::Notarization, None); } for proposal in candidates { + let ancestry = self.state.ancestry(proposal.parent).expect("certification ancestry ready"); let round = proposal.round; let view = round.view(); debug!(%view, "attempting certification"); @@ -1135,7 +1150,7 @@ impl< view = view.traced() ); #[allow(clippy::async_yields_async)] - let receiver = async { self.automaton.certify(round, proposal.payload).await } + let receiver = async { self.automaton.certify(round, proposal.payload, ancestry).await } .instrument(span.clone()) .await; let handle = certify_pool.push(async move { (round, span, receiver.await) }); @@ -1316,3 +1331,129 @@ impl< .expect("unable to sync journal"); } } + +#[cfg(test)] +mod tests { + use super::*; + use crate::{ + simplex::{ + elector::{Config as _, RoundRobin}, + mocks, + scheme::ed25519, + }, + types::{Epoch, TermLength, ViewDelta}, + }; + use commonware_cryptography::{ + Sha256, certificate::Scheme as _, ed25519::PublicKey, sha256::Digest as Sha256Digest, + }; + use commonware_parallel::Sequential; + use commonware_runtime::{Runner, Supervisor as _, deterministic}; + use commonware_utils::{NZU16, NZU32, NZUsize, non_empty}; + use core::panic; + use std::{sync::Arc, time::Duration}; + + #[derive(Clone)] + struct UnusedBlocker; + + impl Blocker for UnusedBlocker { + type PublicKey = PublicKey; + + fn block(&mut self, _: PublicKey) -> commonware_actor::Feedback { + panic!("proposal repair cannot block peers"); + } + + fn blocked(&mut self) -> commonware_p2p::BlockedSubscription { + panic!("proposal repair does not subscribe to blocked peers"); + } + } + + #[test] + fn proposal_repairs_dispatch_without_ready_context() { + deterministic::Runner::default().start(|mut context| async move { + let fixture = ed25519::fixture(&mut context, b"proposal_repairs", 4); + let scheme = fixture.schemes[2].clone(); + let epoch = Epoch::new(9); + let genesis = mocks::application::genesis::(epoch); + let elector = RoundRobin::::default().with_term( + TermLength::new(NZU32!(9)), + Duration::from_secs(4), + ViewDelta::new(8), + ); + let reporter = mocks::reporter::Reporter::new( + context.child("reporter"), + mocks::reporter::Config { + participants: fixture.participants.clone().try_into().unwrap(), + scheme: scheme.clone(), + elector: elector.clone(), + }, + ); + let (_application_actor, application) = mocks::application::Application::new( + context.child("application"), + mocks::application::Config:: { + relay: Arc::new(mocks::relay::Relay::new()), + me: fixture.participants[2].clone(), + propose_latency: (1.0, 0.0), + verify_latency: (1.0, 0.0), + certify_latency: (1.0, 0.0), + should_certify: mocks::application::Certifier::Always, + }, + ); + let (mut actor, _mailbox) = Actor::new( + context.child("voter"), + Config { + elector: elector.build(scheme.participants()), + scheme, + blocker: UnusedBlocker, + automaton: application.clone(), + relay: application, + reporter, + partition: "proposal_repairs".into(), + epoch, + floor: Floor::Genesis(genesis), + mailbox_size: NZUsize!(16), + leader_timeout: Duration::from_secs(1), + certification_timeout: Duration::from_secs(2), + timeout_retry: Duration::from_secs(3), + skip_budget: 4, + view_retention: ViewDelta::new(10), + replay_buffer: NZUsize!(1024), + write_buffer: NZUsize!(1024), + page_cache: CacheRef::from_pooler(&context, NZU16!(1024), NZUsize!(10)), + }, + ); + actor.state.set_genesis(genesis); + assert_eq!(actor.state.try_propose().0.unwrap().view(), View::new(1)); + for view in [2, 4, 6] { + let proposal = Proposal::new( + Rnd::new(epoch, View::new(view)), + View::new(view - 1), + Sha256Digest::from([view as u8; 32]), + ); + let votes: Vec<_> = fixture + .schemes + .iter() + .map(|scheme| Notarize::sign(scheme, proposal.clone()).unwrap()) + .collect(); + let certificate = Notarization::from_notarizes( + &fixture.verifier, + non_empty![@&votes], + &Sequential, + ) + .unwrap(); + assert!(actor.state.add_notarization(certificate).0); + } + let (sender, mut receiver) = mailbox::new(context.child("resolver"), NZUsize!(16)); + let mut resolver = resolver::Mailbox::new(sender); + assert!(actor.try_propose(&mut resolver).await.is_none()); + for (proposal, view) in [(3, 1), (5, 3), (7, 5)] { + assert!( + matches!(receiver.try_recv().unwrap(), resolver::MailboxMessage::Resolve { + proposal: candidate, view: missing, kind: Kind::Notarization, target: None, .. + } if candidate == View::new(proposal) && missing == View::new(view)) + ); + } + assert!(actor.try_propose(&mut resolver).await.is_none()); + assert!(receiver.try_recv().is_err()); + }); + } +} diff --git a/consensus/src/simplex/actors/voter/mod.rs b/consensus/src/simplex/actors/voter/mod.rs index 0bf6d202431..71e71071578 100644 --- a/consensus/src/simplex/actors/voter/mod.rs +++ b/consensus/src/simplex/actors/voter/mod.rs @@ -1723,7 +1723,7 @@ mod tests { _ => panic!("unexpected batcher message"), } - let view = View::new(2); + let view = View::new(1); let proposal_a = Proposal::new( Round::new(Epoch::new(333), view), view.previous().unwrap(), @@ -1910,7 +1910,7 @@ mod tests { _ => panic!("unexpected batcher message"), } - let view = View::new(2); + let view = View::new(1); let proposal = Proposal::new( Round::new(Epoch::new(333), view), view.previous().unwrap(), @@ -4108,6 +4108,142 @@ mod tests { missed_notarization_is_fetched(ed25519::fixture); } + fn certification_repairs_cross_term_parent(conflicting_local_parent: bool) { + deterministic::Runner::timed(Duration::from_secs(30)).start(|mut context| async move { + let epoch = Epoch::new(333); + let Fixture { participants, schemes, .. } = + ed25519::fixture(&mut context, b"_COMMONWARE_CONSENSUS_TEST_CERTIFICATION_PARENT_REPAIR", 4); + let oracle = start_test_network_with_peers( + context.child("network"), participants.clone(), true, + ).await; + let genesis = mocks::application::genesis::(epoch); + let local_round = Round::new(epoch, View::new(1)); + let parent = Proposal::new( + local_round, View::zero(), Sha256::hash(&[b"selected-B"]), + ); + let candidate = Proposal::new( + Round::new(epoch, View::new(6)), parent.view(), Sha256::hash(&[b"candidate-C"]), + ); + let repaired = Arc::new(Mutex::new(false)); + let observed = Arc::new(Mutex::new(0usize)); + let repair_observer = repaired.clone(); + let call_observer = observed.clone(); + let expected = vec![genesis, parent.payload]; + let candidate_view = candidate.view(); + let candidate_payload = candidate.payload; + let elector = RoundRobin::::default().with_term( + TermLength::new(NZU32!(5)), Duration::from_secs(30), ViewDelta::new(2), + ); + let (mut mailbox, mut batcher, mut resolver, relay, _) = setup_voter( + &context, &oracle, &participants, &schemes, elector, + VoterOptions { + local_index: 1, + leader_timeout: Duration::from_secs(20), + certifier: mocks::application::Certifier::WithAncestry(Box::new( + move |round, payload, ancestry| { + if round.view() == candidate_view { + assert!(*repair_observer.lock(), + "certification ran before its untargeted parent repair"); + assert_eq!(payload, candidate_payload); + assert_eq!(&*ancestry, expected.as_slice()); + *call_observer.lock() += 1; + } + true + }, + )), + ..VoterOptions::default() + }, + ).await; + if conflicting_local_parent { + let local_contents = (local_round, genesis, 0u64).encode(); + let local = Proposal::new( + local_round, View::zero(), Sha256::hash(&[&local_contents]), + ); + + // Epoch 333 plus one-based term 1 elects participant 2. + let leader = participants[2].clone(); + relay.broadcast( + &leader, Recipients::All, + (local.payload, local_contents), + ); + mailbox.proposal(local.clone()); + loop { + select! { + message = batcher.recv() => { + if let batcher::Message::Constructed(Vote::Notarize(vote)) = message.unwrap() + && vote.view() == local.view() + { + assert_eq!(vote.proposal, local); + break; + } + }, + _message = resolver.recv() => {}, + _ = context.sleep(Duration::from_secs(10)) => + panic!("local A was not verified and voted"), + } + } + } + + let quorum_schemes = [schemes[0].clone(), schemes[2].clone(), schemes[3].clone()]; + let (_, skipped) = build_nullification( + &quorum_schemes, Round::new(epoch, View::new(2)), quorum(4), + ); + let (_, parent_certificate) = + build_notarization(&quorum_schemes, &parent, quorum(4)); + let (_, candidate_certificate) = + build_notarization(&quorum_schemes, &candidate, quorum(4)); + mailbox.recovered(Certificate::Nullification(skipped)); + mailbox.recovered(Certificate::Notarization(candidate_certificate)); + loop { + select! { + message = resolver.recv() => { + match message.unwrap() { + MailboxMessage::Resolve { + proposal, view, + kind: crate::simplex::actors::Kind::Notarization, + target: None, .. + } if proposal == candidate.view() && view == parent.view() => { + assert!(!*repaired.lock(), "repair must be deduplicated"); + assert_eq!(*observed.lock(), 0); + *repaired.lock() = true; + mailbox.recovered(Certificate::Notarization( + parent_certificate.clone(), + )); + } + MailboxMessage::Certified { view, success, .. } + if view == candidate.view() => { + assert!(*repaired.lock()); + assert!(success); + } + _ => {} + } + }, + message = batcher.recv() => { + if let batcher::Message::Constructed(Vote::Finalize(vote)) = message.unwrap() + && vote.view() == candidate.view() + { + assert!(*repaired.lock()); + assert_eq!(*observed.lock(), 1); + break; + } + }, + _ = context.sleep(Duration::from_secs(10)) => + panic!("candidate failed to certify after exact parent repair"), + } + } + }); + } + + #[test] + fn test_certification_repairs_conflicting_cross_term_parent() { + certification_repairs_cross_term_parent(true); + } + + #[test] + fn test_certification_repairs_missing_cross_term_ancestry() { + certification_repairs_cross_term_parent(false); + } + /// Tests that when proposal verification fails, the voter emits a nullify vote /// immediately rather than waiting for the timeout. fn verification_failure_emits_nullify_immediately(mut fixture: F, elector: L) diff --git a/consensus/src/simplex/actors/voter/state.rs b/consensus/src/simplex/actors/voter/state.rs index dd2851aa359..75e54f39c75 100644 --- a/consensus/src/simplex/actors/voter/state.rs +++ b/consensus/src/simplex/actors/voter/state.rs @@ -21,8 +21,9 @@ use commonware_runtime::{ use commonware_utils::futures::Aborter; use rand_core::CryptoRng; use std::{ - collections::{BTreeMap, BTreeSet}, + collections::{BTreeMap, BTreeSet, btree_map::Entry}, mem::{replace, take}, + sync::Arc, time::{Duration, SystemTime}, }; use tracing::{Span, debug, warn}; @@ -94,8 +95,7 @@ pub enum Verify, D: Digest> { Wait, } -/// A certificate fetch justified by a blocked certification (see -/// [`State::certify_candidates`]). +/// A certificate needed to prepare an application request. pub struct CertificateFetch { /// View of the candidate that exposed the missing certificate. pub proposal: View, @@ -103,6 +103,17 @@ pub struct CertificateFetch { pub view: View, } +/// Why selected application ancestry cannot yet be supplied. +#[derive(Debug, PartialEq, Eq, thiserror::Error)] +pub enum AncestryError { + /// The selected path needs the proposal at this view. + #[error("missing proposal at view {0}")] + Missing(View), + /// The path is malformed or skips the current finalized anchor. + #[error("invalid selected ancestry")] + Invalid, +} + /// Configuration for initializing [`State`]. pub struct Config> { pub scheme: S, @@ -160,6 +171,9 @@ pub struct State, L: Elector, D: certification_candidates: BTreeSet, outstanding_certifications: BTreeSet, + /// Eligible certifications waiting for an arbitrary older proposal. The + /// in-term certification wakeup only covers the immediate child. + ancestry_waiters: BTreeMap>, current_view: Gauge, tracked_views: Gauge, @@ -246,6 +260,7 @@ impl, L: Elector, D: Digest> Sta failed_certifications: BTreeSet::new(), certification_candidates: BTreeSet::new(), outstanding_certifications: BTreeSet::new(), + ancestry_waiters: BTreeMap::new(), current_view, tracked_views, issuance_window_probes, @@ -580,6 +595,9 @@ impl, L: Elector, D: Digest> Sta if view > self.last_finalized { self.certification_candidates.insert(view); } + if let Some(waiters) = self.ancestry_waiters.remove(&view) { + self.certification_candidates.extend(waiters); + } self.slide_optimistic_frontier(view); } result @@ -633,6 +651,11 @@ impl, L: Elector, D: Digest> Sta // Prune certification candidates at or below finalized view. // Finalization is definitive, so these certifications are no longer relevant. self.certification_candidates.retain(|v| *v > view); + let pending = self.ancestry_waiters.split_off(&view.next()); + for waiters in replace(&mut self.ancestry_waiters, pending).into_values() { + self.certification_candidates + .extend(waiters.into_iter().filter(|candidate| *candidate > view)); + } // Abort outstanding certifications at or below finalized view for the same reason. let keep = self.outstanding_certifications.split_off(&view.next()); @@ -877,12 +900,13 @@ impl, L: Elector, D: Digest> Sta } /// Returns proposal context for the lowest locally admissible tracked view - /// ready to propose. - pub fn try_propose(&mut self) -> Option> { + /// ready to propose, plus missing-ancestry repairs encountered during selection. + pub fn try_propose(&mut self) -> (Option>, Vec) { // Nothing above the next term start is admissible (see // [`Self::admits_outbound`]), so bound the scan rather than walking every // tracked future round (certificates can land arbitrarily far ahead). // Ascending order gives the current view precedence over optimistic work. + let mut fetches = Vec::new(); let limit = self.view.next_term_start(self.term_length()); let mut cursor = self.view; while let Some(view) = self.next_tracked_view(cursor) { @@ -913,6 +937,21 @@ impl, L: Elector, D: Digest> Sta continue; } }; + if let Err(err) = self.visit_ancestry(parent_view, |_| {}) { + if let AncestryError::Missing(missing) = err + && self + .views + .get_mut(&view) + .expect("tracked round") + .request(missing) + { + fetches.push(CertificateFetch { + proposal: view, + view: missing, + }); + } + continue; + } let Some(leader) = self .views .get_mut(&view) @@ -920,13 +959,16 @@ impl, L: Elector, D: Digest> Sta else { continue; }; - return Some(Context { - round: Rnd::new(self.epoch, view), - leader: leader.key, - parent: (parent_view, parent_payload), - }); + return ( + Some(Context { + round: Rnd::new(self.epoch, view), + leader: leader.key, + parent: (parent_view, parent_payload), + }), + fetches, + ); } - None + (None, fetches) } /// Records a locally constructed proposal once the automaton finishes building it. @@ -1033,6 +1075,23 @@ impl, L: Elector, D: Digest> Sta }; } }; + if let Err(err) = self.visit_ancestry(proposal.parent, |_| {}) { + if let AncestryError::Missing(missing) = err + && self + .views + .get_mut(&view) + .expect("tracked round") + .request(missing) + { + return Verify::Resolve { + proposal: view, + view: missing, + kind: Kind::Notarization, + target: leader.key, + }; + } + continue; + } let Some(round) = self.views.get_mut(&view) else { continue; }; @@ -1143,9 +1202,8 @@ impl, L: Elector, D: Digest> Sta let Some(proposal) = self .views - .get(&view) - .and_then(|round| round.proposal()) - .cloned() + .get_mut(&view) + .and_then(|round| round.try_certify()) else { continue; }; @@ -1153,7 +1211,7 @@ impl, L: Elector, D: Digest> Sta if err.invalid_proposal() { warn!(round = ?proposal.round, ?err, "proposal failed certification precheck"); } else { - // Dormant candidates wake only through + // Candidates blocked on local certification wake through // [`Self::wake_certification_child`]. Therefore, // [`Self::certification_parent_ready`] may block only on an // uncertified parent. @@ -1166,13 +1224,26 @@ impl, L: Elector, D: Digest> Sta continue; } - if let Some(candidate) = self - .views - .get_mut(&view) - .and_then(|round| round.try_certify()) - { - ready.push(candidate); + // The candidate's notarization anchors every parent link to its selected branch. + if let Err(err) = self.visit_ancestry(proposal.view(), |_| {}) { + if let AncestryError::Missing(missing) = err { + match self.ancestry_waiters.entry(missing) { + Entry::Vacant(entry) => { + entry.insert(BTreeSet::from([view])); + fetches.push(CertificateFetch { + proposal: view, + view: missing, + }); + } + Entry::Occupied(mut entry) => { + entry.get_mut().insert(view); + } + } + } + continue; } + + ready.push(proposal); } (ready, fetches) } @@ -1278,6 +1349,68 @@ impl, L: Elector, D: Digest> Sta // which delegates to `optimistic_ancestry_payload` so issuance and proposal // construction share one ancestry rule. + /// Returns the complete selected path from the current finalized anchor + /// through `parent`. Consecutive re-proposals contribute one commitment. + pub fn ancestry(&self, parent: View) -> Result, AncestryError> { + let mut ancestry = Vec::new(); + self.visit_ancestry(parent, |payload| ancestry.push(*payload))?; + ancestry.reverse(); + ancestry.dedup(); + Ok(ancestry.into()) + } + + /// Visits parent links without allocating so input readiness can be checked + /// before consuming an application request's one-shot latch. + fn visit_ancestry( + &self, + mut parent: View, + mut visit: impl FnMut(&D), + ) -> Result<(), AncestryError> { + // A certificate's ancestors require certificates: a local vote at an + // earlier view may name a different proposal on an equivocating branch. + let mut optimistic = true; + loop { + if parent < self.last_finalized { + return Err(AncestryError::Invalid); + } + if parent == GENESIS_VIEW { + visit(self.genesis.as_ref().expect("genesis must be present")); + return Ok(()); + } + let round = self + .views + .get(&parent) + .ok_or(AncestryError::Missing(parent))?; + let proposal = if round.is_directly_notarized() { + optimistic = false; + round + .finalization() + .map(|certificate| &certificate.proposal) + .or_else(|| { + round + .notarization() + .map(|certificate| &certificate.proposal) + }) + } else { + (optimistic + && round.has_unequivocated_proposal() + && round.broadcast_notarize() + && round.is_verified()) + .then(|| round.proposal()) + .flatten() + }; + let proposal = proposal.ok_or(AncestryError::Missing(parent))?; + if proposal.parent >= parent { + return Err(AncestryError::Invalid); + } + visit(&proposal.payload); + if parent == self.last_finalized { + return Ok(()); + } + parent = proposal.parent; + } + } + /// Returns the payload of `view`'s explicitly certified (or finalized) /// proposal, the strongest form of ancestry. fn explicit_ancestry_payload(&self, view: View) -> Option<&D> { @@ -1860,6 +1993,466 @@ mod tests { proposal } + #[test] + fn ancestry_follows_selected_links_and_keeps_finalized_anchor() { + deterministic::Runner::default().start(|mut context| async move { + let ( + Fixture { + schemes, verifier, .. + }, + mut state, + ) = setup_state(&mut context, 4, 9, 0, 1, 4); + let anchor = fetch_proposal(1, 0, 11); + let sibling = fetch_proposal(2, 0, 12); + let parent = fetch_proposal(3, 1, 13); + for proposal in [&anchor, &sibling, &parent] { + assert!( + state + .add_notarization(build_notarization(&verifier, &schemes, proposal)) + .0 + ); + } + assert!(state.add_nullification(build_nullification( + &verifier, + &schemes, + sibling.round + ))); + assert_eq!( + state.ancestry(parent.view()).unwrap().to_vec(), + vec![test_genesis(), anchor.payload, parent.payload] + ); + + assert!( + state + .add_finalization(build_finalization(&verifier, &schemes, &anchor)) + .0 + ); + state.prune(); + assert_eq!( + state.ancestry(parent.view()).unwrap().to_vec(), + vec![anchor.payload, parent.payload] + ); + + let selected = state.ancestry(parent.view()).unwrap(); + assert!( + state + .add_finalization(build_finalization(&verifier, &schemes, &parent)) + .0 + ); + assert_eq!(&*selected, &[anchor.payload, parent.payload]); + assert_eq!(state.ancestry(anchor.view()), Err(AncestryError::Invalid)); + }); + } + + #[test] + fn ancestry_includes_optimistic_local_proposals() { + deterministic::Runner::default().start(|mut context| async move { + let (_, mut state) = setup_state_with( + &mut context, + 4, + 2, + 9, + 10, + TermLength::new(NZU32!(5)), + ViewDelta::new(2), + 4, + ); + let first = propose_and_notarize_view1(&mut state, 21); + let next = state.try_propose().0.expect("optimistic child ready"); + assert_eq!( + state.ancestry(next.parent.0).unwrap().to_vec(), + vec![test_genesis(), first.payload] + ); + let second = fetch_proposal(2, 1, 22); + assert!(state.proposed(second.clone())); + assert!(state.construct_notarize(second.view()).is_some()); + assert!(state.notarization(first.view()).is_none()); + assert!(state.notarization(second.view()).is_none()); + assert_eq!( + state.ancestry(second.view()).unwrap().to_vec(), + vec![test_genesis(), first.payload, second.payload] + ); + }); + } + + #[test] + fn ancestry_certification_without_verification_repairs_missing_prefix() { + deterministic::Runner::default().start(|mut context| async move { + let ( + Fixture { + schemes, verifier, .. + }, + mut state, + ) = setup_state(&mut context, 4, 9, 10, 1, 4); + let parent = fetch_proposal(3, 2, 31); + let candidate = fetch_proposal(4, 3, 32); + for proposal in [&parent, &candidate] { + assert!( + state + .add_notarization(build_notarization(&verifier, &schemes, proposal)) + .0 + ); + } + let (ready, fetches) = state.certify_candidates(); + assert_eq!(fetches.len(), 1); + assert_eq!(fetches[0].view, View::new(2)); + assert!(ready.is_empty()); + assert!(!state.views.get(&candidate.view()).unwrap().is_verified()); + assert_eq!( + state.ancestry(candidate.parent), + Err(AncestryError::Missing(View::new(2))) + ); + let missing = fetch_proposal(2, 0, 30); + assert!( + state + .add_notarization(build_notarization(&verifier, &schemes, &missing)) + .0 + ); + let (ready, fetches) = state.certify_candidates(); + assert!(fetches.is_empty()); + assert!(ready.contains(&candidate)); + assert_eq!( + state.ancestry(candidate.parent).unwrap().to_vec(), + vec![test_genesis(), missing.payload, parent.payload] + ); + }); + } + + #[test] + fn ancestry_uses_certificate_over_local_proposal() { + deterministic::Runner::default().start(|mut context| async move { + let ( + Fixture { + schemes, verifier, .. + }, + mut state, + ) = setup_state_with( + &mut context, + 4, + 2, + 9, + 10, + TermLength::new(NZU32!(5)), + ViewDelta::new(2), + 4, + ); + let local = propose_and_notarize_view1(&mut state, 61); + let selected = fetch_proposal(1, 0, 62); + assert!( + state + .add_notarization(build_notarization(&verifier, &schemes, &selected)) + .0 + ); + assert_eq!( + state.ancestry(local.view()).unwrap().to_vec(), + vec![test_genesis(), selected.payload] + ); + assert!(state.certified(selected.view(), false).is_some()); + assert_eq!( + state.ancestry(selected.view()).unwrap().to_vec(), + vec![test_genesis(), selected.payload] + ); + assert!( + state + .add_finalization(build_finalization(&verifier, &schemes, &selected)) + .0 + ); + assert_eq!( + state.ancestry(selected.view()).unwrap().to_vec(), + vec![selected.payload] + ); + }); + } + + #[test] + fn ancestry_collapses_only_consecutive_reproposals() { + deterministic::Runner::default().start(|mut context| async move { + let ( + Fixture { + schemes, verifier, .. + }, + mut state, + ) = setup_state(&mut context, 4, 9, 10, 1, 4); + let first = fetch_proposal(1, 0, 41); + let repeated = fetch_proposal(2, 1, 41); + let distinct = fetch_proposal(3, 2, 42); + let nonconsecutive = fetch_proposal(4, 3, 41); + for proposal in [&first, &repeated, &distinct, &nonconsecutive] { + assert!( + state + .add_notarization(build_notarization(&verifier, &schemes, proposal)) + .0 + ); + } + assert_eq!( + state.ancestry(repeated.view()).unwrap().to_vec(), + vec![test_genesis(), first.payload] + ); + assert_eq!( + state.ancestry(nonconsecutive.view()).unwrap().to_vec(), + vec![ + test_genesis(), + first.payload, + distinct.payload, + nonconsecutive.payload + ] + ); + }); + } + + #[test] + fn ancestry_stops_at_nonpreceding_certificate_parent() { + deterministic::Runner::default().start(|mut context| async move { + let ( + Fixture { + schemes, verifier, .. + }, + mut state, + ) = setup_state(&mut context, 4, 9, 10, 1, 4); + let malformed = fetch_proposal(1, 1, 51); + assert!( + state + .add_notarization(build_notarization(&verifier, &schemes, &malformed)) + .0 + ); + assert_eq!( + state.ancestry(malformed.view()), + Err(AncestryError::Invalid) + ); + }); + } + + #[test] + fn ancestry_propose_returns_all_missing_metadata_repairs() { + deterministic::Runner::default().start(|mut context| async move { + let ( + Fixture { + schemes, verifier, .. + }, + mut state, + ) = setup_state_with( + &mut context, + 4, + 2, + 9, + 10, + TermLength::new(NZU32!(9)), + ViewDelta::new(8), + 4, + ); + let (initial, fetches) = state.try_propose(); + assert_eq!(initial.unwrap().view(), View::new(1)); + assert!(fetches.is_empty()); + for view in [2, 4, 6] { + let proposal = fetch_proposal(view, view - 1, view as u8); + assert!( + state + .add_notarization(build_notarization(&verifier, &schemes, &proposal)) + .0 + ); + } + + let (selected, fetches) = state.try_propose(); + assert!(selected.is_none()); + assert_eq!( + fetches + .into_iter() + .map(|fetch| (fetch.proposal, fetch.view)) + .collect::>(), + [ + (View::new(3), View::new(1)), + (View::new(5), View::new(3)), + (View::new(7), View::new(5)) + ] + ); + let (selected, fetches) = state.try_propose(); + assert!(selected.is_none()); + assert!(fetches.is_empty()); + for view in [3, 5, 7] { + assert!(state.views.get(&View::new(view)).unwrap().should_propose()); + } + + let parent = fetch_proposal(1, 0, 1); + assert!( + state + .add_notarization(build_notarization(&verifier, &schemes, &parent)) + .0 + ); + let (selected, fetches) = state.try_propose(); + assert_eq!(selected.unwrap().view(), View::new(3)); + assert!(fetches.is_empty()); + assert!(!state.views.get(&View::new(3)).unwrap().should_propose()); + for view in [5, 7] { + assert!(state.views.get(&View::new(view)).unwrap().should_propose()); + } + }); + } + + #[test] + fn ancestry_inputs_wait_before_consuming_proposal_or_verification() { + for signer in [1, 2] { + deterministic::Runner::default().start(|mut context| async move { + let (Fixture { schemes, verifier, .. }, mut state) = setup_state_with( + &mut context, 4, signer, 9, 10, TermLength::new(NZU32!(5)), ViewDelta::new(2), 4, + ); + if signer == 2 { + assert_eq!(state.try_propose().0.unwrap().view(), View::new(1)); + } + let missing = fetch_proposal(1, 0, 71); + let parent = fetch_proposal(2, 1, 72); + assert!(state.add_notarization(build_notarization(&verifier, &schemes, &parent)).0); + let candidate = fetch_proposal(3, 2, 73); + if signer == 2 { + let (context, fetches) = state.try_propose(); + assert!(context.is_none()); + assert!(state.views.get(&candidate.view()).unwrap().should_propose()); + assert!(fetches.iter().any(|fetch| fetch.proposal == candidate.view() && fetch.view == missing.view())); + } else { + assert!(state.set_proposal(candidate.view(), candidate.clone())); + assert!(matches!(state.try_verify(), Verify::Resolve { view, kind: Kind::Notarization, .. } if view == missing.view())); + assert!(state.views.get(&candidate.view()).unwrap().pending_verification().is_some()); + } + assert!(state.add_notarization(build_notarization(&verifier, &schemes, &missing)).0); + assert!(state.certified(missing.view(), true).is_some()); + assert!(state.certified(parent.view(), true).is_some()); + let selected = if signer == 2 { + state.try_propose().0.expect("metadata repair must preserve build request") + } else { + let Verify::Ready(context, _) = state.try_verify() else { + panic!("metadata repair must preserve verification request"); + }; + context + }; + assert_eq!(selected.view(), candidate.view()); + assert_eq!(&*state.ancestry(selected.parent.0).unwrap(), &[test_genesis(), missing.payload, parent.payload]); + }); + } + } + + #[test] + fn ancestry_finalization_wakes_metadata_waiters_with_selected_anchor() { + deterministic::Runner::default().start(|mut context| async move { + let ( + Fixture { + schemes, verifier, .. + }, + mut state, + ) = setup_state_with( + &mut context, + 4, + 1, + 9, + 10, + TermLength::new(NZU32!(5)), + ViewDelta::new(2), + 4, + ); + let parent = fetch_proposal(3, 2, 81); + let candidate = fetch_proposal(6, 3, 82); + for proposal in [&parent, &candidate] { + assert!( + state + .add_notarization(build_notarization(&verifier, &schemes, proposal)) + .0 + ); + } + assert!(state.certify_candidates().0.is_empty()); + assert!( + state + .ancestry_waiters + .get(&View::new(2)) + .unwrap() + .contains(&candidate.view()) + ); + let finalized = fetch_proposal(3, 2, 83); + assert!( + state + .add_finalization(build_finalization(&verifier, &schemes, &finalized)) + .0 + ); + let (ready, fetches) = state.certify_candidates(); + assert_eq!(ready, vec![candidate]); + assert!(fetches.is_empty()); + assert!(state.ancestry_waiters.is_empty()); + assert_eq!( + &*state.ancestry(finalized.view()).unwrap(), + &[finalized.payload] + ); + }); + } + + #[test] + fn ancestry_malformed_optimistic_anchor_does_not_claim_build() { + deterministic::Runner::default().start(|mut context| async move { + let ( + Fixture { + schemes, verifier, .. + }, + mut state, + ) = setup_state_with( + &mut context, + 4, + 2, + 9, + 10, + TermLength::new(NZU32!(5)), + ViewDelta::new(2), + 4, + ); + assert_eq!(state.try_propose().0.unwrap().view(), View::new(1)); + let malformed = fetch_proposal(2, 2, 91); + assert!( + state + .add_notarization(build_notarization(&verifier, &schemes, &malformed)) + .0 + ); + assert!(state.try_propose().0.is_none()); + assert!(state.views.get(&View::new(3)).unwrap().should_propose()); + assert_eq!( + state.ancestry(malformed.view()), + Err(AncestryError::Invalid) + ); + let finalized = fetch_proposal(2, 1, 92); + assert!( + state + .add_finalization(build_finalization(&verifier, &schemes, &finalized)) + .0 + ); + let next = state + .try_propose() + .0 + .expect("selected finalization must release build input"); + assert_eq!(next.parent, (finalized.view(), finalized.payload)); + assert_eq!( + &*state.ancestry(next.parent.0).unwrap(), + &[finalized.payload] + ); + }); + } + + #[test] + fn ancestry_replay_skips_completed_certification_inputs() { + deterministic::Runner::default().start(|mut context| async move { + let ( + Fixture { + schemes, verifier, .. + }, + mut state, + ) = setup_state(&mut context, 4, 9, 10, 1, 4); + let proposal = fetch_proposal(6, 5, 101); + let notarization = build_notarization(&verifier, &schemes, &proposal); + state.replay(&Artifact::Notarization(notarization.clone())); + assert!(state.add_notarization(notarization).0); + state.replay(&Artifact::Certification(proposal.round, true)); + assert!(!state.views.contains_key(&proposal.parent)); + assert!(state.certification_candidates.contains(&proposal.view())); + let (ready, fetches) = state.certify_candidates(); + assert!(ready.is_empty()); + assert!(fetches.is_empty()); + assert!(state.ancestry_waiters.is_empty()); + }); + } + /// An elector that panics if asked to elect a leader without a certificate /// (past view 1). Optimistic successors must inherit the stable leader /// instead of running a fresh election. @@ -3500,8 +4093,11 @@ mod tests { let notarization = build_notarization(&verifier, &schemes, &parent); assert!(state.add_notarization(notarization).0); assert!(state.certified(View::new(2), true).is_some()); + assert!(matches!(state.try_verify(), Verify::Resolve { view, kind: Kind::Notarization, .. } if view == View::new(1))); + let ancestor = fetch_proposal(1, 0, 42); + assert!(state.add_notarization(build_notarization(&verifier, &schemes, &ancestor)).0); let Verify::Ready(ctx, proposal) = state.try_verify() else { - panic!("proposal should verify once the parent certifies"); + panic!("proposal should verify once the selected metadata is complete"); }; assert_eq!(ctx.parent, (View::new(2), parent.payload)); assert_eq!(proposal, child); @@ -3772,8 +4368,9 @@ mod tests { /// Certification exempts term-start candidates from the parent precheck /// (see [`State::certification_parent_ready`]). A term-start proposal - /// dispatches even when its cross-term parent is uncertified and the - /// skipped views' nullifications are not held. + /// remains eligible when its cross-term parent is uncertified and the + /// skipped views' nullifications are not held. Its application input + /// still requires the parent's commitment metadata. #[test] fn certify_candidates_exempts_term_start_from_parent_precheck() { let runtime = deterministic::Runner::default(); @@ -3803,8 +4400,10 @@ mod tests { .0 ); let (ready, fetches) = state.certify_candidates(); - assert_eq!(ready, vec![term_start]); - assert!(fetches.is_empty()); + assert!(state.certification_parent_ready(&term_start).is_ok()); + assert!(ready.is_empty()); + assert_eq!(fetches.len(), 1); + assert_eq!(fetches[0].view, View::new(1)); }); } @@ -4090,6 +4689,147 @@ mod tests { }); } + #[test] + fn certification_repairs_conflicting_optimistic_parent() { + deterministic::Runner::default().start(|mut context| async move { + let ( + Fixture { + schemes, verifier, .. + }, + mut state, + ) = setup_state_with( + &mut context, + 4, + 1, + 9, + 10, + TermLength::new(NZU32!(5)), + ViewDelta::new(2), + 4, + ); + let local = fetch_proposal(1, 0, 130); + assert!(state.set_proposal(local.view(), local.clone())); + assert!(matches!(state.try_verify(), Verify::Ready(..))); + assert!(state.verified(local.view())); + assert!(state.construct_notarize(local.view()).is_some()); + + let quorum = [schemes[0].clone(), schemes[2].clone(), schemes[3].clone()]; + let selected = fetch_proposal(1, 0, 131); + let candidate = fetch_proposal(6, 1, 132); + assert!(state.add_nullification(build_nullification( + &verifier, + &quorum, + Rnd::new(Epoch::new(9), View::new(2)), + ))); + assert!( + state + .add_notarization(build_notarization(&verifier, &quorum, &candidate)) + .0 + ); + assert!(state.certification_parent_ready(&candidate).is_ok()); + + let (ready, fetches) = state.certify_candidates(); + assert!( + ready.is_empty(), + "local A cannot supply notarized C's parent" + ); + assert_eq!(fetches.len(), 1); + assert_eq!(fetches[0].proposal, candidate.view()); + assert_eq!(fetches[0].view, selected.view()); + let (ready, fetches) = state.certify_candidates(); + assert!(ready.is_empty()); + assert!(fetches.is_empty()); + assert_eq!( + state + .views + .get_mut(&candidate.view()) + .unwrap() + .try_certify(), + Some(candidate.clone()), + ); + + assert!( + state + .add_notarization(build_notarization(&verifier, &quorum, &selected)) + .0 + ); + assert!(state.explicit_ancestry_payload(selected.view()).is_none()); + let (ready, fetches) = state.certify_candidates(); + assert!(fetches.is_empty()); + assert!(ready.contains(&candidate)); + assert_eq!( + state.ancestry(candidate.parent).unwrap().as_ref(), + &[test_genesis(), selected.payload], + ); + assert!(state.certified(candidate.view(), true).is_some()); + let (ready, fetches) = state.certify_candidates(); + assert!(ready.is_empty()); + assert!(fetches.is_empty()); + }); + } + + #[test] + fn ancestry_repairs_local_ancestor_below_notarized_parent() { + deterministic::Runner::default().start(|mut context| async move { + let ( + Fixture { + schemes, verifier, .. + }, + mut state, + ) = setup_state_with( + &mut context, + 4, + 1, + 9, + 10, + TermLength::new(NZU32!(5)), + ViewDelta::new(2), + 4, + ); + let local = fetch_proposal(1, 0, 130); + assert!(state.set_proposal(local.view(), local.clone())); + assert!(matches!(state.try_verify(), Verify::Ready(..))); + assert!(state.verified(local.view())); + assert!(state.construct_notarize(local.view()).is_some()); + + // A quorum excluding this validator can select a different proposal + // at view one before its notarization reaches this validator. + let quorum = [schemes[0].clone(), schemes[2].clone(), schemes[3].clone()]; + let selected = fetch_proposal(1, 0, 131); + let parent = fetch_proposal(2, 1, 132); + assert!( + state + .add_notarization(build_notarization(&verifier, &quorum, &parent)) + .0 + ); + let Verify::Ready(_, verifying_parent) = state.try_verify() else { + panic!("notarized parent should start verification"); + }; + assert_eq!(verifying_parent, parent); + let candidate = fetch_proposal(3, 2, 133); + assert!(state.set_proposal(candidate.view(), candidate.clone())); + assert!(matches!( + state.try_verify(), + Verify::Resolve { proposal, view, kind: Kind::Notarization, .. } + if proposal == candidate.view() && view == selected.view() + )); + assert!( + state + .add_notarization(build_notarization(&verifier, &quorum, &selected)) + .0 + ); + let Verify::Ready(context, actual) = state.try_verify() else { + panic!("selected ancestry repair must preserve the verification request"); + }; + assert_eq!(actual, candidate); + assert_eq!(context.parent, (parent.view(), parent.payload)); + assert_eq!( + state.ancestry(parent.view()).unwrap().as_ref(), + &[test_genesis(), selected.payload, parent.payload], + ); + }); + } + #[test] fn pending_optimistic_child_verification_rejects_replaced_parent_notarization() { let runtime = deterministic::Runner::default(); @@ -4415,6 +5155,7 @@ mod tests { let parent = propose_and_notarize_view1(&mut state, 118); let child_context = state .try_propose() + .0 .expect("optimistic child proposal should start"); assert_eq!(child_context.view(), View::new(2)); assert_eq!(child_context.parent, (View::new(1), parent.payload)); @@ -5603,9 +6344,19 @@ mod tests { }, ); state.set_genesis(test_genesis()); + let parent = Proposal::new( + Rnd::new(epoch, View::new(1)), + GENESIS_VIEW, + Sha256Digest::from([41; 32]), + ); + assert!( + state + .add_finalization(build_finalization(&verifier, &schemes, &parent)) + .0 + ); // Enter the view where we are the leader. - assert!(state.enter_view(view)); + assert_eq!(state.current_view(), view); state.set_leader(view, None); assert_eq!(state.leader_index(view), Some(Participant::new(0))); @@ -5747,7 +6498,17 @@ mod tests { }, ); state.set_genesis(test_genesis()); - assert!(state.enter_view(view)); + let parent = Proposal::new( + Rnd::new(epoch, View::new(1)), + GENESIS_VIEW, + Sha256Digest::from([42; 32]), + ); + assert!( + state + .add_finalization(build_finalization(&verifier, &schemes, &parent)) + .0 + ); + assert_eq!(state.current_view(), view); state.set_leader(view, None); assert_eq!(state.leader_index(view), Some(Participant::new(0))); @@ -6144,25 +6905,19 @@ mod tests { let mut state = State::new(context, cfg); state.set_genesis(test_genesis()); - // Helper to create notarization for a view - let make_notarization = |view: View| { - let proposal = Proposal::new( - Rnd::new(Epoch::new(1), view), - GENESIS_VIEW, - Sha256Digest::from([view.get() as u8; 32]), - ); - build_notarization(&verifier, &schemes, &proposal) - }; - - // Helper to create finalization for a view - let make_finalization = |view: View| { - let proposal = Proposal::new( + let proposal = |view: View| { + Proposal::new( Rnd::new(Epoch::new(1), view), - GENESIS_VIEW, + if view == View::new(3) { + GENESIS_VIEW + } else { + view.previous().unwrap() + }, Sha256Digest::from([view.get() as u8; 32]), - ); - build_finalization(&verifier, &schemes, &proposal) + ) }; + let make_notarization = |view| build_notarization(&verifier, &schemes, &proposal(view)); + let make_finalization = |view| build_finalization(&verifier, &schemes, &proposal(view)); let mut pool = AbortablePool::<()>::default(); @@ -6259,10 +7014,10 @@ mod tests { let mut state = State::new(context, cfg); state.set_genesis(test_genesis()); - let make_notarization = |view: View| { + let make_notarization = |view: View, parent: View| { let proposal = Proposal::new( Rnd::new(Epoch::new(1), view), - GENESIS_VIEW, + parent, Sha256Digest::from([view.get() as u8; 32]), ); build_notarization(&verifier, &schemes, &proposal) @@ -6280,8 +7035,8 @@ mod tests { let stale_view = View::new(2); let live_view = View::new(3); - state.add_notarization(make_notarization(stale_view)); - state.add_notarization(make_notarization(live_view)); + state.add_notarization(make_notarization(stale_view, GENESIS_VIEW)); + state.add_notarization(make_notarization(live_view, stale_view)); state.add_finalization(make_finalization(stale_view)); // Reinsert a stale candidate to exercise the defensive finalized-view guard. @@ -6529,7 +7284,7 @@ mod tests { // Before late certification arrives, we cannot build a child because parent ancestry // is still incomplete for this node. - assert!(state.try_propose().is_none()); + assert!(state.try_propose().0.is_none()); // Late certification after nullification is still recorded. assert!(state.certified(parent_view, true).is_some()); @@ -6537,6 +7292,7 @@ mod tests { // Child proposal selection should build on the now-certified parent view. let propose_context = state .try_propose() + .0 .expect("child view should be able to build on certified parent"); assert_eq!(propose_context.round.view(), child_view); assert_eq!(propose_context.parent, (parent_view, payload)); @@ -6655,7 +7411,7 @@ mod tests { assert!(state.enter_view(View::new(3))); state.set_leader(View::new(3), None); assert_eq!(state.leader_index(View::new(3)), Some(Participant::new(2))); - assert!(state.try_propose().is_none()); + assert!(state.try_propose().0.is_none()); }); } @@ -6710,6 +7466,7 @@ mod tests { let proposal = state .try_propose() + .0 .expect("term-start proposal should use prior-term certified parent"); assert_eq!(proposal.round.view(), View::new(6)); assert_eq!(proposal.parent, (parent_view, parent_payload)); @@ -6786,6 +7543,7 @@ mod tests { assert_eq!(state.leader_index(View::new(6)), Some(Participant::new(3))); let proposal = state .try_propose() + .0 .expect("term-start proposal should skip the blocked chain"); assert_eq!(proposal.parent, (View::new(1), payload_v1)); diff --git a/consensus/src/simplex/mocks/application.rs b/consensus/src/simplex/mocks/application.rs index ee25edfd1fd..ee59cf589e8 100644 --- a/consensus/src/simplex/mocks/application.rs +++ b/consensus/src/simplex/mocks/application.rs @@ -40,6 +40,7 @@ pub enum Message { Certify { round: Round, payload: D, + ancestry: Arc<[D]>, response: oneshot::Sender, }, Broadcast { @@ -63,7 +64,11 @@ impl Au for Mailbox { type Digest = D; type Context = Context; - async fn propose(&mut self, context: Self::Context) -> oneshot::Receiver { + async fn propose( + &mut self, + context: Self::Context, + _ancestry: Arc<[Self::Digest]>, + ) -> oneshot::Receiver { let (response, receiver) = oneshot::channel(); self.sender .send_lossy(Message::Propose { context, response }); @@ -74,6 +79,7 @@ impl Au for Mailbox { &mut self, context: Self::Context, payload: Self::Digest, + _ancestry: Arc<[Self::Digest]>, ) -> oneshot::Receiver { let (response, receiver) = oneshot::channel(); self.sender.send_lossy(Message::Verify { @@ -86,11 +92,17 @@ impl Au for Mailbox { } impl CAu for Mailbox { - async fn certify(&mut self, round: Round, payload: Self::Digest) -> oneshot::Receiver { + async fn certify( + &mut self, + round: Round, + payload: Self::Digest, + ancestry: Arc<[Self::Digest]>, + ) -> oneshot::Receiver { let (tx, rx) = oneshot::channel(); self.sender.send_lossy(Message::Certify { round, payload, + ancestry, response: tx, }); rx @@ -128,6 +140,8 @@ type ProposeObserver = Box::Digest, P>) + Sen type VerifyObserver = Box::Digest, P>, ::Digest) + Send + 'static>; +type AncestryCertifier = Box) -> bool + Send + 'static>; + /// Predicate to determine whether a payload should be certified. /// Returning true means certify, false means reject. pub enum Certifier { @@ -135,6 +149,8 @@ pub enum Certifier { Always, /// A custom predicate function that receives the round and payload digest. Custom(Box bool + Send + 'static>), + /// A custom predicate that also receives the selected parent commitments. + WithAncestry(AncestryCertifier), /// Drop the sender without responding, causing the receiver to be cancelled. /// This simulates scenarios where the automaton cannot determine certification /// (e.g., missing verification context in Marshaled). @@ -365,6 +381,7 @@ impl Application { round: Round, payload: H::Digest, _contents: Bytes, + ancestry: Arc<[H::Digest]>, ) -> Option { // Simulate the certify latency let duration = self.certify_latency.sample(self.context.as_mut()); @@ -376,6 +393,7 @@ impl Application { match &self.should_certify { Certifier::Always => Some(true), Certifier::Custom(func) => Some(func(round, payload)), + Certifier::WithAncestry(func) => Some(func(round, payload, ancestry)), Certifier::Cancel | Certifier::Pending => None, } } @@ -464,10 +482,11 @@ impl Application { Message::Certify { round, payload, + ancestry, response, } => { let contents = self.seen.get(&payload).cloned().unwrap_or_default(); - if let Some(certified) = self.certify(round, payload, contents).await { + if let Some(certified) = self.certify(round, payload, contents, ancestry).await { response.send_lossy(certified); } else if matches!(self.should_certify, Certifier::Pending) { // Hold the sender alive so the receiver never resolves. diff --git a/examples/bridge/src/application/ingress.rs b/examples/bridge/src/application/ingress.rs index c7e2752e22f..72842d4b40d 100644 --- a/examples/bridge/src/application/ingress.rs +++ b/examples/bridge/src/application/ingress.rs @@ -13,7 +13,7 @@ use commonware_consensus::{ }; use commonware_cryptography::{Digest, ed25519::PublicKey}; use commonware_utils::channel::oneshot; -use std::collections::VecDeque; +use std::{collections::VecDeque, sync::Arc}; #[allow(clippy::large_enum_variant)] pub enum Message { @@ -57,6 +57,7 @@ impl Au for Mailbox { async fn propose( &mut self, context: Context, + _: Arc<[Self::Digest]>, ) -> oneshot::Receiver { // If we linked payloads to their parent, we would include // the parent in the `Context` in the payload. @@ -77,6 +78,7 @@ impl Au for Mailbox { &mut self, _: Context, payload: Self::Digest, + _: Arc<[Self::Digest]>, ) -> oneshot::Receiver { // If we linked payloads to their parent, we would verify // the parent included in the payload matches the provided `Context`. diff --git a/examples/log/src/application/ingress.rs b/examples/log/src/application/ingress.rs index a0e989db48c..d3854659e82 100644 --- a/examples/log/src/application/ingress.rs +++ b/examples/log/src/application/ingress.rs @@ -8,7 +8,7 @@ use commonware_consensus::{ }; use commonware_cryptography::{Digest, ed25519::PublicKey}; use commonware_utils::channel::oneshot; -use std::collections::VecDeque; +use std::{collections::VecDeque, sync::Arc}; pub enum Message { Propose { response: oneshot::Sender }, @@ -42,6 +42,7 @@ impl Au for Mailbox { async fn propose( &mut self, _: Context, + _: Arc<[Self::Digest]>, ) -> oneshot::Receiver { // If we linked payloads to their parent, we would include // the parent in the `Context` in the payload. @@ -59,6 +60,7 @@ impl Au for Mailbox { &mut self, _: Context, _: Self::Digest, + _: Arc<[Self::Digest]>, ) -> oneshot::Receiver { // Digests are already verified by consensus, so we don't need to check they are valid. // diff --git a/glue/src/dkg/tests/mocks.rs b/glue/src/dkg/tests/mocks.rs index 0a57e176214..2c8fa01b560 100644 --- a/glue/src/dkg/tests/mocks.rs +++ b/glue/src/dkg/tests/mocks.rs @@ -527,7 +527,11 @@ impl Automaton for MockApplication { type Context = TestContext; type Digest = TestDigest; - async fn propose(&mut self, _context: Self::Context) -> oneshot::Receiver { + async fn propose( + &mut self, + _context: Self::Context, + _ancestry: Arc<[Self::Digest]>, + ) -> oneshot::Receiver { let (sender, receiver) = oneshot::channel(); self.proposals.lock().push(_context); sender.send_lossy(Sha256::hash(&[b"proposal"])); @@ -538,6 +542,7 @@ impl Automaton for MockApplication { &mut self, _context: Self::Context, _payload: Self::Digest, + _ancestry: Arc<[Self::Digest]>, ) -> oneshot::Receiver { let (sender, receiver) = oneshot::channel(); sender.send_lossy(true); diff --git a/glue/src/stateful/tests/mod.rs b/glue/src/stateful/tests/mod.rs index 43f21221ca7..14275702432 100644 --- a/glue/src/stateful/tests/mod.rs +++ b/glue/src/stateful/tests/mod.rs @@ -1168,6 +1168,7 @@ async fn build_multi_chain( fn out_of_order_certifications_complete_on_qmdb() { deterministic::Runner::timed(Duration::from_secs(10)).start(|context| async move { let (genesis, blocks) = build_chain(&context, 6).await; + let ancestor_commitments = blocks.iter().map(|block| block.parent).collect::>(); let page_cache = CacheRef::from_pooler(&context, PAGE_SIZE, PAGE_CACHE_SIZE); let mut signing_context = context.child("signing"); let fixture = scheme_mocks::fixture( @@ -1261,7 +1262,15 @@ fn out_of_order_certifications_complete_on_qmdb() { let mut certifications = Vec::with_capacity(blocks.len()); for index in [5, 1, 4, 0, 3, 2] { let block = &blocks[index]; - certifications.push(deferred.certify(block.context.round, block.digest()).await); + certifications.push( + deferred + .certify( + block.context.round, + block.digest(), + ancestor_commitments[..=index].into(), + ) + .await, + ); } select! { @@ -1486,6 +1495,7 @@ fn stable_leader_finalizations_outpace_slow_qmdb_sync() { fn overlapping_finalizations_complete_on_multi_qmdb() { deterministic::Runner::timed(Duration::from_secs(10)).start(|context| async move { let (genesis, blocks) = build_multi_chain(&context, 6).await; + let ancestor_commitments = blocks.iter().map(|block| block.parent).collect::>(); let page_cache = CacheRef::from_pooler(&context, PAGE_SIZE, PAGE_CACHE_SIZE); let mut signing_context = context.child("signing"); let fixture = scheme_mocks::fixture( @@ -1591,8 +1601,14 @@ fn overlapping_finalizations_complete_on_multi_qmdb() { // Cache the batches that will be finalized so the held descendant // verification does not own their replay. - for block in &blocks[..3] { - let certification = deferred.certify(block.context.round, block.digest()).await; + for (index, block) in blocks[..3].iter().enumerate() { + let certification = deferred + .certify( + block.context.round, + block.digest(), + ancestor_commitments[..=index].into(), + ) + .await; assert!( certification .await @@ -1619,7 +1635,13 @@ fn overlapping_finalizations_complete_on_multi_qmdb() { let block = &blocks[index]; certifications.push(( index, - deferred.certify(block.context.round, block.digest()).await, + deferred + .certify( + block.context.round, + block.digest(), + ancestor_commitments[..=index].into(), + ) + .await, )); } for started in verify_started { @@ -1740,6 +1762,7 @@ fn overlapping_finalizations_complete_on_multi_qmdb() { fn pruning_quiesces_and_retries_verification_on_real_qmdbs() { deterministic::Runner::timed(Duration::from_secs(10)).start(|context| async move { let (genesis, blocks) = build_multi_chain(&context, 5).await; + let ancestor_commitments = blocks.iter().map(|block| block.parent).collect::>(); let page_cache = CacheRef::from_pooler(&context, PAGE_SIZE, PAGE_CACHE_SIZE); let mut signing_context = context.child("signing"); let fixture = scheme_mocks::fixture( @@ -1851,8 +1874,14 @@ fn pruning_quiesces_and_retries_verification_on_real_qmdbs() { // Keep the first four batches available so block 5 reaches application // verification without owning ancestor replay. - for block in &blocks[..4] { - let certification = deferred.certify(block.context.round, block.digest()).await; + for (index, block) in blocks[..4].iter().enumerate() { + let certification = deferred + .certify( + block.context.round, + block.digest(), + ancestor_commitments[..=index].into(), + ) + .await; assert!( certification .await @@ -1897,7 +1926,13 @@ fn pruning_quiesces_and_retries_verification_on_real_qmdbs() { ); let block = &blocks[4]; - let mut certification = reporter.certify(block.context.round, block.digest()).await; + let mut certification = reporter + .certify( + block.context.round, + block.digest(), + ancestor_commitments[..=4].into(), + ) + .await; first_started .await .expect("verification should start before pruning");