commonware_consensus/marshal/standard/

inline.rs

//! Wrapper for standard marshal with inline verification.
//!
//! # Overview
//!
//! [`Inline`] adapts any [`Application`] to the marshal/consensus interfaces
//! while keeping block validation in the [`Automaton::verify`] path. Unlike
//! [`super::Deferred`], it does not defer application verification to certification.
//! Instead, it only reports `true` from `verify` after parent/height checks and
//! application verification complete.
//!
//! # Epoch Boundaries
//!
//! As with [`super::Deferred`], when the parent is the last block of the epoch,
//! [`Inline`] re-proposes that boundary block instead of building a new block.
//! This prevents proposing blocks that would be excluded by epoch transition.
//!
//! # Verification Model
//!
//! Inline mode intentionally avoids relying on embedded block context. This allows
//! usage with block types that implement [`crate::Block`] but not
//! [`crate::CertifiableBlock`].
//!
//! Because verification is completed inline, `certify` must only wait for data
//! availability in marshal. No additional deferred verification state needs to
//! be awaited at certify time.
//!
//! All validators must run the same wrapper for all views in a given epoch (see
//! [`crate::marshal::standard`] for more details).
//!
//! # Usage
//!
//! Wrap your [`Application`] implementation with [`Inline::new`] and provide it to your
//! consensus engine for the [`Automaton`], [`Relay`], and [`Reporter`]. As the reporter, the
//! wrapper forwards consensus activity to marshal and releases the state it keeps for each round
//! once consensus finalizes that round.
//!
//! ```rust,ignore
//! let application = Inline::new(
//!     context,
//!     my_application,
//!     marshal_mailbox,
//!     epocher,
//! );
//! ```
//!
//! # When to Use
//!
//! Prefer this wrapper when:
//! - Your application block type is not certifiable.
//! - You prefer simpler verification semantics over deferred verification latency hiding.
//! - You are willing to perform full application verification before casting a notarize vote.

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};

/// Waits for a marshal block subscription while allowing consensus to cancel the work.
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()
        },
    }
}

/// Standard marshal wrapper that verifies blocks inline in `verify`.
///
/// # Ancestry Validation
///
/// [`Inline`] always validates immediate ancestry before invoking application
/// verification:
/// - Parent digest matches consensus context's expected parent
/// - Child height is exactly parent height plus one
///
/// This is sufficient because the parent was either notarized by consensus or verified locally
/// before this participant voted for it, so its own linkage was already checked.
///
/// # Certifiability
///
/// This wrapper requires [`crate::Block`] and [`Codec`] for `B`. It does not
/// require [`crate::CertifiableBlock`]. It is designed for applications that cannot
/// recover consensus context directly from block payloads.
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,
{
    /// Creates a new inline-verification wrapper.
    ///
    /// Registers a `build_duration` histogram for proposal latency.
    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,
        }
    }

    /// Runs the proposal checks for `consensus_context` and yields what the proposal path does
    /// next: re-propose the epoch boundary block, decline a parent the marshal cannot build on,
    /// or build on the fetched parent.
    ///
    /// The returned future borrows nothing from `self`, so it can move into the spawned prepare
    /// or propose task.
    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 {
            // The parent for any consensus context is in the same epoch: the
            // boundary block of the previous epoch is the genesis block of the
            // current epoch.
            //
            // Proposal context carries the parent view and commitment but not
            // the parent height. The parent may sit above the finalized tip and
            // may still be uncertified, so this must stay round-bound until the
            // block is returned.
            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);

            // At epoch boundary, re-propose the parent block.
            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>;

    /// Proposes a new block or re-proposes an epoch boundary block.
    ///
    /// Proposal runs in a spawned task and returns a receiver for the resulting digest. The
    /// block is staged before the digest is delivered and handed to marshal when consensus
    /// requests the relay broadcast, which persists it after the send. The resulting sync
    /// handle is awaited only at certification so it overlaps consensus voting. The digest does
    /// not imply durability on its own. [`CertifiableAutomaton::certify`] awaits the registered
    /// certification gate before the finalize vote.
    #[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)),
        )
    }

    /// Performs complete verification inline.
    ///
    /// This method:
    /// 1. Waits for the block by digest
    /// 2. Enforces epoch/re-proposal rules
    /// 3. Fetches and validates the parent relationship
    /// 4. Runs application verification over ancestry
    ///
    /// The notarize vote is cast as soon as application verification completes. The block's
    /// durable sync is deferred (it runs concurrently with consensus voting) and its
    /// completion is registered in `gates` for [`Self::certify`] to await before
    /// the finalize vote.
    #[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;

        // Verification needs the full block but waits only for local delivery. Certification starts
        // recovery only when the block is not buffered. If a buffered block is evicted before
        // verification registers its wait, verification is left with neither the block nor an
        // active fetch. Register the wait before publishing the gate so it receives the buffered
        // block or is waiting when recovery delivers it.
        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 {
                // Start the parent fetch immediately: its commitment and certified
                // round are known from the consensus context, so it can proceed in
                // parallel with broadcast delivery of the candidate block.
                // Reproposals (digest == context.parent.1) skip parent validation
                // entirely, so they must not fetch: the "parent" is the candidate
                // itself, and candidate acquisition is deliberately local-only.
                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;
                };

                // Shared pre-checks:
                // - Blocks are invalid if they are not in the expected epoch and are
                //   not a valid boundary re-proposal.
                // - Re-proposals are detected when `digest == context.parent.1`.
                // - Re-proposals skip normal parent/height checks because:
                //   1) consensus settles their validity when certifying the view that
                //      first carried the block
                //   2) parent-child checks would fail by construction when parent == block
                let Some(decision) =
                    precheck_epoch_and_reproposal(&epocher, &marshal, &context, digest, block)
                        .await
                else {
                    return;
                };
                let block = match decision {
                    Decision::Complete(valid) => {
                        // Re-proposal: a valid precheck already persisted the block
                        // (durable), an invalid one is an epoch rejection. The re-proposal
                        // reading is an artifact of this header's declared parent: a
                        // conflicting header naming the block's real parent lets honest
                        // validators verify the same block as a normal proposal and
                        // notarize `(round, digest)`, so certification must recover
                        // rather than adopt the rejection.
                        tx.send_lossy(valid);
                        durable_tx.send_lossy(if valid {
                            GateOutcome::Ready(true)
                        } else {
                            GateOutcome::Recover
                        });
                        return;
                    }
                    Decision::Continue(block) => block,
                };

                // `Continue` implies a non-reproposal, so the parent subscription
                // was started above.
                let parent_request =
                    parent_request.expect("non-reproposal has a parent subscription");

                // Start the candidate store immediately: it depends on neither the
                // parent fetch (which may hit the network) nor the verdict below.
                // Storing before validation is intentional: these caches provide
                // candidate availability/recovery, not a validity decision. The
                // notarize vote follows the app verdict, while certify independently
                // awaits the registered durability gate.
                //
                // The verify future below aborts when consensus drops its receiver
                // (the view exited via nullification or finalization), even though
                // certification can still fire for a nullified view. That is
                // deliberate: inline certification does not need the local app
                // verdict (a notarization implies f+1 honest validators already
                // verified), and the store still completes through the join.
                let store = async {
                    if marshal.verified(round, block.clone()).await {
                        durable_tx.send_lossy(GateOutcome::Ready(true));
                    }
                };
                let verify_then_vote = async {
                    // Non-reproposal path: validate the parent we already started
                    // fetching.
                    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
    }
}

/// Inline certification consumes a registered certification gate when present, and
/// falls back to a round-bound fetch/persist path when the gate is missing (after
/// an unclean restart) or cannot speak for the notarized proposal.
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,
{
    /// Prepares a term-start proposal on an uncertified parent.
    ///
    /// Runs the same checks as [`Self::propose`], in the same order, and hands the application the
    /// fetched ancestry. When the checks find the epoch boundary block, the marshal re-proposes it
    /// without the application and answers [`Handoff::Stage`].
    #[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> {
        // `prepare`, `propose`, and `verify` register an in-flight certification gate whose
        // result resolves once the block's sync handle completes. Awaiting it here is the
        // durability barrier for the finalize vote, and it lets the sync overlap consensus voting
        // instead of freezing certify with a fresh fsync. A proposal still staged here is this
        // node's own proposal, so `verified_deferred` stores it and defers only its durable-sync
        // handle, which it delivers through the staged ack.
        let task = self.gates.claim(round, digest, |block, ack| {
            self.marshal.verified_deferred(round, block, ack)
        });

        // `verify()` waits only on local broadcast delivery, so nudge a
        // round-bound notarized fetch that can unblock the existing waiter
        // if local broadcast never arrives. For the standard variant, the
        // digest is also the variant commitment.
        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 {
                // A ready gate proves the notarized block is durable. A gate that
                // cannot speak for the notarized proposal falls through to recovery.
                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(_) => {}
                    }
                }

                // No local certification gate task (for example after an unclean restart):
                // fetch the notarized block and persist it. A Byzantine leader can form a
                // notarization after sending the proposal to only f+1 honest validators, so
                // the validators left without the block must fetch it here to certify and
                // avoid getting stuck.
                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)
    }
}

/// The wrapper is the consensus engine's reporter, as it is its automaton and relay: it forwards
/// every activity to marshal and first releases the state it keeps per round (certification
/// gates, staged proposals, and sent marks). A finalization releases its round and the rounds
/// below it. Any activity of a later epoch releases every earlier epoch, whose engine has been
/// retired, which covers finalized progress that marshal imported while no engine reported it.
/// The voter reports from the same task that requests the lock-in broadcast and certification,
/// so pruning is ordered with both.
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;

    // Compile-time assertion only: inline standard wrapper must not require `CertifiableBlock`.
    #[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),
            );

            // Seed the parent and child blocks in marshal so verify can fetch locally.
            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);

            // Complete verify first so the block is already available locally.
            let verify_rx = inline.verify(verify_context, digest).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;

            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),
            );

            // Seed the parent and child blocks in marshal without starting a verify task.
            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);

            // Certify should still resolve by waiting on marshal block availability directly.
            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");
                },
            }
        });
    }

    /// A header can only read as a re-proposal under its declared parent, so an
    /// invalid-precheck rejection (non-boundary block) is scoped to that header,
    /// not to the notarized `(round, digest)`. Certification must route through
    /// recovery instead of adopting the rejection as its verdict.
    #[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),
            );

            // A non-boundary block notarized at view 2 and available locally.
            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);

            // The view-3 header names the block as its own parent, so the
            // precheck reads it as a re-proposal and rejects it (not at the
            // epoch boundary).
            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"
            );

            // The header-scoped rejection must not become the certification
            // verdict for the notarized digest.
            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");
                },
            }
        });
    }

    /// Regression: `certify` resolving true drives the finalize vote in inline
    /// mode, so it must imply the block is durably persisted even when the
    /// certify path subscribed before `verify()` finished.
    #[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})"
            );
        });
    }

    /// Regression: in inline mode `propose` registers a certification gate for the
    /// built block that `certify` awaits. After the leader certifies its own proposal,
    /// the block must be durably recoverable. This is the >=f+1 guarantee: the leader
    /// certifies its own block through marshal so it awaits durability before the
    /// finalize vote.
    #[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);

            // Seed the parent at its round so `propose` can fetch it locally.
            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);

            // The leader builds the child via `app.propose`.
            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"
            );

            // The leader certifies its own proposal, which awaits the deferred sync handle.
            assert!(
                inline
                    .certify(round, child_digest)
                    .await
                    .await
                    .expect("certify result missing"),
                "certify must succeed for the leader's own proposal"
            );

            // After certify, the block must be durable across an unclean restart.
            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"
            );
        });
    }

    /// Dropping the verify receiver before the block is available closes the
    /// synchronously-registered certification gate. `certify` must recover through
    /// the fetch/certified path instead of returning the closed gate to consensus.
    #[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);

            // Give the verify task a chance to observe the dropped receiver while its
            // block subscription is still pending.
            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");
                },
            }
        });
    }

    /// The store request runs concurrently with `app.verify`, not after the
    /// notarize vote: while gated application verification is still blocked, the
    /// block has already reached marshal and is locally queryable even though the
    /// sync handle may still be pending. Releasing verification then lets the
    /// notarize vote resolve and certification await the registered certification
    /// gate. Separate restart tests cover durable recovery after certification.
    #[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;

            // Application verification is now blocked. The store request runs concurrently
            // with it, so the block is locally queryable even though the notarize vote has
            // not been cast and the sync handle may still be pending.
            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"
            );

            // Releasing verification resolves the notarize vote and lets certification
            // succeed (valid and durable).
            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");
                },
            }
        });
    }

    /// Inline analog of the deferred equivocation test. A proposal that names
    /// the certified view-1 parent for a block built on the view-2 block
    /// fails structural parent validation and is rejected for voting. The
    /// candidate is nevertheless stored durably, and certification of the
    /// honest notarization for the same `(round, digest)` must use that
    /// durability result rather than the context-scoped verification verdict.
    #[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();

            // The view-1 block: the parent this validator last certified.
            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);

            // The view-2 block: notarized by the network but only nullified
            // here, so this validator never certified it.
            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);

            // The view-3 block builds on the view-2 block. The leader's
            // broadcast delivered it into the local buffer without a local
            // verification.
            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;

            // The equivocating proposal names the certified view-1 block as
            // parent, which this validator's parent selection accepts
            // (view 1 certified, view 2 nullified). Refusing to notarize it
            // is correct.
            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"
            );

            // The honest notarization for the same `(round, digest)` arrives.
            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> {
        /// Exposes the gates this wrapper prunes to the wrapper tests shared by the standard
        /// variants.
        pub(in crate::marshal::standard) const fn gates(&self) -> &Gates<D, Digested<B>> {
            &self.gates
        }
    }
}