use super::{
    super::{Ask, Kind, Until},
    Config,
    ingress::{Handler, HandlerMessage, Mailbox, MailboxMessage},
    state::{Effect, FetchReason},
};
use crate::{
    Epochable, Viewable,
    simplex::{
        actors::{resolver::state::State, voter},
        scheme::Scheme,
        types::{Certificate, Notarization},
    },
    types::{Epoch, View},
};
use bytes::Bytes;
use commonware_actor::mailbox;
use commonware_codec::Decode;
use commonware_cryptography::Digest;
use commonware_macros::select_loop;
use commonware_p2p::{Blocker, Receiver, Sender, utils::StaticProvider};
use commonware_parallel::Strategy;
use commonware_resolver::{Fetch, Outcome, Resolution, Resolver, TargetedResolver, p2p};
use commonware_runtime::{
    Clock, ContextCell, Handle, Metrics, Spawner, spawn_cell, telemetry::traces::TracedExt as _,
};
use commonware_utils::{
    NonZeroDuration, channel::fallible::OneshotExt, ordered::Quorum, sequence::U64,
    vec::NonEmptyVec,
};
use rand_core::CryptoRng;
use std::{num::NonZeroUsize, time::Duration};
use tracing::{debug, info_span};

/// Requests are made concurrently to multiple peers.
pub struct Actor<
    E: Clock + CryptoRng + Metrics + Spawner,
    S: Scheme<D>,
    B: Blocker<PublicKey = S::PublicKey>,
    D: Digest,
    T: Strategy,
> {
    context: ContextCell<E>,
    scheme: S,
    blocker: Option<B>,
    strategy: T,

    epoch: Epoch,
    mailbox_size: NonZeroUsize,
    fetch_timeout: NonZeroDuration,
    fetch_retry_delay: Duration,

    /// Certificates held and served, and the repair progress built on them.
    /// Emits the [Effect]s the actor applies to the resolver (see
    /// [Self::apply_effects]).
    state: State,

    mailbox_receiver: mailbox::Receiver<MailboxMessage<S, D>>,
}

impl<
    E: Clock + CryptoRng + Metrics + Spawner,
    S: Scheme<D>,
    B: Blocker<PublicKey = S::PublicKey>,
    D: Digest,
    T: Strategy,
> Actor<E, S, B, D, T>
{
    pub fn new(context: E, cfg: Config<S, B, T>) -> (Self, Mailbox<S, D>) {
        let (sender, receiver) = mailbox::new(context.child("mailbox"), cfg.mailbox_size);
        (
            Self {
                context: ContextCell::new(context),
                scheme: cfg.scheme,
                blocker: Some(cfg.blocker),
                strategy: cfg.strategy,

                epoch: cfg.epoch,
                mailbox_size: cfg.mailbox_size,
                fetch_timeout: cfg.fetch_timeout,
                fetch_retry_delay: cfg.fetch_retry_delay,

                state: State::new(cfg.term_length),

                mailbox_receiver: receiver,
            },
            Mailbox::new(sender),
        )
    }

    pub fn start(
        mut self,
        voter: voter::Mailbox<S, D>,
        sender: impl Sender<PublicKey = S::PublicKey>,
        receiver: impl Receiver<PublicKey = S::PublicKey>,
    ) -> Handle<()> {
        spawn_cell!(self.context, self.run(voter, sender, receiver))
    }

    async fn run(
        mut self,
        mut voter: voter::Mailbox<S, D>,
        sender: impl Sender<PublicKey = S::PublicKey>,
        receiver: impl Receiver<PublicKey = S::PublicKey>,
    ) {
        let participants = self.scheme.participants().clone();
        let me = self
            .scheme
            .me()
            .and_then(|m| participants.key(m.index()))
            .cloned();

        // Every handler entry is one message the network admitted on the resolver channel, and
        // peer requests are dropped once the ready queue is full, so hold as many as the channel.
        let (handler_tx, mut handler_rx) =
            mailbox::new(self.context.as_ref().child("handler"), receiver.capacity());
        let handler = Handler::new(handler_tx);

        let (resolver_engine, mut resolver) = p2p::Engine::new(
            self.context.child("resolver"),
            p2p::Config {
                peer_provider: StaticProvider::new(self.epoch.get(), participants),
                blocker: self.blocker.take().expect("blocker must be set"),
                consumer: handler.clone(),
                producer: handler,
                mailbox_size: self.mailbox_size,
                me,
                fetch_timeout: self.fetch_timeout,
                fetch_retry_delay: self.fetch_retry_delay,
                priority_requests: true,
                priority_responses: false,
            },
        );
        let mut resolver_task = resolver_engine.start((sender, receiver));

        select_loop! {
            self.context,
            on_stopped => {
                debug!("context shutdown, stopping resolver");
            },
            _ = &mut resolver_task => {
                break;
            },
            // It is possible that a voter proposes some block before the certificate(s) that back
            // it reach the resolver, allowing an external request to arrive in the meantime. We
            // accept this unlikely race, which requires the resolver to lag well behind the voter,
            // rather than block broadcast on an acknowledgment here.
            Some(message) = self.mailbox_receiver.recv() else break => {
                let span = info_span!(
                    parent: message.span(),
                    "simplex.resolver.process",
                    operation = message.name(),
                    epoch = self.epoch.traced(),
                    view = message.view().traced()
                );
                let _guard = span.entered();
                match message {
                    MailboxMessage::Updated { certificate, .. } => {
                        self.updated(&mut resolver, &certificate);
                    }
                    MailboxMessage::Certified {
                        notarization,
                        success,
                        ..
                    } => {
                        self.certified(&mut resolver, notarization, success);
                    }
                    MailboxMessage::Resolve {
                        proposal,
                        view,
                        kind,
                        target,
                        ..
                    } => {
                        self.resolve(&mut resolver, proposal, view, kind, target);
                    }
                    MailboxMessage::Replayed { .. } => {
                        self.replayed(&mut resolver);
                    }
                }
            },
            Some(message) = handler_rx.recv() else break => {
                if message.response_closed() {
                    continue;
                }
                self.handle_resolver(message, &mut voter, &mut resolver);
            },
        }
    }

    /// Records a certificate and applies its effects.
    fn updated<R: Resolver<Key = U64, Subscriber = Ask>>(
        &mut self,
        resolver: &mut R,
        certificate: &Certificate<S, D>,
    ) {
        let effects = self.state.updated(certificate);
        self.apply_effects(resolver, effects);
    }

    /// Records a certification verdict and applies its effects.
    fn certified<R: Resolver<Key = U64, Subscriber = Ask>>(
        &mut self,
        resolver: &mut R,
        notarization: Notarization<S, D>,
        success: bool,
    ) {
        let effects = self.state.certified(notarization, success);
        self.apply_effects(resolver, effects);
    }

    /// Starts background repair once journal replay is complete.
    fn replayed<R: Resolver<Key = U64, Subscriber = Ask>>(&mut self, resolver: &mut R) {
        let effects = self.state.replayed();
        self.apply_effects(resolver, effects);
    }

    /// Applies the [Effect]s returned by [State] to the resolver.
    fn apply_effects<R: Resolver<Key = U64, Subscriber = Ask>>(
        &self,
        resolver: &mut R,
        effects: Vec<Effect>,
    ) {
        let mut fetches = Vec::with_capacity(effects.len());
        for effect in effects {
            match effect {
                Effect::Fetch {
                    view,
                    cause,
                    reason,
                } => fetches.push(self.fetch(view, cause, reason)),
                Effect::Settled { kind, start, end } => {
                    Self::retire(resolver, move |view, ask| {
                        ask.kind == kind && (start..=end).contains(&view)
                    });
                }
                Effect::Finalized(finalized) => {
                    Self::retire(resolver, move |view, _| view <= finalized);
                }
                Effect::Raised(floor) => {
                    // Resolver state does not repair at or below its floor, so a
                    // background ask there has nothing left to do. Only
                    // background asks retire here, because a proposal may name
                    // ancestry below the floor and only finalization rules that
                    // out.
                    Self::retire(resolver, move |view, ask| {
                        ask.until == Until::Floor && view <= floor
                    });
                }
            }
        }

        // Issue the fetches as one batch after the retires. No retire in these
        // effects matches a fetch in the batch, because each fetch asks for a
        // nullification that is still missing above the floor. Every
        // [Self::retire] predicate is also monotonic, so issuing the batch last
        // leaves the same asks open.
        if !fetches.is_empty() {
            let _ = resolver.fetch_all(fetches);
        }
    }

    /// Retires every ask for which `retired` holds.
    ///
    /// Every predicate tests a monotonic condition: settlement, finalization, or
    /// the floor. A retirement therefore stays true however the resolver orders it
    /// against fetches, which it reorders under backpressure.
    fn retire<R: Resolver<Key = U64, Subscriber = Ask>>(
        resolver: &mut R,
        retired: impl Fn(View, Ask) -> bool + Send + 'static,
    ) {
        let _ = resolver.retain(move |key, ask| !retired(View::new(u64::from(key)), *ask));
    }

    /// Builds a background fetch for the nullification covering `view`.
    ///
    /// Both [FetchReason]s want the same certificate: [State] only reports
    /// a view whose covering nullification is missing, whether the gap was found
    /// by scanning below the current view or opened by a failed certification.
    fn fetch(&self, view: View, cause: View, reason: FetchReason) -> Fetch<U64, Ask> {
        let ask = Ask::backfill();

        // State only emits a background fetch for a gap that is unsettled at
        // issuance. Later evidence may settle and retire the queued fetch.
        assert!(!self.state.settled(view, ask.kind));
        let span = info_span!(
            "simplex.resolver.fetch",
            epoch = self.epoch.traced(),
            cause = cause.traced(),
            view = view.traced(),
            reason = reason.as_str(),
            kind = ask.kind.as_str()
        );
        Fetch {
            key: U64::from(view),
            subscriber: ask,
            span,
        }
    }

    /// Fetches missing proposal ancestry. If `target` is provided, the resolver
    /// queries only that peer.
    fn resolve<R>(
        &self,
        resolver: &mut R,
        proposal: View,
        view: View,
        kind: Kind,
        target: Option<S::PublicKey>,
    ) where
        R: TargetedResolver<Key = U64, Subscriber = Ask, PublicKey = S::PublicKey>,
    {
        if view >= proposal || self.state.settled(view, kind) {
            return;
        }
        let ask = Ask::ancestry(kind);
        let span = info_span!(
            "simplex.resolver.fetch",
            epoch = self.epoch.traced(),
            cause = proposal.traced(),
            view = view.traced(),
            reason = "proposal_ancestry",
            kind = ask.kind.as_str()
        );
        let fetch = Fetch {
            key: U64::from(view),
            subscriber: ask,
            span,
        };
        let _ = match target {
            Some(target) => resolver.fetch_targeted(fetch, NonEmptyVec::new(target)),
            None => resolver.fetch(fetch),
        };
    }

    /// Validates an incoming message, returning the parsed message if valid.
    ///
    /// Validity is judged against `view` alone, because that is all the request
    /// named. Any certificate an honest peer could serve for the view is accepted,
    /// including one that answers the other kind: rejecting it would fault a peer
    /// that answered the only question the wire key asked. Whether it
    /// settles the ask is a separate check (see [State::settled]).
    fn validate(&mut self, view: View, data: Bytes) -> Option<Certificate<S, D>> {
        let incoming =
            Certificate::<S, D>::decode_cfg(data, &self.scheme.certificate_codec_config()).ok()?;

        // A nullification covers the rest of its term. A notarization or
        // finalization is servable from a peer's floor, which may sit above the
        // requested view.
        let servable = match &incoming {
            Certificate::Nullification(nullification) => {
                nullification.view().covers(view, self.state.term_length())
            }
            Certificate::Notarization(notarization) => notarization.view() >= view,
            Certificate::Finalization(finalization) => finalization.view() >= view,
        };
        if !servable {
            debug!(%view, received = %incoming.view(), "certificate below requested view");
            return None;
        }

        if incoming.epoch() != self.epoch {
            debug!(
                %view,
                epoch = %incoming.epoch(),
                expected = %self.epoch,
                "rejecting certificate from different epoch"
            );
            return None;
        }

        if !incoming.verify(self.context.as_mut(), &self.scheme, &self.strategy) {
            debug!(%view, "certificate failed verification");
            return None;
        }

        debug!(%view, received = %incoming.view(), "received certificate for request");
        Some(incoming)
    }

    /// Handles a message from the [p2p::Engine].
    fn handle_resolver<R: Resolver<Key = U64, Subscriber = Ask>>(
        &mut self,
        message: HandlerMessage,
        voter: &mut voter::Mailbox<S, D>,
        resolver: &mut R,
    ) {
        match message {
            HandlerMessage::Deliver {
                span,
                view,
                data,
                asks,
                response,
            } => {
                let span = info_span!(
                    parent: &span,
                    "simplex.resolver.deliver",
                    epoch = self.epoch.traced(),
                    view = view.traced()
                );
                let _guard = span.entered();

                // Ignoring is key-wide, so only skip validation after every
                // certificate kind that can share this view is settled.
                if self.state.key_settled(view) {
                    response.send_lossy(Outcome::Ignored);
                    return;
                }

                // Validate incoming message
                let validate = info_span!(
                    "simplex.resolver.validate",
                    epoch = self.epoch.traced(),
                    view = view.traced()
                );
                let Some(parsed) = validate.in_scope(|| self.validate(view, data)) else {
                    response.send_lossy(Outcome::Invalid);
                    return;
                };
                let settled_before = asks
                    .iter()
                    .filter(|ask| self.state.settled(view, ask.kind))
                    .count();

                // A failed notarization remains valid protocol evidence, but
                // replaying it cannot satisfy any outstanding repair ask.
                let obsolete = matches!(
                    &parsed,
                    Certificate::Notarization(notarization)
                        if self.state.uncertifiable(notarization.view())
                );

                // Record evidence before judging the snapshot, but defer resolver
                // effects until after reporting the verdict.
                let effects = if obsolete {
                    Vec::new()
                } else {
                    self.state.updated(&parsed)
                };

                // Dropping the verdict retires this settled snapshot without scoring
                // the peer or removing subscribers that joined after this delivery.
                if settled_before == asks.len().get() {
                    drop(response);
                } else {
                    let settled_after = asks
                        .iter()
                        .filter(|ask| self.state.settled(view, ask.kind))
                        .count();
                    let outcome = if settled_after == asks.len().get() {
                        Outcome::Resolved(Resolution::Complete)
                    } else if settled_after > settled_before {
                        Outcome::Resolved(Resolution::Partial)
                    } else {
                        Outcome::Unresolved
                    };
                    response.send_lossy(outcome);
                }

                self.apply_effects(resolver, effects);
                if !obsolete {
                    let resolved = info_span!(
                        "simplex.resolver.resolved",
                        epoch = self.epoch.traced(),
                        view = view.traced(),
                        certificate_view = parsed.view().traced()
                    );
                    resolved.in_scope(|| voter.recovered(parsed));
                }
            }
            HandlerMessage::Produce { view, response } => {
                let span = info_span!(
                    "simplex.resolver.produce",
                    epoch = self.epoch.traced(),
                    view = view.traced()
                );
                let _guard = span.entered();

                // Produce message for view
                let Some(certificate) = self.state.produce(view) else {
                    // If we drop the response channel, the resolver will automatically
                    // send an error response to the caller (so they don't need to wait
                    // the full timeout)
                    return;
                };
                response.send_lossy(certificate);
            }
        }
    }
}

#[cfg(test)]
mod tests {
    use super::{super::test_helpers::*, *};
    use crate::{
        simplex::{
            elector::{Config as _, Elector as _, RoundRobin, RoundRobinElector},
            scheme::ed25519,
            types::{Notarization, Notarize},
        },
        types::{Round, TermLength, ViewDelta},
    };
    use commonware_actor::Feedback;
    use commonware_codec::Encode;
    use commonware_cryptography::{
        Sha256,
        certificate::{Scheme as _, mocks::Fixture},
        ed25519::PublicKey,
        sha256::Digest as Sha256Digest,
    };
    use commonware_macros::{select, test_async, test_collect_traces};
    use commonware_p2p::{
        LimitedSender, Recipients,
        simulated::{Config as NetworkConfig, Link, Network},
        utils::mocks::{InertCheckedSender, inert_channel},
    };
    use commonware_parallel::Sequential;
    use commonware_resolver::{Consumer as _, Delivery};
    use commonware_runtime::{
        IoBuf, Quota, Runner, Supervisor, deterministic,
        telemetry::{metrics::has_metric_value, traces::collector::TraceStorage},
    };
    use commonware_utils::{
        NZDuration, NZU32, NZUsize,
        channel::{mpsc, oneshot},
        non_empty, non_empty_vec, probability,
        sync::Mutex,
    };
    use std::{collections::BTreeSet, num::NonZeroU32, sync::Arc, time::SystemTime};

    const NAMESPACE: &[u8] = b"resolver-actor";
    const EPOCH: Epoch = Epoch::new(9);
    const TERM_LENGTH: TermLength = TermLength::new(NZU32!(5));

    type TestScheme = ed25519::Scheme;
    type TestActor =
        Actor<deterministic::Context, TestScheme, NoopBlocker, Sha256Digest, Sequential>;

    #[derive(Clone, Default)]
    struct NoopBlocker;

    impl Blocker for NoopBlocker {
        type PublicKey = PublicKey;

        fn block(&mut self, _peer: Self::PublicKey) -> Feedback {
            Feedback::Ok
        }

        fn blocked(&mut self) -> commonware_p2p::BlockedSubscription<Self::PublicKey> {
            let (_, receiver) =
                commonware_utils::channel::ring::channel(commonware_utils::NZUsize!(1));
            receiver
        }
    }

    #[derive(Clone)]
    struct RateLimitedSender {
        checked: mpsc::UnboundedSender<PublicKey>,
        ready_at: SystemTime,
    }

    impl LimitedSender for RateLimitedSender {
        type PublicKey = PublicKey;
        type Checked<'a> = InertCheckedSender<PublicKey>;

        fn check(
            &mut self,
            recipients: Recipients<Self::PublicKey>,
        ) -> Result<Self::Checked<'_>, SystemTime> {
            let Recipients::One(peer) = recipients else {
                panic!("resolver must check one recipient at a time");
            };
            self.checked
                .send(peer)
                .expect("candidate receiver must remain open");
            Err(self.ready_at)
        }
    }

    /// Tracks the set of pending requests the way the resolver engine would.
    #[derive(Clone, Default)]
    struct RecordingResolver {
        outstanding: Arc<Mutex<BTreeSet<(U64, Ask)>>>,
        targeted: Arc<Mutex<Vec<(U64, Ask, PublicKey)>>>,
        batches: Arc<Mutex<Vec<Vec<u64>>>>,
        singles: Arc<Mutex<Vec<u64>>>,
    }

    impl RecordingResolver {
        fn outstanding(&self) -> Vec<u64> {
            self.outstanding
                .lock()
                .iter()
                .map(|(key, _)| u64::from(key))
                .collect::<BTreeSet<_>>()
                .into_iter()
                .collect()
        }

        fn targeted(&self) -> Vec<(u64, Ask, PublicKey)> {
            self.targeted
                .lock()
                .iter()
                .map(|(key, subscription, target)| (u64::from(key), *subscription, target.clone()))
                .collect()
        }

        fn subscriptions(&self, view: u64) -> Vec<Ask> {
            self.outstanding
                .lock()
                .iter()
                .filter_map(|(key, subscription)| (u64::from(key) == view).then_some(*subscription))
                .collect()
        }

        /// Keys of each untargeted [Resolver::fetch_all] call, in call order.
        fn batches(&self) -> Vec<Vec<u64>> {
            self.batches.lock().clone()
        }

        /// Keys of each untargeted [Resolver::fetch] call, in call order.
        fn singles(&self) -> Vec<u64> {
            self.singles.lock().clone()
        }
    }

    impl Resolver for RecordingResolver {
        type Key = U64;
        type Subscriber = Ask;

        fn fetch<F>(&mut self, key: F) -> Feedback
        where
            F: Into<Fetch<U64, Ask>> + Send,
        {
            let fetch = key.into();
            self.singles.lock().push(u64::from(&fetch.key));
            self.outstanding
                .lock()
                .insert((fetch.key, fetch.subscriber));
            Feedback::Ok
        }

        fn fetch_all<F>(&mut self, keys: Vec<F>) -> Feedback
        where
            F: Into<Fetch<U64, Ask>> + Send,
        {
            let fetches: Vec<Fetch<U64, Ask>> = keys.into_iter().map(Into::into).collect();
            self.batches
                .lock()
                .push(fetches.iter().map(|fetch| u64::from(&fetch.key)).collect());
            self.outstanding.lock().extend(
                fetches
                    .into_iter()
                    .map(|fetch| (fetch.key, fetch.subscriber)),
            );
            Feedback::Ok
        }

        fn retain(&mut self, predicate: impl Fn(&U64, &Ask) -> bool + Send + 'static) -> Feedback {
            self.outstanding
                .lock()
                .retain(|(key, subscription)| predicate(key, subscription));
            Feedback::Ok
        }
    }

    impl TargetedResolver for RecordingResolver {
        type PublicKey = PublicKey;

        fn fetch_targeted(
            &mut self,
            fetch: impl Into<Fetch<U64, Ask>> + Send,
            targets: NonEmptyVec<PublicKey>,
        ) -> Feedback {
            let fetch = fetch.into();
            self.targeted.lock().push((
                fetch.key.clone(),
                fetch.subscriber,
                targets.first().clone(),
            ));
            self.outstanding
                .lock()
                .insert((fetch.key, fetch.subscriber));
            Feedback::Ok
        }

        fn fetch_all_targeted<F>(&mut self, fetches: Vec<(F, NonEmptyVec<PublicKey>)>) -> Feedback
        where
            F: Into<Fetch<U64, Ask>> + Send,
        {
            for (fetch, targets) in fetches {
                self.fetch_targeted(fetch, targets);
            }
            Feedback::Ok
        }
    }

    fn build_actor(
        context: deterministic::Context,
        scheme: TestScheme,
        term_length: TermLength,
    ) -> TestActor {
        let (mut actor, _) = Actor::new(
            context,
            Config {
                scheme,
                blocker: NoopBlocker,
                strategy: Sequential,
                epoch: EPOCH,
                mailbox_size: NZUsize!(8),
                fetch_timeout: NZDuration!(Duration::from_secs(1)),
                fetch_retry_delay: Duration::from_millis(100),
                term_length,
            },
        );
        assert!(actor.state.replayed().is_empty());
        actor
    }

    fn assert_targeted_fetch_does_not_restrict_existing_backfill(target_index: usize) {
        let runtime = deterministic::Runner::timed(Duration::from_secs(10));
        runtime.start(|mut context| async move {
            let Fixture {
                participants,
                schemes,
                verifier,
                ..
            } = ed25519::fixture(&mut context, NAMESPACE, 4);
            let (network, oracle) = Network::new_with_peers(
                context.child("network"),
                NetworkConfig {
                    max_size: 1024 * 1024,
                    max_peers_per_set: NZUsize!(participants.len()),
                    disconnect_on_block: true,
                    tracked_peer_sets: NZUsize!(1),
                },
                participants.clone(),
            )
            .await;
            network.start();

            let mut connections = Vec::new();
            for participant in &participants {
                connections.push(
                    oracle
                        .control(participant.clone())
                        .register(2, Quota::per_second(NZU32!(1_000)))
                        .await
                        .unwrap(),
                );
            }
            let mut connections = connections.into_iter();
            let requester_connection = connections.next().unwrap();
            let (_silent_sender, mut silent_receiver) = connections.next().unwrap();
            let responder_connection = connections.next().unwrap();
            let _unused_connection = connections.next().unwrap();

            let link = Link {
                latency: Duration::from_millis(10),
                jitter: Duration::from_millis(1),
                success_rate: probability!(1.0),
            };
            oracle
                .add_link(
                    participants[0].clone(),
                    participants[1].clone(),
                    link.clone(),
                )
                .await
                .unwrap();
            oracle
                .add_link(
                    participants[1].clone(),
                    participants[0].clone(),
                    link.clone(),
                )
                .await
                .unwrap();

            let (requester_voter_sender, mut requester_voter_receiver) =
                mailbox::new(context.child("requester_voter"), NZUsize!(8));

            // Use the network control as the blocker so the requester's fetches skip
            // peers blocked on the network.
            let mut requester_control = oracle.control(participants[0].clone());
            let (requester, mut requester_mailbox) = Actor::<_, _, _, Sha256Digest, _>::new(
                context.child("requester"),
                Config {
                    scheme: schemes[0].clone(),
                    blocker: requester_control.clone(),
                    strategy: Sequential,
                    epoch: EPOCH,
                    mailbox_size: NZUsize!(8),
                    fetch_timeout: NZDuration!(Duration::from_millis(200)),
                    fetch_retry_delay: Duration::from_millis(200),
                    term_length: TermLength::ONE,
                },
            );
            requester_mailbox.replayed();
            let _requester = requester.start(
                voter::Mailbox::new(requester_voter_sender),
                requester_connection.0,
                requester_connection.1,
            );

            let (responder_voter_sender, _responder_voter_receiver) =
                mailbox::new(context.child("responder_voter"), NZUsize!(8));
            let (responder, mut responder_mailbox) = TestActor::new(
                context.child("responder"),
                Config {
                    scheme: schemes[2].clone(),
                    blocker: NoopBlocker,
                    strategy: Sequential,
                    epoch: EPOCH,
                    mailbox_size: NZUsize!(8),
                    fetch_timeout: NZDuration!(Duration::from_millis(200)),
                    fetch_retry_delay: Duration::from_millis(200),
                    term_length: TermLength::ONE,
                },
            );
            responder_mailbox.replayed();
            let _responder = responder.start(
                voter::Mailbox::new(responder_voter_sender),
                responder_connection.0,
                responder_connection.1,
            );

            let requested = View::new(1);
            let available = build_nullification(&schemes, &verifier, EPOCH, requested);
            responder_mailbox.updated(Certificate::Nullification(available.clone()));
            context.sleep(Duration::from_millis(10)).await;

            // Advancing to view 2 creates an unrestricted background ask for
            // view 1. The only connected peer is silent, so seeing
            // its request proves the background fetch is already in flight.
            requester_mailbox.updated(Certificate::Nullification(build_nullification(
                &schemes,
                &verifier,
                EPOCH,
                View::new(2),
            )));
            let (requester_key, _) = select! {
                request = silent_receiver.recv() => request.expect("silent peer channel closed"),
                _ = context.sleep(Duration::from_secs(2)) => {
                    panic!("background request did not reach silent peer");
                },
            };
            assert_eq!(requester_key, participants[0]);

            // A later targeted ancestry request may name a remote peer or the
            // requester itself. It must attach its subscriber without
            // narrowing the in-flight unrestricted fetch.
            requester_mailbox.resolve(
                View::new(3),
                requested,
                Kind::Nullification,
                Some(participants[target_index].clone()),
            );
            context.sleep(Duration::from_millis(10)).await;

            // Block the silent and unlinked peers, then link a different responder. If
            // the targeted ask narrowed the fetch, recovery cannot finish.
            commonware_p2p::block!(requester_control, participants[1].clone(), "unavailable");
            commonware_p2p::block!(requester_control, participants[3].clone(), "unavailable");
            oracle
                .add_link(
                    participants[0].clone(),
                    participants[2].clone(),
                    link.clone(),
                )
                .await
                .unwrap();
            oracle
                .add_link(participants[2].clone(), participants[0].clone(), link)
                .await
                .unwrap();

            let recovered = select! {
                message = requester_voter_receiver.recv() => message.expect("voter mailbox closed"),
                _ = context.sleep(Duration::from_secs(2)) => {
                    panic!("unrestricted fetch was narrowed by the targeted request");
                },
            };
            assert!(matches!(
                recovered,
                voter::Message::Verified {
                    certificate: Certificate::Nullification(nullification),
                    ..
                } if nullification == available
            ));
        });
    }

    #[test_async]
    async fn targeted_fetch_does_not_restrict_existing_backfill() {
        assert_targeted_fetch_does_not_restrict_existing_backfill(1);
    }

    #[test_async]
    async fn self_targeted_fetch_does_not_restrict_existing_backfill() {
        assert_targeted_fetch_does_not_restrict_existing_backfill(0);
    }

    /// Request expiry and retry backoff use their independently configured durations.
    #[test_async]
    async fn retry_delay_is_independent_of_request_timeout() {
        let runtime = deterministic::Runner::timed(Duration::from_secs(5));
        runtime.start(|mut context| async move {
            let Fixture {
                participants,
                schemes,
                ..
            } = ed25519::fixture(&mut context, NAMESPACE, 4);
            let fetch_timeout = NZDuration!(Duration::from_secs(1));
            let fetch_retry_delay = Duration::from_millis(100);
            let start = context.current();

            // The peer accepts requests but remains silent.
            let (sender, receiver) = inert_channel(&participants);
            let (voter_sender, _voter_receiver) = mailbox::new(context.child("voter"), NZUsize!(8));
            let (requester, mut requester_mailbox) = TestActor::new(
                context.child("requester"),
                Config {
                    scheme: schemes[0].clone(),
                    blocker: NoopBlocker,
                    strategy: Sequential,
                    epoch: EPOCH,
                    mailbox_size: NZUsize!(8),
                    fetch_timeout,
                    fetch_retry_delay,
                    term_length: TermLength::ONE,
                },
            );
            let _requester = requester.start(voter::Mailbox::new(voter_sender), sender, receiver);
            requester_mailbox.replayed();
            requester_mailbox.resolve(
                View::new(2),
                View::new(1),
                Kind::Notarization,
                Some(participants[1].clone()),
            );
            context.sleep_until(start + Duration::from_millis(50)).await;
            assert!(has_metric_value(
                &context.encode(),
                "requests_sent_total",
                1
            ));

            // Expiry alone cannot dispatch the next request while its backoff is pending.
            context
                .sleep_until(start + fetch_timeout.get() + fetch_retry_delay / 2)
                .await;
            assert!(has_metric_value(
                &context.encode(),
                "requests_sent_total",
                1
            ));

            // The shorter retry delay releases the request before another timeout elapses.
            context
                .sleep_until(start + fetch_timeout.get() + fetch_retry_delay * 2)
                .await;
            assert!(has_metric_value(
                &context.encode(),
                "requests_sent_total",
                2
            ));
        });
    }

    /// A committee member whose scheme cannot sign never selects itself as a fetch target.
    #[test_async]
    async fn member_without_share_does_not_fetch_from_itself() {
        let runtime = deterministic::Runner::timed(Duration::from_secs(10));
        runtime.start(|mut context| async move {
            let Fixture {
                participants,
                schemes,
                verifier,
                ..
            } = ed25519::fixture(&mut context, NAMESPACE, 4);
            let member = ed25519::Scheme::member(
                NAMESPACE,
                verifier.participants().clone(),
                &participants[0],
            )
            .expect("requester must be a participant");
            let fetch_timeout = NZDuration!(Duration::from_secs(5));
            let (checked_sender, mut checked_receiver) = mpsc::unbounded_channel();
            let sender = RateLimitedSender {
                checked: checked_sender,
                ready_at: context.current() + fetch_timeout.get(),
            };
            let (_, receiver) = inert_channel(&participants);
            let (requester_voter_sender, _requester_voter_receiver) =
                mailbox::new(context.child("requester_voter"), NZUsize!(8));
            let (requester, mut requester_mailbox) = TestActor::new(
                context.child("requester"),
                Config {
                    scheme: member,
                    blocker: NoopBlocker,
                    strategy: Sequential,
                    epoch: EPOCH,
                    mailbox_size: NZUsize!(8),
                    fetch_timeout,
                    fetch_retry_delay: fetch_timeout.get(),
                    term_length: TermLength::ONE,
                },
            );
            requester_mailbox.replayed();
            let _requester = requester.start(
                voter::Mailbox::new(requester_voter_sender),
                sender,
                receiver,
            );

            // Advancing to view 2 opens a background fetch for view 1. Rate limiting
            // every peer makes it inspect the entire candidate set in one synchronous
            // pass, regardless of selection order.
            requester_mailbox.updated(Certificate::Nullification(build_nullification(
                &schemes,
                &verifier,
                EPOCH,
                View::new(2),
            )));
            let mut checked = vec![
                checked_receiver
                    .recv()
                    .await
                    .expect("background fetch must check a candidate"),
            ];
            while let Ok(peer) = checked_receiver.try_recv() {
                checked.push(peer);
            }
            checked.sort();
            assert_eq!(
                checked,
                participants.iter().skip(1).cloned().collect::<Vec<_>>()
            );
        });
    }

    /// Certification recovery fetches a missing mid-term parent from an
    /// available holder when the term leader is unavailable.
    #[test_async]
    async fn untargeted_parent_fetch_uses_available_holder() {
        let runtime = deterministic::Runner::timed(Duration::from_secs(10));
        runtime.start(|mut context| async move {
            let Fixture {
                participants,
                schemes,
                verifier,
                ..
            } = ed25519::fixture(&mut context, NAMESPACE, 4);
            let elector: RoundRobinElector<TestScheme> = RoundRobin::<Sha256>::default()
                .with_term(TERM_LENGTH, Duration::from_secs(1), ViewDelta::new(0))
                .build(schemes[0].participants());
            let leader = usize::from(elector.elect(Round::new(EPOCH, View::new(1)), ()));
            assert_eq!(
                leader, 2,
                "fixture must leave the stable leader unavailable"
            );

            let (network, oracle) = Network::new_with_peers(
                context.child("network"),
                NetworkConfig {
                    max_size: 1024 * 1024,
                    max_peers_per_set: NZUsize!(participants.len()),
                    disconnect_on_block: true,
                    tracked_peer_sets: NZUsize!(1),
                },
                participants.clone(),
            )
            .await;
            network.start();

            let mut connections = Vec::new();
            for participant in &participants {
                connections.push(
                    oracle
                        .control(participant.clone())
                        .register(2, Quota::per_second(NZU32!(1_000)))
                        .await
                        .unwrap(),
                );
            }
            let mut connections = connections.into_iter();
            let requester_connection = connections.next().unwrap();
            let first_holder_connection = connections.next().unwrap();
            let _unavailable_leader_connection = connections.next().unwrap();
            let second_holder_connection = connections.next().unwrap();

            let link = Link {
                latency: Duration::from_millis(10),
                jitter: Duration::from_millis(1),
                success_rate: probability!(1.0),
            };
            for holder in [&participants[1], &participants[3]] {
                oracle
                    .add_link(participants[0].clone(), holder.clone(), link.clone())
                    .await
                    .unwrap();
                oracle
                    .add_link(holder.clone(), participants[0].clone(), link.clone())
                    .await
                    .unwrap();
            }

            let (requester_voter_sender, mut requester_voter_receiver) =
                mailbox::new(context.child("requester_voter"), NZUsize!(8));
            let (requester, mut requester_mailbox) = TestActor::new(
                context.child("requester"),
                Config {
                    scheme: schemes[0].clone(),
                    blocker: NoopBlocker,
                    strategy: Sequential,
                    epoch: EPOCH,
                    mailbox_size: NZUsize!(8),
                    fetch_timeout: NZDuration!(Duration::from_millis(200)),
                    fetch_retry_delay: Duration::from_millis(200),
                    term_length: TERM_LENGTH,
                },
            );
            requester_mailbox.replayed();
            let _requester = requester.start(
                voter::Mailbox::new(requester_voter_sender),
                requester_connection.0,
                requester_connection.1,
            );

            let parent = build_notarization(&schemes, &verifier, EPOCH, View::new(2));
            let mut holders = Vec::new();
            for (index, connection) in [(1, first_holder_connection), (3, second_holder_connection)]
            {
                let (voter_sender, voter_receiver) =
                    mailbox::new(context.child("holder_voter"), NZUsize!(8));
                let (holder, mut holder_mailbox) = TestActor::new(
                    context.child("holder"),
                    Config {
                        scheme: schemes[index].clone(),
                        blocker: NoopBlocker,
                        strategy: Sequential,
                        epoch: EPOCH,
                        mailbox_size: NZUsize!(8),
                        fetch_timeout: NZDuration!(Duration::from_millis(200)),
                        fetch_retry_delay: Duration::from_millis(200),
                        term_length: TERM_LENGTH,
                    },
                );
                holder_mailbox.replayed();
                let handle = holder.start(
                    voter::Mailbox::new(voter_sender),
                    connection.0,
                    connection.1,
                );
                holder_mailbox.updated(Certificate::Notarization(parent.clone()));
                holder_mailbox.certified(parent.clone(), true);

                // Keep the holder and its mailboxes alive until the fetch completes.
                holders.push((handle, holder_mailbox, voter_receiver));
            }

            // Request the parent from any peer. The voter test covers the child
            // that triggers this request.
            context.sleep(Duration::from_millis(10)).await;
            requester_mailbox.resolve(View::new(3), View::new(2), Kind::Notarization, None);

            let recovered = select! {
                message = requester_voter_receiver.recv() => {
                    message.expect("voter mailbox closed")
                },
                _ = context.sleep(Duration::from_secs(5)) => {
                    panic!("missing parent was not fetched from an available validator");
                },
            };
            assert!(matches!(
                recovered,
                voter::Message::Verified {
                    certificate: Certificate::Notarization(fetched),
                    ..
                } if fetched == parent
            ));
        });
    }

    /// Peer requests the network admits are all served even when every peer bursts at once.
    #[test_async]
    async fn served_requests_hold_channel_capacity() {
        /// Requests sent by each peer before the holder starts.
        const BURST: u32 = 2;

        let runtime = deterministic::Runner::timed(Duration::from_secs(10));
        runtime.start(|mut context| async move {
            let Fixture {
                participants,
                schemes,
                verifier,
                ..
            } = ed25519::fixture(&mut context, NAMESPACE, 4);
            let (network, oracle) = Network::new_with_peers(
                context.child("network"),
                NetworkConfig {
                    max_size: 1024 * 1024,
                    max_peers_per_set: NZUsize!(participants.len()),
                    disconnect_on_block: true,
                    tracked_peer_sets: NZUsize!(1),
                },
                participants.clone(),
            )
            .await;
            network.start();

            // One holder serves raw requests from every other participant.
            let quota = Quota::per_second(NonZeroU32::new(BURST).unwrap());
            let holder = participants[1].clone();
            let holder_connection = oracle
                .control(holder.clone())
                .register(2, quota)
                .await
                .unwrap();
            let link = Link {
                latency: Duration::from_millis(10),
                jitter: Duration::from_millis(0),
                success_rate: probability!(1.0),
            };
            let mut requesters = Vec::new();
            for requester in participants.iter().filter(|p| **p != holder) {
                let connection = oracle
                    .control(requester.clone())
                    .register(2, quota)
                    .await
                    .unwrap();
                oracle
                    .add_link(requester.clone(), holder.clone(), link.clone())
                    .await
                    .unwrap();
                oracle
                    .add_link(holder.clone(), requester.clone(), link.clone())
                    .await
                    .unwrap();
                requesters.push(connection);
            }

            // Each requester sends its full burst before the holder starts, so the holder's
            // resolver engine forwards every request to the handler in one pass.
            for (index, (sender, _)) in requesters.iter_mut().enumerate() {
                for id in 0..BURST {
                    let mut request = (index as u64 * u64::from(BURST) + u64::from(id))
                        .to_be_bytes()
                        .to_vec();
                    request.push(0);
                    request.extend_from_slice(&1u64.to_be_bytes());
                    sender.send(Recipients::One(holder.clone()), IoBuf::from(request), true);
                }
            }
            context.sleep(Duration::from_millis(50)).await;

            // Start the holder with the view 1 nullification every request asks for. Its handler
            // mailbox holds the channel's capacity, so every request fits even with a
            // `mailbox_size` of one.
            let (voter_sender, _voter_receiver) =
                mailbox::new(context.child("holder_voter"), NZUsize!(8));
            let (actor, mut holder_mailbox) = TestActor::new(
                context.child("holder"),
                Config {
                    scheme: schemes[1].clone(),
                    blocker: NoopBlocker,
                    strategy: Sequential,
                    epoch: EPOCH,
                    mailbox_size: NZUsize!(1),
                    fetch_timeout: NZDuration!(Duration::from_millis(200)),
                    fetch_retry_delay: Duration::from_millis(200),
                    term_length: TERM_LENGTH,
                },
            );
            holder_mailbox.replayed();
            holder_mailbox.updated(Certificate::Nullification(build_nullification(
                &schemes,
                &verifier,
                EPOCH,
                View::new(1),
            )));
            let _holder = actor.start(
                voter::Mailbox::new(voter_sender),
                holder_connection.0,
                holder_connection.1,
            );

            // Every request is answered with the certificate rather than an error.
            let mut answered = BTreeSet::new();
            for (_, receiver) in requesters.iter_mut() {
                for _ in 0..BURST {
                    let (_, reply) = select! {
                        reply = receiver.recv() => reply.expect("requester channel must stay open"),
                        _ = context.sleep(Duration::from_secs(2)) => {
                            panic!("every request in the burst must be answered");
                        },
                    };
                    let reply = reply.as_ref();
                    let id = u64::from_be_bytes(reply[..8].try_into().unwrap());
                    assert_eq!(reply[8], 1, "request {id} must be served");
                    assert!(answered.insert(id));
                }
            }
        });
    }

    /// A valid notarization that does not settle the ask does not prevent a
    /// different peer from supplying the requested nullification.
    #[test_async]
    async fn unresolved_delivery_does_not_block_nullification_fetch() {
        let runtime = deterministic::Runner::timed(Duration::from_secs(10));
        runtime.start(|mut context| async move {
            let Fixture {
                participants,
                schemes,
                verifier,
                ..
            } = ed25519::fixture(&mut context, NAMESPACE, 4);
            let (network, oracle) = Network::new_with_peers(
                context.child("network"),
                NetworkConfig {
                    max_size: 1024 * 1024,
                    max_peers_per_set: NZUsize!(participants.len()),
                    disconnect_on_block: true,
                    tracked_peer_sets: NZUsize!(1),
                },
                participants.clone(),
            )
            .await;
            network.start();

            let mut connections = Vec::new();
            for participant in &participants {
                connections.push(
                    oracle
                        .control(participant.clone())
                        .register(2, Quota::per_second(NZU32!(1_000)))
                        .await
                        .unwrap(),
                );
            }
            let mut connections = connections.into_iter();
            let requester_connection = connections.next().unwrap();
            let first_responder_connection = connections.next().unwrap();
            let nullification_holder_connection = connections.next().unwrap();
            let _unused_connection = connections.next().unwrap();

            let link = Link {
                latency: Duration::from_millis(10),
                jitter: Duration::from_millis(1),
                success_rate: probability!(1.0),
            };
            oracle
                .add_link(
                    participants[0].clone(),
                    participants[1].clone(),
                    link.clone(),
                )
                .await
                .unwrap();
            oracle
                .add_link(
                    participants[1].clone(),
                    participants[0].clone(),
                    link.clone(),
                )
                .await
                .unwrap();

            let (requester_voter_sender, mut requester_voter_receiver) =
                mailbox::new(context.child("requester_voter"), NZUsize!(8));

            // Use the network control as the blocker so the requester's fetches skip
            // peers blocked on the network.
            let mut requester_control = oracle.control(participants[0].clone());
            let (requester, mut requester_mailbox) = Actor::<_, _, _, Sha256Digest, _>::new(
                context.child("requester"),
                Config {
                    scheme: schemes[0].clone(),
                    blocker: requester_control.clone(),
                    strategy: Sequential,
                    epoch: EPOCH,
                    mailbox_size: NZUsize!(8),
                    fetch_timeout: NZDuration!(Duration::from_millis(200)),
                    fetch_retry_delay: Duration::from_millis(200),
                    term_length: TermLength::ONE,
                },
            );
            requester_mailbox.replayed();
            let _requester = requester.start(
                voter::Mailbox::new(requester_voter_sender),
                requester_connection.0,
                requester_connection.1,
            );

            let (first_voter_sender, _first_voter_receiver) =
                mailbox::new(context.child("first_voter"), NZUsize!(8));
            let (first_responder, mut first_responder_mailbox) = TestActor::new(
                context.child("first_responder"),
                Config {
                    scheme: schemes[1].clone(),
                    blocker: NoopBlocker,
                    strategy: Sequential,
                    epoch: EPOCH,
                    mailbox_size: NZUsize!(8),
                    fetch_timeout: NZDuration!(Duration::from_millis(200)),
                    fetch_retry_delay: Duration::from_millis(200),
                    term_length: TermLength::ONE,
                },
            );
            first_responder_mailbox.replayed();
            let _first_responder = first_responder.start(
                voter::Mailbox::new(first_voter_sender),
                first_responder_connection.0,
                first_responder_connection.1,
            );

            let (holder_voter_sender, _holder_voter_receiver) =
                mailbox::new(context.child("holder_voter"), NZUsize!(8));
            let (nullification_holder, mut nullification_holder_mailbox) = TestActor::new(
                context.child("nullification_holder"),
                Config {
                    scheme: schemes[2].clone(),
                    blocker: NoopBlocker,
                    strategy: Sequential,
                    epoch: EPOCH,
                    mailbox_size: NZUsize!(8),
                    fetch_timeout: NZDuration!(Duration::from_millis(200)),
                    fetch_retry_delay: Duration::from_millis(200),
                    term_length: TermLength::ONE,
                },
            );
            nullification_holder_mailbox.replayed();
            let _nullification_holder = nullification_holder.start(
                voter::Mailbox::new(holder_voter_sender),
                nullification_holder_connection.0,
                nullification_holder_connection.1,
            );

            let requested = View::new(1);
            let notarized = requested.next();
            let notarization = build_notarization(&schemes, &verifier, EPOCH, notarized);
            first_responder_mailbox.updated(Certificate::Notarization(notarization.clone()));
            // Certifying the notarization makes it the responder's floor, so
            // the responder answers a lower-view request with it.
            first_responder_mailbox.certified(notarization.clone(), true);
            let nullification = build_nullification(&schemes, &verifier, EPOCH, requested);
            nullification_holder_mailbox.updated(Certificate::Nullification(nullification.clone()));
            context.sleep(Duration::from_millis(10)).await;

            // A later nullification exposes a gap and starts an unrestricted
            // background fetch. The only connected peer answers with a valid
            // notarization that does not settle the nullification ask.
            requester_mailbox.updated(Certificate::Nullification(build_nullification(
                &schemes, &verifier, EPOCH, notarized,
            )));
            let first = select! {
                message = requester_voter_receiver.recv() => {
                    message.expect("voter mailbox closed")
                },
                _ = context.sleep(Duration::from_secs(2)) => {
                    panic!("notarization was not fetched");
                },
            };
            assert!(matches!(
                first,
                voter::Message::Verified {
                    certificate: Certificate::Notarization(ref fetched),
                    ..
                } if fetched == &notarization
            ));

            // Block the other peers and link the nullification holder so retries reach
            // only that peer.
            commonware_p2p::block!(requester_control, participants[1].clone(), "unavailable");
            commonware_p2p::block!(requester_control, participants[3].clone(), "unavailable");
            oracle
                .add_link(
                    participants[0].clone(),
                    participants[2].clone(),
                    link.clone(),
                )
                .await
                .unwrap();
            oracle
                .add_link(participants[2].clone(), participants[0].clone(), link)
                .await
                .unwrap();
            context.sleep(Duration::from_millis(10)).await;

            // The unresolved delivery releases the network slot. An ancestry subscriber
            // for the same view can fetch the needed nullification from another peer.
            requester_mailbox.resolve(
                View::new(3),
                requested,
                Kind::Nullification,
                Some(participants[2].clone()),
            );
            let recovered = select! {
                message = requester_voter_receiver.recv() => {
                    message.expect("voter mailbox closed")
                },
                _ = context.sleep(Duration::from_secs(2)) => {
                    panic!("unresolved delivery blocked ancestry repair");
                },
            };
            assert!(matches!(
                recovered,
                voter::Message::Verified {
                    certificate: Certificate::Nullification(fetched),
                    ..
                } if fetched == nullification
            ));
        });
    }

    /// One scan reaches the resolver as a single batch. An update that only
    /// settles an outstanding fetch retires it and sends no batch.
    #[test_async]
    async fn scan_fetches_issue_as_one_batch() {
        let runtime = deterministic::Runner::default();
        runtime.start(|mut context| async move {
            let Fixture {
                schemes, verifier, ..
            } = ed25519::fixture(&mut context, NAMESPACE, 4);
            let mut actor = build_actor(context, verifier.clone(), TERM_LENGTH);
            let mut resolver = RecordingResolver::default();

            let nullification = build_nullification(&schemes, &verifier, EPOCH, View::new(20));
            actor.updated(&mut resolver, &Certificate::Nullification(nullification));
            assert_eq!(resolver.batches(), vec![vec![1, 6, 11, 16]]);

            let nullification = build_nullification(&schemes, &verifier, EPOCH, View::new(6));
            actor.updated(&mut resolver, &Certificate::Nullification(nullification));
            assert_eq!(resolver.batches(), vec![vec![1, 6, 11, 16]]);
            assert!(resolver.singles().is_empty());
            assert_eq!(resolver.outstanding(), vec![1, 11, 16]);
        });
    }

    #[test_async]
    async fn updates_maintain_resolver_pending_set() {
        let runtime = deterministic::Runner::default();
        runtime.start(|mut context| async move {
            let Fixture {
                schemes, verifier, ..
            } = ed25519::fixture(&mut context, NAMESPACE, 4);
            let mut actor = build_actor(context, verifier.clone(), TERM_LENGTH);
            let mut resolver = RecordingResolver::default();

            // The first certificate emits requests at the term anchors.
            let nullification = build_nullification(&schemes, &verifier, EPOCH, View::new(20));
            actor.updated(&mut resolver, &Certificate::Nullification(nullification));
            assert_eq!(resolver.outstanding(), vec![1, 6, 11, 16]);

            // A covering nullification retains out only its own term's requests
            // (here, the request at its own view): views below its start and
            // above its term end stay pending.
            let nullification = build_nullification(&schemes, &verifier, EPOCH, View::new(6));
            actor.updated(&mut resolver, &Certificate::Nullification(nullification));
            assert_eq!(resolver.outstanding(), vec![1, 11, 16]);

            // A mid-term floor raise drops the requests below it and re-scans
            // the stranded term tail (view 5).
            let finalization = build_finalization(&schemes, &verifier, EPOCH, View::new(4));
            actor.updated(&mut resolver, &Certificate::Finalization(finalization));
            assert_eq!(resolver.outstanding(), vec![5, 11, 16]);

            // A below-floor nullification covering the floor's term retains
            // out the request at its term end (view 5), not just the request
            // at its own view.
            let nullification = build_nullification(&schemes, &verifier, EPOCH, View::new(2));
            actor.updated(&mut resolver, &Certificate::Nullification(nullification));
            assert_eq!(resolver.outstanding(), vec![11, 16]);

            // A floor raise drops the request at the floor view itself and
            // re-scans the stranded term tail (view 12).
            let finalization = build_finalization(&schemes, &verifier, EPOCH, View::new(11));
            actor.updated(&mut resolver, &Certificate::Finalization(finalization));
            assert_eq!(resolver.outstanding(), vec![12, 16]);

            // A nullification at the floor covering the floor's term retains
            // out mid-term requests strictly inside its range.
            let nullification = build_nullification(&schemes, &verifier, EPOCH, View::new(11));
            actor.updated(&mut resolver, &Certificate::Nullification(nullification));
            assert_eq!(resolver.outstanding(), vec![16]);
        });
    }

    /// A resolve without a target must open an untargeted fetch, not be
    /// dropped.
    #[test_async]
    async fn resolve_without_target_falls_back_to_untargeted_fetch() {
        let runtime = deterministic::Runner::default();
        runtime.start(|mut context| async move {
            let Fixture { verifier, .. } = ed25519::fixture(&mut context, NAMESPACE, 4);
            let actor = build_actor(context, verifier, TERM_LENGTH);
            let mut resolver = RecordingResolver::default();
            let requested = View::new(9);

            actor.resolve(
                &mut resolver,
                View::new(10),
                requested,
                Kind::Notarization,
                None,
            );
            assert!(resolver.targeted().is_empty());
            assert_eq!(resolver.outstanding(), vec![9]);
            assert_eq!(
                resolver.subscriptions(9),
                vec![Ask::ancestry(Kind::Notarization)]
            );
        });
    }

    #[test_async]
    async fn targeted_fetches_drop_only_when_ask_is_satisfied() {
        let runtime = deterministic::Runner::default();
        runtime.start(|mut context| async move {
            let Fixture {
                participants,
                schemes,
                verifier,
                ..
            } = ed25519::fixture(&mut context, NAMESPACE, 4);
            let mut actor = build_actor(context, verifier.clone(), TERM_LENGTH);
            let mut resolver = RecordingResolver::default();
            let requested = View::new(3);

            actor.resolve(
                &mut resolver,
                View::new(10),
                requested,
                Kind::Nullification,
                Some(participants[0].clone()),
            );
            actor.resolve(
                &mut resolver,
                View::new(11),
                requested,
                Kind::Notarization,
                Some(participants[1].clone()),
            );
            resolver.fetch(Fetch {
                key: U64::from(requested),
                subscriber: Ask::backfill(),
                span: tracing::Span::none(),
            });
            assert_eq!(
                resolver.targeted(),
                vec![
                    (
                        3,
                        Ask::ancestry(Kind::Nullification),
                        participants[0].clone()
                    ),
                    (
                        3,
                        Ask::ancestry(Kind::Notarization),
                        participants[1].clone()
                    ),
                ]
            );
            assert_eq!(
                resolver.subscriptions(3),
                vec![
                    Ask::backfill(),
                    Ask::ancestry(Kind::Nullification),
                    Ask::ancestry(Kind::Notarization),
                ]
            );

            // A covering nullification completes both background repair and
            // the targeted nullification request. It must preserve the
            // targeted parent request. That opposite evidence is the reason
            // the parent certificate is still needed.
            actor.updated(
                &mut resolver,
                &Certificate::Nullification(build_nullification(
                    &schemes, &verifier, EPOCH, requested,
                )),
            );
            assert_eq!(
                resolver.subscriptions(3),
                vec![Ask::ancestry(Kind::Notarization)]
            );
            actor.resolve(
                &mut resolver,
                View::new(14),
                requested,
                Kind::Nullification,
                Some(participants[2].clone()),
            );
            actor.resolve(
                &mut resolver,
                View::new(14),
                requested.next(),
                Kind::Nullification,
                Some(participants[2].clone()),
            );
            assert_eq!(resolver.targeted().len(), 2);

            // Exercise the mirror case at another key. Holding the exact
            // parent completes only the targeted parent request. It does not
            // provide the nullification needed to skip that view.
            let second_requested = requested.next_term_start(actor.state.term_length());
            actor.resolve(
                &mut resolver,
                View::new(12),
                second_requested,
                Kind::Nullification,
                Some(participants[2].clone()),
            );
            actor.resolve(
                &mut resolver,
                View::new(13),
                second_requested,
                Kind::Notarization,
                Some(participants[3].clone()),
            );
            resolver.fetch(Fetch {
                key: U64::from(second_requested),
                subscriber: Ask::backfill(),
                span: tracing::Span::none(),
            });
            let second_notarization =
                build_notarization(&schemes, &verifier, EPOCH, second_requested);
            actor.updated(
                &mut resolver,
                &Certificate::Notarization(second_notarization.clone()),
            );
            actor.certified(&mut resolver, second_notarization, true);

            // The certified notarization is now the floor, which satisfies
            // background repair at that view. The targeted nullification ask
            // survives: a notarization is not a nullification covering it.
            assert_eq!(
                resolver.subscriptions(6),
                vec![Ask::ancestry(Kind::Nullification)]
            );
            actor.resolve(
                &mut resolver,
                View::new(14),
                second_requested,
                Kind::Notarization,
                Some(participants[0].clone()),
            );
            assert_eq!(resolver.targeted().len(), 4);

            // A numeric floor raise alone does not identify which ancestry
            // requirement it satisfies. Both remaining targeted requests
            // survive until matching evidence or finalization arrives.
            actor.apply_effects(&mut resolver, vec![Effect::Raised(second_requested)]);
            assert_eq!(resolver.outstanding(), vec![3, 6]);

            // Finalization is the universal boundary. No valid proposal can
            // require ancestry at or below it. It removes every such key and
            // prevents a delayed mailbox request from recreating stale work.
            let finalized = second_requested;
            actor.updated(
                &mut resolver,
                &Certificate::Finalization(build_finalization(
                    &schemes, &verifier, EPOCH, finalized,
                )),
            );
            assert!(resolver.outstanding().is_empty());
            actor.resolve(
                &mut resolver,
                finalized.next(),
                requested,
                Kind::Notarization,
                Some(participants[0].clone()),
            );
            assert!(resolver.outstanding().is_empty());
            assert_eq!(resolver.targeted().len(), 4);
        });
    }

    #[test_async]
    async fn exact_certified_notarization_is_preferred_over_covering_nullification_and_floor() {
        let runtime = deterministic::Runner::default();
        runtime.start(|mut context| async move {
            let Fixture {
                schemes, verifier, ..
            } = ed25519::fixture(&mut context, NAMESPACE, 4);
            let mut responder =
                build_actor(context.child("responder"), verifier.clone(), TERM_LENGTH);
            let mut responder_resolver = RecordingResolver::default();

            let requested = View::new(3);
            let nullification = Certificate::Nullification(build_nullification(
                &schemes, &verifier, EPOCH, requested,
            ));
            responder.updated(&mut responder_resolver, &nullification);
            let parent = build_notarization(&schemes, &verifier, EPOCH, requested);
            let expected_parent = Certificate::Notarization(parent.clone()).encode();
            responder.updated(
                &mut responder_resolver,
                &Certificate::Notarization(parent.clone()),
            );
            responder.certified(&mut responder_resolver, parent, true);
            let floor = View::new(4);
            let floor_notarization = build_notarization(&schemes, &verifier, EPOCH, floor);
            responder.updated(
                &mut responder_resolver,
                &Certificate::Notarization(floor_notarization.clone()),
            );
            responder.certified(&mut responder_resolver, floor_notarization, true);

            // Both a covering nullification and the exact certified parent are
            // held for the requested view.
            assert!(responder.state.settled(requested, Kind::Nullification));
            assert!(responder.state.settled(requested, Kind::Notarization));

            // The exact certified parent is preferred to both the covering
            // nullification and a nonmatching higher floor.
            assert_eq!(responder.state.produce(requested), Some(expected_parent));
        });
    }

    #[test_async]
    async fn higher_certified_floor_does_not_hide_older_finalization() {
        let runtime = deterministic::Runner::default();
        runtime.start(|mut context| async move {
            let Fixture {
                schemes, verifier, ..
            } = ed25519::fixture(&mut context, NAMESPACE, 4);
            let mut responder =
                build_actor(context.child("responder"), verifier.clone(), TERM_LENGTH);
            let mut responder_resolver = RecordingResolver::default();

            let requested = View::new(3);
            let finalization = Certificate::Finalization(build_finalization(
                &schemes, &verifier, EPOCH, requested,
            ));
            responder.updated(&mut responder_resolver, &finalization);

            let floor = View::new(6);
            let notarization = build_notarization(&schemes, &verifier, EPOCH, floor);
            responder.updated(
                &mut responder_resolver,
                &Certificate::Notarization(notarization.clone()),
            );
            responder.certified(&mut responder_resolver, notarization, true);

            // The notarization is now the construction floor, but returning it
            // for view 3 would leave the targeted ask unresolved. The retained
            // finalization settles that ask instead.
            let data = responder
                .state
                .produce(requested)
                .expect("responder should retain settling finalization");

            let (voter_tx, _voter_rx) = mailbox::new(context.child("requester_voter"), NZUsize!(8));
            let mut voter = voter::Mailbox::new(voter_tx);
            let mut requester = build_actor(context.child("requester"), verifier, TERM_LENGTH);
            let mut requester_resolver = RecordingResolver::default();
            let (response, receiver) = oneshot::channel();
            requester.handle_resolver(
                HandlerMessage::Deliver {
                    span: tracing::Span::none(),
                    view: requested,
                    data,
                    asks: non_empty_vec![Ask::ancestry(Kind::Notarization)],
                    response,
                },
                &mut voter,
                &mut requester_resolver,
            );
            assert_eq!(
                receiver.await.unwrap(),
                Outcome::Resolved(Resolution::Complete)
            );
        });
    }

    #[test_async]
    async fn notarization_is_served_only_after_certification() {
        let runtime = deterministic::Runner::default();
        runtime.start(|mut context| async move {
            let Fixture {
                schemes, verifier, ..
            } = ed25519::fixture(&mut context, NAMESPACE, 4);
            let (voter_tx, _voter_rx) = mailbox::new(context.child("voter"), NZUsize!(8));
            let mut voter = voter::Mailbox::new(voter_tx);
            let mut actor = build_actor(context.child("actor"), verifier.clone(), TERM_LENGTH);
            let mut resolver = RecordingResolver::default();
            let view = View::new(3);
            let notarization = build_notarization(&schemes, &verifier, EPOCH, view);
            let encoded = Certificate::Notarization(notarization.clone()).encode();

            let (response, receiver) = oneshot::channel();
            actor.handle_resolver(
                HandlerMessage::Deliver {
                    span: tracing::Span::none(),
                    view,
                    data: encoded.clone(),
                    asks: non_empty_vec![Ask::ancestry(Kind::Notarization)],
                    response,
                },
                &mut voter,
                &mut resolver,
            );
            assert_eq!(
                receiver.await.unwrap(),
                Outcome::Resolved(Resolution::Complete)
            );
            assert!(actor.state.produce(view).is_none());

            actor.certified(&mut resolver, notarization, true);
            assert_eq!(actor.state.produce(view), Some(encoded));
        });
    }

    /// Regression test for https://github.com/commonwarexyz/monorepo/pull/5097.
    ///
    /// The resolver engine can release its handles to a fetch span after the
    /// actor checks that a delivery is still wanted. The delivery then holds the
    /// only handle, and processing must still run under that span.
    #[test_collect_traces]
    fn pr_5097_regression(traces: TraceStorage) {
        let runtime = deterministic::Runner::default();
        runtime.start(|mut context| async move {
            let Fixture {
                schemes, verifier, ..
            } = ed25519::fixture(&mut context, NAMESPACE, 4);
            let (voter_tx, _voter_rx) = mailbox::new(context.child("voter"), NZUsize!(8));
            let mut voter = voter::Mailbox::new(voter_tx);
            let (handler_tx, mut handler_rx) = mailbox::new(context.child("handler"), NZUsize!(8));
            let mut handler = Handler::new(handler_tx);
            let mut actor = build_actor(context.child("actor"), verifier.clone(), TERM_LENGTH);
            let mut resolver = RecordingResolver::default();
            let view = View::new(3);
            let notarization = build_notarization(&schemes, &verifier, EPOCH, view);

            // The handler keeps only a clone of the first subscriber's span and
            // drops the delivery, so the queued message holds the only handle.
            let _outcome = handler.deliver(
                Delivery {
                    key: U64::from(view),
                    subscribers: non_empty_vec![(
                        Ask::ancestry(Kind::Notarization),
                        tracing::info_span!("test.fetch"),
                    )],
                },
                Certificate::Notarization(notarization).encode(),
            );
            let message = handler_rx.recv().await.unwrap();
            actor.handle_resolver(message, &mut voter, &mut resolver);
        });

        // Processing runs under a delivery span whose parent is the fetch span.
        traces
            .get_all()
            .expect_event(|event| {
                event.spans.windows(2).any(|pair| {
                    pair[0].content == "simplex.resolver.deliver" && pair[1].content == "test.fetch"
                })
            })
            .unwrap();
    }

    #[test_async]
    async fn nullification_satisfies_target_at_certified_floor() {
        let runtime = deterministic::Runner::default();
        runtime.start(|mut context| async move {
            let Fixture {
                participants,
                schemes,
                verifier,
                ..
            } = ed25519::fixture(&mut context, NAMESPACE, 4);
            let mut actor = build_actor(context, verifier.clone(), TermLength::ONE);
            let mut resolver = RecordingResolver::default();
            let view = View::new(3);

            // Certify the view first, then request its coexisting
            // nullification as ancestry for a proposal that skips it.
            let notarization = build_notarization(&schemes, &verifier, EPOCH, view);
            actor.updated(
                &mut resolver,
                &Certificate::Notarization(notarization.clone()),
            );
            actor.certified(&mut resolver, notarization, true);
            let floor = View::new(8);
            let certified_floor = build_notarization(&schemes, &verifier, EPOCH, floor);
            actor.updated(
                &mut resolver,
                &Certificate::Notarization(certified_floor.clone()),
            );
            actor.certified(&mut resolver, certified_floor.clone(), true);
            actor.resolve(
                &mut resolver,
                View::new(10),
                view,
                Kind::Nullification,
                Some(participants[0].clone()),
            );
            assert_eq!(
                resolver.subscriptions(3),
                vec![Ask::ancestry(Kind::Nullification)]
            );

            // The certified floor stays in place, but locally observing the
            // nullification must retire the exact targeted ask it satisfies.
            actor.updated(
                &mut resolver,
                &Certificate::Nullification(build_nullification(&schemes, &verifier, EPOCH, view)),
            );
            assert!(resolver.outstanding().is_empty());
            assert_eq!(
                actor.state.produce(View::new(6)),
                Some(Certificate::Notarization(certified_floor).encode())
            );
            actor.resolve(
                &mut resolver,
                View::new(11),
                view,
                Kind::Nullification,
                Some(participants[1].clone()),
            );
            assert_eq!(resolver.targeted().len(), 1);
        });
    }

    #[test_async]
    async fn higher_certified_floor_does_not_suppress_exact_parent_fetch() {
        let runtime = deterministic::Runner::default();
        runtime.start(|mut context| async move {
            let Fixture {
                participants,
                schemes,
                verifier,
                ..
            } = ed25519::fixture(&mut context, NAMESPACE, 4);
            let mut actor = build_actor(context, verifier.clone(), TERM_LENGTH);
            let mut resolver = RecordingResolver::default();
            let floor = View::new(6);
            let floor_notarization = build_notarization(&schemes, &verifier, EPOCH, floor);

            actor.updated(
                &mut resolver,
                &Certificate::Notarization(floor_notarization.clone()),
            );
            actor.certified(&mut resolver, floor_notarization, true);

            actor.resolve(
                &mut resolver,
                View::new(8),
                View::new(4),
                Kind::Notarization,
                Some(participants[0].clone()),
            );
            assert_eq!(
                resolver.targeted(),
                vec![(
                    4,
                    Ask::ancestry(Kind::Notarization),
                    participants[0].clone()
                )]
            );
        });
    }

    #[test_async]
    async fn certification_failure_retires_parent_target() {
        let runtime = deterministic::Runner::default();
        runtime.start(|mut context| async move {
            let Fixture {
                participants,
                schemes,
                verifier,
                ..
            } = ed25519::fixture(&mut context, NAMESPACE, 4);
            let mut actor = build_actor(context, verifier.clone(), TERM_LENGTH);
            let mut resolver = RecordingResolver::default();
            let view = View::new(5);

            actor.resolve(
                &mut resolver,
                View::new(10),
                view,
                Kind::Notarization,
                Some(participants[0].clone()),
            );
            let notarization = build_notarization(&schemes, &verifier, EPOCH, view);
            actor.updated(
                &mut resolver,
                &Certificate::Notarization(notarization.clone()),
            );
            actor.certified(&mut resolver, notarization, false);

            // A false certification verdict is permanent. Parent repair is
            // retired, while ordinary repair asks for the nullification that
            // can now cover the failed view.
            assert_eq!(resolver.subscriptions(5), vec![Ask::backfill()]);
            actor.resolve(
                &mut resolver,
                View::new(11),
                view,
                Kind::Notarization,
                Some(participants[1].clone()),
            );
            assert_eq!(resolver.targeted().len(), 1);
        });
    }

    #[test_async]
    async fn parent_delivery_completes_on_possession_not_certification() {
        let runtime = deterministic::Runner::default();
        runtime.start(|mut context| async move {
            let Fixture {
                schemes, verifier, ..
            } = ed25519::fixture(&mut context, NAMESPACE, 4);
            let (voter_tx, _voter_rx) = mailbox::new(context.child("voter"), NZUsize!(8));
            let mut voter = voter::Mailbox::new(voter_tx);
            let mut actor = build_actor(context.child("actor"), verifier.clone(), TERM_LENGTH);
            let mut resolver = RecordingResolver::default();
            let view = View::new(6);

            // The exact parent was asked for and arrived. Certification is an
            // application verdict on evidence already in hand, so the fetch has
            // nothing left to retrieve and must not stay open waiting for it.
            let (response, receiver) = oneshot::channel();
            actor.handle_resolver(
                HandlerMessage::Deliver {
                    span: tracing::Span::none(),
                    view,
                    data: Certificate::<TestScheme, Sha256Digest>::Notarization(
                        build_notarization(&schemes, &verifier, EPOCH, view),
                    )
                    .encode(),
                    asks: non_empty_vec![Ask::ancestry(Kind::Notarization)],
                    response,
                },
                &mut voter,
                &mut resolver,
            );
            assert_eq!(
                receiver.await.unwrap(),
                Outcome::Resolved(Resolution::Complete)
            );

            // Possession settles the ask, but the notarization is not served
            // until it certifies.
            assert!(actor.state.settled(view, Kind::Notarization));
            assert!(actor.state.produce(view).is_none());
        });
    }

    #[test_async]
    async fn failed_certification_retires_parent_ask_until_finalization() {
        let runtime = deterministic::Runner::default();
        runtime.start(|mut context| async move {
            let Fixture {
                schemes,
                verifier,
                participants,
                ..
            } = ed25519::fixture(&mut context, NAMESPACE, 4);
            let mut actor = build_actor(context, verifier.clone(), TERM_LENGTH);
            let mut resolver = RecordingResolver::default();
            let view = View::new(6);

            let notarization = build_notarization(&schemes, &verifier, EPOCH, view);
            actor.updated(
                &mut resolver,
                &Certificate::Notarization(notarization.clone()),
            );
            actor.certified(&mut resolver, notarization, false);

            // The notarization is no longer an answer, and the tombstone keeps a
            // delayed request from recreating the ask. A floor raise above the
            // view must not resurrect it.
            assert!(actor.state.uncertifiable(view));
            assert!(actor.state.produce(view).is_none());
            let floor = View::new(9);
            let floor_notarization = build_notarization(&schemes, &verifier, EPOCH, floor);
            actor.updated(
                &mut resolver,
                &Certificate::Notarization(floor_notarization.clone()),
            );
            actor.certified(&mut resolver, floor_notarization, true);

            let targeted = resolver.targeted().len();
            actor.resolve(
                &mut resolver,
                View::new(11),
                view,
                Kind::Notarization,
                Some(participants[0].clone()),
            );
            assert_eq!(resolver.targeted().len(), targeted);
        });
    }

    #[test_async]
    async fn tombstoned_notarization_delivery_retires_without_recording() {
        let runtime = deterministic::Runner::default();
        runtime.start(|mut context| async move {
            let Fixture {
                schemes, verifier, ..
            } = ed25519::fixture(&mut context, NAMESPACE, 4);
            let (voter_tx, mut voter_rx) = mailbox::new(context.child("voter"), NZUsize!(8));
            let mut voter = voter::Mailbox::new(voter_tx);
            let mut actor = build_actor(context, verifier.clone(), TERM_LENGTH);
            let mut resolver = RecordingResolver::default();
            let view = View::new(6);

            let notarization = build_notarization(&schemes, &verifier, EPOCH, view);
            actor.updated(
                &mut resolver,
                &Certificate::Notarization(notarization.clone()),
            );
            actor.certified(&mut resolver, notarization.clone(), false);

            // The failed verdict already settled this snapshot's ask. Its
            // notarization cannot become pending or notify the voter again.
            let (response, receiver) = oneshot::channel();
            actor.handle_resolver(
                HandlerMessage::Deliver {
                    span: tracing::Span::none(),
                    view,
                    data: Certificate::<TestScheme, Sha256Digest>::Notarization(notarization)
                        .encode(),
                    asks: non_empty_vec![Ask::ancestry(Kind::Notarization)],
                    response,
                },
                &mut voter,
                &mut resolver,
            );
            assert!(receiver.await.is_err());
            assert!(actor.state.uncertifiable(view));
            assert!(actor.state.produce(view).is_none());
            drop(voter);
            assert!(voter_rx.recv().await.is_none());

            // Finalization is the tombstone's retirement boundary.
            actor.updated(
                &mut resolver,
                &Certificate::Finalization(build_finalization(
                    &schemes,
                    &verifier,
                    EPOCH,
                    view.next(),
                )),
            );
            assert!(!actor.state.uncertifiable(view));
        });
    }

    #[test_async]
    async fn valid_nullification_does_not_complete_parent_delivery() {
        let runtime = deterministic::Runner::default();
        runtime.start(|mut context| async move {
            let Fixture {
                schemes, verifier, ..
            } = ed25519::fixture(&mut context, NAMESPACE, 4);
            let (voter_tx, _voter_rx) = mailbox::new(context.child("voter"), NZUsize!(8));
            let mut voter = voter::Mailbox::new(voter_tx);
            let mut actor = build_actor(context, verifier.clone(), TERM_LENGTH);
            let mut resolver = RecordingResolver::default();

            // The local requirement is intentionally absent from a delivery.
            // A covering nullification is a valid answer for the U64 key even
            // if the triggering proposal wanted a certified parent.
            let nullification = build_nullification(&schemes, &verifier, EPOCH, View::new(3));
            let (response, receiver) = oneshot::channel();
            actor.handle_resolver(
                HandlerMessage::Deliver {
                    span: tracing::Span::none(),
                    view: View::new(4),
                    data: Certificate::<TestScheme, Sha256Digest>::Nullification(nullification)
                        .encode(),
                    asks: non_empty_vec![Ask::ancestry(Kind::Notarization)],
                    response,
                },
                &mut voter,
                &mut resolver,
            );

            assert_eq!(receiver.await.unwrap(), Outcome::Unresolved);
        });
    }

    #[test_async]
    async fn unrelated_higher_certified_notarization_does_not_complete_parent_delivery() {
        let runtime = deterministic::Runner::default();
        runtime.start(|mut context| async move {
            let Fixture {
                schemes, verifier, ..
            } = ed25519::fixture(&mut context, NAMESPACE, 4);
            let (voter_tx, _voter_rx) = mailbox::new(context.child("voter"), NZUsize!(8));
            let mut voter = voter::Mailbox::new(voter_tx);
            let mut actor = build_actor(context, verifier.clone(), TERM_LENGTH);
            let mut resolver = RecordingResolver::default();

            let requested = View::new(4);
            let notarization = build_notarization(&schemes, &verifier, EPOCH, View::new(6));
            actor.updated(
                &mut resolver,
                &Certificate::Notarization(notarization.clone()),
            );
            actor.certified(&mut resolver, notarization.clone(), true);
            let (response, receiver) = oneshot::channel();
            actor.handle_resolver(
                HandlerMessage::Deliver {
                    span: tracing::Span::none(),
                    view: requested,
                    data: Certificate::<TestScheme, Sha256Digest>::Notarization(notarization)
                        .encode(),
                    asks: non_empty_vec![Ask::ancestry(Kind::Notarization)],
                    response,
                },
                &mut voter,
                &mut resolver,
            );
            assert_eq!(receiver.await.unwrap(), Outcome::Unresolved);
        });
    }

    #[test_async]
    async fn higher_notarization_does_not_complete_backfill_delivery() {
        let runtime = deterministic::Runner::default();
        runtime.start(|mut context| async move {
            let Fixture {
                schemes, verifier, ..
            } = ed25519::fixture(&mut context, NAMESPACE, 4);
            let (voter_tx, _voter_rx) = mailbox::new(context.child("voter"), NZUsize!(8));
            let mut voter = voter::Mailbox::new(voter_tx);
            let mut actor = build_actor(context, verifier.clone(), TERM_LENGTH);
            let mut resolver = RecordingResolver::default();

            // A peer may answer a background fetch above our floor with its own
            // higher floor. That covers nothing at the requested view, so the
            // fetch stays open.
            let notarization = build_notarization(&schemes, &verifier, EPOCH, View::new(6));
            let (response, receiver) = oneshot::channel();
            actor.handle_resolver(
                HandlerMessage::Deliver {
                    span: tracing::Span::none(),
                    view: View::new(4),
                    data: Certificate::<TestScheme, Sha256Digest>::Notarization(notarization)
                        .encode(),
                    asks: non_empty_vec![Ask::backfill()],
                    response,
                },
                &mut voter,
                &mut resolver,
            );
            assert_eq!(receiver.await.unwrap(), Outcome::Unresolved);
        });
    }

    #[test_async]
    async fn mixed_delivery_credits_only_newly_settled_asks() {
        let runtime = deterministic::Runner::default();
        runtime.start(|mut context| async move {
            let Fixture {
                schemes, verifier, ..
            } = ed25519::fixture(&mut context, NAMESPACE, 4);
            let view = View::new(4);
            let certificates = [
                (
                    Kind::Nullification,
                    Certificate::Nullification(build_nullification(
                        &schemes, &verifier, EPOCH, view,
                    )),
                ),
                (
                    Kind::Notarization,
                    Certificate::Notarization(build_notarization(&schemes, &verifier, EPOCH, view)),
                ),
            ];

            for (kind, certificate) in certificates {
                let context = context.child(kind.as_str());
                let (voter_tx, _voter_rx) = mailbox::new(context.child("voter"), NZUsize!(8));
                let mut voter = voter::Mailbox::new(voter_tx);
                let mut actor = build_actor(context.child("actor"), verifier.clone(), TERM_LENGTH);
                let mut resolver = RecordingResolver::default();

                // Coalesce backfill and both ancestry kinds under one key. Either
                // certificate can settle only part of this subscriber set.
                let asks = non_empty_vec![
                    Ask::backfill(),
                    Ask::ancestry(Kind::Nullification),
                    Ask::ancestry(Kind::Notarization),
                ];
                for ask in asks.iter() {
                    resolver.fetch(Fetch {
                        key: U64::from(view),
                        subscriber: *ask,
                        span: tracing::Span::none(),
                    });
                }

                // Credit the first response for settling some asks. Replaying the
                // same snapshot settles nothing new and must not earn more credit.
                for outcome in [Outcome::Resolved(Resolution::Partial), Outcome::Unresolved] {
                    let (response, receiver) = oneshot::channel();
                    actor.handle_resolver(
                        HandlerMessage::Deliver {
                            span: tracing::Span::none(),
                            view,
                            data: certificate.encode(),
                            asks: asks.clone(),
                            response,
                        },
                        &mut voter,
                        &mut resolver,
                    );
                    assert_eq!(receiver.await.unwrap(), outcome);
                }

                // Nullification satisfies backfill and nullification ancestry.
                // Notarization satisfies only notarization ancestry.
                assert!(actor.state.settled(view, kind));
                let remaining = match kind {
                    Kind::Nullification => vec![Ask::ancestry(Kind::Notarization)],
                    Kind::Notarization => {
                        vec![Ask::backfill(), Ask::ancestry(Kind::Nullification)]
                    }
                };
                assert_eq!(resolver.subscriptions(view.get()), remaining);
            }
        });
    }

    #[test_async]
    async fn stale_delivery_records_new_evidence_without_credit() {
        let runtime = deterministic::Runner::default();
        runtime.start(|mut context| async move {
            let Fixture {
                schemes, verifier, ..
            } = ed25519::fixture(&mut context, NAMESPACE, 4);
            let (voter_tx, mut voter_rx) = mailbox::new(context.child("voter"), NZUsize!(8));
            let mut voter = voter::Mailbox::new(voter_tx);
            let mut actor = build_actor(context.child("actor"), verifier.clone(), TERM_LENGTH);
            let mut resolver = RecordingResolver::default();

            // Settle the delivery's notarization ask while keeping a nullification
            // ask for the same view outstanding.
            let requested = View::new(4);
            actor.updated(
                &mut resolver,
                &Certificate::Notarization(build_notarization(
                    &schemes, &verifier, EPOCH, requested,
                )),
            );
            let higher = requested.next_term_start(TERM_LENGTH);
            actor.resolve(&mut resolver, higher, requested, Kind::Nullification, None);

            // The stale delivery contains a certificate for a later view, so its
            // evidence is still useful even though the original ask is settled.
            let certificate =
                Certificate::Notarization(build_notarization(&schemes, &verifier, EPOCH, higher));
            let (response, receiver) = oneshot::channel();
            actor.handle_resolver(
                HandlerMessage::Deliver {
                    span: tracing::Span::none(),
                    view: requested,
                    data: certificate.encode(),
                    asks: non_empty_vec![Ask::ancestry(Kind::Notarization)],
                    response,
                },
                &mut voter,
                &mut resolver,
            );

            // A stale snapshot produces no scoring verdict and leaves the
            // outstanding nullification subscriber attached to the key.
            assert!(receiver.await.is_err());
            assert!(!actor.state.settled(requested, Kind::Nullification));
            assert_eq!(
                resolver.subscriptions(requested.get()),
                vec![Ask::ancestry(Kind::Nullification)]
            );

            // The new evidence must reach both the resolver state and the voter.
            assert!(actor.state.settled(higher, Kind::Notarization));
            assert!(matches!(
                voter_rx.recv().await.unwrap(),
                voter::Message::Verified { certificate: received, .. }
                    if received.encode() == certificate.encode()
            ));
        });
    }

    #[test_async]
    async fn already_certified_notarization_delivery_retires_snapshot() {
        let runtime = deterministic::Runner::default();
        runtime.start(|mut context| async move {
            let Fixture {
                schemes, verifier, ..
            } = ed25519::fixture(&mut context, NAMESPACE, 4);
            let (voter_tx, _voter_rx) = mailbox::new(context.child("voter"), NZUsize!(8));
            let mut voter = voter::Mailbox::new(voter_tx);
            let mut actor = build_actor(context, verifier.clone(), TERM_LENGTH);
            let mut resolver = RecordingResolver::default();

            let view = View::new(6);
            let notarization = build_notarization(&schemes, &verifier, EPOCH, view);
            actor.updated(
                &mut resolver,
                &Certificate::Notarization(notarization.clone()),
            );
            actor.certified(&mut resolver, notarization.clone(), true);

            // Build a second valid aggregate for the same proposal from a
            // different signer subset. Certification belongs to the proposal,
            // not to one encoding of its aggregate certificate.
            let votes: Vec<_> = schemes
                .iter()
                .take(3)
                .map(|scheme| Notarize::sign(scheme, notarization.proposal.clone()).unwrap())
                .collect();
            let alternate =
                Notarization::from_notarizes(&verifier, non_empty![@&votes], &Sequential).unwrap();
            assert_ne!(notarization, alternate);

            // Certification already settled this snapshot's parent ask, so
            // another aggregate for the same proposal cannot earn credit.
            let (response, receiver) = oneshot::channel();
            actor.handle_resolver(
                HandlerMessage::Deliver {
                    span: tracing::Span::none(),
                    view,
                    data: Certificate::<TestScheme, Sha256Digest>::Notarization(alternate).encode(),
                    asks: non_empty_vec![Ask::ancestry(Kind::Notarization)],
                    response,
                },
                &mut voter,
                &mut resolver,
            );
            assert!(receiver.await.is_err());
            assert!(!actor.state.settled(view, Kind::Nullification));
        });
    }

    #[test_async]
    async fn finalization_completes_every_delivery_ask() {
        let runtime = deterministic::Runner::default();
        runtime.start(|mut context| async move {
            let Fixture {
                schemes, verifier, ..
            } = ed25519::fixture(&mut context, NAMESPACE, 4);
            let (voter_tx, _voter_rx) = mailbox::new(context.child("voter"), NZUsize!(8));
            let mut voter = voter::Mailbox::new(voter_tx);
            let mut actor = build_actor(context.child("actor"), verifier.clone(), TERM_LENGTH);
            let mut resolver = RecordingResolver::default();

            let requested = View::new(4);
            let (response, receiver) = oneshot::channel();
            actor.handle_resolver(
                HandlerMessage::Deliver {
                    span: tracing::Span::none(),
                    view: requested,
                    data: Certificate::<TestScheme, Sha256Digest>::Finalization(
                        build_finalization(&schemes, &verifier, EPOCH, View::new(6)),
                    )
                    .encode(),
                    asks: non_empty_vec![
                        Ask::backfill(),
                        Ask::ancestry(Kind::Nullification),
                        Ask::ancestry(Kind::Notarization),
                    ],
                    response,
                },
                &mut voter,
                &mut resolver,
            );
            assert_eq!(
                receiver.await.unwrap(),
                Outcome::Resolved(Resolution::Complete)
            );
        });
    }

    #[test_async]
    async fn settled_key_ignores_delivery_without_validation() {
        let runtime = deterministic::Runner::default();
        runtime.start(|mut context| async move {
            let Fixture {
                schemes, verifier, ..
            } = ed25519::fixture(&mut context, NAMESPACE, 4);
            let (voter_tx, mut voter_rx) = mailbox::new(context.child("voter"), NZUsize!(8));
            let mut voter = voter::Mailbox::new(voter_tx);
            let mut actor = build_actor(context.child("actor"), verifier.clone(), TERM_LENGTH);
            let mut resolver = RecordingResolver::default();

            let requested = View::new(4);
            let finalization = Certificate::Finalization(build_finalization(
                &schemes,
                &verifier,
                EPOCH,
                View::new(6),
            ));
            actor.updated(&mut resolver, &finalization);
            assert!(actor.state.settled(requested, Kind::Nullification));
            assert!(actor.state.settled(requested, Kind::Notarization));

            // Garbage queued before local settlement is ignored without
            // decoding, verification, or peer penalty.
            let (response, receiver) = oneshot::channel();
            actor.handle_resolver(
                HandlerMessage::Deliver {
                    span: tracing::Span::none(),
                    view: requested,
                    data: Bytes::from_static(b"unverifiable"),
                    asks: non_empty_vec![
                        Ask::backfill(),
                        Ask::ancestry(Kind::Nullification),
                        Ask::ancestry(Kind::Notarization),
                    ],
                    response,
                },
                &mut voter,
                &mut resolver,
            );

            assert_eq!(receiver.await.unwrap(), Outcome::Ignored);
            assert!(!actor.state.settled(View::new(7), Kind::Nullification));
            assert_eq!(
                actor.state.produce(View::new(6)),
                Some(finalization.encode())
            );
            drop(voter);
            assert!(voter_rx.recv().await.is_none());
        });
    }

    #[test_async]
    async fn nullification_only_settled_key_still_validates_delivery() {
        let runtime = deterministic::Runner::default();
        runtime.start(|mut context| async move {
            let Fixture {
                schemes, verifier, ..
            } = ed25519::fixture(&mut context, NAMESPACE, 4);
            let (voter_tx, _voter_rx) = mailbox::new(context.child("voter"), NZUsize!(8));
            let mut voter = voter::Mailbox::new(voter_tx);
            let mut actor = build_actor(context.child("actor"), verifier.clone(), TERM_LENGTH);
            let mut resolver = RecordingResolver::default();

            let requested = View::new(4);
            actor.updated(
                &mut resolver,
                &Certificate::Nullification(build_nullification(
                    &schemes, &verifier, EPOCH, requested,
                )),
            );
            assert!(actor.state.settled(requested, Kind::Nullification));
            assert!(!actor.state.settled(requested, Kind::Notarization));

            let (response, receiver) = oneshot::channel();
            actor.handle_resolver(
                HandlerMessage::Deliver {
                    span: tracing::Span::none(),
                    view: requested,
                    data: Bytes::from_static(b"unverifiable"),
                    asks: non_empty_vec![Ask::ancestry(Kind::Nullification)],
                    response,
                },
                &mut voter,
                &mut resolver,
            );

            assert_eq!(receiver.await.unwrap(), Outcome::Invalid);
        });
    }

    #[test_async]
    async fn notarization_only_settled_key_still_validates_delivery() {
        let runtime = deterministic::Runner::default();
        runtime.start(|mut context| async move {
            let Fixture {
                schemes, verifier, ..
            } = ed25519::fixture(&mut context, NAMESPACE, 4);
            let (voter_tx, _voter_rx) = mailbox::new(context.child("voter"), NZUsize!(8));
            let mut voter = voter::Mailbox::new(voter_tx);
            let mut actor = build_actor(context.child("actor"), verifier.clone(), TERM_LENGTH);
            let mut resolver = RecordingResolver::default();

            let requested = View::new(4);
            actor.updated(
                &mut resolver,
                &Certificate::Notarization(build_notarization(
                    &schemes, &verifier, EPOCH, requested,
                )),
            );
            assert!(!actor.state.settled(requested, Kind::Nullification));
            assert!(actor.state.settled(requested, Kind::Notarization));

            let (response, receiver) = oneshot::channel();
            actor.handle_resolver(
                HandlerMessage::Deliver {
                    span: tracing::Span::none(),
                    view: requested,
                    data: Bytes::from_static(b"unverifiable"),
                    asks: non_empty_vec![Ask::ancestry(Kind::Notarization)],
                    response,
                },
                &mut voter,
                &mut resolver,
            );

            assert_eq!(receiver.await.unwrap(), Outcome::Invalid);
        });
    }

    #[test_async]
    async fn validate_accepts_nullification_covering_requested_view_in_term() {
        let runtime = deterministic::Runner::default();
        runtime.start(|mut context| async move {
            let Fixture {
                schemes, verifier, ..
            } = ed25519::fixture(&mut context, NAMESPACE, 4);
            let nullification = build_nullification(&schemes, &verifier, EPOCH, View::new(6));
            assert!(View::new(6).same_term(View::new(10), TERM_LENGTH));
            let mut actor = build_actor(context, verifier, TERM_LENGTH);

            let validated = actor.validate(
                View::new(10),
                Certificate::<TestScheme, Sha256Digest>::Nullification(nullification.clone())
                    .encode(),
            );

            assert!(matches!(
                validated,
                Some(Certificate::Nullification(parsed)) if parsed.view() == nullification.view()
            ));
        });
    }

    #[test_async]
    async fn validate_rejects_nullification_from_different_term() {
        let runtime = deterministic::Runner::default();
        runtime.start(|mut context| async move {
            let Fixture {
                schemes, verifier, ..
            } = ed25519::fixture(&mut context, NAMESPACE, 4);
            let mut actor = build_actor(context, verifier.clone(), TERM_LENGTH);
            let nullification = build_nullification(&schemes, &verifier, EPOCH, View::new(10));

            let validated = actor.validate(
                View::new(11),
                Certificate::<TestScheme, Sha256Digest>::Nullification(nullification).encode(),
            );

            assert!(validated.is_none());
        });
    }

    #[test_async]
    async fn validate_rejects_nullification_above_requested_view() {
        let runtime = deterministic::Runner::default();
        runtime.start(|mut context| async move {
            let Fixture {
                schemes, verifier, ..
            } = ed25519::fixture(&mut context, NAMESPACE, 4);
            let mut actor = build_actor(context, verifier.clone(), TERM_LENGTH);
            let nullification = build_nullification(&schemes, &verifier, EPOCH, View::new(9));

            let validated = actor.validate(
                View::new(8),
                Certificate::<TestScheme, Sha256Digest>::Nullification(nullification).encode(),
            );

            assert!(validated.is_none());
        });
    }

    #[test_async]
    async fn failed_verdict_survives_higher_certified_floor() {
        let runtime = deterministic::Runner::default();
        runtime.start(|mut context| async move {
            let Fixture {
                schemes, verifier, ..
            } = ed25519::fixture(&mut context, NAMESPACE, 4);
            let mut actor = build_actor(context, verifier.clone(), TERM_LENGTH);
            let mut resolver = RecordingResolver::default();
            let failed = View::new(7);
            let failed_notarization = build_notarization(&schemes, &verifier, EPOCH, failed);
            actor.updated(
                &mut resolver,
                &Certificate::Notarization(failed_notarization.clone()),
            );
            actor.certified(&mut resolver, failed_notarization, false);

            let floor = View::new(9);
            let floor_notarization = build_notarization(&schemes, &verifier, EPOCH, floor);
            actor.updated(
                &mut resolver,
                &Certificate::Notarization(floor_notarization.clone()),
            );
            actor.certified(&mut resolver, floor_notarization.clone(), true);

            // The tombstone survives the higher floor, which is served in place
            // of the failed notarization.
            let notarization = build_notarization(&schemes, &verifier, EPOCH, failed);
            assert!(actor.state.uncertifiable(failed));
            assert_eq!(
                actor.state.produce(failed),
                Some(Certificate::Notarization(floor_notarization).encode())
            );
            assert!(
                actor
                    .validate(failed, Certificate::Notarization(notarization).encode())
                    .is_some(),
                "failed certification does not make signed protocol evidence invalid",
            );
        });
    }

    #[test_async]
    async fn validate_rejects_finalization_from_different_epoch() {
        let runtime = deterministic::Runner::default();
        runtime.start(|mut context| async move {
            let Fixture {
                schemes, verifier, ..
            } = ed25519::fixture(&mut context, NAMESPACE, 4);
            let mut actor = build_actor(context, verifier.clone(), TERM_LENGTH);
            let finalization =
                build_finalization(&schemes, &verifier, Epoch::new(10), View::new(7));

            let validated = actor.validate(
                View::new(7),
                Certificate::Finalization(finalization).encode(),
            );

            assert!(validated.is_none());
        });
    }
}