commonware_consensus/marshal/core/

actor.rs

use super::{
    Buffer, Retirement, Variant,
    acks::{PendingAck, PendingAcks},
    cache,
    delivery::PendingVerification,
    durability::{DispatchGate, Durable as _},
    floor::{Floor, State as FloorState},
    mailbox::{CommitmentFallback, Mailbox, Message},
    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, Epocher, Height, Round, ViewDelta},
};
use bytes::Bytes;
use commonware_actor::mailbox;
use commonware_codec::{Decode, Encode, Read};
use commonware_cryptography::{
    Digestible,
    certificate::{Provider, Scoped, Verifier},
};
use commonware_macros::{boxed, select_loop};
use commonware_p2p::Recipients;
use commonware_parallel::Strategy;
use commonware_resolver::{Delivery, Resolver, TargetedResolver};
use commonware_runtime::{
    BufferPooler, 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::{join, join_all},
    try_join,
};
use rand_core::CryptoRng;
use std::{collections::BTreeMap, future::Future, num::NonZeroUsize, sync::Arc};
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<bool>,
}

/// 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: BufferPooler + 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 ----------
    context: ContextCell<E>,

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

    // ---------- Configuration ----------
    // 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 blocks to repair at once
    max_repair: NonZeroUsize,
    // Codec configuration for block type
    block_codec_config: <V::ApplicationBlock as Read>::Cfg,
    // Strategy for parallel operations
    strategy: T,

    // ---------- State ----------
    // Current durable floor and any update awaiting its anchor block
    floor: FloorState<P::Scheme, V::Commitment>,
    // Application delivery cursor
    stream: Stream<E>,
    // Pending application acknowledgements
    pending_acks: PendingAcks<V, A>,
    // Acknowledgements cleared while a floor transition owns application progress
    cleared_acks: Vec<(Height, V::Commitment)>,
    // Highest known finalized height
    tip: Height,
    // Outstanding subscriptions for blocks
    block_subscriptions: Subscriptions<V>,
    // Defers application dispatch of finalized-archive writes until a sync
    // covering them completes
    dispatch_gate: DispatchGate,

    // ---------- Storage ----------
    // 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,

    // ---------- Metrics ----------
    // 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: BufferPooler + 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 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 pending_floor_anchor = 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)
                        .await;
                None
            }
            Start::Floor(finalization) => Some(finalization),
        };
        let last_processed_round = Self::latest_processed_round(
            &finalizations_by_height,
            &finalized_blocks,
            last_processed_height,
        )
        .await;

        // 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 = pending_floor_anchor.map_or_else(
            || FloorState::resolved(last_processed_height, last_processed_round),
            |finalization| {
                FloorState::awaiting_anchor(
                    last_processed_height,
                    last_processed_round,
                    finalization,
                )
            },
        );
        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()),
                cleared_acks: Vec::new(),
                tip: Height::zero(),
                block_subscriptions: Subscriptions::new(),
                dispatch_gate: DispatchGate::default(),
                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>,
    ) -> 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::Block = stored.into();
                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.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<Arc<V::Block>, SubscriptionKeyFor<V>>>::default();

        // Observe durable syncs that no consensus caller awaits (the
        // notarization and finalization paths). A flush failure inside
        // `start_sync` is reported only through the returned handle, so every
        // handle must be observed to apply the fatal policy. This pool does
        // so without blocking the actor on a sync.
        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) = async move {
            // Get tip and send to application
            let tip = self.get_latest().await;
            if let Some((height, digest, round)) = tip {
                application.report(Update::Tip(round, height, digest));
                self.tip = height;
                let _ = self.finalized_height.try_set(height.get());
            }

            // 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 fetch the anchor block.
            if let Some(finalization) = self.floor.take_pending_anchor() {
                self = self
                    .install_floor(
                        finalization,
                        false,
                        &mut resolver,
                        &mut buffer,
                        &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)
        }
        .instrument(info_span!("marshal.actor.start"))
        .await;

        select_loop! {
            self.context,
            on_start => {
                // 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));
        let mut processed_commitments = Vec::new();
        let processed_round = loop {
            let (height, commitment, 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.
                    self.update_processed_height(height, resolver);
                    self = self
                        .update_processed_round(height, buffer, application, resolver)
                        .await;
                }
                Err(e) => return Err((height, e)),
            }
            processed_commitments.push(commitment);

            // 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 self.floor.round(),
            }
        };

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

        // The round is an inclusive floor. Retire every exact commitment even if
        // sparse certificates leave it above that floor.
        buffer.retire(Retirement {
            round_floor: processed_round,
            exact_retirements: processed_commitments,
        });

        // 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<Arc<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) => self
                        .finalized_blocks
                        .get(ArchiveID::Key(&digest))
                        .await
                        .ok()
                        .flatten()
                        .map(|b| (b.height(), digest)),
                    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 = self.cache.get_verified(round).await.map(Into::into);
                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 may broadcast a conflicting block for the same
                // round after restart. This is tolerated: extra block bytes
                // cannot form a conflicting certificate (unlike votes), block
                // storage tolerates multiple candidates per round (see
                // [Mailbox::get_verified]), and the propose paths skip or
                // reuse a recovered block on restart.
                buffer.send(round, Arc::clone(&block), recipients);
                self = self
                    .persist_verified(round, block, ack, buffer, application, resolver)
                    .await;
            }
            Message::Verified {
                round, block, ack, ..
            } => {
                self = self
                    .persist_verified(round, block, ack, buffer, application, resolver)
                    .await;
            }
            Message::Certified {
                round, block, ack, ..
            } => {
                (self, _) = self
                    .ingest(Arc::clone(&block), 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, Arc::unwrap_or_clone(block).into())
                        .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::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(Arc::clone(&block), 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, Arc::unwrap_or_clone(block).into())
                            .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.clone())
                    .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(Arc::clone(&block), buffer, application, resolver)
                        .await;
                    if anchored {
                        return self;
                    }

                    let height = block.height();
                    let stored;
                    (self, stored) = self
                        .update_processed_round_floor(height, round, buffer, application, resolver)
                        .await
                        .store_finalization(
                            height,
                            digest,
                            Arc::unwrap_or_clone(block),
                            Some(finalization),
                            application,
                        )
                        .await;
                    if stored {
                        // 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_block_by_round(commitment, round),
                        )
                        .ignore();
                }
            }
            Message::GetBlock {
                identifier,
                response,
                ..
            } => match identifier {
                BlockID::Digest(digest) => {
                    let result = self
                        .find_block_by_digest(buffer, digest)
                        .await
                        .map(Arc::unwrap_or_clone);
                    response.send_lossy(result);
                }
                BlockID::Height(height) => {
                    let result = self.get_finalized_block(height).await;
                    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,
                    }
                    .map(Arc::unwrap_or_clone);
                    response.send_lossy(block);
                }
            },
            Message::GetFinalization {
                height, response, ..
            } => {
                let finalization = self.get_finalization_by_height(height).await;
                response.send_lossy(finalization);
            }
            Message::GetProcessedHeight { response, .. } => {
                response.send_lossy(self.stream.processed_height());
            }
            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, true, resolver, buffer, 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 node's processed round when it started.
        let round = self.floor.round();
        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(block) = self.get_finalized_block(height).await else {
                    debug!(%height, "finalized block missing on request");
                    return;
                };
                response.send_lossy((finalization, V::into_inner(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<Arc<V::Block>>,
        resolver: &mut impl Resolver<Key = ResolverRequestFor<V>, Subscriber = Annotation>,
        waiters: &mut AbortablePool<'_, Result<Arc<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 {
            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")
                    }
                };

                // This path is only for accepted ancestry or finalized repair,
                // never for a candidate block's immediate parent.
                self.floor
                    .fetch_if_permitted(resolver, Request::certified_block(commitment, height))
                    .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, fetching the anchor block if needed.
    async fn install_floor<Buf, R>(
        mut self: Box<Self>,
        finalization: Finalization<P::Scheme, V::Commitment>,
        skip_if_superseded: bool,
        resolver: &mut R,
        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 round = finalization.round();
        let processed_round = self.floor.round();
        if round <= processed_round {
            warn!(
                ?round,
                floor = ?processed_round,
                "floor not updated, below 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.clone())
            .await;

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

        if let Some(block) = self.find_block_by_commitment(buffer, commitment).await {
            self.floor.await_anchor(finalization);
            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,
        // but retain their heights and commitments until the anchor makes the floor active.
        self.cleared_acks.extend(self.pending_acks.clear());

        debug!(?round, ?commitment, "starting fetch for floor block");
        self.floor.await_anchor(finalization);
        self.floor
            .fetch_if_permitted(
                resolver,
                Request::finalized_block_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: Arc<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(Arc::clone(&block), buffer, application, resolver)
            .await;
        let digest = block.digest();
        let handle;
        (self.cache, handle) = self
            .cache
            .put_verified(round, digest, Arc::unwrap_or_clone(block).into())
            .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: Arc<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(Arc::clone(&block));

        if !self.floor.matches_pending_anchor(V::commitment(&block)) {
            return (self, false);
        }

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

    /// Applies the pending floor transition using its matching anchor block.
    ///
    /// # Panics
    ///
    /// Panics if no pending floor anchor is installed.
    async fn apply_pending_floor<Buf: Buffer<V>>(
        mut self: Box<Self>,
        block: Arc<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.
        if height <= self.floor.processed_height() {
            warn!(
                %height,
                existing = %self.floor.processed_height(),
                "floor not updated, at or below existing"
            );
            let finalization = self
                .floor
                .take_pending_anchor()
                .expect("pending floor anchor missing");
            self = self
                .update_processed_round_floor(
                    height,
                    finalization.round(),
                    buffer,
                    application,
                    resolver,
                )
                .await;
            let commitments = self.take_superseded_ack_commitments();
            buffer.retire(Retirement {
                round_floor: self.floor.round(),
                exact_retirements: commitments,
            });
            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 finalization = self
            .floor
            .take_pending_anchor()
            .expect("pending floor anchor missing");
        let round = finalization.round();
        (self.finalized_blocks, self.finalizations_by_height) = try_join!(
            self.finalized_blocks
                .put(Arc::unwrap_or_clone(block).into())
                .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;

        if height > self.tip {
            application.report(Update::Tip(round, height, digest));
            self.tip = height;
            let _ = self.finalized_height.try_set(height.get());
        }

        // 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");
        self.update_processed_height(dispatch_floor, resolver);
        self = self
            .update_processed_round_floor(dispatch_floor, round, 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.cleared_acks.extend(self.pending_acks.clear());

        // The active floor retires round-bound entries and every commitment whose
        // acknowledgement it superseded.
        let commitments = self.take_superseded_ack_commitments();
        buffer.retire(Retirement {
            round_floor: self.floor.round(),
            exact_retirements: commitments,
        });

        // The floor is durable, so cache/finalized data below it can be pruned.
        self = self.prune_after_floor(height).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
    }

    /// Takes cleared acknowledgement commitments covered by the active processed-height floor.
    ///
    /// Cleared acknowledgements above the floor are re-dispatched and remain live.
    fn take_superseded_ack_commitments(&mut self) -> Vec<V::Commitment> {
        let processed_height = self.floor.processed_height();
        std::mem::take(&mut self.cleared_acks)
            .into_iter()
            .filter_map(|(height, commitment)| (height <= processed_height).then_some(commitment))
            .collect()
    }

    /// 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.
    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) => {
                let block_cfg = V::block_cfg(&self.block_codec_config, commitment);
                let Ok(block) = V::Block::decode_cfg(value.as_ref(), &block_cfg) else {
                    response.send_lossy(false);
                    return self;
                };
                if V::commitment(&block) != commitment {
                    response.send_lossy(false);
                    return self;
                }

                // 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 block = Arc::new(block);
                let anchored;
                (self, anchored) = self
                    .ingest(Arc::clone(&block), buffer, application, resolver)
                    .await;
                if anchored {
                    response.send_lossy(true);
                    return self;
                }

                // The peer-visible request only says "give me this block".
                // Local annotations explain why the block was requested and
                // therefore where, if anywhere, it should be stored.
                let height = block.height();
                let digest = block.digest();
                let annotations = subscribers
                    .map_into(|(annotation, _)| annotation)
                    .into_vec();

                // 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
                        .update_processed_round_floor(
                            height,
                            finalization.round(),
                            buffer,
                            application,
                            resolver,
                        )
                        .await;
                }
                if finalization.is_some()
                    || annotations
                        .iter()
                        .any(|annotation| matches!(annotation, Annotation::Finalized(_)))
                {
                    (self, _) = self
                        .store_finalization(
                            height,
                            digest,
                            Arc::unwrap_or_clone(block),
                            finalization,
                            application,
                        )
                        .await;
                } else if annotations.iter().any(|annotation| {
                    matches!(
                        annotation,
                        Annotation::Certified { 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,
                            Arc::unwrap_or_clone(block).into(),
                        )
                        .await;
                }
                debug!(?digest, %height, "received block");
                response.send_lossy(true);
            }
            Key::Finalized { height } => {
                let Some((epoch, scoped)) = self.scoped_for_height(height) else {
                    debug!(
                        %height,
                        floor = %self.floor.processed_height(),
                        "ignoring stale delivery"
                    );
                    response.send_lossy(true);
                    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(false);
                    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(false);
                    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(false);
                    return self;
                };

                // In contrast to the `Block` and `Notarization` deliveries, the finalization delivery
                // is guaranteed to be certified (assuming the certificate verifies). Because of this,
                // we can skip broader payload checks and just check that the application block matches
                // the commitment in the finalization proposal.
                //
                // TODO(https://github.com/commonwarexyz/monorepo/issues/3938): Apply this pattern
                // conditionally to `Request::Block` and `Request::Notarized`, if the requester knows
                // the requested block is certified.
                let commitment = finalization.proposal.payload;
                if block.height() != height || block.digest() != V::commitment_to_inner(commitment)
                {
                    response.send_lossy(false);
                    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
                // delivery is acknowledged as stale.
                let Some(scheme) = self.provider.scheme(round.epoch()) else {
                    debug!(
                        ?round,
                        floor = %self.floor.processed_height(),
                        "ignoring stale delivery"
                    );
                    response.send_lossy(true);
                    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(false);
                    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(false);
                    return self;
                }

                // Use the notarization payload to derive the block decode config. Below, the
                // decoded block is checked against the same payload.
                let commitment = notarization.proposal.payload;
                if !V::check_payload(scheme.as_ref(), commitment) {
                    response.send_lossy(false);
                    return self;
                }
                let block_cfg = V::block_cfg(&self.block_codec_config, commitment);
                let Ok(block) = V::Block::decode_cfg(value, &block_cfg) else {
                    response.send_lossy(false);
                    return self;
                };

                if V::commitment(&block) != notarization.proposal.payload {
                    response.send_lossy(false);
                    return self;
                }
                delivers.push(PendingVerification::Notarized {
                    scoped: Scoped::scheme(scheme),
                    notarization,
                    block,
                    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(false);
                    }
                }
                continue;
            }
            match item {
                PendingVerification::Finalized {
                    finalization,
                    block,
                    response,
                    ..
                } => {
                    // Valid finalization received.
                    response.send_lossy(true);
                    let block = Arc::new(V::from_application_block(
                        block,
                        finalization.proposal.payload,
                    ));
                    let round = finalization.round();
                    let height = block.height();
                    let digest = block.digest();
                    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(Arc::clone(&block), buffer, application, resolver)
                        .await;
                    if anchored {
                        continue;
                    }

                    (self, _) = self
                        .update_processed_round_floor(height, round, buffer, application, resolver)
                        .await
                        .store_finalization(
                            height,
                            digest,
                            Arc::unwrap_or_clone(block),
                            Some(finalization),
                            application,
                        )
                        .await;
                }
                PendingVerification::Notarized {
                    notarization,
                    block,
                    response,
                    ..
                } => {
                    // Valid notarization received.
                    response.send_lossy(true);
                    let round = notarization.round();
                    let commitment = notarization.proposal.payload;
                    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 = Arc::new(block);
                    let block_sync;
                    (self.cache, block_sync) = self
                        .cache
                        .put_notarized(round, digest, block.as_ref().clone().into())
                        .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(Arc::clone(&block), 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
                            .update_processed_round_floor(
                                height,
                                finalization.round(),
                                buffer,
                                application,
                                resolver,
                            )
                            .await;

                        // SAFETY: `digest` identifies a unique `commitment`, so this
                        // cached finalization payload must match `V::commitment(&block)`.
                        (self, _) = self
                            .store_finalization(
                                height,
                                digest,
                                Arc::unwrap_or_clone(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))
    }

    // -------------------- Application Dispatch --------------------

    /// 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_height  ->  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;
            }
            let Some(block) = self.get_finalized_block(next_height).await else {
                return self;
            };
            assert_eq!(
                block.height(),
                next_height,
                "finalized block height mismatch"
            );

            let (height, commitment) = (block.height(), V::commitment(&block));
            let (ack, ack_waiter) = A::handle();
            application.report(Update::Block(V::owned_into_inner_shared(block), ack));
            self.pending_acks.enqueue(PendingAck {
                height,
                commitment,
                receiver: ack_waiter,
            });
        }
        self
    }

    // -------------------- Prunable Storage --------------------

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

    // -------------------- Immutable Storage --------------------

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

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

        self.get_finalized_block(height)
            .await
            .map(|block| (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 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.
    async fn store_finalization(
        mut self: Box<Self>,
        height: Height,
        digest: <V::Block as Digestible>::Digest,
        block: V::Block,
        finalization: Option<Finalization<P::Scheme, V::Commitment>>,
        application: &mut impl Reporter<Activity = Update<V::ApplicationBlock, A>>,
    ) -> (Box<Self>, bool) {
        // Blocks below the last processed height are not useful to us, so we ignore them (this
        // has the nice byproduct of ensuring we don't call a backing store with a block below the
        // pruning boundary)
        if height <= self.floor.processed_height() {
            debug!(
                %height,
                floor = %self.floor.processed_height(),
                ?digest,
                "dropping finalization at or below processed height floor"
            );
            return (self, false);
        }

        // Convert block to storage format
        let stored: V::StoredBlock = block.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(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);

        // Update metrics and application
        if let Some(round) = round.filter(|_| height > self.tip) {
            application.report(Update::Tip(round, height, digest));
            self.tip = height;
            let _ = self.finalized_height.try_set(height.get());
        }

        (self, true)
    }

    /// 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(),
        ))
    }

    // -------------------- Mixed Storage --------------------

    /// 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.into());
        }
        // Check finalized blocks.
        match self.finalized_blocks.get(ArchiveID::Key(&digest)).await {
            Ok(stored) => stored.map(|stored| stored.into()),
            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, |stored| V::stored_commitment(stored) == commitment)
            .await
        {
            return Some(block.into());
        }

        match self.finalized_blocks.get(ArchiveID::Key(&digest)).await {
            Ok(Some(stored)) => {
                (V::stored_commitment(&stored) == commitment).then(|| stored.into())
            }
            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<Arc<V::Block>> {
        if let Some(block) = buffer.find_by_digest(digest).await {
            return Some(block);
        }
        self.find_block_in_storage(digest).await.map(Arc::new)
    }

    /// 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<Arc<V::Block>> {
        if let Some(block) = buffer.find_by_commitment(commitment).await {
            return Some(block);
        }
        self.find_block_in_storage_by_commitment(commitment)
            .await
            .map(Arc::new)
    }

    /// Attempt to repair any identified gaps in the finalized blocks archive. The total
    /// number of missing heights that can be repaired at once is bounded by `self.max_repair`,
    /// though multiple gaps may be spanned.
    ///
    /// 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 digest = block.digest();
                    let stored;
                    (self, stored) = self
                        .store_finalization(
                            last_finalized,
                            digest,
                            Arc::unwrap_or_clone(block),
                            Some(finalization),
                            application,
                        )
                        .await;
                    wrote |= stored;
                } else {
                    // Request the missing block.
                    self.floor
                        .fetch_if_permitted(
                            resolver,
                            Request::finalized_block_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::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,
                            parent_digest,
                            Arc::unwrap_or_clone(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_block_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_height(
        &mut self,
        height: Height,
        resolver: &mut impl Resolver<Key = ResolverRequestFor<V>, Subscriber = Annotation>,
    ) {
        self.stream.acknowledge(height);
        self.floor.set_processed_height(height);
        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));
    }

    /// Returns the latest recoverable round at or immediately after the processed height.
    ///
    /// A finalization above the processed height advances the round floor only when its matching
    /// block is also durable, so recovery never suppresses a fetch for a certificate-only successor.
    async fn latest_processed_round(
        finalizations_by_height: &FC,
        finalized_blocks: &FB,
        height: Option<Height>,
    ) -> Round {
        let processed_round = height.and_then(|height| {
            finalizations_by_height
                .ranges_from(Height::zero())
                .filter_map(|(start, end)| (start <= height).then_some(end.min(height)))
                .max()
        });
        let processed_round = match processed_round {
            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(),
        };

        let successor = match height {
            Some(height) if height.get() == u64::MAX => return processed_round,
            Some(height) => height.next(),
            None => Height::zero(),
        };
        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 processed_round;
        };
        assert!(
            V::stored_commitment(&block) == finalization.proposal.payload,
            "successor block does not match stored finalization"
        );
        processed_round.max(finalization.round())
    }

    /// Buffers a processed round update in memory and prunes round-bound requests.
    async fn update_processed_round<Buf: Buffer<V>>(
        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> {
        let Some(finalization) = self.get_finalization_by_height(height).await else {
            return self;
        };
        self.update_processed_round_floor(
            height,
            finalization.round(),
            buffer,
            application,
            resolver,
        )
        .await
    }

    /// Buffers a processed round floor update in memory and prunes round-bound requests.
    ///
    /// A pending floor anchor whose round the new floor covers is released: the
    /// same retention that prunes its fetch proves it sits at or below the
    /// processed height, so repair and dispatch resume without it.
    async fn update_processed_round_floor<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> {
        let processed_round = self.floor.round();
        if height > self.floor.processed_height() || round <= processed_round {
            return self;
        }

        self.floor.set_processed_round(round);

        // Retain view-indexed cache data for a window behind the previously
        // processed finalized block.
        let prune_round = Round::new(
            processed_round.epoch(),
            processed_round.view().saturating_sub(self.view_retention),
        );
        self.cache = self.cache.prune_by_view(prune_round).await;

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

        // A superseded anchor is an ancestor of a processed block, so the floor
        // it announced is already active. Retire the acks it displaced and resume.
        if self.floor.take_superseded_anchor(round).is_none() {
            return self;
        }
        let commitments = self.take_superseded_ack_commitments();
        buffer.retire(Retirement {
            round_floor: round,
            exact_retirements: commitments,
        });
        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 and certificates below the given height.
    async fn prune_finalized_archives(mut self: Box<Self>, height: Height) -> Box<Self> {
        // 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(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.
    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
    }
}