use super::{
    Buffer, ExpectedCommitment, Variant,
    acks::{PendingAck, PendingAcks},
    cache,
    certified::Certified,
    delivery::PendingVerification,
    durability::{DispatchGate, Durable as _},
    floor::{Cutoff, Floor, Processed, State as FloorState, closed_epoch},
    mailbox::{CommitmentFallback, Mailbox, Message},
    staged::Staged,
    stream::Stream,
    subscriptions::{Key as SubscriptionKey, KeyFor as SubscriptionKeyFor, Subscriptions},
    variant::NoBuffer,
};
use crate::{
    Block, Epochable, Heightable, Reporter,
    marshal::{
        Config, Identifier as BlockID, Start, Update,
        resolver::handler::{self, Annotation, Key, Request},
        store::{Blocks, Certificates},
    },
    simplex::{
        scheme::Scheme,
        types::{Finalization, Notarization, Subject, verify_certificates},
    },
    types::{Epoch, EpochDelta, Epocher, Height, Round, ViewDelta},
};
use bytes::Bytes;
use commonware_actor::mailbox;
use commonware_codec::{Decode, Encode, Read};
use commonware_cryptography::{
    Digested, Digestible,
    certificate::{Provider, Scoped, Verifier},
};
use commonware_macros::{boxed, select_loop};
use commonware_p2p::Recipients;
use commonware_parallel::Strategy;
use commonware_resolver::{Delivery, Outcome, Resolution, Resolver, TargetedResolver};
use commonware_runtime::{
    Clock, ContextCell, Handle, Metrics, Spawner, Storage, spawn_cell,
    telemetry::{
        metrics::{Gauge, GaugeExt, MetricsExt as _},
        traces::TracedExt as _,
    },
};
use commonware_storage::archive::Identifier as ArchiveID;
use commonware_utils::{
    Acknowledgement, BoxedError,
    acknowledgement::Exact,
    channel::{fallible::OneshotExt, oneshot},
    futures::{AbortablePool, Pool},
};
use futures::{
    FutureExt as _, TryFutureExt as _,
    future::{self, join, join_all},
    try_join,
};
use rand_core::CryptoRng;
use std::{collections::BTreeMap, future::Future, num::NonZeroUsize};
use tracing::{Instrument as _, Span, debug, info_span, warn};

// Resolver request keys are expressed in the variant commitment type, which
// may differ from the block digest for coded variants.
type ResolverRequestFor<V> = Key<<V as Variant>::Commitment>;

// A resolver delivery plus the peer-validity response channel. Local
// annotations on the delivery decide how accepted data is used.
struct ResolverDelivery<V: Variant> {
    delivery: Delivery<ResolverRequestFor<V>, Annotation>,
    value: Bytes,
    response: oneshot::Sender<Outcome>,
}

/// Completion marker for entries in the actor's durability sync pool.
enum PooledSync {
    /// A sync that requires no action on completion.
    Observed,
    /// A finalized-archive sync batch became durable. Carries the sequence
    /// assigned by [`Actor::start_finalized_sync`] so the completion arm can
    /// release every batch the sync covers (see [`DispatchGate::release`]).
    Finalized(u64),
}

/// The [Actor] is responsible for receiving uncertified blocks from the broadcast mechanism,
/// receiving notarizations and finalizations from consensus, and reconstructing a total order
/// of blocks.
///
/// The actor is designed to be used in a view-based model. Each view corresponds to a
/// potential block in the chain. The actor will only finalize a block if it has a
/// corresponding finalization.
///
/// The actor also provides a backfill mechanism for missing blocks. If the actor receives a
/// finalization for a block that is ahead of its current view, it will request the missing blocks
/// from its peers. This ensures that the actor can catch up to the rest of the network if it falls
/// behind.
pub struct Actor<E, V, P, FC, FB, ES, T, A = Exact>
where
    E: CryptoRng + Spawner + Metrics + Clock + Storage,
    V: Variant,
    P: Provider<Scope = Epoch, Scheme: Scheme<V::Commitment>>,
    FC: Certificates<
            BlockDigest = <V::Block as Digestible>::Digest,
            Commitment = V::Commitment,
            Scheme = P::Scheme,
        >,
    FB: Blocks<Block = V::StoredBlock>,
    ES: Epocher,
    T: Strategy,
    A: Acknowledgement,
{
    context: ContextCell<E>,

    // Mailbox
    mailbox: mailbox::Receiver<Message<P::Scheme, V>>,

    // Provider for epoch-specific signing schemes
    provider: P,
    // Epoch configuration
    epocher: ES,
    // Minimum number of views to retain temporary data after the application processes a block
    view_retention: ViewDelta,
    // Maximum number of missing finalized heights fetched from peers at a time
    max_repair: NonZeroUsize,
    // Codec configuration for block type
    block_codec_config: <V::ApplicationBlock as Read>::Cfg,
    // Strategy for parallel operations
    strategy: T,

    // Durable floor and floor updates not yet applied
    floor: FloorState<P::Scheme, V::Commitment>,
    // Application delivery cursor
    stream: Stream<E>,
    // Pending application acknowledgements
    pending_acks: PendingAcks<A>,
    // Highest known finalized height
    tip: Height,
    // Outstanding subscriptions for blocks
    block_subscriptions: Subscriptions<V>,
    // Commitments known certified above the finalized tip
    certified: Certified<V::Commitment>,
    // Defers application dispatch of finalized-archive writes until a sync
    // covering them completes
    dispatch_gate: DispatchGate,
    // Finalized blocks awaiting dispatch, capped at twice the pending-ack
    // capacity to absorb finalization bursts while earlier blocks are processed
    staged: Staged<V::Block>,

    // Prunable cache
    cache: cache::Manager<E, V, P::Scheme>,
    // Finalizations stored by height
    finalizations_by_height: FC,
    // Finalized blocks stored by height
    finalized_blocks: FB,

    // Latest height metric
    finalized_height: Gauge,
    // Latest processed height
    processed_height: Gauge,
}

impl<E, V, P, FC, FB, ES, T, A> Actor<E, V, P, FC, FB, ES, T, A>
where
    E: CryptoRng + Spawner + Metrics + Clock + Storage,
    V: Variant,
    P: Provider<Scope = Epoch, Scheme: Scheme<V::Commitment>>,
    FC: Certificates<
            BlockDigest = <V::Block as Digestible>::Digest,
            Commitment = V::Commitment,
            Scheme = P::Scheme,
        >,
    FB: Blocks<Block = V::StoredBlock>,
    ES: Epocher,
    T: Strategy,
    A: Acknowledgement,
{
    /// Create a new application actor.
    #[boxed]
    pub async fn init(
        context: E,
        finalizations_by_height: FC,
        mut finalized_blocks: FB,
        config: Config<P, ES, T, V::ApplicationBlock, V::Block, V::Commitment>,
    ) -> (Self, Mailbox<P::Scheme, V>, Floor) {
        // Initialize cache
        let prunable_config = cache::Config {
            partition_prefix: format!("{}-cache", config.partition_prefix),
            prunable_items_per_section: config.prunable_items_per_section,
            replay_buffer: config.replay_buffer,
            key_write_buffer: config.key_write_buffer,
            value_write_buffer: config.value_write_buffer,
            key_page_cache: config.page_cache.clone(),
        };
        let cache = cache::Manager::init(
            context.child("cache"),
            prunable_config,
            config.block_codec_config.clone(),
        )
        .await;

        // The application metadata name is retained for legacy support.
        let application_metadata_partition =
            format!("{}-application-metadata", config.partition_prefix);
        let mut stream =
            Stream::new(context.child("stream"), &application_metadata_partition).await;
        let last_processed_height = stream.processed_height();

        // Genesis is a local anchor. A floor finalization is verified and
        // resolved after `run` receives the resolver and buffer.
        let configured = match config.start {
            Start::Genesis(anchor) => {
                assert_eq!(
                    anchor.height(),
                    Height::zero(),
                    "genesis anchor must be at height zero"
                );
                finalized_blocks = Self::ensure_genesis_anchor(
                    finalized_blocks,
                    anchor,
                    last_processed_height,
                    &config.strategy,
                )
                .await;
                None
            }
            Start::Floor(finalization) => Some(finalization),
        };

        // The metadata's round floor can be lower than what the archives and the epocher justify.
        // An earlier marshal may have written no round floor, or one that never records a closed
        // epoch. Marshal also archives each block and its finalization before the application
        // acknowledges it, and records the new processed height only after the acknowledgement. The
        // round floor covers only processed blocks and installed floor anchors, so a restart in
        // between can find the next block archived with its finalization.
        //
        // That archived block raises the floor to its finalization round and covers the epochs
        // ending at or below it. Init syncs a higher floor to the metadata, so a later start reads
        // it even after the application retires those epochs.
        let (derived, witnessed) = Self::derive_cutoff(
            &finalizations_by_height,
            &finalized_blocks,
            last_processed_height,
            &config.epocher,
            &config.strategy,
        )
        .await;
        if derived > stream.cutoff() {
            stream.set_cutoff(derived);
            stream = stream.sync().await.expect("restored round floor must sync");
        }
        let last_processed =
            last_processed_height.map(|height| Self::processed(&finalized_blocks, height));

        // Create metrics
        let finalized_height = context.gauge("finalized_height", "Finalized height of application");
        let processed_height = context.gauge("processed_height", "Processed height of application");
        if let Some(last_processed_height) = last_processed_height {
            let _ = processed_height.try_set(last_processed_height.get());
        }
        let floor_state = FloorState::new(
            configured,
            Floor::new(last_processed, stream.cutoff()),
            witnessed,
        );
        let floor = floor_state.snapshot();

        // Initialize mailbox
        let (sender, mailbox) = mailbox::new(context.child("mailbox"), config.mailbox_size);
        (
            Self {
                context: ContextCell::new(context),
                mailbox,
                provider: config.provider,
                epocher: config.epocher,
                view_retention: config.view_retention,
                max_repair: config.max_repair,
                block_codec_config: config.block_codec_config,
                strategy: config.strategy,
                floor: floor_state,
                stream,
                pending_acks: PendingAcks::new(config.max_pending_acks.get()),
                tip: Height::zero(),
                block_subscriptions: Subscriptions::new(),
                certified: Certified::new(),
                dispatch_gate: DispatchGate::default(),
                staged: Staged::new(config.max_pending_acks.get().saturating_mul(2)),
                cache,
                finalizations_by_height,
                finalized_blocks,
                finalized_height,
                processed_height,
            },
            Mailbox::new(sender, config.max_pending_acks),
            floor,
        )
    }

    async fn ensure_genesis_anchor(
        mut finalized_blocks: FB,
        anchor: V::Block,
        last_processed_height: Option<Height>,
        strategy: &T,
    ) -> FB {
        let anchor_height = anchor.height();
        let anchor_commitment = V::commitment(&anchor);
        match finalized_blocks
            .get(ArchiveID::Index(anchor_height.get()))
            .await
        {
            Ok(Some(stored)) => {
                let stored = V::restore_across(stored, strategy).await;
                assert_eq!(
                    stored.height(),
                    anchor_height,
                    "stored genesis block height mismatch"
                );
                assert!(
                    V::commitment(&stored) == anchor_commitment,
                    "stored genesis block does not match configured anchor"
                );
            }
            Ok(None) => {
                if let Some(existing) =
                    last_processed_height.filter(|height| anchor_height < *height)
                {
                    warn!(
                        height = %anchor_height,
                        %existing,
                        "ignoring stale anchor"
                    );
                    return finalized_blocks;
                }

                finalized_blocks = finalized_blocks
                    .put(anchor.digest(), &anchor.into())
                    .await
                    .expect("failed to store startup anchor")
                    .sync()
                    .await
                    .expect("failed to sync startup anchor");
                debug!(height = %anchor_height, "stored genesis block");
            }
            Err(err) => panic!("failed to check startup anchor: {err}"),
        }
        finalized_blocks
    }

    /// Start the actor.
    pub fn start<R, Buf>(
        self,
        application: impl Reporter<Activity = Update<V::ApplicationBlock, A>>,
        buffer: Buf,
        resolver: (handler::Receiver<V::Commitment>, R),
    ) -> Handle<()>
    where
        R: TargetedResolver<
                Key = ResolverRequestFor<V>,
                Subscriber = Annotation,
                PublicKey = <P::Scheme as Verifier>::PublicKey,
            >,
        Buf: Buffer<V, PublicKey = <P::Scheme as Verifier>::PublicKey>,
    {
        let mut actor = Box::new(self);
        spawn_cell!(actor.context, actor.run(application, buffer, resolver))
    }

    /// Start the actor without a broadcast buffer.
    pub fn start_unbuffered<R>(
        self,
        application: impl Reporter<Activity = Update<V::ApplicationBlock, A>>,
        resolver: (handler::Receiver<V::Commitment>, R),
    ) -> Handle<()>
    where
        R: TargetedResolver<
                Key = ResolverRequestFor<V>,
                Subscriber = Annotation,
                PublicKey = <P::Scheme as Verifier>::PublicKey,
            >,
    {
        self.start(
            application,
            NoBuffer::<<P::Scheme as Verifier>::PublicKey>::new(),
            resolver,
        )
    }

    /// Run the application actor.
    async fn run<R, Buf>(
        mut self: Box<Self>,
        mut application: impl Reporter<Activity = Update<V::ApplicationBlock, A>>,
        mut buffer: Buf,
        (mut resolver_rx, mut resolver): (handler::Receiver<V::Commitment>, R),
    ) where
        R: TargetedResolver<
                Key = ResolverRequestFor<V>,
                Subscriber = Annotation,
                PublicKey = <P::Scheme as Verifier>::PublicKey,
            >,
        Buf: Buffer<V, PublicKey = <P::Scheme as Verifier>::PublicKey>,
    {
        // Create a local pool for waiter futures.
        let mut waiters = AbortablePool::<Result<V::Block, SubscriptionKeyFor<V>>>::default();

        // Observe durable syncs that no consensus caller awaits (the
        // notarization and finalization paths), so a failure applies the fatal
        // policy without waiting for a later archive call or blocking the actor.
        let mut syncs = Pool::<PooledSync>::default();

        // Anchor all startup work under a single root span. Tip recovery, floor
        // installation, gap repair, and the initial dispatch all run before any
        // mailbox message arrives, so without this root their work would emit as
        // orphan traces.
        (self, application, buffer, resolver, waiters) = async move {
            // Get tip and send to application
            let tip = self.get_latest().await;
            if let Some((height, digest, round)) = tip {
                self.advance_tip(round, height, digest, &mut application);
            }

            // Load persisted cache epochs so find_block can discover blocks
            // written before the last shutdown.
            self.cache = self.cache.load_persisted_epochs().await;

            // A configured floor follows the same path as `SetFloor`: verify it,
            // then apply a local anchor or await it from the buffer or peers.
            if let Some(finalization) = self.floor.take_configured() {
                self = self
                    .install_floor(
                        finalization,
                        &mut resolver,
                        &mut buffer,
                        &mut waiters,
                        &mut application,
                    )
                    .await;
            }

            // Attempt to repair any gaps in the finalized blocks archive, if there are any.
            let repaired;
            (self, repaired) = self
                .try_repair_gaps(&mut buffer, &mut resolver, &mut application)
                .await;
            if repaired {
                self = self.sync_finalized().await;
            }

            // Attempt to dispatch the next finalized block to the application, if it is ready.
            self = self.try_dispatch_blocks(&mut application).await;

            (self, application, buffer, resolver, waiters)
        }
        .instrument(info_span!("marshal.actor.start"))
        .await;

        select_loop! {
            self.context,
            on_start => {
                // Every arm that buffers stream metadata syncs it before yielding, so a raised
                // round floor or acknowledged height is durable before the next event is handled.
                assert!(
                    self.stream.is_synced(),
                    "stream must be synced before the actor handles its next event"
                );

                // Remove any dropped subscribers. If all subscribers dropped, abort the waiter.
                self.block_subscriptions.retain_open();
            },
            on_stopped => {
                debug!("context shutdown, stopping marshal");
            },
            // Drive durability syncs: a real sync failure panics inside the
            // pooled future (the fatal policy), aborting the actor. A completed
            // finalized-archive sync additionally releases the dispatch barrier
            // for the batches it covers and resumes application dispatch.
            sync = syncs.next_completed() => {
                if let PooledSync::Finalized(seq) = sync {
                    self.dispatch_gate.release(seq);
                    self = self.try_dispatch_blocks(&mut application).await;
                }
            },
            // Handle waiter completions first
            Ok(completion) = waiters.next_completed() else continue => match completion {
                Ok(block) => {
                    (self, _) = self
                        .ingest(block, &mut buffer, &mut application, &mut resolver)
                        .await;
                }
                Err(key) => {
                    // A closed buffer subscription marks the key as permanently unavailable.
                    match key {
                        SubscriptionKey::Digest(digest) => {
                            debug!(
                                ?digest,
                                "buffer subscription closed, canceling local subscribers"
                            );
                        }
                        SubscriptionKey::Commitment(commitment) => {
                            debug!(
                                ?commitment,
                                "buffer subscription closed, canceling local subscribers"
                            );
                        }
                    }
                    self.block_subscriptions.remove(&key);
                }
            },
            // Handle application acknowledgements (drain all ready acks, sync once)
            result = self.pending_acks.current() => {
                let next = match self
                    .handle_ack(result, &mut application, &mut buffer, &mut resolver)
                    .await
                {
                    Ok(next) => next,
                    Err((height, e)) => {
                        debug!(
                            ?e,
                            %height,
                            "application acknowledgement dropped, stopping marshal"
                        );
                        return;
                    }
                };
                self = next;
            },
            // Handle consensus inputs before backfill or resolver traffic
            Some(message) = self.mailbox.recv() else {
                debug!("mailbox closed, shutting down");
                break;
            } => {
                let span = info_span!(
                    parent: message.span(),
                    "marshal.actor.process",
                    operation = message.name(),
                );
                self = self
                    .handle_mailbox_message(
                        message,
                        &mut resolver,
                        &mut waiters,
                        &mut syncs,
                        &mut buffer,
                        &mut application,
                    )
                    .instrument(span)
                    .await;
            },
            // Handle resolver messages last (batched up to max_repair, sync once)
            Some(message) = resolver_rx.recv() else {
                debug!("handler closed, shutting down");
                return;
            } => {
                self = self
                    .handle_resolver_message(
                        message,
                        &mut resolver_rx,
                        &mut resolver,
                        &mut syncs,
                        &mut buffer,
                        &mut application,
                    )
                    .await;
            },
        }
    }

    /// Handles one ready application acknowledgement and drains any queued acks
    /// that are already complete.
    async fn handle_ack<Buf, R>(
        mut self: Box<Self>,
        result: <A::Waiter as Future>::Output,
        application: &mut impl Reporter<Activity = Update<V::ApplicationBlock, A>>,
        buffer: &mut Buf,
        resolver: &mut R,
    ) -> Result<Box<Self>, (Height, A::Error)>
    where
        Buf: Buffer<V>,
        R: Resolver<Key = ResolverRequestFor<V>, Subscriber = Annotation>,
    {
        // Start with the ack that woke this `select_loop!` arm.
        let mut pending = Some(self.pending_acks.complete_current(result));
        loop {
            let (height, result) = pending.take().expect("pending ack must exist");
            match result {
                Ok(()) => {
                    // Apply in-memory progress updates for this acknowledged
                    // block. The metadata sync below makes drained updates durable.
                    // Only archived blocks are dispatched, so the block is stored.
                    self.update_processed(Processed::Block(height), resolver);
                    self = self
                        .cover_processed(height, buffer, application, resolver)
                        .await;
                }
                Err(e) => return Err((height, e)),
            }

            // Opportunistically drain any additional already-ready acks so we
            // can persist one metadata sync for the whole batch below.
            match self.pending_acks.pop_ready() {
                Some(next) => pending = Some(next),
                None => break,
            }
        }

        // Persist buffered progress updates once after draining all ready acks.
        self.stream = self
            .stream
            .sync()
            .await
            .expect("failed to sync application progress");

        // Refill the application dispatch pipeline.
        Ok(self.try_dispatch_blocks(application).await)
    }

    /// Handles a single mailbox message from local consensus/application callers.
    async fn handle_mailbox_message<Buf, R>(
        mut self: Box<Self>,
        message: Message<P::Scheme, V>,
        resolver: &mut R,
        waiters: &mut AbortablePool<'_, Result<V::Block, SubscriptionKeyFor<V>>>,
        syncs: &mut Pool<'_, PooledSync>,
        buffer: &mut Buf,
        application: &mut impl Reporter<Activity = Update<V::ApplicationBlock, A>>,
    ) -> Box<Self>
    where
        Buf: Buffer<V, PublicKey = <P::Scheme as Verifier>::PublicKey>,
        R: TargetedResolver<
                Key = ResolverRequestFor<V>,
                Subscriber = Annotation,
                PublicKey = <P::Scheme as Verifier>::PublicKey,
            >,
    {
        if message.response_closed() {
            return self;
        }

        match message {
            Message::GetInfo {
                identifier,
                response,
                ..
            } => {
                let info = match identifier {
                    // TODO: Instead of pulling out the entire block, determine the
                    // height directly from the archive by mapping the digest to
                    // the index, which is the same as the height.
                    BlockID::Digest(digest) => {
                        match self.finalized_blocks.get(ArchiveID::Key(&digest)).await {
                            Ok(stored) => stored.map(|b| (b.height(), digest)),
                            Err(e) => panic!("failed to get block: {e}"),
                        }
                    }
                    BlockID::Height(height) => self.get_info_by_height(height).await,
                    BlockID::Latest => self.get_latest().await.map(|(h, d, _)| (h, d)),
                };
                response.send_lossy(info);
            }
            Message::GetVerified {
                round, response, ..
            } => {
                let block = match self.cache.get_verified(round).await {
                    Some(stored) => Some(V::restore_across(stored, &self.strategy).await),
                    None => None,
                };
                response.send_lossy(block);
            }
            Message::Forward {
                round,
                commitment,
                recipients,
                ..
            } => {
                if matches!(&recipients, Recipients::Some(peers) if peers.is_empty()) {
                    return self;
                }
                let Some(block) = self.find_block_by_commitment(buffer, commitment).await else {
                    debug!(?commitment, "block not found for forwarding");
                    return self;
                };
                buffer.send(round, block, recipients);
            }
            Message::Proposed {
                round,
                block,
                recipients,
                ack,
                ..
            } => {
                // To lower view latency as much as possible while preserving
                // safety, we broadcast the block before persisting it
                // (durability is not required until certify). A leader that
                // crashes here rebuilds after restart and may broadcast a
                // different block for the same round. This is tolerated: a
                // block that no journaled vote names cannot be notarized, extra
                // block bytes cannot form a conflicting certificate (unlike
                // votes), and block storage holds multiple candidates per
                // round (see [Mailbox::get_verified]).
                buffer.send(round, block.clone(), recipients);
                self = self
                    .persist_verified(round, block, ack, buffer, application, resolver)
                    .await;
            }
            Message::Prepared {
                round,
                block,
                recipients,
                ..
            } => {
                // A held candidate is only sent. A later [Message::Verified],
                // requested by the lock-in or by certification, stores it, so
                // an abandoned candidate costs no storage write.
                buffer.send(round, block, recipients);
            }
            Message::Verified {
                round, block, ack, ..
            } => {
                self = self
                    .persist_verified(round, block, ack, buffer, application, resolver)
                    .await;
            }
            Message::Certified {
                round, block, ack, ..
            } => {
                (self, _) = self
                    .ingest(block.clone(), buffer, application, resolver)
                    .await;
                let digest = block.digest();

                // A block the verified archive already holds needs no second copy:
                // the verified archive's covering sync handle vouches for it. At
                // most one notarization exists per round, so the notarized slot can
                // never belong to a different payload: a duplicate put is a no-op
                // whose handle still covers the original write. If the round has
                // already been pruned by tip advancement, both writes are no-ops
                // because the round is below the retention floor.
                let block_sync;
                if self.cache.has_verified(round, &digest).await {
                    debug!(?round, "certified block covered by verified write");
                    (self.cache, block_sync) = self.cache.start_sync_verified(round).await;
                } else {
                    (self.cache, block_sync) =
                        self.cache.put_notarized(round, digest, &block).await;
                }

                // Hold the certify barrier until the round's notarization
                // certificate (when one was accepted before this message) is
                // durable alongside the block.
                let notarization_sync;
                (self.cache, notarization_sync) = self.cache.start_sync_notarizations(round).await;
                let handle = Handle::from_future(async move {
                    let (notarization, block) = join(notarization_sync, block_sync).await;
                    notarization.and(block)
                });
                ack.send_lossy(handle);
            }
            Message::Certification { notarization, .. } => {
                // Certification also authenticates the parent commitment
                let commitment = notarization.proposal.payload;
                let Some(block) = self.find_block_by_commitment(buffer, commitment).await else {
                    debug!(?commitment, "certified block unavailable locally");
                    return self;
                };
                let height = block.height();
                self.certified.insert(height, commitment);
                if let Some(parent) = height.previous() {
                    self.certified.insert(parent, V::parent_commitment(&block));
                }
            }
            Message::Notarization { notarization, .. } => {
                let round = notarization.round();
                let commitment = notarization.proposal.payload;
                let digest = V::commitment_to_inner(commitment);

                // Persist the notarization; the certify barrier folds in its
                // durability via `start_sync_notarizations`. The archive keeps a
                // single notarization per round, so a re-delivery is a no-op whose
                // handle still covers the original write. No consensus caller
                // awaits this handle, so the pool observes it (applying the fatal
                // policy) without blocking the actor.
                let handle;
                (self.cache, handle) = self
                    .cache
                    .put_notarization(round, digest, &notarization)
                    .await;
                syncs.push(async move {
                    handle.durable(round, "notarization").await;
                    PooledSync::Observed
                });

                // A notarization alone is not enough to fetch missing proposal
                // data. If the block is not locally available, remember the
                // certificate and wait for a later finalization/repair path.
                if let Some(block) = self.find_block_by_commitment(buffer, commitment).await {
                    (self, _) = self
                        .ingest(block.clone(), buffer, application, resolver)
                        .await;
                    if self.cache.has_verified(round, &digest).await {
                        debug!(?round, "notarized block covered by verified write");
                    } else {
                        let handle;
                        (self.cache, handle) =
                            self.cache.put_notarized(round, digest, &block).await;
                        syncs.push(async move {
                            handle.durable(round, "notarized").await;
                            PooledSync::Observed
                        });
                    }
                } else {
                    debug!(?round, "notarized block unavailable locally");
                }
            }
            Message::Finalization { finalization, .. } => {
                let round = finalization.round();
                let commitment = finalization.proposal.payload;
                let digest = V::commitment_to_inner(commitment);

                // Cache finalization by round.
                self.cache = self
                    .cache
                    .put_finalization(round, digest, &finalization)
                    .await;

                // Search for the finalized block locally, otherwise fetch it remotely.
                if let Some(block) = self.find_block_by_commitment(buffer, commitment).await {
                    // The anchor path stores the floor block and finalization,
                    // advances floors, prunes below them, and resumes dispatch.
                    let anchored;
                    (self, anchored) = self
                        .ingest(block.clone(), buffer, application, resolver)
                        .await;
                    if anchored {
                        return self;
                    }

                    let height = block.height();
                    self = self
                        .witness_round(height, round, buffer, application, resolver)
                        .await;
                    self.stream = self
                        .stream
                        .sync()
                        .await
                        .expect("round floor sync must succeed");
                    let stored;
                    (self, stored) = self
                        .store_finalization(height, &block, Some(finalization), application)
                        .await;
                    if stored {
                        self.staged.insert(height, block);

                        // If a floor anchor is pending, repair and dispatch are
                        // no-ops until the anchor block is stored.
                        (self, _) = self.try_repair_gaps(buffer, resolver, application).await;
                        self = self.start_finalized_sync(round, syncs).await;
                        debug!(?round, %height, "finalized block stored");
                    }
                } else {
                    // The finalization carries a round and commitment, but not a
                    // height. Keep the request round-bound until the block is decoded.
                    debug!(?round, ?commitment, "finalized block missing");
                    self.floor
                        .fetch_if_permitted(
                            resolver,
                            Request::finalized_by_round(commitment, round),
                        )
                        .ignore();
                }
            }
            Message::GetBlock {
                identifier,
                response,
                ..
            } => match identifier {
                BlockID::Digest(digest) => {
                    let result = self.find_block_by_digest(buffer, digest).await;
                    response.send_lossy(result);
                }
                BlockID::Height(height) => {
                    let result = match self.get_finalized_block(height).await {
                        Some(stored) => Some(V::restore_across(stored, &self.strategy).await),
                        None => None,
                    };
                    response.send_lossy(result);
                }
                BlockID::Latest => {
                    let block = match self.get_latest().await {
                        Some((_, digest, _)) => self.find_block_by_digest(buffer, digest).await,
                        None => None,
                    };
                    response.send_lossy(block);
                }
            },
            Message::GetFinalization {
                height, response, ..
            } => {
                let finalization = self.get_finalization_by_height(height).await;
                response.send_lossy(finalization);
            }
            Message::GetProcessed { response, .. } => {
                response.send_lossy(self.floor.processed());
            }
            Message::GetAnchor { response, .. } => {
                let anchor = match self.floor.processed() {
                    Some(processed) => {
                        let stored = self
                            .get_finalized_block(processed.anchor())
                            .await
                            .expect("processed position must be backed by a stored block");
                        let block = V::restore_across(stored, &self.strategy).await;
                        Some((processed, block))
                    }
                    None => None,
                };
                response.send_lossy(anchor);
            }
            Message::HintFinalized {
                height, targets, ..
            } => {
                // Skip if finalization is already available locally.
                if self.has_finalization_by_height(height).await {
                    return self;
                }

                self.floor
                    .fetch_targeted_if_permitted(resolver, Request::finalized(height), targets)
                    .ignore();
            }
            Message::SubscribeByDigest {
                span,
                digest,
                fallback,
                response,
            } => {
                self.handle_subscribe(
                    span,
                    fallback.into(),
                    SubscriptionKey::Digest(digest),
                    response,
                    resolver,
                    waiters,
                    buffer,
                )
                .await;
            }
            Message::SubscribeByCommitment {
                span,
                commitment,
                fallback,
                response,
            } => {
                self.handle_subscribe(
                    span,
                    fallback,
                    SubscriptionKey::Commitment(commitment),
                    response,
                    resolver,
                    waiters,
                    buffer,
                )
                .await;
            }
            Message::HintNotarized {
                round, commitment, ..
            } => {
                if self
                    .find_block_by_commitment(buffer, commitment)
                    .await
                    .is_none()
                {
                    self.floor
                        .fetch_if_permitted(resolver, Request::notarized(round))
                        .ignore();
                }
            }
            Message::SetFloor { finalization, .. } => {
                self = self
                    .install_floor(finalization, resolver, buffer, waiters, application)
                    .await;
            }
            Message::Prune { height, .. } => {
                // Only allow pruning at or below the current floor.
                if height > self.floor.processed_height() {
                    warn!(%height, floor = %self.floor.processed_height(), "prune height above floor, ignoring");
                    return self;
                }

                self = self.prune_finalized_archives(height).await;
            }
        }
        self
    }

    /// Handles a batch of resolver messages, starting one pooled
    /// finalized-archive sync if any accepted delivery buffered a write.
    async fn handle_resolver_message<Buf, R>(
        mut self: Box<Self>,
        message: handler::Message<V::Commitment>,
        resolver_rx: &mut handler::Receiver<V::Commitment>,
        resolver: &mut R,
        syncs: &mut Pool<'_, PooledSync>,
        buffer: &mut Buf,
        application: &mut impl Reporter<Activity = Update<V::ApplicationBlock, A>>,
    ) -> Box<Self>
    where
        Buf: Buffer<V, PublicKey = <P::Scheme as Verifier>::PublicKey>,
        R: Resolver<Key = ResolverRequestFor<V>, Subscriber = Annotation>,
    {
        let mut handled = false;
        let mut produces = Vec::new();
        let mut delivers = Vec::new();

        // Drain up to max_repair resolver messages. Block deliveries are handled
        // immediately, certificate-bearing deliveries are batched for verification,
        // and produce responses wait until repair has had a chance to fill gaps.
        for msg in std::iter::once(message)
            .chain(std::iter::from_fn(|| resolver_rx.try_recv().ok()))
            .take(self.max_repair.get())
        {
            if msg.response_closed() {
                continue;
            }
            handled = true;

            match msg {
                handler::Message::Produce { key, response } => {
                    produces.push((key, response));
                }
                handler::Message::Deliver {
                    delivery,
                    value,
                    response,
                } => {
                    let span = info_span!(
                        parent: &delivery.subscribers.first().1,
                        "marshal.resolver.deliver",
                        key = %delivery.key
                    );
                    for (_, subscriber_span) in delivery.subscribers.iter().skip(1) {
                        span.follows_from(subscriber_span.id());
                    }
                    self = self
                        .handle_deliver(
                            ResolverDelivery {
                                delivery,
                                value,
                                response,
                            },
                            &mut delivers,
                            buffer,
                            application,
                            resolver,
                        )
                        .instrument(span)
                        .await;
                }
            }
        }
        if !handled {
            return self;
        }

        // Batch verify and process all certificate-bearing deliveries.
        self = self
            .verify_delivered(delivers, buffer, application, resolver)
            .await;

        // Attempt to fill gaps before handling produce requests so we can serve
        // data received earlier in the same batch.
        (self, _) = self.try_repair_gaps(buffer, resolver, application).await;

        // Start a pooled sync so any writes buffered by this batch become
        // durable without blocking the mailbox. Dispatch of the written
        // heights resumes when the sync completes. A batch has no single
        // round, so the label is the newest witnessed finalization round.
        let round = self.floor.witnessed();
        self = self.start_finalized_sync(round, syncs).await;

        // Handle produce requests in parallel.
        join_all(
            produces
                .into_iter()
                .filter(|(_, response)| !response.is_closed())
                .map(|(key, response)| self.handle_produce(key, response, buffer)),
        )
        .await;

        self
    }

    /// Handle a produce request from a remote peer.
    #[tracing::instrument(name = "marshal.resolver.produce", level = "debug", skip_all, fields(key = %key))]
    async fn handle_produce<Buf: Buffer<V>>(
        &self,
        key: ResolverRequestFor<V>,
        response: oneshot::Sender<Bytes>,
        buffer: &Buf,
    ) {
        match key {
            Key::Block(commitment) => {
                let Some(block) = self.find_block_by_commitment(buffer, commitment).await else {
                    debug!(?commitment, "block missing on request");
                    return;
                };
                response.send_lossy(block.encode());
            }
            Key::Finalized { height } => {
                let Some(finalization) = self.get_finalization_by_height(height).await else {
                    debug!(%height, "finalization missing on request");
                    return;
                };
                let Some(stored) = self.get_finalized_block(height).await else {
                    debug!(%height, "finalized block missing on request");
                    return;
                };

                // The block at a height is stored under the digest its finalization commits to.
                let digest = V::commitment_to_inner(finalization.proposal.payload);
                let block = V::restore(stored, digest);
                response.send_lossy((finalization, V::into_shared(block)).encode());
            }
            Key::Notarized { round } => {
                let Some(notarization) = self.cache.get_notarization(round).await else {
                    debug!(?round, "notarization missing on request");
                    return;
                };
                let commitment = notarization.proposal.payload;
                let Some(block) = self.find_block_by_commitment(buffer, commitment).await else {
                    debug!(?commitment, "block missing on request");
                    return;
                };
                response.send_lossy((notarization, block).encode());
            }
        }
    }

    /// Handle a local subscription request for a block.
    #[allow(clippy::too_many_arguments)]
    async fn handle_subscribe<Buf: Buffer<V>>(
        &mut self,
        span: Span,
        fallback: CommitmentFallback,
        key: SubscriptionKeyFor<V>,
        response: oneshot::Sender<V::Block>,
        resolver: &mut impl Resolver<Key = ResolverRequestFor<V>, Subscriber = Annotation>,
        waiters: &mut AbortablePool<'_, Result<V::Block, SubscriptionKeyFor<V>>>,
        buffer: &mut Buf,
    ) {
        let digest = match key {
            SubscriptionKey::Digest(digest) => digest,
            SubscriptionKey::Commitment(commitment) => V::commitment_to_inner(commitment),
        };

        let block = match key {
            SubscriptionKey::Digest(digest) => self.find_block_by_digest(buffer, digest).await,
            SubscriptionKey::Commitment(commitment) => {
                self.find_block_by_commitment(buffer, commitment).await
            }
        };
        if let Some(block) = block {
            // An ancestor may be certified through a descendant without its own local notification.
            // Its digest binds the parent commitment, so extend certification evidence here.
            if let Some(parent) = block.height().previous()
                && match key {
                    SubscriptionKey::Commitment(commitment) => {
                        self.certified.contains(block.height(), &commitment)
                    }
                    SubscriptionKey::Digest(digest) => self
                        .certified
                        .contains_matching(block.height(), |commitment| {
                            V::commitment_to_inner(*commitment) == digest
                        }),
                }
            {
                self.certified.insert(parent, V::parent_commitment(&block));
            }
            response.send_lossy(block);
            return;
        }

        // Resolver admission controls remote acquisition. Every caller remains
        // registered for later local availability.
        //
        // Round-based fetching is for notarized proposal lookups whose height is
        // not known before the request. Height-based fetching is only for callers
        // that have a validated block height for resolver retention.
        match fallback {
            CommitmentFallback::FetchByRound { round } => {
                // Fetch the notarized proposal for this round. The response
                // must include a certificate so the commitment is tied to the
                // certified round context. The decoded block is heightable, but
                // that height is not known soon enough to key, coalesce, or prune
                // the in-flight resolver request.
                self.floor
                    .fetch_if_permitted(resolver, Request::notarized(round))
                    .ignore();
                debug!(?round, ?digest, "notarized block unavailable");
            }
            CommitmentFallback::FetchByCommitment { height } => {
                let commitment = match key {
                    SubscriptionKey::Commitment(commitment) => commitment,
                    SubscriptionKey::Digest(_) => {
                        unreachable!("digest subscriptions cannot request commitment fallback")
                    }
                };

                // Ancestry may be only notarized, so skipping commitment recomputation
                // requires certification evidence
                let request = if self.certified.contains(height, &commitment) {
                    Request::certified(commitment, height)
                } else {
                    Request::untrusted(commitment, height)
                };
                self.floor.fetch_if_permitted(resolver, request).ignore();
                debug!(%height, ?commitment, ?digest, "certified ancestry block unavailable");
            }
            CommitmentFallback::Wait => {}
        }

        // Register subscriber.
        match key {
            SubscriptionKey::Digest(digest) => {
                debug!(?fallback, ?digest, "registering subscriber");
            }
            SubscriptionKey::Commitment(commitment) => {
                debug!(?fallback, ?commitment, ?digest, "registering subscriber");
            }
        }
        self.block_subscriptions
            .insert(span, key, response, waiters, buffer);
    }

    /// Verifies and installs a floor, awaiting the anchor block from the buffer or peers if needed.
    async fn install_floor<Buf, R>(
        mut self: Box<Self>,
        finalization: Finalization<P::Scheme, V::Commitment>,
        resolver: &mut R,
        buffer: &mut Buf,
        waiters: &mut AbortablePool<'_, Result<V::Block, SubscriptionKeyFor<V>>>,
        application: &mut impl Reporter<Activity = Update<V::ApplicationBlock, A>>,
    ) -> Box<Self>
    where
        Buf: Buffer<V, PublicKey = <P::Scheme as Verifier>::PublicKey>,
        R: Resolver<Key = ResolverRequestFor<V>, Subscriber = Annotation>,
    {
        let round = finalization.round();
        if self.floor.cutoff().covers(round) {
            warn!(
                %round,
                floor = %self.floor.cutoff(),
                "floor not updated, covered by existing round floor"
            );
            return self;
        }

        let Some(scoped) = self.provider.scoped(finalization.epoch()) else {
            panic!("floor finalization epoch unavailable");
        };
        assert!(
            finalization.verify(self.context.as_mut(), &scoped, &self.strategy),
            "floor finalization must verify"
        );

        let commitment = finalization.proposal.payload;
        let digest = V::commitment_to_inner(commitment);
        self.cache = self
            .cache
            .put_finalization(round, digest, &finalization)
            .await;

        // A pending floor at the same or a newer round already blocks
        // progress. Keep waiting for it instead of replacing it.
        if let Some(pending) = self.floor.pending()
            && pending.round() >= round
        {
            return self;
        }

        // A local anchor replaces any older pending floor and installs through ingest.
        if let Some(block) = self.find_block_by_commitment(buffer, commitment).await {
            self.floor.set_pending(finalization, None);
            let anchored;
            (self, anchored) = self.ingest(block, buffer, application, resolver).await;
            assert!(anchored, "failed to ingest pending floor anchor");
            return self;
        }

        // The pending floor owns the next application sync point. Drop any
        // in-flight acks before they can advance the processed height past it.
        self.pending_acks.clear();

        // The pending floor holds the waiter, which is released when the floor is
        // replaced, applied, or superseded. Reporting a closed subscription as an
        // error would cancel every caller subscription on the anchor, so the waiter
        // stays pending instead. The fetch below is issued either way.
        let aborter = buffer
            .subscribe_by_commitment(commitment)
            .map(|rx| waiters.push(rx.or_else(|_| future::pending())));
        self.floor.set_pending(finalization, aborter);

        debug!(?round, ?commitment, "starting fetch for floor block");
        self.floor
            .fetch_if_permitted(resolver, Request::finalized_by_round(commitment, round))
            .ignore();
        self
    }

    /// Ingests `block` and persists it as a verify-stage candidate for `round`,
    /// delivering the write's durable-sync handle through `ack`.
    ///
    /// If the round has already been pruned by tip advancement, `put_verified`
    /// is a no-op because the round is below the retention floor (and no longer
    /// is required by consensus to make progress). A duplicate delivery is also
    /// a no-op, with the handle still covering the original write's durability.
    async fn persist_verified<Buf: Buffer<V>>(
        mut self: Box<Self>,
        round: Round,
        block: V::Block,
        ack: oneshot::Sender<Handle<()>>,
        buffer: &mut Buf,
        application: &mut impl Reporter<Activity = Update<V::ApplicationBlock, A>>,
        resolver: &mut impl Resolver<Key = ResolverRequestFor<V>, Subscriber = Annotation>,
    ) -> Box<Self> {
        (self, _) = self
            .ingest(block.clone(), buffer, application, resolver)
            .await;
        let digest = block.digest();
        let handle;
        (self.cache, handle) = self.cache.put_verified(round, digest, &block).await;
        ack.send_lossy(handle);
        self
    }

    /// Notifies subscribers of a validated block and applies it to any
    /// pending floor transition.
    ///
    /// Subscribers are notified before the block is persisted. This is not
    /// observable while running because mailbox requests are only served
    /// after the current `select_loop!` arm completes. After an unclean
    /// shutdown, however, a subscriber may hold a block that marshal never
    /// durably stored. Subscriptions make no durability promise. Durable
    /// height-ordered delivery is provided by application dispatch, which
    /// only sends blocks once the finalized archives are durable (see
    /// [`Self::try_dispatch_blocks`]).
    ///
    /// Returns true if the block was consumed as the floor anchor.
    async fn ingest<Buf: Buffer<V>>(
        mut self: Box<Self>,
        block: V::Block,
        buffer: &mut Buf,
        application: &mut impl Reporter<Activity = Update<V::ApplicationBlock, A>>,
        resolver: &mut impl Resolver<Key = ResolverRequestFor<V>, Subscriber = Annotation>,
    ) -> (Box<Self>, bool) {
        self.block_subscriptions.notify(&block);

        let Some(finalization) = self.floor.take_matching(V::commitment(&block)) else {
            return (self, false);
        };

        self = self
            .apply_floor(finalization, block, buffer, application, resolver)
            .await;
        (self, true)
    }

    /// Applies the floor transition that `finalization` announces using its anchor block.
    async fn apply_floor<Buf: Buffer<V>>(
        mut self: Box<Self>,
        finalization: Finalization<P::Scheme, V::Commitment>,
        block: V::Block,
        buffer: &mut Buf,
        application: &mut impl Reporter<Activity = Update<V::ApplicationBlock, A>>,
        resolver: &mut impl Resolver<Key = ResolverRequestFor<V>, Subscriber = Annotation>,
    ) -> Box<Self> {
        // Floor anchors can bypass the local proposal-verification path. Check
        // the parent relationship before using a non-genesis anchor for walkback.
        let height = block.height();
        if height > Height::zero() {
            let parent_commitment = V::parent_commitment(&block);
            assert!(
                block.parent() == V::commitment_to_inner(parent_commitment),
                "floor block parent commitment mismatch"
            );
        }

        // This anchor cannot move the application sync point, but its finalization round can still
        // prune round-bound resolver work. Keep pending acks intact because processed_height is
        // unchanged. The raised round floor is synced so a restart with the same configured floor
        // treats it as superseded, even after its anchor block is pruned and its epoch retired
        // from the provider.
        if height <= self.floor.processed_height() {
            warn!(
                %height,
                existing = %self.floor.processed_height(),
                "floor not updated, at or below existing"
            );
            self = self
                .witness_round(height, finalization.round(), buffer, application, resolver)
                .await;
            self.stream = self
                .stream
                .sync()
                .await
                .expect("round floor sync must succeed");
            let repaired;
            (self, repaired) = self.try_repair_gaps(buffer, resolver, application).await;
            if repaired {
                self = self.sync_finalized().await;
            }
            return self.try_dispatch_blocks(application).await;
        }

        let digest = block.digest();
        let round = finalization.round();
        let stored: V::StoredBlock = block.into();
        (self.finalized_blocks, self.finalizations_by_height) = try_join!(
            self.finalized_blocks
                .put(digest, &stored)
                .map_err(BoxedError::from),
            self.finalizations_by_height
                .put(height, digest, &finalization)
                .map_err(BoxedError::from),
        )
        .expect("failed to store floor anchor");
        self = self.sync_finalized().await;
        self.advance_tip(round, height, digest, application);

        // The anchor is durable, but the application still needs to process it.
        // Record the previous height so dispatch resumes at the anchor itself.
        let dispatch_floor = height
            .previous()
            .expect("floor anchor above processed height must have predecessor");
        let processed = Self::processed(&self.finalized_blocks, dispatch_floor);
        self.update_processed(processed, resolver);

        // Release staged blocks skipped by the floor transition
        self.staged.retain(height);

        // Raise the round floor through the anchor's round and every epoch that ends at or below
        // the anchor, and persist it with the dispatch floor before pruning finalized data.
        self = self
            .witness_round(dispatch_floor, round, buffer, application, resolver)
            .await
            .cover_closed_epoch(height, buffer, application, resolver)
            .await;
        self.stream = self
            .stream
            .sync()
            .await
            .expect("failed to sync floor metadata");

        // Drop all pending acknowledgement waiters so any in-flight application
        // acks for blocks below the new floor cannot rewrite the processed floor.
        self.pending_acks.clear();

        // Keep the processed block so the application can restart from it.
        self = self.prune_after_floor(dispatch_floor).await;

        // Keep caller-owned block subscriptions alive across the floor update. Resolver pruning
        // stops obsolete network work, but later local ingress can still satisfy these waiters,
        // and callers do not retry a closed subscription.
        let repaired;
        (self, repaired) = self.try_repair_gaps(buffer, resolver, application).await;
        if repaired {
            self = self.sync_finalized().await;
        }
        self.try_dispatch_blocks(application).await
    }

    /// Handle a deliver message from the resolver. Block delivers are handled
    /// immediately. Finalized/Notarized delivers are parsed and structurally
    /// validated, then collected into `delivers` for batch certificate verification.
    /// A notarized delivery's block is decoded only after its certificate verifies.
    async fn handle_deliver<Buf: Buffer<V>>(
        mut self: Box<Self>,
        message: ResolverDelivery<V>,
        delivers: &mut Vec<PendingVerification<P::Scheme, V>>,
        buffer: &mut Buf,
        application: &mut impl Reporter<Activity = Update<V::ApplicationBlock, A>>,
        resolver: &mut impl Resolver<Key = ResolverRequestFor<V>, Subscriber = Annotation>,
    ) -> Box<Self> {
        let ResolverDelivery {
            delivery,
            mut value,
            response,
        } = message;
        let Delivery {
            key, subscribers, ..
        } = delivery;
        match key {
            Key::Block(commitment) => {
                // Local annotations determine commitment checks and block storage
                let annotations = subscribers
                    .map_into(|(annotation, _)| annotation)
                    .into_vec();

                // Any `Certified` or `Finalized` subscriber authenticates the shared commitment
                let expected = if annotations.iter().any(|annotation| {
                    matches!(
                        annotation,
                        Annotation::Certified { .. } | Annotation::Finalized(_)
                    )
                }) {
                    ExpectedCommitment::Trusted(commitment)
                } else {
                    ExpectedCommitment::Untrusted(commitment)
                };
                let Ok(block) =
                    V::decode_block(value, &self.block_codec_config, expected, &self.strategy)
                        .await
                else {
                    response.send_lossy(Outcome::Invalid);
                    return self;
                };
                response.send_lossy(Outcome::Resolved(Resolution::Complete));

                // This block may match the pending floor request. Whether it
                // installs or is rejected as the floor anchor, do not also
                // process it as an ordinary block delivery.
                let anchored;
                (self, anchored) = self
                    .ingest(block.clone(), buffer, application, resolver)
                    .await;
                if anchored {
                    return self;
                }

                let height = block.height();
                let digest = block.digest();

                // A certified block's parent link was checked by the validators
                // that certified it, so the walk keeps trusting as it descends.
                if annotations
                    .iter()
                    .any(|annotation| matches!(annotation, Annotation::Certified { .. }))
                    && let Some(parent) = height.previous()
                {
                    self.certified.insert(parent, V::parent_commitment(&block));
                }

                // Round-bound proposal-parent fetches are `Key::Notarized`
                // deliveries and are handled below. In this block-keyed path,
                // `Finalized` means the block belongs in the finalized chain.
                let finalization = self.cache.get_finalization_for(digest).await;
                if let Some(finalization) = &finalization {
                    self = self
                        .witness_round(height, finalization.round(), buffer, application, resolver)
                        .await;
                    self.stream = self
                        .stream
                        .sync()
                        .await
                        .expect("round floor sync must succeed");
                }
                if finalization.is_some()
                    || annotations
                        .iter()
                        .any(|annotation| matches!(annotation, Annotation::Finalized(_)))
                {
                    (self, _) = self
                        .store_finalization(height, &block, finalization, application)
                        .await;
                } else if annotations.iter().any(|annotation| {
                    matches!(
                        annotation,
                        Annotation::Certified { height: bound }
                        | Annotation::Untrusted { height: bound } if height <= *bound
                    )
                }) && height > self.floor.processed_height()
                    && let Some(bounds) = self.epocher.containing(height)
                {
                    self.cache = self
                        .cache
                        .put_certified(bounds.epoch(), height, digest, &block)
                        .await;
                }
                debug!(?digest, %height, "received block");
            }
            Key::Finalized { height } => {
                // Without an epoch for the height or a scope for that epoch, the response cannot
                // be validated, so the key is retired without crediting or blaming its source.
                let Some((epoch, scoped)) = self.scoped_for_height(height) else {
                    debug!(
                        %height,
                        floor = %self.floor.processed_height(),
                        "ignoring stale delivery"
                    );
                    response.send_lossy(Outcome::Ignored);
                    return self;
                };
                let certificate_codec_config = scoped.certificate_codec_config();

                let Ok(finalization) =
                    Finalization::read_cfg(&mut value, &certificate_codec_config)
                else {
                    response.send_lossy(Outcome::Invalid);
                    return self;
                };

                // We decoded the certificate with the codec config for the height's epoch, so the
                // finalization must claim that same epoch. A mismatch means the bytes were bounded
                // against the wrong participant set, so reject before verification.
                if finalization.epoch() != epoch {
                    response.send_lossy(Outcome::Invalid);
                    return self;
                }

                // Decode the block carried with the finalization. Below, it is checked against
                // the requested height and the finalization payload.
                let Ok(block) =
                    V::ApplicationBlock::decode_cfg(&mut value, &self.block_codec_config)
                else {
                    response.send_lossy(Outcome::Invalid);
                    return self;
                };

                // Once the certificate verifies, the finalization authenticates the application
                // block, so only the height and digest are checked here. The working block is then
                // rebuilt from the finalized commitment.
                let commitment = finalization.proposal.payload;
                if block.height() != height {
                    response.send_lossy(Outcome::Invalid);
                    return self;
                }
                let block = Digested::compute(block, &self.strategy).await;
                if block.digest() != V::commitment_to_inner(commitment) {
                    response.send_lossy(Outcome::Invalid);
                    return self;
                }
                delivers.push(PendingVerification::Finalized {
                    scoped,
                    finalization,
                    block,
                    response,
                });
            }
            Key::Notarized { round } => {
                // The payload check below needs the epoch's participant set, so a
                // scope without the full scheme counts as unavailable and the key is
                // retired without crediting or blaming its source.
                let Some(scheme) = self.provider.scheme(round.epoch()) else {
                    debug!(
                        ?round,
                        floor = %self.floor.processed_height(),
                        "ignoring stale delivery"
                    );
                    response.send_lossy(Outcome::Ignored);
                    return self;
                };
                let certificate_codec_config = scheme.certificate_codec_config();
                let Ok(notarization) =
                    Notarization::read_cfg(&mut value, &certificate_codec_config)
                else {
                    response.send_lossy(Outcome::Invalid);
                    return self;
                };

                // The resolver key binds this response to `round`; a certificate for any other
                // round is a bad response even if it decodes correctly.
                if notarization.round() != round {
                    response.send_lossy(Outcome::Invalid);
                    return self;
                }

                // The payload must be valid under the epoch's scheme.
                if !V::check_payload(scheme.as_ref(), notarization.proposal.payload) {
                    response.send_lossy(Outcome::Invalid);
                    return self;
                }
                delivers.push(PendingVerification::Notarized {
                    scoped: Scoped::scheme(scheme),
                    notarization,
                    block: value,
                    response,
                });
            }
        }
        self
    }

    /// Batch verify pending certificates and process valid items.
    #[tracing::instrument(name = "marshal.actor.verify_delivered", level = "info", skip_all, fields(count = delivers.len().traced()))]
    async fn verify_delivered<Buf: Buffer<V>>(
        mut self: Box<Self>,
        mut delivers: Vec<PendingVerification<P::Scheme, V>>,
        buffer: &mut Buf,
        application: &mut impl Reporter<Activity = Update<V::ApplicationBlock, A>>,
        resolver: &mut impl Resolver<Key = ResolverRequestFor<V>, Subscriber = Annotation>,
    ) -> Box<Self> {
        delivers.retain(|item| !item.response_closed());
        if delivers.is_empty() {
            return self;
        }

        // Extract (subject, certificate) pairs for batch verification.
        let certs: Vec<_> = delivers
            .iter()
            .map(|item| match item {
                PendingVerification::Finalized { finalization, .. } => (
                    Subject::Finalize {
                        proposal: &finalization.proposal,
                    },
                    &finalization.certificate,
                ),
                PendingVerification::Notarized { notarization, .. } => (
                    Subject::Notarize {
                        proposal: &notarization.proposal,
                    },
                    &notarization.certificate,
                ),
            })
            .collect();

        // Group indices by epoch.
        let mut by_epoch: BTreeMap<Epoch, Vec<usize>> = BTreeMap::new();
        for (i, item) in delivers.iter().enumerate() {
            let epoch = match item {
                PendingVerification::Notarized { notarization, .. } => notarization.epoch(),
                PendingVerification::Finalized { finalization, .. } => finalization.epoch(),
            };
            by_epoch.entry(epoch).or_default().push(i);
        }

        // Verify each epoch group under the scope captured at admission, so a
        // provider that has since retired the epoch cannot fail the delivery.
        let mut verified = vec![false; delivers.len()];
        for indices in by_epoch.values() {
            let scoped = delivers[indices[0]].scoped();
            let group: Vec<_> = indices.iter().map(|&i| certs[i]).collect();
            let results =
                verify_certificates(self.context.as_mut(), scoped, &group, &self.strategy);
            for (j, &idx) in indices.iter().enumerate() {
                verified[idx] = results[j];
            }
        }

        // Process each verified item, rejecting unverified ones.
        for (index, item) in delivers.drain(..).enumerate() {
            if !verified[index] {
                match item {
                    PendingVerification::Finalized { response, .. }
                    | PendingVerification::Notarized { response, .. } => {
                        response.send_lossy(Outcome::Invalid);
                    }
                }
                continue;
            }
            match item {
                PendingVerification::Finalized {
                    finalization,
                    block,
                    response,
                    ..
                } => {
                    // Valid finalization received.
                    response.send_lossy(Outcome::Resolved(Resolution::Complete));
                    let block = V::from_application_block(block, finalization.proposal.payload);
                    let round = finalization.round();
                    let height = block.height();
                    debug!(?round, %height, "received finalization");

                    // The floor-anchor path fully handles this finalization
                    // and moves the lower bound past it.
                    let anchored;
                    (self, anchored) = self
                        .ingest(block.clone(), buffer, application, resolver)
                        .await;
                    if anchored {
                        continue;
                    }

                    self = self
                        .witness_round(height, round, buffer, application, resolver)
                        .await;
                    self.stream = self
                        .stream
                        .sync()
                        .await
                        .expect("round floor sync must succeed");
                    (self, _) = self
                        .store_finalization(height, &block, Some(finalization), application)
                        .await;
                }
                PendingVerification::Notarized {
                    notarization,
                    block,
                    response,
                    ..
                } => {
                    // Notarization alone does not prove the commitment encodes the block,
                    // so decoding must recompute it.
                    let commitment = notarization.proposal.payload;
                    let Ok(block) = V::decode_block(
                        block,
                        &self.block_codec_config,
                        ExpectedCommitment::Untrusted(commitment),
                        &self.strategy,
                    )
                    .await
                    else {
                        response.send_lossy(Outcome::Invalid);
                        continue;
                    };

                    // Valid notarization received.
                    response.send_lossy(Outcome::Resolved(Resolution::Complete));
                    let round = notarization.round();
                    let digest = V::commitment_to_inner(commitment);
                    debug!(?round, ?digest, "received notarization");

                    // Cache the notarization and block, blocking until both are
                    // durable (or the runtime is shutting down) so the repair
                    // bookkeeping below never runs ahead of storage.
                    let height = block.height();
                    let block_sync;
                    (self.cache, block_sync) =
                        self.cache.put_notarized(round, digest, &block).await;
                    let notarization_sync;
                    (self.cache, notarization_sync) = self
                        .cache
                        .put_notarization(round, digest, &notarization)
                        .await;
                    join(
                        block_sync.durable(round, "notarized"),
                        notarization_sync.durable(round, "notarization"),
                    )
                    .await;

                    // A notarized delivery can carry the pending floor block
                    // after the finalization is cached.
                    let anchored;
                    (self, anchored) = self
                        .ingest(block.clone(), buffer, application, resolver)
                        .await;
                    if anchored {
                        continue;
                    }

                    // If there exists a finalization certificate for this block, we
                    // should finalize it. This could finalize the block faster when
                    // a notarization then a finalization are received via consensus
                    // and we resolve the notarization request before the block request.
                    if let Some(finalization) = self.cache.get_finalization_for(digest).await {
                        self = self
                            .witness_round(
                                height,
                                finalization.round(),
                                buffer,
                                application,
                                resolver,
                            )
                            .await;
                        self.stream = self
                            .stream
                            .sync()
                            .await
                            .expect("round floor sync must succeed");

                        // SAFETY: `digest` identifies a unique `commitment`, so this
                        // cached finalization payload must match `V::commitment(&block)`.
                        (self, _) = self
                            .store_finalization(height, &block, Some(finalization), application)
                            .await;
                    }
                }
            }
        }
        self
    }

    /// Returns the epoch containing `height` and the scope that verifies its certificates.
    fn scoped_for_height(&self, height: Height) -> Option<(Epoch, Scoped<P::Scheme>)> {
        let epoch = self.epocher.containing(height)?.epoch();
        let scoped = self.provider.scoped(epoch)?;
        Some((epoch, scoped))
    }

    /// Attempt to dispatch the next finalized block to the application if ready.
    ///
    /// Dispatch finalized blocks to the application until the pipeline is full
    /// or no more blocks are available.
    ///
    /// This does NOT advance the processed floor height or sync metadata. It only
    /// sends blocks to the application and enqueues pending acks. Metadata is
    /// updated later, in a subsequent `select_loop!` iteration, when the ack
    /// handler updates the processed height.
    ///
    /// Blocks are dispatched only once durable. Every buffered
    /// finalized-archive write freezes dispatch at or above its height until
    /// a sync covering it completes (see [`DispatchGate`]). Dispatch is
    /// re-attempted by the pool-completion arm for pooled syncs and by the
    /// caller itself after a blocking sync. Callers that buffer writes must
    /// still call [`Self::sync_finalized`] or [`Self::start_finalized_sync`]
    /// before yielding to the `select_loop!` so the freeze is released.
    ///
    /// Acks are processed in FIFO order so the processed floor height always
    /// advances sequentially.
    ///
    /// # Crash safety
    ///
    /// Because `select_loop!` arms run to completion, archive data is always
    /// durable before the ack handler advances the processed floor height:
    ///
    /// ```text
    /// Iteration N (caller):
    ///   store_finalization   ->  Archive::put (buffered)
    ///   sync_finalized       ->  archive durable
    ///   try_dispatch_blocks  ->  sends durable blocks to app, enqueues pending acks
    ///
    /// Iteration M (ack handler, M > N):
    ///   ack handler       ->  update_processed         ->  metadata buffered
    ///   stream.sync       ->  metadata durable
    /// ```
    async fn try_dispatch_blocks(
        mut self: Box<Self>,
        application: &mut impl Reporter<Activity = Update<V::ApplicationBlock, A>>,
    ) -> Box<Self> {
        // Dispatch resumes after the floor anchor is durably stored.
        if self.floor.blocks_progress() {
            return self;
        }

        // Durability barrier: buffered writes are readable from the archives
        // before they are durable. Never dispatch at or above the lowest
        // write not yet covered by a completed sync.
        let barrier = self.dispatch_gate.barrier();
        while self.pending_acks.has_capacity() {
            let next_height = self
                .pending_acks
                .next_dispatch_height(self.stream.next_height());
            if barrier.is_some_and(|lowest| next_height >= lowest) {
                return self;
            }

            // Prefer the staged block and fall back to the archive
            let block = match self.staged.remove(next_height) {
                Some(block) => block,
                None => match self.get_finalized_block(next_height).await {
                    Some(stored) => V::restore_across(stored, &self.strategy).await,
                    None => return self,
                },
            };
            let height = block.height();
            assert_eq!(height, next_height, "finalized block height mismatch");

            let (ack, ack_waiter) = A::handle();
            application.report(Update::Block(V::into_shared(block), ack));
            self.pending_acks.enqueue(PendingAck {
                height,
                receiver: ack_waiter,
            });
            self.staged.retain(height.next());
        }
        self
    }

    /// Sync both finalization archives to durable storage, blocking the actor
    /// until they are durable.
    ///
    /// Must be called within the same `select_loop!` arm as any preceding
    /// [`Self::store_finalization`] / [`Self::try_repair_gaps`] writes, before yielding back
    /// to the loop. This is the durability barrier for application delivery:
    /// [`Self::try_dispatch_blocks`] must run only after this sync completes.
    /// It also ensures archives are durable before the ack handler advances
    /// the processed floor height. See [`Self::try_dispatch_blocks`] for details.
    ///
    /// Blocking the actor stalls every mailbox caller behind the sync.
    /// Prefer [`Self::start_finalized_sync`] unless work later in the same
    /// arm requires the writes to already be durable.
    #[tracing::instrument(name = "marshal.actor.sync_finalized", level = "info", skip_all)]
    async fn sync_finalized(mut self: Box<Self>) -> Box<Self> {
        (self.finalized_blocks, self.finalizations_by_height) = try_join!(
            self.finalized_blocks.sync().map_err(BoxedError::from),
            self.finalizations_by_height
                .sync()
                .map_err(BoxedError::from),
        )
        .unwrap_or_else(|e| panic!("failed to sync finalization archives: {e}"));

        // Everything accepted before this sync is now durable, so nothing
        // remains to gate dispatch.
        self.dispatch_gate.clear();
        self
    }

    /// Start a non-blocking sync of both finalization archives on the
    /// durability pool. A no-op if nothing was written since the last sync
    /// (blocking or pooled) started.
    ///
    /// The pooled entry resolves to [`PooledSync::Finalized`] once every write
    /// accepted before this call is durable. The sync adopts every deferred
    /// write (see [`DispatchGate::adopt`]), and until the pool-completion arm
    /// observes the completion, [`Self::try_dispatch_blocks`] will not
    /// dispatch at or above the lowest height a pending batch wrote. This
    /// preserves the durability barrier described there without blocking the
    /// mailbox on a sync like [`Self::sync_finalized`].
    ///
    /// Like [`Self::sync_finalized`], this must be called within the same
    /// `select_loop!` arm as the writes it covers, before yielding back to the
    /// loop. `round` only labels the sync in diagnostics.
    #[tracing::instrument(name = "marshal.actor.start_finalized_sync", level = "info", skip_all)]
    async fn start_finalized_sync(
        mut self: Box<Self>,
        round: Round,
        syncs: &mut Pool<'_, PooledSync>,
    ) -> Box<Self> {
        // If no write needs syncing, every accepted write is already covered
        // by a blocking or in-flight sync.
        let Some(seq) = self.dispatch_gate.adopt() else {
            return self;
        };

        let (blocks, finalizations);
        (
            (self.finalized_blocks, blocks),
            (self.finalizations_by_height, finalizations),
        ) = try_join!(
            self.finalized_blocks.start_sync().map_err(BoxedError::from),
            self.finalizations_by_height
                .start_sync()
                .map_err(BoxedError::from),
        )
        .unwrap_or_else(|e| panic!("failed to start finalization archive sync: {e}"));
        syncs.push(async move {
            let (blocks, finalizations) = join(
                blocks.durable(round, "finalized blocks"),
                finalizations.durable(round, "finalizations"),
            )
            .await;
            if blocks && finalizations {
                PooledSync::Finalized(seq)
            } else {
                // Runtime shutdown before the sync completed: nothing may be
                // released for dispatch.
                PooledSync::Observed
            }
        });
        self
    }

    /// Get a finalized block from the immutable archive in its stored form.
    async fn get_finalized_block(&self, height: Height) -> Option<V::StoredBlock> {
        match self
            .finalized_blocks
            .get(ArchiveID::Index(height.get()))
            .await
        {
            Ok(stored) => stored,
            Err(e) => panic!("failed to get block: {e}"),
        }
    }

    /// Check whether a finalized block exists in the archive at `height` without
    /// fetching it.
    fn has_finalized_block(&self, height: Height) -> bool {
        self.finalized_blocks.next_gap(height).0.is_some()
    }

    /// Get a finalization from the archive by height.
    async fn get_finalization_by_height(
        &self,
        height: Height,
    ) -> Option<Finalization<P::Scheme, V::Commitment>> {
        match self
            .finalizations_by_height
            .get(ArchiveID::Index(height.get()))
            .await
        {
            Ok(finalization) => finalization,
            Err(e) => panic!("failed to get finalization: {e}"),
        }
    }

    /// Check whether a finalization exists in the archive at `height` without
    /// fetching it.
    async fn has_finalization_by_height(&self, height: Height) -> bool {
        match self.finalizations_by_height.has(height).await {
            Ok(has) => has,
            Err(e) => panic!("failed to check finalization: {e}"),
        }
    }

    /// Get finalized block information from either the finalization archive or
    /// the finalized-block archive.
    async fn get_info_by_height(
        &self,
        height: Height,
    ) -> Option<(Height, <V::Block as Digestible>::Digest)> {
        if let Some(finalization) = self.get_finalization_by_height(height).await {
            return Some((
                height,
                V::commitment_to_inner(finalization.proposal.payload),
            ));
        }

        let stored = self.get_finalized_block(height).await?;
        let block = V::restore_across(stored, &self.strategy).await;
        Some((block.height(), block.digest()))
    }

    /// Add a finalized block, and optionally a finalization, to the archive.
    ///
    /// After persisting the block, the caller must sync finalized archives
    /// before dispatching the next contiguous block to the application. The
    /// buffered archive writes from this method are not a sufficient durability
    /// guarantee for downstream application state transitions on their own.
    ///
    /// Writes for a block above the processed height are buffered and not synced. The caller must
    /// call [sync_finalized](Self::sync_finalized) (blocking) or
    /// [start_finalized_sync](Self::start_finalized_sync) (pooled) before yielding to the
    /// `select_loop!` so that archive data is durable before the ack handler advances the
    /// processed floor height. See [`Self::try_dispatch_blocks`] for the crash safety invariant.
    ///
    /// A block at or below the processed height is not stored or dispatched. When the archive
    /// holds that block without its finalization, the finalization is stored and synced. A
    /// finalization for a block the archive lacks is dropped, because that block was pruned or
    /// sits below a state-sync floor.
    ///
    /// Walkback repair stores a block without its finalization when the cache lacks one, and an
    /// unclean shutdown during the concurrent sync of both archives can leave a block durable
    /// without its finalization. In either case the application can process the block before its
    /// finalization is reported.
    ///
    /// Returns whether the block is above the processed height.
    async fn store_finalization(
        mut self: Box<Self>,
        height: Height,
        block: &V::Block,
        finalization: Option<Finalization<P::Scheme, V::Commitment>>,
        application: &mut impl Reporter<Activity = Update<V::ApplicationBlock, A>>,
    ) -> (Box<Self>, bool) {
        let digest = block.digest();
        if height <= self.floor.processed_height() {
            // Both existence checks read archive indices rather than stored values, so a
            // finalization replayed after restart or reported again adds no archive read here.
            let kept = if let Some(finalization) = finalization
                && self.has_finalized_block(height)
                && !self.has_finalization_by_height(height).await
            {
                Some(finalization)
            } else {
                None
            };
            let Some(finalization) = kept else {
                debug!(
                    %height,
                    floor = %self.floor.processed_height(),
                    ?digest,
                    "dropping finalization at or below processed height floor"
                );
                return (self, false);
            };

            // Every finalization at or below the processed height must be durable, because
            // [`Self::prune_finalized_archives`] keeps the newest one as the round-floor evidence
            // that restart reads. Syncing only the certificate archive leaves earlier deferred
            // writes in the dispatch gate, so the caller's sync still covers them and their
            // dispatch resumes when it completes.
            let round = finalization.round();
            self.finalizations_by_height = self
                .finalizations_by_height
                .put(height, digest, &finalization)
                .await
                .unwrap_or_else(|e| panic!("failed to store finalization: {e}"))
                .sync()
                .await
                .unwrap_or_else(|e| panic!("failed to sync finalization: {e}"));
            self.advance_tip(round, height, digest, application);
            debug!(%height, ?digest, "stored finalization for processed block");
            return (self, false);
        }

        // Convert block to storage format
        let stored: V::StoredBlock = block.clone().into();
        let round = finalization.as_ref().map(|f| f.round());

        // In parallel, update the finalized blocks and finalizations archives
        let finalizations_by_height = self.finalizations_by_height;
        (self.finalized_blocks, self.finalizations_by_height) = try_join!(
            // Update the finalized blocks archive
            self.finalized_blocks
                .put(digest, &stored)
                .map_err(BoxedError::from),
            // Update the finalizations archive (if provided)
            async {
                let store = if let Some(finalization) = finalization {
                    finalizations_by_height
                        .put(height, digest, &finalization)
                        .await
                        .map_err(BoxedError::from)?
                } else {
                    finalizations_by_height
                };
                Ok::<_, BoxedError>(store)
            }
        )
        .unwrap_or_else(|e| panic!("failed to finalize: {e}"));

        // The write above is buffered and readable before it is durable, so
        // hold dispatch at or above it until a sync covers it.
        self.dispatch_gate.defer(height);

        // The tip is the newest stored finalization, so a block repaired without its finalization
        // leaves it unchanged.
        if let Some(round) = round {
            self.advance_tip(round, height, digest, application);
        }
        (self, true)
    }

    /// Records `height` as the finalized tip and reports it to the application when it is above
    /// the current tip.
    fn advance_tip(
        &mut self,
        round: Round,
        height: Height,
        digest: <V::Block as Digestible>::Digest,
        application: &mut impl Reporter<Activity = Update<V::ApplicationBlock, A>>,
    ) {
        if height <= self.tip {
            return;
        }
        application.report(Update::Tip(round, height, digest));
        self.tip = height;
        self.certified.retain(height.next());
        let _ = self.finalized_height.try_set(height.get());
    }

    /// Get the latest finalized block information (height and digest tuple).
    ///
    /// Blocks are only finalized directly with a finalization or indirectly via a descendant
    /// block's finalization. Thus, the highest known finalized block must itself have a direct
    /// finalization.
    ///
    /// We return the height and digest using the highest known finalization that we know the
    /// block height for. While it's possible that we have a later finalization, if we do not have
    /// the full block for that finalization, we do not know its height and therefore it would not
    /// yet be found in the `finalizations_by_height` archive. While not checked explicitly, we
    /// should have the associated block (in the `finalized_blocks` archive) for the information
    /// returned.
    async fn get_latest(&self) -> Option<(Height, <V::Block as Digestible>::Digest, Round)> {
        let height = self.finalizations_by_height.last_index()?;
        let finalization = self
            .get_finalization_by_height(height)
            .await
            .expect("finalization missing");
        Some((
            height,
            V::commitment_to_inner(finalization.proposal.payload),
            finalization.round(),
        ))
    }

    /// Looks for a block in cache and finalized storage by digest.
    async fn find_block_in_storage(
        &self,
        digest: <V::Block as Digestible>::Digest,
    ) -> Option<V::Block> {
        // Check verified / notarized blocks via cache manager.
        if let Some(block) = self.cache.find_block_matching(digest, |_| true).await {
            return Some(block);
        }
        // Check finalized blocks.
        match self.finalized_blocks.get(ArchiveID::Key(&digest)).await {
            Ok(stored) => stored.map(|stored| V::restore(stored, digest)),
            Err(e) => panic!("failed to get block: {e}"),
        }
    }

    /// Looks for a block in cache and finalized storage by full consensus commitment.
    async fn find_block_in_storage_by_commitment(
        &self,
        commitment: V::Commitment,
    ) -> Option<V::Block> {
        let digest = V::commitment_to_inner(commitment);
        if let Some(block) = self
            .cache
            .find_block_matching(digest, |block| V::commitment(block) == commitment)
            .await
        {
            return Some(block);
        }

        match self.finalized_blocks.get(ArchiveID::Key(&digest)).await {
            Ok(Some(stored)) => {
                let block = V::restore(stored, digest);
                (V::commitment(&block) == commitment).then_some(block)
            }
            Ok(None) => None,
            Err(e) => panic!("failed to get block: {e}"),
        }
    }

    /// Looks for a block anywhere in local storage using only the digest.
    ///
    /// This is used when we only have a digest (during gap repair following
    /// parent links).
    async fn find_block_by_digest<Buf: Buffer<V>>(
        &self,
        buffer: &Buf,
        digest: <V::Block as Digestible>::Digest,
    ) -> Option<V::Block> {
        if let Some(block) = buffer.find_by_digest(digest).await {
            return Some(block);
        }
        self.find_block_in_storage(digest).await
    }

    /// Looks for a block anywhere in local storage using the full commitment.
    ///
    /// This is used when we have a full commitment (from notarizations/finalizations).
    /// Having the full commitment may enable additional retrieval mechanisms.
    async fn find_block_by_commitment<Buf: Buffer<V>>(
        &self,
        buffer: &Buf,
        commitment: V::Commitment,
    ) -> Option<V::Block> {
        if let Some(block) = buffer.find_by_commitment(commitment).await {
            return Some(block);
        }
        self.find_block_in_storage_by_commitment(commitment).await
    }

    /// Attempt to repair any identified gaps in the finalized blocks archive. Gaps are walked back
    /// from their ends, lowest first, through locally available blocks; the first block a walk
    /// cannot find is requested by commitment, which ends the walk for this pass. Up to
    /// `self.max_repair` of the lowest missing heights, possibly spanning several gaps, are
    /// requested by finalization.
    ///
    /// This also handles the "trailing" case where finalizations exist beyond
    /// the last stored block (the block data was lost before a crash). The
    /// trailing block is anchored first so that backward gap repair can fill
    /// inward from it.
    ///
    /// Writes are buffered. Returns `true` if this call wrote repaired blocks and
    /// needs a subsequent [`sync_finalized`](Self::sync_finalized).
    #[tracing::instrument(name = "marshal.actor.try_repair_gaps", level = "info", skip_all)]
    async fn try_repair_gaps<Buf: Buffer<V>>(
        mut self: Box<Self>,
        buffer: &mut Buf,
        resolver: &mut impl Resolver<Key = ResolverRequestFor<V>, Subscriber = Annotation>,
        application: &mut impl Reporter<Activity = Update<V::ApplicationBlock, A>>,
    ) -> (Box<Self>, bool) {
        // Gap repair needs a known processed floor. A floor transition may
        // jump the lower bound once its anchor block arrives.
        if self.floor.blocks_progress() {
            return (self, false);
        }

        let mut wrote = false;
        let start = self.floor.processed_height().next();

        // If finalizations extend beyond the last stored block, anchor the
        // trailing block so the gap repair loop below can walk backward from it.
        if let Some(last_finalized) = self.finalizations_by_height.last_index() {
            let have_block = self
                .finalized_blocks
                .last_index()
                .is_some_and(|last| last >= last_finalized);
            if last_finalized > self.floor.processed_height() && !have_block {
                // Get the finalization for the last finalized block.
                let finalization = self
                    .get_finalization_by_height(last_finalized)
                    .await
                    .expect("finalization missing");
                let commitment = finalization.proposal.payload;
                if let Some(block) = self.find_block_by_commitment(buffer, commitment).await {
                    // If found, persist the block.
                    let stored;
                    (self, stored) = self
                        .store_finalization(last_finalized, &block, Some(finalization), application)
                        .await;
                    wrote |= stored;
                } else {
                    // Request the missing block.
                    self.floor
                        .fetch_if_permitted(
                            resolver,
                            Request::finalized_by_height(commitment, last_finalized),
                        )
                        .ignore();
                }
            }
        }

        // Fill internal gaps by walking backward from each gap's end block.
        'cache_repair: loop {
            let (gap_start, Some(gap_end)) = self.finalized_blocks.next_gap(start) else {
                // No gaps detected
                return (self, wrote);
            };

            // Attempt to repair the gap backwards from the end of the gap, using
            // blocks from our local storage. The walkback only needs each
            // block's height and parent linkage.
            let Some(cursor) = self.get_finalized_block(gap_end).await else {
                panic!("gapped block missing that should exist: {gap_end}");
            };
            let (mut height, mut parent_digest, mut parent_commitment) = (
                cursor.height(),
                cursor.parent(),
                V::stored_parent_commitment(&cursor),
            );

            // Compute the lower bound of the recursive repair. `gap_start` is `Some`
            // if `start` is not in a gap. We add one to it to ensure we don't
            // re-persist it to the database in the repair loop below.
            let gap_start = gap_start.map(Height::next).unwrap_or(start);

            // Iterate backwards, repairing blocks as we go.
            while height > gap_start {
                if let Some(block) = self
                    .find_block_by_commitment(buffer, parent_commitment)
                    .await
                {
                    let finalization = self.cache.get_finalization_for(parent_digest).await;
                    let next = (block.height(), block.parent(), V::parent_commitment(&block));
                    let stored;
                    (self, stored) = self
                        .store_finalization(next.0, &block, finalization, application)
                        .await;
                    wrote |= stored;
                    debug!(height = %next.0, "repaired block");
                    (height, parent_digest, parent_commitment) = next;
                } else {
                    // Request the next missing commitment.
                    //
                    // SAFETY: Finalized blocks are archived only after the
                    // parent relationship needed for walkback has been
                    // validated by marshal.
                    let parent_height = height
                        .previous()
                        .expect("cursor above gap start has a parent");
                    self.floor
                        .fetch_if_permitted(
                            resolver,
                            Request::finalized_by_height(parent_commitment, parent_height),
                        )
                        .ignore();
                    break 'cache_repair;
                }
            }
        }

        // Request any finalizations for missing items in the archive, up to
        // the `max_repair` quota. This may help shrink the size of the gap
        // closest to the application's processed height if finalizations
        // for the requests' heights exist. If not, we rely on the recursive
        // digest fetches above.
        let missing_items = self
            .finalized_blocks
            .missing_items(start, self.max_repair.get());
        let requests: Vec<_> = missing_items.into_iter().map(Request::finalized).collect();
        if !requests.is_empty() {
            self.floor
                .fetch_all_if_permitted(resolver, requests)
                .ignore();
        }
        (self, wrote)
    }

    /// Buffers a processed height update in memory and metrics. Does NOT sync
    /// to durable storage. Sync metadata after buffered updates to make them durable.
    fn update_processed(
        &mut self,
        processed: Processed,
        resolver: &mut impl Resolver<Key = ResolverRequestFor<V>, Subscriber = Annotation>,
    ) {
        let height = processed.height();
        self.stream.acknowledge(height);
        self.floor.set_processed(processed);
        let _ = self
            .processed_height
            .try_set(self.floor.processed_height().get());

        // Resolver request retention is independent of caller-owned block subscriptions.
        resolver.retain(handler::above_height_floor::<V::Commitment>(height));
    }

    /// Classifies a processed height by whether its block is stored.
    ///
    /// A floor installed above a missing predecessor records that predecessor as processed. The
    /// floor block is stored before the processed height is recorded.
    fn processed(finalized_blocks: &FB, height: Height) -> Processed {
        if finalized_blocks.next_gap(height).0.is_some() {
            Processed::Block(height)
        } else {
            Processed::Absent(height)
        }
    }

    /// Returns the height of the newest stored finalization at or below `height`.
    fn latest_finalization_height(finalizations_by_height: &FC, height: Height) -> Option<Height> {
        finalizations_by_height
            .ranges_from(Height::zero())
            .filter_map(|(start, end)| (start <= height).then_some(end.min(height)))
            .max()
    }

    /// Returns the round floor that the archives and `epocher` justify, and the newest archived
    /// finalization round of a processed block or its durable successor.
    ///
    /// The floor covers that round and every epoch that ends at or below the processed height, or
    /// at or below the successor when the successor block and its finalization are both archived.
    /// A certificate-only successor is not durable progress, so it neither raises the floor nor
    /// closes an epoch, and recovery never suppresses the fetch of its block.
    async fn derive_cutoff(
        finalizations_by_height: &FC,
        finalized_blocks: &FB,
        processed_height: Option<Height>,
        epocher: &ES,
        strategy: &T,
    ) -> (Cutoff, Round) {
        // The processed round is the round of the newest archived finalization at or below the
        // processed height, or zero when there is none.
        let processed_round = match processed_height
            .and_then(|height| Self::latest_finalization_height(finalizations_by_height, height))
        {
            Some(finalization_height) => match finalizations_by_height
                .get(ArchiveID::Index(finalization_height.get()))
                .await
            {
                Ok(Some(finalization)) => finalization.round(),
                Ok(None) => panic!("processed finalization missing from stored range"),
                Err(err) => panic!("failed to get processed finalization: {err}"),
            },
            None => Round::zero(),
        };

        // A durable successor raises the witnessed round to at least its finalization round and
        // becomes the anchor that closes epochs. Without one, the processed height is the anchor.
        let (witnessed, anchor) = match Self::durable_successor(
            finalizations_by_height,
            finalized_blocks,
            processed_height,
            strategy,
        )
        .await
        {
            Some((successor, round)) => (processed_round.max(round), Some(successor)),
            None => (processed_round, processed_height),
        };

        // The cutoff covers the witnessed round and every epoch that ends at or below the anchor.
        let cutoff = Cutoff::Round(witnessed);
        let cutoff = anchor
            .and_then(|anchor| closed_epoch(epocher, anchor))
            .map_or(cutoff, |epoch| cutoff.max(Cutoff::Epoch(epoch)));
        (cutoff, witnessed)
    }

    /// Returns the height and finalization round of the block just above `processed_height`, or of
    /// height zero when nothing is processed, if that block and a matching finalization are both
    /// archived.
    async fn durable_successor(
        finalizations_by_height: &FC,
        finalized_blocks: &FB,
        processed_height: Option<Height>,
        strategy: &T,
    ) -> Option<(Height, Round)> {
        let successor = match processed_height {
            None => Height::zero(),
            Some(height) if height.get() < u64::MAX => height.next(),
            Some(_) => return None,
        };
        let (block, finalization) = join(
            finalized_blocks.get(ArchiveID::Index(successor.get())),
            finalizations_by_height.get(ArchiveID::Index(successor.get())),
        )
        .await;
        let block = block.unwrap_or_else(|err| panic!("failed to get successor block: {err}"));
        let finalization = finalization
            .unwrap_or_else(|err| panic!("failed to get successor finalization: {err}"));
        let (Some(block), Some(finalization)) = (block, finalization) else {
            return None;
        };
        let block = V::restore_across(block, strategy).await;
        assert!(
            V::commitment(&block) == finalization.proposal.payload,
            "successor block does not match stored finalization"
        );
        Some((successor, finalization.round()))
    }

    /// Raises the round floor for an acknowledged block to cover its archived finalization, if
    /// any, and every epoch that ends at or below it. The new floor is buffered like
    /// [`Self::raise_cutoff`].
    async fn cover_processed<Buf: Buffer<V>>(
        mut self: Box<Self>,
        height: Height,
        buffer: &mut Buf,
        application: &mut impl Reporter<Activity = Update<V::ApplicationBlock, A>>,
        resolver: &mut impl Resolver<Key = ResolverRequestFor<V>, Subscriber = Annotation>,
    ) -> Box<Self> {
        if let Some(finalization) = self.get_finalization_by_height(height).await {
            self = self
                .witness_round(height, finalization.round(), buffer, application, resolver)
                .await;
        }
        self.cover_closed_epoch(height, buffer, application, resolver)
            .await
    }

    /// Raises the round floor to cover every epoch that ends at or below `anchor`.
    ///
    /// The block at `anchor` must be processed or stored just above the processed height. The new
    /// floor is buffered like [`Self::raise_cutoff`].
    async fn cover_closed_epoch<Buf: Buffer<V>>(
        self: Box<Self>,
        anchor: Height,
        buffer: &mut Buf,
        application: &mut impl Reporter<Activity = Update<V::ApplicationBlock, A>>,
        resolver: &mut impl Resolver<Key = ResolverRequestFor<V>, Subscriber = Annotation>,
    ) -> Box<Self> {
        let Some(epoch) = closed_epoch(&self.epocher, anchor) else {
            return self;
        };
        self.raise_cutoff(Cutoff::Epoch(epoch), buffer, application, resolver)
            .await
    }

    /// Witnesses `round` and raises the processed round floor through it, provided `height` is at
    /// or below the processed height. A newer round replaces the witnessed finalization round.
    ///
    /// The new floor is buffered in the stream metadata. The caller must sync the stream before
    /// yielding to the `select_loop!`, so a raised floor is durable before the actor handles its
    /// next event.
    async fn witness_round<Buf: Buffer<V>>(
        mut self: Box<Self>,
        height: Height,
        round: Round,
        buffer: &mut Buf,
        application: &mut impl Reporter<Activity = Update<V::ApplicationBlock, A>>,
        resolver: &mut impl Resolver<Key = ResolverRequestFor<V>, Subscriber = Annotation>,
    ) -> Box<Self> {
        if height > self.floor.processed_height() {
            return self;
        }

        // Retain view-indexed cache data for `view_retention` views behind the previously
        // witnessed finalization round.
        if let Some(previous) = self.floor.witness(round) {
            // Views restart with each epoch and the number of views earlier epochs spanned is
            // unknown here, so the window is measured within the previous round's epoch and
            // stops at its first view. The views it could not cover there are owed by whole
            // earlier epochs, one per view, because every epoch spans at least one view; this
            // keeps at least the window and usually more. With a window of ten views, (3, 15)
            // prunes behind (3, 5), while (8, 3) takes three views from epoch 8, owes seven to
            // epochs 7 through 1, and prunes behind (1, 0).
            let view = previous.view().saturating_sub(self.view_retention);
            let owed = self
                .view_retention
                .get()
                .saturating_sub(previous.view().get());
            let epoch = previous.epoch().saturating_sub(EpochDelta::new(owed));

            // Destroy the caches of every epoch below `epoch` and prune the view-indexed archives
            // of `epoch` below `view`. Pruning removes whole sections, so entries below `view` in
            // the section that contains it can remain.
            self.cache = self.cache.prune_by_view(Round::new(epoch, view)).await;
        }

        // The round floor covers the witnessed round, which retires round-bound requests at or
        // below it.
        self.raise_cutoff(Cutoff::Round(round), buffer, application, resolver)
            .await
    }

    /// Raises the processed round floor to `cutoff`, buffers it in the stream metadata, and prunes
    /// round-bound requests it covers.
    ///
    /// A pending floor anchor whose round `cutoff` covers is released, and repair and dispatch
    /// resume without it.
    async fn raise_cutoff<Buf: Buffer<V>>(
        mut self: Box<Self>,
        cutoff: Cutoff,
        buffer: &mut Buf,
        application: &mut impl Reporter<Activity = Update<V::ApplicationBlock, A>>,
        resolver: &mut impl Resolver<Key = ResolverRequestFor<V>, Subscriber = Annotation>,
    ) -> Box<Self> {
        if cutoff <= self.floor.cutoff() {
            return self;
        }

        self.floor.set_cutoff(cutoff);
        self.stream.set_cutoff(cutoff);

        // Resolver request retention is independent of caller-owned block subscriptions.
        resolver.retain(handler::above_round_floor::<V::Commitment>(cutoff));

        // Finalized rounds and heights increase together along the finalized chain, so a pending
        // anchor at a round the cutoff covers sits at or below the processed height. The floor it
        // announced is already active, so drop it and resume repair and dispatch.
        match self.floor.pending() {
            Some(pending) if cutoff.covers(pending.round()) => self.floor.clear_pending(),
            _ => return self,
        }
        let repaired;
        (self, repaired) = self.try_repair_gaps(buffer, resolver, application).await;
        if repaired {
            self = self.sync_finalized().await;
        }
        self.try_dispatch_blocks(application).await
    }

    /// Prunes finalized blocks below `height` and finalization certificates below the lower of
    /// `height` and the newest finalization at or below the processed height.
    async fn prune_finalized_archives(mut self: Box<Self>, height: Height) -> Box<Self> {
        // The newest finalization at or below the processed height stays queryable by height after
        // its block is pruned. Restart reads its round through [`Self::derive_cutoff`] to seed the
        // witnessed round and to raise the persisted round floor when it is missing or lower.
        // Blocks repaired by walking back from a later finalization may be stored without
        // certificates, so the retained finalization can sit far below the processed height.
        let certificate_height = Self::latest_finalization_height(
            &self.finalizations_by_height,
            self.floor.processed_height(),
        )
        .map_or(height, |retained| retained.min(height));

        // Prune the finalized block and finalization certificate archives in parallel.
        (self.finalized_blocks, self.finalizations_by_height) = try_join!(
            self.finalized_blocks
                .prune(height)
                .map_err(BoxedError::from),
            self.finalizations_by_height
                .prune(certificate_height)
                .map_err(BoxedError::from),
        )
        .unwrap_or_else(|e| panic!("failed to prune finalized archives: {e}"));
        self
    }

    /// Prunes finalized archives and height-indexed certified cache data below the durable floor.
    ///
    /// The floor block and its finalization sit at `height + 1`, so restart restores the floor's
    /// round from them as the processed height's successor until acknowledgements move the
    /// processed height onto the floor block, after which that finalization is the newest one
    /// at or below the processed height (see [`Self::prune_finalized_archives`]).
    async fn prune_after_floor(mut self: Box<Self>, height: Height) -> Box<Self> {
        (
            self.cache,
            self.finalized_blocks,
            self.finalizations_by_height,
        ) = try_join!(
            self.cache.prune_by_height(height).map(Ok::<_, BoxedError>),
            self.finalized_blocks
                .prune(height)
                .map_err(BoxedError::from),
            self.finalizations_by_height
                .prune(height)
                .map_err(BoxedError::from),
        )
        .unwrap_or_else(|e| panic!("failed to prune data below floor: {e}"));
        self
    }
}