use super::{
    Config, Mailbox,
    ingress::Message,
    state::{CertificateFetch, Config as StateConfig, ProposalRequest, State, Verify},
};
use crate::{
    CertifiableAutomaton, Handoff, LATENCY, Relay, Reporter, Viewable,
    simplex::{
        Floor, Plan,
        actors::{Kind, batcher, resolver},
        elector::Elector,
        metrics::{
            self, HandoffAbandoned, HandoffAbandonedReason, HandoffEvent, HandoffEventKind,
            Outbound, TimeoutReason,
        },
        scheme::Scheme,
        types::{
            Activity, Artifact, Certificate, Context, Finalization, Finalize, Notarization,
            Notarize, Nullification, Nullify, Vote,
        },
    },
    types::{Round as Rnd, View},
};
use commonware_actor::mailbox;
use commonware_codec::Read;
use commonware_cryptography::Digest;
use commonware_macros::select_loop;
use commonware_p2p::{Blocker, Recipients, Sender, utils::codec::WrappedSender};
use commonware_runtime::{
    Clock, ContextCell, Handle, Metrics, ReadOptions, Spawner, Storage,
    buffer::paged::CacheRef,
    spawn_cell,
    telemetry::{
        metrics::{CounterFamily, Histogram, MetricsExt as _},
        traces::TracedExt as _,
    },
};
use commonware_storage::journal::segmented::variable::{Config as JConfig, Journal};
use commonware_utils::{
    channel::oneshot,
    futures::{AbortablePool, rebind},
};
use core::{
    future::{Future, pending},
    panic,
};
use rand_core::CryptoRng;
use std::{
    num::NonZeroUsize,
    pin::Pin,
    task::{self, Poll},
};
use tracing::{Instrument as _, Span, debug, info, info_span, trace, warn};

/// Messages built and recorded during an event loop iteration, staged for
/// broadcast after the journal sync barrier (see [Actor::construct] and
/// [Actor::notify]).
#[allow(clippy::type_complexity)]
struct Staged<S: Scheme<D>, D: Digest> {
    nullify: Option<(Nullify<S>, Option<Certificate<S, D>>)>,
    certification: Option<(bool, Notarization<S, D>)>,
    notarize: Option<Notarize<S, D>>,
    notarization: Option<Notarization<S, D>>,
    nullification: Option<Nullification<S>>,
    finalize: Option<Finalize<S, D>>,
    finalization: Option<Finalization<S, D>>,
}

/// An outstanding request to the automaton.
struct Request<V: Viewable, F>(
    /// Attached context for the pending item. Must yield a view.
    V,
    /// Span tracking the request from issuance to processed response.
    Span,
    /// Pending response, or the retained state of a handoff request.
    F,
);

impl<V: Viewable, F> Viewable for Request<V, F> {
    fn view(&self) -> View {
        self.0.view()
    }
}

/// Adapter that polls an [Option<Request<V, F>>] in place.
struct Waiter<'a, V: Viewable, F>(&'a mut Option<Request<V, F>>);

impl<'a, V: Viewable, F: Future + Unpin> Future for Waiter<'a, V, F> {
    type Output = (V, Span, F::Output);

    fn poll(self: Pin<&mut Self>, cx: &mut task::Context<'_>) -> Poll<Self::Output> {
        let Waiter(slot) = self.get_mut();
        let res = match slot.as_mut() {
            Some(Request(_, _, receiver)) => match Pin::new(receiver).poll(cx) {
                Poll::Ready(res) => res,
                Poll::Pending => return Poll::Pending,
            },
            None => return Poll::Pending,
        };
        let Request(v, span, _) = slot.take().expect("request must exist");
        Poll::Ready((v, span, res))
    }
}

/// Unified response from a prepare or propose request.
enum ProposalResponse<D> {
    /// The response to a handoff request.
    Prepared(Handoff<D>),
    /// An ordinary candidate.
    Proposed(D),
    /// A held candidate released by parent certification. It was relayed when held, so only
    /// the [`Plan::Propose`] broadcast and the vote remain.
    Released(D),
}

/// Lifecycle of the pending proposal slot.
enum ProposalState<D> {
    /// The automaton is answering a handoff request.
    Preparing(oneshot::Receiver<Handoff<D>>),
    /// The automaton is answering an ordinary proposal request.
    Proposing(oneshot::Receiver<D>),
    /// A handoff without a pending build: the application responded with [`Handoff::Wait`],
    /// or we cancelled its build (see [`State::pending_handoff_abandonment`]). Once the exact
    /// parent certifies or finalizes, an ordinary request for the same context follows, unless
    /// we voted to nullify the request's view. A replacement parent can supersede it first.
    Waiting,
    /// A relayed candidate whose proposal broadcast and notarize vote require parent
    /// certification.
    ///
    /// Once its parent has certified, the select loop consumes the result only
    /// after the journal sync that follows that certification.
    Held(D),
    /// A handoff whose response closed. It forfeits the view once the exact parent
    /// certifies or finalizes, unless a replacement parent supersedes it first.
    Closed,
}

/// The proposal request pending or retained for the current or a future view, if any.
type PendingProposal<D, P> = Option<Request<ProposalRequest<D, P>, ProposalState<D>>>;

/// The verification request in flight for the current or an optimistic future view, if any.
type PendingVerification = Option<Request<Rnd, oneshot::Receiver<bool>>>;

/// Actor responsible for driving participation in the consensus protocol.
pub struct Actor<
    E: Clock + CryptoRng + Spawner + Storage + Metrics,
    S: Scheme<D>,
    L: Elector<S>,
    B: Blocker<PublicKey = S::PublicKey>,
    D: Digest,
    A: CertifiableAutomaton<Digest = D, Context = Context<D, S::PublicKey>>,
    R: Relay,
    F: Reporter<Activity = Activity<S, D>>,
> {
    context: ContextCell<E>,
    state: State<E, S, L, D>,
    blocker: B,
    automaton: A,
    relay: R,
    reporter: F,
    floor: Option<Floor<S, D>>,

    certificate_config: <S::Certificate as Read>::Cfg,
    partition: String,
    replay_buffer: NonZeroUsize,
    write_buffer: NonZeroUsize,
    page_cache: CacheRef,
    journal: Option<Journal<E, Artifact<S, D>>>,
    dirty_section: Option<View>,

    mailbox_receiver: mailbox::Receiver<Message<S, D>>,

    outbound_messages: CounterFamily<Outbound>,
    handoff_events: CounterFamily<HandoffEvent>,
    handoff_abandoned: CounterFamily<HandoffAbandoned>,
    notarization_latency: Histogram,
    finalization_latency: Histogram,
    notarization_latency_from_view_entry: Histogram,
    finalization_latency_from_view_entry: Histogram,
}

impl<
    E: Clock + CryptoRng + Spawner + Storage + Metrics,
    S: Scheme<D>,
    L: Elector<S>,
    B: Blocker<PublicKey = S::PublicKey>,
    D: Digest,
    A: CertifiableAutomaton<Digest = D, Context = Context<D, S::PublicKey>>,
    R: Relay<Digest = D, PublicKey = S::PublicKey, Plan = Plan<S::PublicKey>>,
    F: Reporter<Activity = Activity<S, D>>,
> Actor<E, S, L, B, D, A, R, F>
{
    pub fn new(context: E, cfg: Config<S, L, B, D, A, R, F>) -> (Self, Mailbox<S, D>) {
        // Initialize metrics
        let outbound_messages = context.family("outbound_messages", "number of outbound messages");
        let handoff_events = context.family(
            "handoff_events",
            "number of handoff lifecycle events, where one request can count several",
        );
        let handoff_abandoned = context.family(
            "handoff_abandoned",
            "number of handoff requests, candidates, and builds abandoned before the proposer votes",
        );
        let notarization_latency =
            context.histogram("notarization_latency", "notarization latency", LATENCY);
        let finalization_latency =
            context.histogram("finalization_latency", "finalization latency", LATENCY);

        // A certificate that is ready before this node enters its view has no entry time, so
        // the from-entry histograms take no sample for it.
        let notarization_latency_from_view_entry = context.histogram(
            "notarization_latency_from_view_entry",
            "leader seconds from first local view entry to local notarization certificate readiness",
            LATENCY,
        );
        let finalization_latency_from_view_entry = context.histogram(
            "finalization_latency_from_view_entry",
            "leader seconds from first local view entry to local finalization certificate readiness",
            LATENCY,
        );

        // Initialize store
        let (mailbox_sender, mailbox_receiver) =
            mailbox::new(context.child("mailbox"), cfg.mailbox_size);
        let mailbox = Mailbox::new(mailbox_sender);
        let certificate_config = cfg.scheme.certificate_codec_config();
        let state = State::new(
            context.child("state"),
            StateConfig {
                scheme: cfg.scheme,
                elector: cfg.elector,
                epoch: cfg.epoch,
                view_retention: cfg.view_retention,
                leader_timeout: cfg.leader_timeout,
                certification_timeout: cfg.certification_timeout,
                timeout_retry: cfg.timeout_retry,
                skip_budget: cfg.skip_budget,
            },
        );
        (
            Self {
                context: ContextCell::new(context),
                state,
                blocker: cfg.blocker,
                automaton: cfg.automaton,
                relay: cfg.relay,
                reporter: cfg.reporter,
                floor: Some(cfg.floor),

                certificate_config,
                partition: cfg.partition,
                replay_buffer: cfg.replay_buffer,
                write_buffer: cfg.write_buffer,
                page_cache: cfg.page_cache,
                journal: None,
                dirty_section: None,

                mailbox_receiver,

                outbound_messages,
                handoff_events,
                handoff_abandoned,
                notarization_latency,
                finalization_latency,
                notarization_latency_from_view_entry,
                finalization_latency_from_view_entry,
            },
            mailbox,
        )
    }

    /// Returns the wall-clock seconds since we proposed in `view` (or entered
    /// it, when we led without a local proposal). None when we are not the
    /// view's leader or never started it (no meaningful sample).
    fn leader_elapsed(&self, view: View) -> Option<f64> {
        let elapsed = self.state.elapsed_since_start(view)?;
        let leader = self.state.leader_index(view)?;
        if !self.state.is_me(leader) {
            return None;
        }
        Some(elapsed.as_secs_f64())
    }

    /// Drops views and journal entries that are below the activity floor.
    async fn prune_views(mut self) -> Self {
        let removed = self.state.prune();
        if removed.is_empty() {
            return self;
        }
        for view in &removed {
            debug!(
                %view,
                last_finalized = %self.state.last_finalized(),
                "pruned view"
            );
        }
        let min_active = self.state.min_active();
        if self.journal.is_some() {
            let span = info_span!(
                "simplex.voter.journal.prune",
                epoch = self.state.epoch().traced(),
                min = min_active.traced()
            );
            rebind(&mut self.journal, |journal| {
                journal.prune(min_active.get()).instrument(span)
            })
            .await
            .expect("unable to prune journal");
        }
        self
    }

    /// Appends a verified message to the journal.
    ///
    /// The append is not immediately durable. All appends in an event loop
    /// iteration target the view being processed and are synced together by
    /// [Self::sync_journal].
    ///
    /// Inert until replay attaches the journal, so handlers called during
    /// replay do not re-journal the artifacts they restore.
    async fn append_journal(mut self, artifact: Artifact<S, D>) -> Self {
        if self.journal.is_some() {
            let view = artifact.view();
            rebind(&mut self.journal, |journal| {
                journal.append(view.get(), &artifact)
            })
            .await
            .expect("unable to append to journal");
            match self.dirty_section {
                Some(dirty) => assert_eq!(
                    dirty, view,
                    "one voter iteration must append to a single journal section"
                ),
                None => self.dirty_section = Some(view),
            }
        }
        self
    }

    /// Syncs the journal section written by this iteration, if any.
    ///
    /// Called after construction and before publication so every appended artifact
    /// is durable by the end of the iteration. Proposal builds may start before
    /// this sync, but the voter consumes their responses and dispatches child
    /// certification only in the next iteration, so a durable child certification
    /// implies its parent anchor is durable. A single sync coalesces all appends.
    async fn sync_journal(mut self) -> Self {
        let Some(view) = self.dirty_section else {
            return self;
        };
        let span = info_span!(
            "simplex.voter.journal.sync",
            epoch = self.state.epoch().traced(),
            view = view.traced()
        );
        rebind(&mut self.journal, |journal| {
            journal.sync(view.get()).instrument(span)
        })
        .await
        .expect("unable to sync journal");
        self.dirty_section = None;
        self
    }

    /// Publishes a durable local vote to the batcher and every peer.
    ///
    /// Callers must first sync pending journal appends. Otherwise a restart may
    /// forget the vote and sign a conflicting one, allowing conflicting certificates.
    fn publish_vote<T: Sender>(
        &mut self,
        batcher: &mut batcher::Mailbox<S, D>,
        sender: &mut WrappedSender<T, Vote<S, D>>,
        vote: Vote<S, D>,
    ) {
        // Every nullify retry refreshes volatile batcher state. The first retry after
        // replay may be the batcher's first copy of the vote.
        batcher.constructed(vote.clone());

        // Update outbound metrics
        let metric = match &vote {
            Vote::Notarize(_) => metrics::Outbound::notarize(),
            Vote::Nullify(_) => metrics::Outbound::nullify(),
            Vote::Finalize(_) => metrics::Outbound::finalize(),
        };
        self.outbound_messages.get_or_create(metric).inc();

        // Broadcast vote
        sender.send(Recipients::All, vote, true);
    }

    /// Send a certificate to every peer.
    ///
    /// Callers must sync pending journal appends first (via [Self::sync_journal])
    /// so any state we advertise to the network survives a restart.
    fn broadcast_certificate<T: Sender>(
        &mut self,
        sender: &mut WrappedSender<T, Certificate<S, D>>,
        certificate: Certificate<S, D>,
    ) {
        // Update outbound metrics
        let metric = match &certificate {
            Certificate::Notarization(_) => metrics::Outbound::notarization(),
            Certificate::Nullification(_) => metrics::Outbound::nullification(),
            Certificate::Finalization(_) => metrics::Outbound::finalization(),
        };
        self.outbound_messages.get_or_create(metric).inc();

        // Broadcast certificate
        sender.send(Recipients::All, certificate, true);
    }

    /// Blocks an equivocator.
    fn block_equivocator(&mut self, equivocator: Option<S::PublicKey>) {
        let Some(equivocator) = equivocator else {
            return;
        };
        commonware_p2p::block!(self.blocker, equivocator, "blocking equivocator");
    }

    /// Counts a handoff lifecycle event.
    fn record_handoff_event(&self, event: HandoffEventKind) {
        self.handoff_events
            .get_or_create(&HandoffEvent { event })
            .inc();
    }

    /// Counts a handoff request, candidate, or build abandoned before the proposer votes.
    fn record_handoff_abandoned(&self, reason: HandoffAbandonedReason) {
        self.handoff_abandoned
            .get_or_create(&HandoffAbandoned { reason })
            .inc();
    }

    /// Attempt to propose a new block.
    async fn try_propose(&mut self) -> PendingProposal<D, S::PublicKey> {
        // Check if we are ready to propose
        let request = self.state.try_propose()?;
        let context = request.context().clone();

        // Request proposal from application
        let span = info_span!(
            parent: self.state.view_span(context.view()),
            "simplex.voter.propose",
            epoch = context.round.epoch().traced(),
            view = context.view().traced()
        );
        let state = async {
            debug!(
                round = ?context.round,
                handoff = request.is_handoff(),
                "requested proposal from automaton"
            );
            match &request {
                ProposalRequest::Prepare(_) => {
                    self.record_handoff_event(HandoffEventKind::Requested);
                    ProposalState::Preparing(self.automaton.prepare(context).await)
                }
                ProposalRequest::Propose(_) => {
                    ProposalState::Proposing(self.automaton.propose(context).await)
                }
            }
        }
        .instrument(span.clone())
        .await;
        Some(Request(request, span, state))
    }

    /// Attempt to verify a proposed block.
    #[allow(clippy::async_yields_async)]
    async fn try_verify(&mut self, resolver: &mut resolver::Mailbox<S, D>) -> PendingVerification {
        // Check if we are ready to verify
        let (context, proposal) = match self.state.try_verify() {
            Verify::Ready(context, proposal) => (context, proposal),
            Verify::Resolve {
                proposal,
                view,
                kind,
                target,
            } => {
                resolver.resolve(proposal, view, kind, Some(target));
                return None;
            }
            Verify::Wait => return None,
        };

        // Request verification
        let span = info_span!(
            parent: self.state.view_span(context.view()),
            "simplex.voter.verify",
            epoch = context.round.epoch().traced(),
            view = context.view().traced()
        );
        let round = context.round;
        let receiver = async {
            debug!(?proposal, "requested proposal verification");
            self.automaton.verify(context, proposal.payload).await
        }
        .instrument(span.clone())
        .await;
        Some(Request(round, span, receiver))
    }

    /// Drops obsolete pending application requests and dispatches eligible new ones.
    async fn reconcile_application_requests(
        &mut self,
        resolver: &mut resolver::Mailbox<S, D>,
        pending_propose: &mut PendingProposal<D, S::PublicKey>,
        pending_verify: &mut PendingVerification,
    ) {
        // Retain requests for the current and optimistic future views. Drop requests for exited
        // views, and a proposal request whose captured parent is no longer valid once a
        // replacement parent is selectable. Certification of an exited view continues after its
        // verification receiver is dropped.
        let current_view = self.state.current_view();
        if let Some(request) = pending_propose.as_ref() {
            let reason = if request.view() < current_view {
                Some(HandoffAbandonedReason::ViewExit)
            } else if self.state.supersede_proposal_request(request.0.context()) {
                Some(HandoffAbandonedReason::AncestrySuperseded)
            } else {
                None
            };
            if let Some(reason) = reason {
                if request.0.is_handoff() {
                    self.record_handoff_abandoned(reason);
                }
                *pending_propose = None;
            }
        }
        if pending_verify
            .as_ref()
            .is_some_and(|request| request.view() < current_view)
        {
            *pending_verify = None;
        }

        // Cancel a pending handoff build once State::pending_handoff_abandonment reports a reason.
        // Dropping the receiver cancels the build, and the request then waits (see
        // [`ProposalState::Waiting`]).
        if let Some(Request(ProposalRequest::Prepare(context), _, state)) = pending_propose.as_mut()
            && matches!(state, ProposalState::Preparing(_))
            && let Some(reason) = self.state.pending_handoff_abandonment(context)
        {
            self.record_handoff_abandoned(reason);
            *state = ProposalState::Waiting;
        }

        // Resolve a waiting or closed handoff once its exact parent certifies or finalizes.
        //
        // A waiting handoff releases its build latch, so the ordinary request below asks for the
        // same context on the certified parent. If we voted to nullify the request's view, no
        // proposal can be recorded there, so no ordinary request follows. A closed response
        // forfeits the view.
        //
        // Certification and finalization are recorded only in iterations that end with a journal
        // sync, and responses are polled only in later iterations, so nothing built on the
        // parent is consumed before its evidence is durable.
        if let Some(Request(request, _, state)) = pending_propose.as_ref()
            && matches!(state, ProposalState::Waiting | ProposalState::Closed)
            && self.state.parent_certified(request.context().parent)
        {
            let view = request.view();
            if matches!(state, ProposalState::Closed) {
                self.record_handoff_abandoned(HandoffAbandonedReason::ResponseClosed);
                self.state
                    .trigger_timeout(view, TimeoutReason::MissingProposal);
            } else {
                if self.state.voted_nullify(view) {
                    self.record_handoff_abandoned(HandoffAbandonedReason::ViewNullify);
                }
                self.state.release_proposal_request(view);
            }
            *pending_propose = None;
        }

        // State and Round prevent duplicate requests when both checkpoints
        // observe the same view.
        if pending_propose.is_none() {
            *pending_propose = self.try_propose().await;
        }
        if pending_verify.is_none() {
            *pending_verify = self.try_verify(resolver).await;
        }
    }

    /// Handle a timeout.
    ///
    /// Builds a nullify vote for the current view (as many times as required
    /// until we exit the view) and records it on the first attempt. Returns
    /// the vote for broadcast by [Self::notify], along with the best entry
    /// certificate for the current view (on retry) to help others enter it
    /// (see [State::get_best_certificate]).
    async fn timeout(
        mut self,
        reason: TimeoutReason,
    ) -> (Self, Option<(Nullify<S>, Option<Certificate<S, D>>)>) {
        // Construct a nullify vote for the current view
        let view = self.state.current_view();
        let Some((retry, nullify)) = self.state.construct_nullify(view, reason) else {
            return (self, None);
        };

        // Persist the nullify if it is a first attempt
        if !retry {
            self = self
                .append_journal(Artifact::Nullify(nullify.clone()))
                .await;
            return (self, Some((nullify, None)));
        }

        // Include entry to help others enter the view (if on retry).
        //
        // We don't worry about recording this certificate because it must've already existed (and thus
        // we must've already broadcast and persisted it).
        let entry = self.state.get_best_certificate();
        (self, Some((nullify, entry)))
    }

    /// Tracks a verified nullification certificate and returns whether it is new.
    async fn handle_nullification(mut self, nullification: Nullification<S>) -> (Self, bool) {
        let artifact = Artifact::Nullification(nullification.clone());

        // Add verified nullification to journal
        if !self.state.add_nullification(nullification) {
            return (self, false);
        }
        (self.append_journal(artifact).await, true)
    }

    /// Records a notarization certificate, blocks any equivocating leader, and
    /// returns whether it is new.
    async fn handle_notarization(mut self, notarization: Notarization<S, D>) -> (Self, bool) {
        let artifact = Artifact::Notarization(notarization.clone());
        let (added, equivocator) = self.state.add_notarization(notarization);
        if added {
            self = self.append_journal(artifact).await;
        }
        self.block_equivocator(equivocator);
        (self, added)
    }

    /// Handles the certification of a proposal.
    ///
    /// If certification succeeds, the proposal can be used in future views. If it fails, we
    /// should nullify the view as fast as possible.
    async fn handle_certification(
        mut self,
        view: View,
        success: bool,
    ) -> (Self, Option<Notarization<S, D>>) {
        // Get the notarization before advancing state
        let Some(notarization) = self.state.certified(view, success) else {
            return (self, None);
        };

        // Record the certification result for recovery. It is synced before this
        // iteration's broadcast phase. If lost to a crash before then, certification
        // is re-requested on restart.
        let artifact = Artifact::Certification(Rnd::new(self.state.epoch(), view), success);
        self = self.append_journal(artifact).await;

        (self, Some(notarization))
    }

    /// Stores a finalization certificate, guards against leader equivocation, and
    /// returns whether it is new.
    ///
    /// The finalization is appended to the journal without an immediate sync.
    /// If a crash loses a finalization that healed the same-term finalize
    /// gate, replay restores the blocked gate (which is safe) and it heals
    /// again as soon as peers redeliver any covering finalization.
    async fn handle_finalization(mut self, finalization: Finalization<S, D>) -> (Self, bool) {
        let artifact = Artifact::Finalization(finalization.clone());
        let (added, equivocator) = self.state.add_finalization(finalization);
        if added {
            self = self.append_journal(artifact).await;
        }
        self.block_equivocator(equivocator);
        (self, added)
    }

    /// Builds and records a notarize vote when this view is ready.
    async fn prepare_notarize(mut self, view: View) -> (Self, Option<Notarize<S, D>>) {
        // Construct a notarize vote
        let Some(notarize) = self.state.construct_notarize(view) else {
            return (self, None);
        };

        // Record the vote locally before sharing it.
        self = self
            .append_journal(Artifact::Notarize(notarize.clone()))
            .await;
        (self, Some(notarize))
    }

    /// Returns the notarization certificate to broadcast, at most once per view.
    fn prepare_notarization(&mut self, view: View) -> Option<Notarization<S, D>> {
        let notarization = self.state.broadcast_notarization(view)?;

        // Record leader-local latency at certificate readiness.
        if let Some(elapsed) = self.leader_elapsed(view) {
            self.notarization_latency.observe(elapsed);
            if let Some(since_entry) = self.state.elapsed_since_entry(view) {
                self.notarization_latency_from_view_entry
                    .observe(since_entry.as_secs_f64());
            }
        }
        Some(notarization)
    }

    /// Builds and records a finalize vote if the round provides a candidate.
    async fn prepare_finalize(mut self, view: View) -> (Self, Option<Finalize<S, D>>) {
        // Construct the finalize vote.
        let Some(finalize) = self.state.construct_finalize(view) else {
            return (self, None);
        };

        // Record the vote locally before sharing it.
        self = self
            .append_journal(Artifact::Finalize(finalize.clone()))
            .await;
        (self, Some(finalize))
    }

    /// Returns the finalization certificate to broadcast, at most once per view.
    fn prepare_finalization(&mut self, view: View) -> Option<Finalization<S, D>> {
        let finalization = self.state.broadcast_finalization(view)?;

        // Only record latency if we are the current leader.
        if let Some(elapsed) = self.leader_elapsed(view) {
            self.finalization_latency.observe(elapsed);
            if let Some(since_entry) = self.state.elapsed_since_entry(view) {
                self.finalization_latency_from_view_entry
                    .observe(since_entry.as_secs_f64());
            }
        }
        Some(finalization)
    }

    /// Processes the automaton's response to a proposal request.
    ///
    /// `released` marks a held candidate whose relay was counted when it was held, so only its
    /// vote is counted here.
    ///
    /// Returns the view to notify if the proposal was recorded.
    fn process_proposed(
        &mut self,
        request: ProposalRequest<D, S::PublicKey>,
        proposed: Result<D, oneshot::error::RecvError>,
        released: bool,
    ) -> Option<View> {
        // Try to use result
        let is_handoff = request.is_handoff();
        let context = request.into_context();
        let proposed = match proposed {
            Ok(proposed) => proposed,
            Err(err) => {
                debug!(?err, round = ?context.round, "failed to propose container");
                self.state
                    .trigger_timeout(context.view(), TimeoutReason::MissingProposal);
                return None;
            }
        };

        // If we have already moved past this view, drop the response as we
        // will not broadcast it. Proposals for the current or optimistic
        // future views are kept.
        if context.view() < self.state.current_view() {
            debug!(round = ?context.round, current = ?self.state.current_view(), "dropping requested proposal");
            return None;
        }

        if !self.state.proposed(&context, proposed) {
            if is_handoff {
                self.record_handoff_abandoned(HandoffAbandonedReason::IneligibleAtRecording);
            }
            warn!(round = ?context.round, "dropped our proposal");
            return None;
        }
        let view = context.view();

        // Notify the application of the proposal. To lower view latency as
        // much as possible while preserving safety, this precedes the notarize
        // vote's journal sync: unlike votes (which can form a conflicting
        // certificate), extra payload bytes are harmless, and the worst a
        // crash can do is relay a different payload for the same round after
        // restart (see [Plan::Propose]).
        //
        // A released held candidate, already relayed under [Plan::Prepare], also reaches this
        // broadcast (see [Plan::Propose]).
        let _ = self.relay.broadcast(
            proposed,
            Plan::Propose {
                round: context.round,
            },
        );
        if is_handoff {
            let certified = self.state.parent_certified(context.parent);
            if !released {
                self.record_handoff_event(if certified {
                    HandoffEventKind::RelayedAfterCertification
                } else {
                    HandoffEventKind::RelayedBeforeCertification
                });
            }
            self.record_handoff_event(if certified {
                HandoffEventKind::VotedAfterCertification
            } else {
                HandoffEventKind::VotedBeforeCertification
            });
        }
        Some(view)
    }

    /// Processes the automaton's response to a verification request.
    ///
    /// Returns the view to notify.
    fn process_verified(
        &mut self,
        round: Rnd,
        verified: Result<bool, oneshot::error::RecvError>,
    ) -> View {
        let view = round.view();
        match verified {
            Ok(true) => {
                // Mark verification complete
                self.state.verified(view);
            }
            Ok(false) => {
                warn!(?round, "proposal failed verification");
                self.state
                    .verification_failed(view, TimeoutReason::InvalidProposal);
            }
            Err(err) => {
                debug!(?err, ?round, "failed to verify proposal");
                self.state
                    .verification_failed(view, TimeoutReason::IgnoredProposal);
            }
        };
        view
    }

    /// Processes the automaton's response to a certification request.
    ///
    /// Returns whether the round was still active (false if it was already
    /// pruned) and, if the result was recorded, the certification outcome to
    /// stage for [Self::notify].
    async fn process_certified(
        mut self,
        round: Rnd,
        certified: Result<bool, oneshot::error::RecvError>,
    ) -> (Self, bool, Option<(bool, Notarization<S, D>)>) {
        // Unlike propose/verify (where failing to act will lead to a timeout
        // and subsequent nullification), failing to certify can lead to a halt
        // because we'll never exit the view without a notarization + certification.
        //
        // We do not assume failure here because we recover on restart: a synced
        // certification result is replayed from the journal and a missing one
        // causes certification to be re-requested.
        let certified = match certified {
            Ok(certified) => certified,
            Err(err) => {
                debug!(?err, ?round, "failed to certify proposal");
                return (self, true, None);
            }
        };
        if !certified {
            warn!(?round, "proposal failed certification");
        }
        let notarization;
        (self, notarization) = self.handle_certification(round.view(), certified).await;
        let Some(notarization) = notarization else {
            return (self, false, None);
        };
        (self, true, Some((certified, notarization)))
    }

    /// Processes a message from the resolver or batcher.
    ///
    /// Returns the view to notify.
    async fn process_message(
        mut self,
        resolver: &mut resolver::Mailbox<S, D>,
        msg: Message<S, D>,
    ) -> (Self, Option<View>) {
        match msg {
            Message::Proposal { proposal, .. } => {
                let view = proposal.view();
                if !self.state.admits_vote(view) {
                    trace!(%view, "proposal outside viewport");
                    return (self, None);
                }
                trace!(%view, "received proposal");
                if !self.state.set_proposal(view, &proposal) {
                    return (self, None);
                }
                (self, Some(view))
            }
            Message::Verified { certificate, .. } => {
                // Certificates can come from future views (they advance our view)
                let view = certificate.view();
                if !self.state.admits_certificate(view) {
                    trace!(%view, "certificate outside viewport");
                    return (self, None);
                }

                // Record the certificate, keeping a copy to report if it is new.
                // This includes certificates the resolver delivered, which it
                // accepts again (see `resolver::Mailbox::updated`).
                let forward = certificate.clone();
                let added;
                match certificate {
                    Certificate::Notarization(notarization) => {
                        trace!(%view, "received notarization");
                        (self, added) = self.handle_notarization(notarization).await;
                    }
                    Certificate::Nullification(nullification) => {
                        trace!(%view, "received nullification");
                        (self, added) = self.handle_nullification(nullification).await;
                    }
                    Certificate::Finalization(finalization) => {
                        trace!(%view, "received finalization");
                        (self, added) = self.handle_finalization(finalization).await;
                    }
                }
                if added {
                    resolver.updated(forward);
                }
                (self, Some(view))
            }
            Message::Timeout { round, reason, .. } => {
                let view = round.view();
                debug!(%view, ?reason, "timing out view");
                self.state.trigger_timeout(view, reason);
                (self, Some(view))
            }
        }
    }

    /// Builds and records any votes that became available for `view`, and takes
    /// any certificates for `view` not yet broadcast.
    ///
    /// Returned artifacts must be synced through [Self::sync_journal] before
    /// [Self::notify] publishes them.
    ///
    /// We don't need to iterate over all views to check for new actions because messages we receive
    /// only affect a single view. In particular, healing the same-term finalize gate does not
    /// proactively retry finalize votes for views certified while the gate was blocked: such a view
    /// only emits its vote if a later message touches it again (see the module documentation on
    /// same-term vote safety for the consequences when none arrives).
    async fn construct(mut self, view: View) -> (Self, Staged<S, D>) {
        let (notarize, finalize);
        (self, notarize) = self.prepare_notarize(view).await;
        let notarization = self.prepare_notarization(view);
        let nullification = self.state.broadcast_nullification(view);
        (self, finalize) = self.prepare_finalize(view).await;
        let finalization = self.prepare_finalization(view);
        (
            self,
            Staged {
                nullify: None,
                certification: None,
                notarize,
                notarization,
                nullification,
                finalize,
                finalization,
            },
        )
    }

    /// Publishes the staged votes and certificates.
    ///
    /// Callers must first sync pending journal appends so locally constructed votes
    /// are durable before reaching the batcher, reporter, or network.
    fn notify<Sp: Sender, Sr: Sender>(
        &mut self,
        batcher: &mut batcher::Mailbox<S, D>,
        resolver: &mut resolver::Mailbox<S, D>,
        vote_sender: &mut WrappedSender<Sp, Vote<S, D>>,
        certificate_sender: &mut WrappedSender<Sr, Certificate<S, D>>,
        staged: Staged<S, D>,
    ) {
        assert!(
            self.dirty_section.is_none(),
            "journal must be synced before broadcast"
        );

        if let Some((certified, notarization)) = staged.certification {
            // Always forward certification outcomes to resolver. This can happen
            // after a nullification for the same view because certification is
            // asynchronous; finalization is the boundary that cancels in-flight
            // certification and suppresses late reporting.
            resolver.certified(notarization.clone(), certified);
            if certified {
                self.reporter.report(Activity::Certification(notarization));
            }
        }

        if let Some((nullify, entry)) = staged.nullify {
            debug!(round=?nullify.round(), "broadcasting nullify");
            self.publish_vote(batcher, vote_sender, Vote::Nullify(nullify));

            // Broadcast entry to help others enter the view (if on retry).
            if let Some(entry) = entry {
                self.broadcast_certificate(certificate_sender, entry);
            }
        }
        if let Some(notarize) = staged.notarize {
            debug!(proposal=?notarize.proposal, "broadcasting notarize");
            self.publish_vote(batcher, vote_sender, Vote::Notarize(notarize));
        }
        if let Some(notarization) = staged.notarization {
            debug!(proposal=?notarization.proposal, "broadcasting notarization");
            self.broadcast_certificate(
                certificate_sender,
                Certificate::Notarization(notarization.clone()),
            );
            self.reporter.report(Activity::Notarization(notarization));
        }
        if let Some(nullification) = staged.nullification {
            debug!(round=?nullification.round(), "broadcasting nullification");
            self.broadcast_certificate(
                certificate_sender,
                Certificate::Nullification(nullification.clone()),
            );
            self.reporter.report(Activity::Nullification(nullification));
        }
        if let Some(finalize) = staged.finalize {
            debug!(proposal=?finalize.proposal, "broadcasting finalize");
            self.publish_vote(batcher, vote_sender, Vote::Finalize(finalize));
        }
        if let Some(finalization) = staged.finalization {
            debug!(proposal=?finalization.proposal, "broadcasting finalization");
            self.broadcast_certificate(
                certificate_sender,
                Certificate::Finalization(finalization.clone()),
            );
            self.reporter.report(Activity::Finalization(finalization));
        }
    }

    /// Spawns the actor event loop with the provided channels.
    pub fn start(
        mut self,
        batcher: batcher::Mailbox<S, D>,
        resolver: resolver::Mailbox<S, D>,
        vote_sender: impl Sender<PublicKey = S::PublicKey>,
        certificate_sender: impl Sender<PublicKey = S::PublicKey>,
    ) -> Handle<()> {
        spawn_cell!(
            self.context,
            self.run(batcher, resolver, vote_sender, certificate_sender)
        )
    }

    /// Core event loop that drives proposal, voting, networking, and recovery.
    async fn run(
        mut self,
        mut batcher: batcher::Mailbox<S, D>,
        mut resolver: resolver::Mailbox<S, D>,
        vote_sender: impl Sender<PublicKey = S::PublicKey>,
        certificate_sender: impl Sender<PublicKey = S::PublicKey>,
    ) {
        // Wrap channels
        let mut vote_sender = WrappedSender::new(vote_sender);
        let mut certificate_sender = WrappedSender::new(certificate_sender);

        // Initialize journal
        let journal = Journal::<_, Artifact<S, D>>::init(
            self.context.child("journal"),
            JConfig {
                partition: self.partition.clone(),
                compression: None, // most of the data is not compressible
                codec_config: self.certificate_config.clone(),
                page_cache: self.page_cache.clone(),
                write_buffer: self.write_buffer,
            },
        )
        .await
        .expect("unable to open journal");

        // Add initial view from the configured floor. Replay skips all
        // artifacts at or below the floor's view.
        let floor = self.floor.take().expect("floor not initialized");
        let replay_floor = floor.view();

        // Anchor all startup work under a single root span. The floor
        // finalization and journal replay both run here before any view span
        // exists, so without this root their work would emit as orphan traces.
        let start = self.context.current();
        let epoch = self.state.epoch();
        let start_span = info_span!("simplex.voter.start", epoch = epoch.traced());

        // Apply the configured floor, forwarding and reporting any finalization.
        // The finalization is journaled after replay unless the journal already
        // holds a finalization at or above it, so a later restart with a lower
        // floor does not roll the voter back.
        let mut floor_finalization = start_span.in_scope(|| {
            let finalization = self.state.set_floor(floor)?;
            resolver.updated(Certificate::Finalization(finalization.clone()));
            self.reporter
                .report(Activity::Finalization(finalization.clone()));
            Some(finalization)
        });

        // Rebuild from journal, nested under the startup span.
        // Replayed artifacts become in-memory state, so journal pages need not
        // remain in the OS page cache.
        let replayed;
        (self, replayed) = async {
            let mut replay = journal
                .replay(0, 0, self.replay_buffer, ReadOptions::DONT_CACHE)
                .await
                .expect("unable to replay journal");
            while let Some(artifact) = replay.next().await {
                let (_, _, _, artifact) = artifact.expect("unable to replay journal");

                // A journaled finalization at or above the floor already anchors later restarts
                // there, so the configured floor need not be journaled.
                if matches!(artifact, Artifact::Finalization(_)) && artifact.view() >= replay_floor
                {
                    floor_finalization = None;
                }

                // Dropping our own nullify votes at or below the floor is safe
                // for the same-term finalize gate: the floor finalization
                // covers any such vote (it lies between the vote and any later
                // same-term view), so the gate would treat it as healed anyway.
                // If the gate ever stops keying off last_finalized, this skip
                // must be revisited.
                if artifact.view() <= replay_floor {
                    continue;
                }

                self.state.replay(&artifact);
                match artifact {
                    Artifact::Notarize(notarize) => {
                        self.reporter.report(Activity::Notarize(notarize));
                    }
                    Artifact::Notarization(notarization) => {
                        (self, _) = self.handle_notarization(notarization.clone()).await;
                        resolver.updated(Certificate::Notarization(notarization.clone()));
                        self.reporter.report(Activity::Notarization(notarization));
                    }
                    Artifact::Certification(round, success) => {
                        let notarization;
                        (self, notarization) =
                            self.handle_certification(round.view(), success).await;
                        let Some(notarization) = notarization else {
                            continue;
                        };
                        resolver.certified(notarization.clone(), success);
                        if success {
                            self.reporter.report(Activity::Certification(notarization));
                        }
                    }
                    Artifact::Nullify(nullify) => {
                        self.reporter.report(Activity::Nullify(nullify));
                    }
                    Artifact::Nullification(nullification) => {
                        (self, _) = self.handle_nullification(nullification.clone()).await;
                        resolver.updated(Certificate::Nullification(nullification.clone()));
                        self.reporter.report(Activity::Nullification(nullification));
                    }
                    Artifact::Finalize(finalize) => {
                        self.reporter.report(Activity::Finalize(finalize));
                    }
                    Artifact::Finalization(finalization) => {
                        (self, _) = self.handle_finalization(finalization.clone()).await;
                        resolver.updated(Certificate::Finalization(finalization.clone()));
                        self.reporter.report(Activity::Finalization(finalization));
                    }
                }

                // We deliberately avoid re-seeding the batcher with our
                // own votes (or the votes of other peers) on replay. We assume that
                // whatever view we were in during shutdown is no longer the latest
                // and we'll quickly jump ahead to a new view.
                //
                // If this is not the case (cluster-wide shutdown), we will recover
                // when timing out.
            }
            (self, replay.finish().expect("unable to replay journal"))
        }
        .instrument(info_span!(parent: &start_span, "simplex.voter.replay", epoch = epoch.traced()))
        .await;
        self.journal = Some(replayed);

        // Every recovered certificate has been forwarded, so the resolver can
        // start background repair above the recovered floor.
        resolver.replayed();
        if let Some(finalization) = floor_finalization {
            self = async {
                self.append_journal(Artifact::Finalization(finalization))
                    .await
                    .sync_journal()
                    .await
            }
            .instrument(start_span)
            .await;
        }

        // Log current view after recovery
        let end = self.context.current();
        let elapsed = end.duration_since(start).unwrap_or_default();
        let observed_view = self.state.current_view();
        info!(
            %observed_view,
            ?elapsed,
            "consensus initialized"
        );

        // Initialize batcher with leader for current view
        let leader = self
            .state
            .leader_index(observed_view)
            .expect("leader not set");
        batcher.update(
            self.state.view_span(observed_view).clone(),
            observed_view,
            leader,
            self.state.last_finalized(),
            None,
        );

        // Process messages
        let mut pending_propose: PendingProposal<D, S::PublicKey> = None;
        let mut pending_verify: PendingVerification = None;
        let mut certify_pool = AbortablePool::default();
        select_loop! {
            self.context,
            on_start => {
                // Reconcile application requests before building this iteration's
                // response waiters.
                self.reconcile_application_requests(
                    &mut resolver,
                    &mut pending_propose,
                    &mut pending_verify,
                )
                .await;

                // Attempt to certify any views that we have notarizations for.
                //
                // Even our own proposals are certified through the automaton: that
                // is the durability barrier that makes a block recoverable before
                // we cast a finalize vote for it. Because the prior iteration's
                // journal sync completed before this block runs, a child made
                // eligible by its parent cannot become durable first.
                let (candidates, fetches) = self.state.certify_candidates();
                for CertificateFetch { proposal, view } in fetches {
                    resolver.resolve(proposal, view, Kind::Notarization, None);
                }
                for proposal in candidates {
                    let round = proposal.round;
                    let view = round.view();
                    debug!(%view, "attempting certification");
                    let span = info_span!(
                        parent: self.state.view_span(view),
                        "simplex.voter.certify",
                        epoch = round.epoch().traced(),
                        view = view.traced()
                    );
                    #[allow(clippy::async_yields_async)]
                    let receiver = async { self.automaton.certify(round, proposal.payload).await }
                        .instrument(span.clone())
                        .await;
                    let handle = certify_pool.push(async move { (round, span, receiver.await) });
                    self.state.set_certify_handle(view, handle);
                }

                // Prune views below the activity floor. To lower view latency,
                // this runs after the automaton dispatches above so pruning
                // overlaps proposal building and verification instead of
                // delaying them.
                self = self.prune_views().await;

                // Prepare waiters
                let ready = matches!(
                    pending_propose.as_ref(),
                    Some(Request(request, _, ProposalState::Held(_)))
                        if self.state.parent_certified(request.context().parent)
                );
                let propose_wait = async {
                    let proposed = match pending_propose.as_mut() {
                        Some(Request(_, _, ProposalState::Preparing(receiver))) => {
                            receiver.await.map(ProposalResponse::Prepared)
                        }
                        Some(Request(_, _, ProposalState::Proposing(receiver))) => {
                            receiver.await.map(ProposalResponse::Proposed)
                        }
                        Some(Request(_, _, ProposalState::Held(payload))) if ready => {
                            Ok(ProposalResponse::Released(*payload))
                        }
                        _ => pending().await,
                    };
                    let Request(request, span, _) =
                        pending_propose.take().expect("request must exist");
                    (request, span, proposed)
                };
                let verify_wait = Waiter(&mut pending_verify);
                let certify_wait = certify_pool.next_completed();

                // Wait for a timeout to fire or for a message to arrive
                let (deadline, reason) = self.state.next_timeout();
                let start = self.state.current_view();
                let mut nullify = None;
                let mut certification = None;
                let view;
            },
            on_stopped => {
                debug!("context shutdown, stopping voter");
            },
            _ = self.context.sleep_until(deadline) => {
                // Process the timeout (the constructed nullify is staged for the broadcast phase)
                let current_view = self.state.current_view();
                let span = info_span!(
                    parent: self.state.view_span(current_view),
                    "simplex.voter.timeout",
                    epoch = self.state.epoch().traced(),
                    view = current_view.traced(),
                    reason = reason.as_str()
                );
                (self, nullify) = self.timeout(reason).instrument(span).await;
                view = self.state.current_view();
            },
            (request, span, proposed) = propose_wait => {
                // Clear propose waiter
                pending_propose = None;

                // Retain a waiting, held, or closed handoff outside the round proposal
                // slot until its parent resolves. The captured request and build latch
                // remain active until then.
                let (proposed, released) = match proposed {
                    Ok(ProposalResponse::Prepared(Handoff::Wait)) => {
                        self.record_handoff_event(HandoffEventKind::WaitReturned);
                        pending_propose = Some(Request(request, span, ProposalState::Waiting));
                        continue;
                    }
                    Ok(ProposalResponse::Prepared(Handoff::Stage(payload)))
                        if !self.state.parent_certified(request.context().parent) =>
                    {
                        // Relay the held candidate now so its distribution overlaps parent
                        // certification. The relay names no proposal (see [Plan::Prepare]).
                        // The proposal broadcast and the vote wait for the parent.
                        self.record_handoff_event(HandoffEventKind::CandidateReturned);
                        let _ = self.relay.broadcast(
                            payload,
                            Plan::Prepare {
                                round: request.context().round,
                            },
                        );
                        self.record_handoff_event(HandoffEventKind::RelayedBeforeCertification);
                        pending_propose = Some(Request(request, span, ProposalState::Held(payload)));
                        continue;
                    }
                    Ok(ProposalResponse::Prepared(Handoff::Vote(payload) | Handoff::Stage(payload))) => {
                        self.record_handoff_event(HandoffEventKind::CandidateReturned);
                        (Ok(payload), false)
                    }
                    Ok(ProposalResponse::Proposed(payload)) => (Ok(payload), false),
                    Ok(ProposalResponse::Released(payload)) => (Ok(payload), true),
                    Err(_) if request.is_handoff() => {
                        pending_propose = Some(Request(request, span, ProposalState::Closed));
                        continue;
                    }
                    Err(err) => (Err(err), false),
                };

                // Process the automaton's response
                let Some(proposed_view) =
                    span.in_scope(|| self.process_proposed(request, proposed, released))
                else {
                    continue;
                };
                view = proposed_view;
            },
            (round, span, verified) = verify_wait => {
                // Clear verify waiter
                pending_verify = None;

                // Process the automaton's response
                view = span.in_scope(|| self.process_verified(round, verified));
            },
            // Aborted futures are expected when old views are pruned
            Ok((round, span, certified)) = certify_wait else continue => {
                // Handle response to our certification request.
                view = round.view();
                let (processed, certification_result);
                (self, processed, certification_result) = self
                    .process_certified(round, certified)
                    .instrument(span)
                    .await;
                if !processed {
                    continue;
                }
                certification = certification_result;
            },
            Some(msg) = self.mailbox_receiver.recv() else break => {
                // Handle messages from resolver and batcher
                let span = info_span!(
                    parent: msg.span(),
                    "simplex.voter.process",
                    operation = msg.name(),
                    epoch = self.state.epoch().traced(),
                    view = msg.view().traced()
                );
                let processed;
                (self, processed) = self
                    .process_message(&mut resolver, msg)
                    .instrument(span)
                    .await;
                let Some(processed_view) = processed else {
                    continue;
                };
                view = processed_view;
            },
            on_end => {
                // Attempt to send any new view messages
                //
                // The batcher buffers votes we construct here even before its update to the
                // message's view (own votes are bounded only by the retention floor) and
                // revisits them on the update sent below, so a multi-view jump delays their
                // processing. A vote for a view the batcher skipped is reported but not
                // assembled here. A peer that tracked the view may assemble its certificate,
                // and otherwise a later finalization covers it.
                let span = info_span!(
                    parent: self.state.view_span(view),
                    "simplex.voter.notify",
                    epoch = self.state.epoch().traced(),
                    view = view.traced()
                );
                self = async {
                    // Build and record everything that became available for `view`.
                    let mut staged;
                    (self, staged) = self.construct(view).await;
                    staged.nullify = nullify;
                    staged.certification = certification;

                    // A constructed notarize advances the optimistic frontier and
                    // can make child requests eligible. Start those requests before
                    // journal sync and publication. The next iteration polls their
                    // responses.
                    self.reconcile_application_requests(
                        &mut resolver,
                        &mut pending_propose,
                        &mut pending_verify,
                    ).await;

                    // Sync everything appended this iteration (during message
                    // processing and construction) in a single coalesced sync.
                    // This runs even if there is nothing to broadcast (e.g. a
                    // certification result was recorded) so every artifact is
                    // durable by the end of the iteration that appended it.
                    self = self.sync_journal().await;

                    // Broadcast everything we built (and report it to the application).
                    self.notify(
                        &mut batcher,
                        &mut resolver,
                        &mut vote_sender,
                        &mut certificate_sender,
                        staged,
                    );
                    self
                }
                .instrument(span)
                .await;

                // Close the root span of any view the chain has now decided.
                // This runs after notify so the finalization broadcast and the
                // report into the application still nest under the view span.
                self.state.close_decided_spans();

                // Update the batcher if we have moved to a new view
                let current_view = self.state.current_view();
                if current_view > start {
                    let leader = self
                        .state
                        .leader_index(current_view)
                        .expect("leader not set");

                    // Forward only the previous view's proposal. After a
                    // multi-view jump, the network has moved past anything older.
                    let forwardable_proposal = current_view
                        .previous()
                        .and_then(|view| self.state.forwardable_proposal(view));

                    // If the leader nullified or is inactive, the batcher
                    // responds with a timeout that expires the view immediately
                    batcher.update(
                        self.state.view_span(current_view).clone(),
                        current_view,
                        leader,
                        self.state.last_finalized(),
                        forwardable_proposal,
                    );
                }
            },
        }

        // Sync and drop the journal
        self.journal
            .take()
            .expect("journal missing on voter exit")
            .sync_all()
            .await
            .expect("unable to sync journal");
    }
}