use crate::{
Application, Automaton, Block, CertifiableAutomaton, Epochable, Handoff, Heightable, Relay,
Reporter,
marshal::{
ancestry::Ancestry,
application::{
build::{self, Resolved},
gates::{GateOutcome, Gates},
relay,
},
core::{CommitmentFallback, DigestFallback, Mailbox},
standard::{
Standard,
validation::{
Decision, ParentCheck, await_and_validate_parent, precheck_epoch_and_reproposal,
run_app_verify,
},
},
},
simplex::{
Plan,
types::{Activity, Context},
},
types::{Epocher, Round, View},
};
use commonware_actor::Feedback;
use commonware_codec::Codec;
use commonware_cryptography::{Digested, Digestible, certificate::Scheme};
use commonware_macros::select;
use commonware_runtime::{
Clock, Metrics, Spawner,
telemetry::{
metrics::{
MetricsExt as _,
histogram::{Buckets, Timed},
},
traces::TracedExt as _,
},
};
use commonware_utils::channel::{fallible::OneshotExt, oneshot};
use std::{
future::{Future, ready},
sync::Arc,
};
use tracing::{Instrument as _, debug, info_span};
async fn await_block_subscription<T, D>(
tx: &mut oneshot::Sender<bool>,
block_rx: oneshot::Receiver<T>,
digest: &D,
stage: &'static str,
) -> Option<T>
where
D: std::fmt::Debug,
{
select! {
_ = tx.closed() => {
debug!(
stage,
reason = "consensus dropped receiver",
"skipping block wait"
);
None
},
result = block_rx => {
if result.is_err() {
debug!(
stage,
?digest,
reason = "failed to fetch block",
"skipping block wait"
);
}
result.ok()
},
}
}
pub struct Inline<E, S, A, B, ES>
where
S: Scheme,
A: Application<E>,
B: Block + Codec,
ES: Epocher,
{
context: Arc<E>,
application: A,
marshal: Mailbox<S, Standard<B>>,
epocher: ES,
gates: Gates<B::Digest, Digested<B>>,
build_duration: Timed,
proposal_parent_fetch_duration: Timed,
ancestor_fetch_duration: Timed,
}
impl<E, S, A, B, ES> Clone for Inline<E, S, A, B, ES>
where
S: Scheme,
A: Application<E>,
B: Block + Codec,
ES: Epocher,
{
fn clone(&self) -> Self {
Self {
context: self.context.clone(),
application: self.application.clone(),
marshal: self.marshal.clone(),
epocher: self.epocher.clone(),
gates: self.gates.clone(),
build_duration: self.build_duration.clone(),
proposal_parent_fetch_duration: self.proposal_parent_fetch_duration.clone(),
ancestor_fetch_duration: self.ancestor_fetch_duration.clone(),
}
}
}
impl<E, S, A, B, ES> Inline<E, S, A, B, ES>
where
E: Metrics + Clock,
S: Scheme,
A: Application<
E,
Block = B,
SigningScheme = S,
Context = Context<B::Digest, S::PublicKey>,
Input = (),
>,
B: Block + Codec,
ES: Epocher,
{
pub fn new(context: E, application: A, marshal: Mailbox<S, Standard<B>>, epocher: ES) -> Self {
let build_histogram = context.histogram(
"build_duration",
"Histogram of time taken for the application to build a new block, in seconds",
Buckets::LOCAL,
);
let build_duration = Timed::new(build_histogram);
let parent_fetch_histogram = context.histogram(
"parent_fetch_duration",
"Histogram of time taken to fetch a parent block in proposal, in seconds",
Buckets::LOCAL,
);
let proposal_parent_fetch_duration = Timed::new(parent_fetch_histogram);
let ancestor_fetch_histogram = context.histogram(
"ancestor_fetch_duration",
"Histogram of time taken to fetch a block via the ancestry stream, in seconds",
Buckets::LOCAL,
);
let ancestor_fetch_duration = Timed::new(ancestor_fetch_histogram);
Self {
context: Arc::new(context),
application,
marshal,
epocher,
gates: Gates::new(),
build_duration,
proposal_parent_fetch_duration,
ancestor_fetch_duration,
}
}
fn checks(
&self,
consensus_context: Context<B::Digest, S::PublicKey>,
) -> impl Future<Output = Resolved<B::Digest, Digested<B>, impl Ancestry<B>>> + Send + 'static
{
let context = self.context.clone();
let marshal = self.marshal.clone();
let epocher = self.epocher.clone();
let proposal_parent_fetch_duration = self.proposal_parent_fetch_duration.clone();
let ancestor_fetch_duration = self.ancestor_fetch_duration.clone();
let build_duration = self.build_duration.clone();
async move {
let (parent_view, parent_commitment) = consensus_context.parent;
let parent_request = marshal.subscribe_by_commitment(
parent_commitment,
CommitmentFallback::FetchByRound {
round: Round::new(consensus_context.epoch(), parent_view),
},
);
let parent_timer = proposal_parent_fetch_duration.timer(&*context);
let Ok(parent) = parent_request.await else {
debug!(
?parent_commitment,
reason = "failed to fetch parent block",
"skipping proposal"
);
return Resolved::Skip;
};
parent_timer.observe(&*context);
let last_in_epoch = epocher
.last(consensus_context.epoch())
.expect("current epoch should exist");
if parent.height() == last_in_epoch {
return Resolved::Reuse(parent.digest(), parent);
}
let ancestor_stream =
marshal.ancestor_stream(context.clone(), [parent], ancestor_fetch_duration);
Resolved::Build(ancestor_stream, build_duration.timer(&*context), ())
}
}
}
impl<E, S, A, B, ES> Automaton for Inline<E, S, A, B, ES>
where
E: Spawner + Metrics + Clock,
S: Scheme,
A: Application<
E,
Block = B,
SigningScheme = S,
Context = Context<B::Digest, S::PublicKey>,
Input = (),
>,
B: Block + Codec,
ES: Epocher,
{
type Digest = B::Digest;
type Context = Context<Self::Digest, S::PublicKey>;
#[allow(clippy::async_yields_async)]
#[tracing::instrument(name = "marshal.inline.propose", level = "info", skip_all, fields(round = %consensus_context.round))]
async fn propose(
&mut self,
consensus_context: Context<Self::Digest, S::PublicKey>,
) -> oneshot::Receiver<Self::Digest> {
let checks = self.checks(consensus_context.clone());
let span = info_span!(
"marshal.inline.propose.task",
round = %consensus_context.round
);
let round = consensus_context.round;
let (parent_view, parent_commitment) = consensus_context.parent;
build::propose(
self.context.as_ref(),
&self.application,
self.gates.clone(),
consensus_context,
checks,
span,
move || {
info_span!(
"marshal.inline.application.propose",
round = %round,
parent_view = parent_view.traced(),
parent = %parent_commitment
)
},
|block, ()| ready((block.digest(), block)),
)
}
#[allow(clippy::async_yields_async)]
#[tracing::instrument(name = "marshal.inline.verify", level = "info", skip_all, fields(round = %context.round, digest = %digest))]
async fn verify(
&mut self,
context: Context<Self::Digest, S::PublicKey>,
digest: Self::Digest,
) -> oneshot::Receiver<bool> {
let round = context.round;
let block_request = self
.marshal
.subscribe_by_digest(digest, DigestFallback::Wait);
let (durable_tx, durable_rx) = oneshot::channel();
self.gates.insert(round, digest, durable_rx);
let marshal = self.marshal.clone();
let mut application = self.application.clone();
let epocher = self.epocher.clone();
let ancestor_fetch_duration = self.ancestor_fetch_duration.clone();
let (mut tx, rx) = oneshot::channel();
let runtime_context = self
.context
.child("inline_verify")
.with_attribute("round", round);
let span = info_span!(
"marshal.inline.verify.task",
round = %round,
digest = %digest
);
runtime_context.spawn(move |runtime_context| {
async move {
let parent_request = (digest != context.parent.1).then(|| {
let (parent_view, parent_commitment) = context.parent;
marshal.subscribe_by_commitment(
parent_commitment,
CommitmentFallback::FetchByRound {
round: Round::new(context.epoch(), parent_view),
},
)
});
let Some(block) =
await_block_subscription(&mut tx, block_request, &digest, "verification").await
else {
return;
};
let Some(decision) =
precheck_epoch_and_reproposal(&epocher, &marshal, &context, digest, block)
.await
else {
return;
};
let block = match decision {
Decision::Complete(valid) => {
tx.send_lossy(valid);
durable_tx.send_lossy(if valid {
GateOutcome::Ready(true)
} else {
GateOutcome::Recover
});
return;
}
Decision::Continue(block) => block,
};
let parent_request =
parent_request.expect("non-reproposal has a parent subscription");
let store = async {
if marshal.verified(round, block.clone()).await {
durable_tx.send_lossy(GateOutcome::Ready(true));
}
};
let verify_then_vote = async {
let parent = match await_and_validate_parent(
context.parent.1,
&block,
parent_request,
&mut tx,
)
.await
{
Some(ParentCheck::Valid(parent)) => parent,
Some(ParentCheck::Invalid) => {
tx.send_lossy(false);
return Some(false);
}
None => return None,
};
let valid = run_app_verify(
runtime_context,
context,
block.clone(),
parent,
&mut application,
&marshal,
&mut tx,
ancestor_fetch_duration,
)
.await;
if let Some(valid) = valid {
tx.send_lossy(valid);
}
valid
};
futures::join!(verify_then_vote, store);
}
.instrument(span)
});
rx
}
}
impl<E, S, A, B, ES> CertifiableAutomaton for Inline<E, S, A, B, ES>
where
E: Spawner + Metrics + Clock,
S: Scheme,
A: Application<
E,
Block = B,
SigningScheme = S,
Context = Context<B::Digest, S::PublicKey>,
Input = (),
>,
B: Block + Codec,
ES: Epocher,
{
#[allow(clippy::async_yields_async)]
#[tracing::instrument(name = "marshal.inline.prepare", level = "info", skip_all, fields(round = %consensus_context.round))]
async fn prepare(
&mut self,
consensus_context: Context<Self::Digest, S::PublicKey>,
) -> oneshot::Receiver<Handoff<Self::Digest>> {
let checks = self.checks(consensus_context.clone());
let span = info_span!(
"marshal.inline.prepare.task",
round = %consensus_context.round
);
let round = consensus_context.round;
let (parent_view, parent_commitment) = consensus_context.parent;
build::prepare(
self.context.as_ref(),
&self.application,
self.gates.clone(),
consensus_context,
checks,
span,
move || {
info_span!(
"marshal.inline.application.prepare",
round = %round,
parent_view = parent_view.traced(),
parent = %parent_commitment
)
},
|block, ()| ready((block.digest(), block)),
)
}
#[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<bool> {
let task = self.gates.claim(round, digest, |block, ack| {
self.marshal.verified_deferred(round, block, ack)
});
if task.is_some() {
self.marshal.hint_notarized(round, digest);
}
let marshal = self.marshal.clone();
let (mut tx, rx) = oneshot::channel();
let context = self
.context
.child("inline_certify")
.with_attribute("round", round);
context.spawn(move |_| {
async move {
if let Some(task) = task {
let result = select! {
_ = tx.closed() => {
debug!(
reason = "consensus dropped receiver",
"skipping certification"
);
return;
},
result = task => result,
};
match result {
Ok(GateOutcome::Ready(verdict)) => {
tx.send_lossy(verdict);
return;
}
Ok(GateOutcome::Recover) | Err(_) => {}
}
}
let block_rx =
marshal.subscribe_by_digest(digest, DigestFallback::FetchByRound { round });
let Some(block) =
await_block_subscription(&mut tx, block_rx, &digest, "certification").await
else {
return;
};
if !marshal.certified(round, block).await {
return;
}
tx.send_lossy(true);
}
.instrument(info_span!(
"marshal.inline.certify.task",
round = %round,
digest = %digest
))
});
rx
}
}
impl<E, S, A, B, ES> Relay for Inline<E, S, A, B, ES>
where
E: Send + Sync + 'static,
S: Scheme,
A: Application<E, Block = B, Context = Context<B::Digest, S::PublicKey>>,
B: Block + Codec,
ES: Epocher,
{
type Digest = B::Digest;
type PublicKey = S::PublicKey;
type Plan = Plan<S::PublicKey>;
fn broadcast(&mut self, commitment: Self::Digest, plan: Plan<S::PublicKey>) -> Feedback {
relay::broadcast(&self.gates, &self.marshal, commitment, plan)
}
}
impl<E, S, A, B, ES> Reporter for Inline<E, S, A, B, ES>
where
E: Send + Sync + 'static,
S: Scheme,
A: Application<E, Block = B, Context = Context<B::Digest, S::PublicKey>>,
B: Block + Codec,
ES: Epocher,
{
type Activity = Activity<S, B::Digest>;
fn report(&mut self, activity: Self::Activity) -> Feedback {
let retired = match &activity {
Activity::Finalization(finalization) => finalization.round(),
activity => Round::new(activity.epoch(), View::zero()),
};
self.gates.retain_after(retired);
self.marshal.report(activity)
}
}
#[cfg(test)]
mod tests {
use super::Inline;
use crate::{
Application, Automaton, Block, CertifiableAutomaton, Relay,
marshal::{
application::gates::Gates,
mocks::{
harness::{
B, BLOCKS_PER_EPOCH, Ctx, D, NAMESPACE, NUM_VALIDATORS, S, StandardHarness,
TestHarness, V, default_leader, make_raw_block,
setup_network_with_participants,
},
verifying::{GatedVerifyingApp, MockVerifyingApp},
},
standard::tests::ready,
},
simplex::{scheme::bls12381_threshold::vrf as bls12381_threshold_vrf, types::Context},
types::{Epoch, FixedEpocher, Height, Round, View},
};
use commonware_broadcast::Broadcaster;
use commonware_codec::Codec;
use commonware_cryptography::{
Digested, Digestible, Hasher as _,
certificate::{ConstantProvider, Scheme, mocks::Fixture},
sha256::Sha256,
};
use commonware_macros::{select, test_traced};
use commonware_runtime::{Clock, Metrics, Runner, Spawner, Supervisor as _, deterministic};
use commonware_utils::{NZUsize, channel::fallible::OneshotExt};
use std::time::Duration;
#[allow(dead_code)]
fn assert_non_certifiable_block_supported<E, S, A, B, ES>()
where
E: Spawner + Metrics + Clock,
S: Scheme,
A: Application<
E,
Block = B,
SigningScheme = S,
Context = Context<B::Digest, S::PublicKey>,
Input = (),
>,
B: Block + Codec,
ES: crate::types::Epocher,
{
fn assert_automaton<T: Automaton>() {}
fn assert_certifiable<T: CertifiableAutomaton>() {}
fn assert_relay<T: Relay>() {}
assert_automaton::<Inline<E, S, A, B, ES>>();
assert_certifiable::<Inline<E, S, A, B, ES>>();
assert_relay::<Inline<E, S, A, B, ES>>();
}
#[test_traced("INFO")]
fn test_certify_returns_immediately_after_verify_fetches_block() {
let runner = deterministic::Runner::timed(Duration::from_secs(30));
runner.start(|mut context| async move {
let Fixture {
participants,
schemes,
..
} = bls12381_threshold_vrf::fixture::<V, _>(&mut context, NAMESPACE, NUM_VALIDATORS);
let mut oracle = setup_network_with_participants(
context.child("network"),
NZUsize!(1),
participants.clone(),
1024 * 1024,
)
.await;
let me = participants[0].clone();
let setup = StandardHarness::setup_validator(
context.child("validator").with_attribute("index", 0),
&mut oracle,
me.clone(),
ConstantProvider::new(schemes[0].clone()),
)
.await;
let marshal = setup.mailbox;
let genesis = make_raw_block(Sha256::hash(&[b""]), Height::zero(), 0);
let mock_app: MockVerifyingApp<B, S> = MockVerifyingApp::new();
let mut inline = Inline::new(
context.child("inline"),
mock_app,
marshal.clone(),
FixedEpocher::new(BLOCKS_PER_EPOCH),
);
let parent_round = Round::new(Epoch::zero(), View::new(1));
let parent_ctx = Ctx {
round: parent_round,
leader: default_leader(),
parent: (View::zero(), genesis.digest()),
};
let parent = B::new::<Sha256>(parent_ctx, genesis.digest(), Height::new(1), 100);
let parent_digest = parent.digest();
assert!(marshal.verified(parent_round, ready(parent)).await);
let round = Round::new(Epoch::zero(), View::new(2));
let verify_context = Ctx {
round,
leader: me,
parent: (View::new(1), parent_digest),
};
let block =
B::new::<Sha256>(verify_context.clone(), parent_digest, Height::new(2), 200);
let digest = block.digest();
assert!(marshal.verified(round, ready(block)).await);
let verify_rx = inline.verify(verify_context, digest).await;
assert!(
verify_rx.await.unwrap(),
"verify should complete successfully before certify"
);
let certify_rx = inline.certify(round, digest).await;
select! {
result = certify_rx => {
assert!(
result.unwrap(),
"certify should return immediately once verify has fetched the block"
);
},
_ = context.sleep(Duration::from_secs(5)) => {
panic!("certify should not hang after local verify completed");
},
}
});
}
#[test_traced("INFO")]
fn test_certify_succeeds_without_verify_task() {
let runner = deterministic::Runner::timed(Duration::from_secs(30));
runner.start(|mut context| async move {
let Fixture {
participants,
schemes,
..
} = bls12381_threshold_vrf::fixture::<V, _>(&mut context, NAMESPACE, NUM_VALIDATORS);
let mut oracle = setup_network_with_participants(
context.child("network"),
NZUsize!(1),
participants.clone(),
1024 * 1024,
)
.await;
let me = participants[0].clone();
let setup = StandardHarness::setup_validator(
context.child("validator").with_attribute("index", 0),
&mut oracle,
me.clone(),
ConstantProvider::new(schemes[0].clone()),
)
.await;
let marshal = setup.mailbox;
let genesis = make_raw_block(Sha256::hash(&[b""]), Height::zero(), 0);
let mock_app: MockVerifyingApp<B, S> = MockVerifyingApp::new();
let mut inline = Inline::new(
context.child("inline"),
mock_app,
marshal.clone(),
FixedEpocher::new(BLOCKS_PER_EPOCH),
);
let parent_round = Round::new(Epoch::zero(), View::new(1));
let parent_ctx = Ctx {
round: parent_round,
leader: default_leader(),
parent: (View::zero(), genesis.digest()),
};
let parent = B::new::<Sha256>(parent_ctx, genesis.digest(), Height::new(1), 100);
let parent_digest = parent.digest();
assert!(marshal.verified(parent_round, ready(parent)).await);
let round = Round::new(Epoch::zero(), View::new(2));
let verify_context = Ctx {
round,
leader: me,
parent: (View::new(1), parent_digest),
};
let block =
B::new::<Sha256>(verify_context.clone(), parent_digest, Height::new(2), 200);
let digest = block.digest();
assert!(marshal.verified(round, ready(block)).await);
let certify_rx = inline.certify(round, digest).await;
select! {
result = certify_rx => {
assert!(
result.unwrap(),
"certify should resolve once block availability is known"
);
},
_ = context.sleep(Duration::from_secs(5)) => {
panic!("certify should not hang when block is already available in marshal");
},
}
});
}
#[test_traced("INFO")]
fn test_certify_reproposal_uses_available_blocks_after_verify() {
let runner = deterministic::Runner::timed(Duration::from_secs(30));
runner.start(|mut context| async move {
let Fixture {
participants,
schemes,
..
} = bls12381_threshold_vrf::fixture::<V, _>(&mut context, NAMESPACE, NUM_VALIDATORS);
let mut oracle = setup_network_with_participants(
context.child("network"),
NZUsize!(1),
participants.clone(),
1024 * 1024,
)
.await;
let me = participants[0].clone();
let setup = StandardHarness::setup_validator(
context.child("validator").with_attribute("index", 0),
&mut oracle,
me.clone(),
ConstantProvider::new(schemes[0].clone()),
)
.await;
let marshal = setup.mailbox;
let marshal_actor_handle = setup.actor_handle;
let genesis = make_raw_block(Sha256::hash(&[b""]), Height::zero(), 0);
let mock_app: MockVerifyingApp<B, S> = MockVerifyingApp::new();
let mut inline = Inline::new(context.child("inline"),
mock_app,
marshal.clone(),
FixedEpocher::new(BLOCKS_PER_EPOCH),
);
let boundary_height = Height::new(BLOCKS_PER_EPOCH.get() - 1);
let boundary_round = Round::new(Epoch::zero(), View::new(boundary_height.get()));
let boundary_block = B::new::<Sha256>(
Ctx {
round: boundary_round,
leader: default_leader(),
parent: (View::zero(), genesis.digest()),
},
genesis.digest(),
boundary_height,
1900,
);
let boundary_digest = boundary_block.digest();
assert!(
marshal.verified(boundary_round, ready(boundary_block)).await
);
let reproposal_round = Round::new(Epoch::zero(), View::new(boundary_height.get() + 1));
let reproposal_context = Ctx {
round: reproposal_round,
leader: me,
parent: (View::new(boundary_height.get()), boundary_digest),
};
let verify_rx = inline.verify(reproposal_context, boundary_digest).await;
assert!(
verify_rx.await.unwrap(),
"verify should accept a valid boundary re-proposal"
);
marshal_actor_handle.abort();
drop(marshal);
context.sleep(Duration::from_millis(1)).await;
let certify_rx = inline.certify(reproposal_round, boundary_digest).await;
select! {
result = certify_rx => {
assert!(
result.unwrap(),
"certify should use the available_blocks fast path for verified re-proposals"
);
},
_ = context.sleep(Duration::from_secs(5)) => {
panic!("certify should not depend on marshal after verify cached a re-proposal");
},
}
});
}
#[test_traced("WARN")]
fn test_certify_recovers_from_invalid_reproposal_precheck() {
let runner = deterministic::Runner::timed(Duration::from_secs(30));
runner.start(|mut context| async move {
let Fixture {
participants,
schemes,
..
} = bls12381_threshold_vrf::fixture::<V, _>(&mut context, NAMESPACE, NUM_VALIDATORS);
let mut oracle = setup_network_with_participants(
context.child("network"),
NZUsize!(1),
participants.clone(),
1024 * 1024,
)
.await;
let me = participants[0].clone();
let setup = StandardHarness::setup_validator(
context.child("validator").with_attribute("index", 0),
&mut oracle,
me.clone(),
ConstantProvider::new(schemes[0].clone()),
)
.await;
let marshal = setup.mailbox;
let genesis = make_raw_block(Sha256::hash(&[b""]), Height::zero(), 0);
let mock_app: MockVerifyingApp<B, S> = MockVerifyingApp::new();
let mut inline = Inline::new(
context.child("inline"),
mock_app,
marshal.clone(),
FixedEpocher::new(BLOCKS_PER_EPOCH),
);
let block_round = Round::new(Epoch::zero(), View::new(2));
let block = B::new::<Sha256>(
Ctx {
round: block_round,
leader: default_leader(),
parent: (View::zero(), genesis.digest()),
},
genesis.digest(),
Height::new(2),
200,
);
let digest = block.digest();
assert!(marshal.verified(block_round, ready(block)).await);
let round = Round::new(Epoch::zero(), View::new(3));
let reproposal_context = Ctx {
round,
leader: me,
parent: (View::new(2), digest),
};
let verify_rx = inline.verify(reproposal_context, digest).await;
assert!(
!verify_rx.await.expect("verify result missing"),
"a non-boundary re-proposal must be rejected"
);
let certify_rx = inline.certify(round, digest).await;
select! {
result = certify_rx => {
assert!(
result.expect("certify result missing"),
"certify must recover instead of adopting the precheck rejection"
);
},
_ = context.sleep(Duration::from_secs(5)) => {
panic!("certify should resolve promptly");
},
}
});
}
#[test_traced("WARN")]
fn test_inline_certify_persists_block_before_resolving() {
for seed in 0u64..16 {
inline_certify_persists_block_before_resolving_at(seed);
}
}
fn inline_certify_persists_block_before_resolving_at(seed: u64) {
let runner = deterministic::Runner::new(
deterministic::Config::new()
.with_seed(seed)
.with_timeout(Some(Duration::from_secs(60))),
);
runner.start(|mut context| async move {
let Fixture {
participants,
schemes,
..
} = bls12381_threshold_vrf::fixture::<V, _>(&mut context, NAMESPACE, NUM_VALIDATORS);
let mut oracle = setup_network_with_participants(
context.child("network"),
NZUsize!(1),
participants.clone(),
1024 * 1024,
)
.await;
let me = participants[0].clone();
let setup = StandardHarness::setup_validator(
context.child("validator").with_attribute("index", 0),
&mut oracle,
me.clone(),
ConstantProvider::new(schemes[0].clone()),
)
.await;
let marshal = setup.mailbox;
let buffer = setup.extra;
let actor_handle = setup.actor_handle;
let genesis = make_raw_block(Sha256::hash(&[b""]), Height::zero(), 0);
let mock_app: MockVerifyingApp<B, S> = MockVerifyingApp::new();
let mut inline = Inline::new(
context.child("inline"),
mock_app,
marshal.clone(),
FixedEpocher::new(BLOCKS_PER_EPOCH),
);
let parent = make_raw_block(genesis.digest(), Height::new(1), 100);
let parent_digest = parent.digest();
let child_round = Round::new(Epoch::zero(), View::new(2));
let child_ctx = Ctx {
round: child_round,
leader: me.clone(),
parent: (View::new(1), parent_digest),
};
let child = B::new::<Sha256>(child_ctx.clone(), parent_digest, Height::new(2), 200);
let child_digest = child.digest();
assert!(
buffer
.broadcast(
commonware_p2p::Recipients::Some(vec![]),
ready(parent.clone())
)
.accepted(),
"buffer broadcast for parent should be accepted"
);
assert!(
buffer
.broadcast(
commonware_p2p::Recipients::Some(vec![]),
ready(child.clone())
)
.accepted(),
"buffer broadcast for child should be accepted"
);
let verify_rx = inline.verify(child_ctx, child_digest).await;
let certify_result = inline
.certify(child_round, child_digest)
.await
.await
.expect("certify result missing");
assert!(certify_result, "certify should succeed");
actor_handle.abort();
let _ = actor_handle.await;
drop(verify_rx);
drop(inline);
drop(marshal);
drop(buffer);
let setup2 = StandardHarness::setup_validator(
context
.child("validator_restart")
.with_attribute("index", 0),
&mut oracle,
me.clone(),
ConstantProvider::new(schemes[0].clone()),
)
.await;
let marshal2 = setup2.mailbox;
let post_restart = marshal2.get_block(&child_digest).await;
assert!(
post_restart.is_some(),
"certify resolved true so block must be durably persisted (seed={seed})"
);
});
}
#[test_traced("WARN")]
fn test_inline_propose_then_certify_persists_block() {
let runner = deterministic::Runner::timed(Duration::from_secs(30));
runner.start(|mut context| async move {
let Fixture {
participants,
schemes,
..
} = bls12381_threshold_vrf::fixture::<V, _>(&mut context, NAMESPACE, NUM_VALIDATORS);
let mut oracle = setup_network_with_participants(
context.child("network"),
NZUsize!(1),
participants.clone(),
1024 * 1024,
)
.await;
let me = participants[0].clone();
let setup = StandardHarness::setup_validator(
context.child("validator").with_attribute("index", 0),
&mut oracle,
me.clone(),
ConstantProvider::new(schemes[0].clone()),
)
.await;
let marshal = setup.mailbox;
let actor_handle = setup.actor_handle;
let genesis = make_raw_block(Sha256::hash(&[b""]), Height::zero(), 0);
let parent_round = Round::new(Epoch::zero(), View::new(1));
let parent_ctx = Ctx {
round: parent_round,
leader: default_leader(),
parent: (View::zero(), genesis.digest()),
};
let parent = B::new::<Sha256>(parent_ctx, genesis.digest(), Height::new(1), 100);
let parent_digest = parent.digest();
assert!(marshal.verified(parent_round, ready(parent)).await);
let round = Round::new(Epoch::zero(), View::new(2));
let ctx = Ctx {
round,
leader: me.clone(),
parent: (View::new(1), parent_digest),
};
let child = B::new::<Sha256>(ctx.clone(), parent_digest, Height::new(2), 200);
let child_digest = child.digest();
let mock_app: MockVerifyingApp<B, S> =
MockVerifyingApp::new().with_propose_result(child);
let mut inline = Inline::new(
context.child("inline"),
mock_app,
marshal.clone(),
FixedEpocher::new(BLOCKS_PER_EPOCH),
);
let digest = inline
.propose(ctx)
.await
.await
.expect("propose must return a digest");
assert_eq!(
digest, child_digest,
"propose must return the built block's digest"
);
assert!(
inline
.certify(round, child_digest)
.await
.await
.expect("certify result missing"),
"certify must succeed for the leader's own proposal"
);
actor_handle.abort();
let _ = actor_handle.await;
drop(inline);
drop(marshal);
let setup2 = StandardHarness::setup_validator(
context
.child("validator_restart")
.with_attribute("index", 0),
&mut oracle,
me,
ConstantProvider::new(schemes[0].clone()),
)
.await;
let marshal2 = setup2.mailbox;
assert!(
marshal2.get_block(&child_digest).await.is_some(),
"certify resolved true for the leader's own proposal so the block must be durable"
);
});
}
#[test_traced("WARN")]
fn test_inline_certify_recovers_after_verify_receiver_drop() {
let runner = deterministic::Runner::timed(Duration::from_secs(30));
runner.start(|mut context| async move {
let Fixture {
participants,
schemes,
..
} = bls12381_threshold_vrf::fixture::<V, _>(&mut context, NAMESPACE, NUM_VALIDATORS);
let mut oracle = setup_network_with_participants(
context.child("network"),
NZUsize!(1),
participants.clone(),
1024 * 1024,
)
.await;
let me = participants[0].clone();
let setup = StandardHarness::setup_validator(
context.child("validator").with_attribute("index", 0),
&mut oracle,
me.clone(),
ConstantProvider::new(schemes[0].clone()),
)
.await;
let marshal = setup.mailbox;
let genesis = make_raw_block(Sha256::hash(&[b""]), Height::zero(), 0);
let mock_app: MockVerifyingApp<B, S> = MockVerifyingApp::new();
let mut inline = Inline::new(
context.child("inline"),
mock_app,
marshal.clone(),
FixedEpocher::new(BLOCKS_PER_EPOCH),
);
let round = Round::new(Epoch::zero(), View::new(1));
let block_context = Ctx {
round,
leader: me,
parent: (View::zero(), genesis.digest()),
};
let block =
B::new::<Sha256>(block_context.clone(), genesis.digest(), Height::new(1), 100);
let digest = block.digest();
let verify_rx = inline.verify(block_context, digest).await;
drop(verify_rx);
context.sleep(Duration::from_millis(10)).await;
assert!(marshal.verified(round, ready(block)).await);
let certify_rx = inline.certify(round, digest).await;
select! {
result = certify_rx => {
assert!(
result.expect("certify result missing"),
"certify should recover after verify receiver drop"
);
},
_ = context.sleep(Duration::from_secs(5)) => {
panic!("certify should recover promptly after verify drop");
},
}
});
}
#[test_traced("WARN")]
fn test_inline_store_overlaps_app_verify() {
let runner = deterministic::Runner::timed(Duration::from_secs(30));
runner.start(|mut context| async move {
let Fixture {
participants,
schemes,
..
} = bls12381_threshold_vrf::fixture::<V, _>(&mut context, NAMESPACE, NUM_VALIDATORS);
let mut oracle = setup_network_with_participants(
context.child("network"),
NZUsize!(1),
participants.clone(),
1024 * 1024,
)
.await;
let me = participants[0].clone();
let setup = StandardHarness::setup_validator(
context.child("validator").with_attribute("index", 0),
&mut oracle,
me.clone(),
ConstantProvider::new(schemes[0].clone()),
)
.await;
let marshal = setup.mailbox;
let buffer = setup.extra;
let genesis = make_raw_block(Sha256::hash(&[b""]), Height::zero(), 0);
let (mock_app, verify_started, release_verify): (GatedVerifyingApp<B, S>, _, _) =
GatedVerifyingApp::new();
let mut inline = Inline::new(
context.child("inline"),
mock_app,
marshal.clone(),
FixedEpocher::new(BLOCKS_PER_EPOCH),
);
let parent = make_raw_block(genesis.digest(), Height::new(1), 100);
let parent_digest = parent.digest();
let child_round = Round::new(Epoch::zero(), View::new(2));
let child_ctx = Ctx {
round: child_round,
leader: me.clone(),
parent: (View::new(1), parent_digest),
};
let child = B::new::<Sha256>(child_ctx.clone(), parent_digest, Height::new(2), 200);
let child_digest = child.digest();
assert!(
buffer
.broadcast(commonware_p2p::Recipients::Some(vec![]), ready(parent))
.accepted(),
"buffer broadcast for parent should be accepted"
);
assert!(
buffer
.broadcast(commonware_p2p::Recipients::Some(vec![]), ready(child))
.accepted(),
"buffer broadcast for child should be accepted"
);
let verify_rx = inline.verify(child_ctx, child_digest).await;
verify_started
.await
.expect("verify should reach the gated application");
assert!(
marshal.get_block(&child_digest).await.is_some(),
"the store request runs concurrently with app.verify, so the block is locally queryable while verification is still gated"
);
release_verify.send_lossy(());
assert!(
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;
select! {
result = certify_rx => {
assert!(
result.expect("certify result missing"),
"certify should succeed once verification passes"
);
},
_ = context.sleep(Duration::from_secs(5)) => {
panic!("certify should resolve after verification is released");
},
}
});
}
#[test_traced("WARN")]
fn test_certify_not_poisoned_by_equivocating_parent_verify() {
let runner = deterministic::Runner::timed(Duration::from_secs(30));
runner.start(|mut context| async move {
let Fixture {
participants,
schemes,
..
} = bls12381_threshold_vrf::fixture::<V, _>(&mut context, NAMESPACE, NUM_VALIDATORS);
let mut oracle = setup_network_with_participants(
context.child("network"),
NZUsize!(1),
participants.clone(),
1024 * 1024,
)
.await;
let me = participants[0].clone();
let setup = StandardHarness::setup_validator(
context.child("validator").with_attribute("index", 0),
&mut oracle,
me,
ConstantProvider::new(schemes[0].clone()),
)
.await;
let marshal = setup.mailbox;
let buffer = setup.extra;
let genesis = make_raw_block(Sha256::hash(&[b""]), Height::zero(), 0);
let mock_app: MockVerifyingApp<B, S> = MockVerifyingApp::new();
let mut inline = Inline::new(
context.child("inline"),
mock_app,
marshal.clone(),
FixedEpocher::new(BLOCKS_PER_EPOCH),
);
let leader = participants[1].clone();
let certified_round = Round::new(Epoch::zero(), View::new(1));
let certified_ctx = Ctx {
round: certified_round,
leader: default_leader(),
parent: (View::zero(), genesis.digest()),
};
let certified = B::new::<Sha256>(certified_ctx, genesis.digest(), Height::new(1), 100);
let certified_digest = certified.digest();
assert!(marshal.verified(certified_round, ready(certified)).await);
let notarized_round = Round::new(Epoch::zero(), View::new(2));
let notarized_ctx = Ctx {
round: notarized_round,
leader: leader.clone(),
parent: (View::new(1), certified_digest),
};
let notarized = B::new::<Sha256>(notarized_ctx, certified_digest, Height::new(2), 200);
let notarized_digest = notarized.digest();
assert!(marshal.verified(notarized_round, ready(notarized)).await);
let round = Round::new(Epoch::zero(), View::new(3));
let block_ctx = Ctx {
round,
leader: leader.clone(),
parent: (View::new(2), notarized_digest),
};
let block = B::new::<Sha256>(block_ctx, notarized_digest, Height::new(3), 300);
let digest = block.digest();
assert!(
buffer
.broadcast(commonware_p2p::Recipients::Some(vec![]), ready(block))
.accepted(),
"buffer broadcast for the candidate should be accepted"
);
context.sleep(Duration::from_millis(10)).await;
let equivocating_ctx = Ctx {
round,
leader,
parent: (View::new(1), certified_digest),
};
let verify_rx = inline.verify(equivocating_ctx, digest).await;
assert!(
!verify_rx.await.expect("verify result missing"),
"the equivocating proposal must not be notarized"
);
let certify_rx = inline.certify(round, digest).await;
select! {
result = certify_rx => {
assert!(
result.expect("certify result missing"),
"certify of the notarized digest must not adopt the verdict computed under the equivocating context"
);
},
_ = context.sleep(Duration::from_secs(5)) => {
panic!("certify should resolve promptly");
},
}
});
}
impl Inline<deterministic::Context, S, MockVerifyingApp<B, S>, B, FixedEpocher> {
pub(in crate::marshal::standard) const fn gates(&self) -> &Gates<D, Digested<B>> {
&self.gates
}
}
}