linera_core/chain_worker/

state.rs

// Copyright (c) Zefchain Labs, Inc.
// SPDX-License-Identifier: Apache-2.0

//! The state and functionality of a chain worker.

use std::{
    borrow::Cow,
    collections::{BTreeMap, BTreeSet, HashMap, HashSet},
    sync::{self, Arc},
};

use futures::future::Either;
#[cfg(with_metrics)]
use linera_base::prometheus_util::MeasureLatency as _;
use linera_base::{
    crypto::{CryptoHash, ValidatorPublicKey},
    data_types::{
        ApplicationDescription, ArithmeticError, Blob, BlockHeight, Epoch, Round, Timestamp,
    },
    ensure,
    hashed::Hashed,
    identifiers::{AccountOwner, ApplicationId, BlobId, ChainId, StreamId},
};
use linera_cache::{Arc as CacheArc, UniqueValueCache, ValueCache};
use linera_chain::{
    data_types::{
        BlockProposal, BundleExecutionPolicy, IncomingBundle, MessageAction, MessageBundle,
        OriginalProposal, ProposalContent, ProposedBlock,
    },
    manager::{self, ManagerSafetySnapshot},
    types::{
        Block, ConfirmedBlock, ConfirmedBlockCertificate, TimeoutCertificate,
        ValidatedBlockCertificate,
    },
    BlockExecution, ChainError, ChainExecutionContext, ChainIdSet, ChainStateView, ChainTipState,
    ExecutionResultExt as _,
};
use linera_execution::{
    system::EventSubscriptions, ExecutionRuntimeContext as _, ExecutionStateView, Query,
    QueryContext, QueryOutcome, ResourceTracker, ServiceRuntimeEndpoint,
};
use linera_storage::{Clock as _, Storage};
use linera_views::{
    context::{Context, InactiveContext},
    views::{ReplaceContext as _, RootView as _, View as _},
};
use tokio::sync::oneshot;
use tracing::{debug, info, instrument, trace, warn};

use crate::{
    chain_worker::{handle::AtomicTimestamp, ChainWorkerConfig, DeliveryNotifier},
    client::{ChainModes, ListeningMode},
    data_types::{ChainInfo, ChainInfoQuery, ChainInfoResponse, CrossChainRequest},
    worker::{BatchRequest, NetworkActions, Notification, Reason, WorkerError},
};

/// Type alias for event subscriptions result.
pub(crate) type EventSubscriptionsResult = Vec<((ChainId, StreamId), EventSubscriptions)>;

#[cfg(with_metrics)]
mod metrics {
    use std::sync::LazyLock;

    use linera_base::prometheus_util::{
        exponential_bucket_interval, exponential_bucket_latencies, register_histogram,
        register_histogram_vec, register_int_counter, register_int_counter_vec,
    };
    use prometheus::{Histogram, HistogramVec, IntCounter, IntCounterVec};

    pub static CREATE_NETWORK_ACTIONS_LATENCY: LazyLock<Histogram> = LazyLock::new(|| {
        register_histogram(
            "create_network_actions_latency",
            "Time (ms) to create network actions",
            exponential_bucket_latencies(10_000.0),
        )
    });

    pub static NUM_INBOXES: LazyLock<HistogramVec> = LazyLock::new(|| {
        register_histogram_vec(
            "num_inboxes",
            "Number of inboxes",
            &[],
            exponential_bucket_interval(1.0, 10_000.0),
        )
    });

    pub static BLOCK_PROPOSALS_RECEIVED_TOTAL: LazyLock<IntCounter> = LazyLock::new(|| {
        register_int_counter(
            "block_proposals_received_total",
            "Total number of block proposals received by the worker",
        )
    });

    pub static BLOCK_PROPOSALS_REJECTED_TOTAL: LazyLock<IntCounterVec> = LazyLock::new(|| {
        register_int_counter_vec(
            "block_proposals_rejected_total",
            "Total number of block proposals rejected by the worker, labelled by error type",
            &["error_type"],
        )
    });
}

/// The state of the chain worker.
pub(crate) struct ChainWorkerState<StorageClient>
where
    StorageClient: Storage,
{
    config: ChainWorkerConfig,
    storage: StorageClient,
    chain: ChainStateView<StorageClient::Context>,
    service_runtime_endpoint: Option<ServiceRuntimeEndpoint>,
    /// The background task running the service runtime. Must be kept alive for the
    /// lifetime of the worker: the pool `Guard` wrapper returns the thread-pool slot
    /// when dropped, so dropping this early lets the pool schedule unrelated work on a
    /// thread that is still running the service runtime.
    service_runtime_task: Option<web_thread_pool::Task<()>>,
    /// Timestamp of the last access.
    /// Used by the keep-alive task to determine when the worker has been idle.
    /// Wrapped in `Arc` so the keep-alive task can read it without acquiring
    /// the `RwLock`.
    last_access: Arc<AtomicTimestamp>,
    block_values: Arc<ValueCache<CryptoHash, ConfirmedBlock>>,
    execution_state_cache:
        Option<Arc<UniqueValueCache<CryptoHash, ExecutionStateView<InactiveContext>>>>,
    chain_modes: Option<Arc<sync::RwLock<ChainModes>>>,
    delivery_notifier: DeliveryNotifier,
    knows_chain_is_active: bool,
    /// Set to `true` if a database `save` failure has left storage potentially
    /// inconsistent.
    poisoned: bool,
}

/// The result of processing a cross-chain update.
pub(crate) enum CrossChainUpdateResult {
    /// The update was applied up to the given height. The caller must save.
    Updated(BlockHeight),
    /// All bundles were already received; nothing to do.
    NothingToDo,
    /// A gap was detected in the inbox for messages from `origin`. If
    /// `allow_revert_confirm` is enabled, a `RevertConfirm` request should be sent
    /// to retransmit bundles starting from `retransmit_from`.
    GapDetected {
        origin: ChainId,
        retransmit_from: BlockHeight,
    },
}

/// Whether the block was processed or skipped. Used for metrics.
pub enum BlockOutcome {
    Processed,
    Preprocessed,
    Skipped,
}

/// How to handle a confirmed block.
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum ProcessConfirmedBlockMode {
    /// Execute the block if it is contiguous (or bridgeable via a checkpoint);
    /// otherwise preprocess it. Used by validators and by any caller that
    /// wants graceful fallback when there's a gap.
    Auto,
    /// Execute the block. Fail with [`WorkerError::InvalidBlockChaining`] if
    /// there's a gap that can't be bridged. Use when the caller knows the
    /// block must be contiguous and wants a hard error otherwise.
    Execute,
    /// Only preprocess the block, never execute it — even if it would be
    /// contiguous. Used by the client for sender-chain blocks it is not
    /// otherwise tracking, since their execution state is irrelevant.
    Preprocess,
}

impl<StorageClient> ChainWorkerState<StorageClient>
where
    StorageClient: Storage + Clone + 'static,
{
    /// Creates a new [`ChainWorkerState`] using the provided `storage` client.
    #[instrument(skip_all, fields(
        chain_id = %chain_id
    ))]
    #[expect(clippy::too_many_arguments)]
    pub(crate) async fn load(
        config: ChainWorkerConfig,
        storage: StorageClient,
        block_values: Arc<ValueCache<CryptoHash, ConfirmedBlock>>,
        execution_state_cache: Option<
            Arc<UniqueValueCache<CryptoHash, ExecutionStateView<InactiveContext>>>,
        >,
        chain_modes: Option<Arc<sync::RwLock<ChainModes>>>,
        delivery_notifier: DeliveryNotifier,
        chain_id: ChainId,
        service_runtime_endpoint: Option<ServiceRuntimeEndpoint>,
        service_runtime_task: Option<web_thread_pool::Task<()>>,
    ) -> Result<Self, WorkerError> {
        let chain = storage.load_chain(chain_id).await?;

        Ok(ChainWorkerState {
            config,
            storage,
            chain,
            service_runtime_endpoint,
            service_runtime_task,
            last_access: Arc::new(AtomicTimestamp::now()),
            block_values,
            execution_state_cache,
            chain_modes,
            delivery_notifier,
            knows_chain_is_active: false,
            poisoned: false,
        })
    }

    /// Returns the [`ChainId`] of the chain handled by this worker.
    fn chain_id(&self) -> ChainId {
        self.chain.chain_id()
    }

    /// Returns a reference to the chain state view.
    pub(crate) fn chain(&self) -> &ChainStateView<StorageClient::Context> {
        &self.chain
    }

    /// Returns whether this chain is known to be active (initialized).
    pub(crate) fn knows_chain_is_active(&self) -> bool {
        self.knows_chain_is_active
    }

    /// Rolls back any uncommitted changes to the chain state.
    pub(crate) fn rollback(&mut self) {
        self.chain.rollback();
    }

    /// Returns `WorkerError::PoisonedWorker` if the worker is poisoned due to a database
    /// `save` failure.
    pub(crate) fn check_not_poisoned(&self) -> Result<(), WorkerError> {
        ensure!(!self.poisoned, WorkerError::PoisonedWorker);
        Ok(())
    }

    /// Updates the last-access timestamp to the current time.
    pub(crate) fn touch(&self) {
        self.last_access.store_now();
    }

    /// Returns a clone of the last-access `Arc`, for use by the keep-alive task.
    pub(crate) fn last_access_arc(&self) -> Arc<AtomicTimestamp> {
        Arc::clone(&self.last_access)
    }

    /// Drops the service runtime endpoint, signaling the runtime task to stop.
    /// Returns the runtime task so the caller can await it outside the lock.
    pub(crate) fn clear_service_runtime(&mut self) -> Option<web_thread_pool::Task<()>> {
        self.service_runtime_endpoint.take();
        self.service_runtime_task.take()
    }

    /// Returns the pending cross-chain network actions if the outbox index is already
    /// reconciled to the current tracked set, or `None` if it must first be reconciled (which
    /// needs a write lock).
    pub(crate) async fn cross_chain_network_actions_if_reconciled(
        &self,
    ) -> Result<Option<NetworkActions>, WorkerError> {
        let tracked = self.tracked_full_chains();
        if !self.chain.outbox_index_is_reconciled(tracked.as_deref()) {
            return Ok(None);
        }
        Ok(Some(
            self.build_network_actions(None, tracked.as_deref().map(|h| h.inner()))
                .await?,
        ))
    }

    /// Reconciles the outbox index with the current tracked set, then returns the pending
    /// cross-chain network actions for this chain, without initializing the chain's execution
    /// state. Intended for callers that only need to re-emit cross-chain requests from the
    /// outbox of a sender chain whose `ChainDescription` we may never have needed.
    ///
    /// This is the slow path of [`Self::cross_chain_network_actions_if_reconciled`].
    #[instrument(skip_all, fields(chain_id = %self.chain_id()))]
    pub(crate) async fn reconcile_and_cross_chain_network_actions(
        &mut self,
    ) -> Result<NetworkActions, WorkerError> {
        let tracked = self.tracked_full_chains();
        self.chain
            .reconcile_outbox_index(tracked.as_deref())
            .await?;
        let actions = self
            .build_network_actions(None, tracked.as_deref().map(|h| h.inner()))
            .await?;
        self.save().await?;
        Ok(actions)
    }

    /// Handles a [`ChainInfoQuery`], potentially voting on the next block.
    #[tracing::instrument(level = "debug", skip(self))]
    pub(crate) async fn handle_chain_info_query(
        &mut self,
        query: ChainInfoQuery,
    ) -> Result<(ChainInfoResponse, NetworkActions), WorkerError> {
        let create_network_actions = query.create_network_actions;
        if let Some((height, round)) = query.request_leader_timeout {
            self.vote_for_leader_timeout(height, round).await?;
        }
        if query.request_fallback {
            self.vote_for_fallback().await?;
        }
        let response = self.prepare_chain_info_response(query).await?;
        // Trigger any outgoing cross-chain messages that haven't been confirmed yet.
        let actions = if create_network_actions {
            self.create_network_actions(None).await?
        } else {
            NetworkActions::default()
        };
        Ok((response, actions))
    }

    /// Returns the requested blob, if it belongs to the current locking block or pending proposal.
    #[instrument(skip_all, fields(
        chain_id = %self.chain_id(),
        blob_id = %blob_id
    ))]
    pub(crate) async fn download_pending_blob(
        &self,
        blob_id: BlobId,
    ) -> Result<CacheArc<Blob>, WorkerError> {
        if let Some(blob) = self.chain.manager.pending_blob(&blob_id).await? {
            return Ok(self.storage.cache_blob(blob));
        }
        self.storage
            .read_blob(blob_id)
            .await?
            .ok_or(WorkerError::BlobsNotFound(vec![blob_id]))
    }

    /// Reads the blobs from the chain manager or from storage. Returns an error if any are
    /// missing.
    #[instrument(skip_all, fields(
        chain_id = %self.chain_id()
    ))]
    async fn get_required_blobs(
        &self,
        required_blob_ids: impl IntoIterator<Item = BlobId>,
        created_blobs: BTreeMap<BlobId, Blob>,
    ) -> Result<BTreeMap<BlobId, Blob>, WorkerError> {
        let maybe_blobs = self
            .maybe_get_required_blobs(required_blob_ids, Some(created_blobs))
            .await?;
        let not_found_blob_ids = missing_blob_ids(&maybe_blobs);
        ensure!(
            not_found_blob_ids.is_empty(),
            WorkerError::BlobsNotFound(not_found_blob_ids)
        );
        Ok(maybe_blobs
            .into_iter()
            .filter_map(|(blob_id, maybe_blob)| Some((blob_id, maybe_blob?)))
            .collect())
    }

    /// Tries to read the blobs from the chain manager or storage. Returns `None` if not found.
    #[instrument(skip_all, fields(
        chain_id = %self.chain_id()
    ))]
    async fn maybe_get_required_blobs(
        &self,
        blob_ids: impl IntoIterator<Item = BlobId>,
        created_blobs: Option<BTreeMap<BlobId, Blob>>,
    ) -> Result<BTreeMap<BlobId, Option<Blob>>, WorkerError> {
        let maybe_blobs = blob_ids.into_iter().collect::<BTreeSet<_>>();
        let mut maybe_blobs = maybe_blobs
            .into_iter()
            .map(|x| (x, None))
            .collect::<Vec<(BlobId, Option<Blob>)>>();

        if let Some(mut blob_map) = created_blobs {
            for (blob_id, value) in &mut maybe_blobs {
                if let Some(blob) = blob_map.remove(blob_id) {
                    *value = Some(blob);
                }
            }
        }

        let (missing_indices, missing_blob_ids) = missing_indices_blob_ids(&maybe_blobs);
        let second_block_blobs = self.chain.manager.pending_blobs(&missing_blob_ids).await?;
        for (index, blob) in missing_indices.into_iter().zip(second_block_blobs) {
            maybe_blobs[index].1 = blob;
        }

        let (missing_indices, missing_blob_ids) = missing_indices_blob_ids(&maybe_blobs);
        let third_block_blobs = self
            .chain
            .pending_validated_blobs
            .multi_get(&missing_blob_ids)
            .await?;
        for (index, blob) in missing_indices.into_iter().zip(third_block_blobs) {
            maybe_blobs[index].1 = blob;
        }

        let (missing_indices, missing_blob_ids) = missing_indices_blob_ids(&maybe_blobs);
        if !missing_indices.is_empty() {
            let all_entries_pending_blobs = self
                .chain
                .pending_proposed_blobs
                .try_load_all_entries()
                .await?;
            for (index, blob_id) in missing_indices.into_iter().zip(missing_blob_ids) {
                for (_, pending_blobs) in &all_entries_pending_blobs {
                    if let Some(blob) = pending_blobs.get(&blob_id).await? {
                        maybe_blobs[index].1 = Some(blob);
                        break;
                    }
                }
            }
        }

        let (missing_indices, missing_blob_ids) = missing_indices_blob_ids(&maybe_blobs);
        let fourth_block_blobs = self.storage.read_blobs(&missing_blob_ids).await?;
        for (index, blob) in missing_indices.into_iter().zip(fourth_block_blobs) {
            maybe_blobs[index].1 = blob.map(CacheArc::unwrap_or_clone);
        }
        Ok(maybe_blobs.into_iter().collect())
    }

    /// Creates cross-chain requests for a single recipient from its outbox.
    #[instrument(skip_all, fields(
        chain_id = %self.chain_id()
    ))]
    async fn create_cross_chain_actions_for_recipient(
        &self,
        recipient: ChainId,
    ) -> Result<NetworkActions, WorkerError> {
        let outbox = self.chain.outboxes.try_load_entry(&recipient).await?;
        let Some(outbox) = outbox else {
            return Ok(NetworkActions::default());
        };
        let heights = outbox.queue.elements().await?;
        if heights.is_empty() {
            return Ok(NetworkActions::default());
        }
        let heights_by_recipient = BTreeMap::from([(recipient, heights)]);
        let cross_chain_requests = self
            .create_cross_chain_requests(heights_by_recipient)
            .await?;
        Ok(NetworkActions {
            cross_chain_requests,
            notifications: Vec::new(),
        })
    }

    /// Returns the set of chains tracked in full mode together with its memoized hash, or `None` if
    /// every chain is implicitely in full-mode (e.g. on validator).
    fn tracked_full_chains(&self) -> Option<Arc<Hashed<ChainIdSet>>> {
        let chain_modes = self.chain_modes.as_ref()?;
        let full = chain_modes
            .read()
            .expect("Panics should not happen while holding a lock to `chain_modes`")
            .full();
        Some(full)
    }

    /// Returns whether the given chain is tracked in full mode (always `true` on a validator),
    /// i.e. whether its outbox should be indexed.
    fn is_tracked(&self, chain_id: &ChainId) -> bool {
        self.chain_modes.as_ref().is_none_or(|chain_modes| {
            chain_modes
                .read()
                .expect("Panics should not happen while holding a lock to `chain_modes`")
                .get(chain_id)
                .is_some_and(ListeningMode::is_full)
        })
    }

    /// Reconciles the chain's `nonempty_outboxes` and `outbox_counters` indices with the current
    /// set of fully-tracked chains.
    async fn reconcile_tracked_outboxes(
        &mut self,
    ) -> Result<Option<Arc<Hashed<ChainIdSet>>>, WorkerError> {
        let full_chains = self.tracked_full_chains();
        self.chain
            .reconcile_outbox_index(full_chains.as_deref())
            .await?;
        Ok(full_chains)
    }

    /// Reconciles the outbox index with the current tracked set, then loads pending cross-chain
    /// requests and adds `NewRound` notifications where appropriate.
    async fn create_network_actions(
        &mut self,
        old_round: Option<Round>,
    ) -> Result<NetworkActions, WorkerError> {
        // Make the outbox index authoritative for the current tracked set first, so it already
        // holds only `is_full` targets and needs no read-time filtering.
        let tracked = self.reconcile_tracked_outboxes().await?;
        self.build_network_actions(old_round, tracked.as_deref().map(|h| h.inner()))
            .await
    }

    /// Builds the pending cross-chain actions from the already-reconciled outbox index.
    async fn build_network_actions(
        &self,
        old_round: Option<Round>,
        tracked: Option<&ChainIdSet>,
    ) -> Result<NetworkActions, WorkerError> {
        #[cfg(with_metrics)]
        let _latency = metrics::CREATE_NETWORK_ACTIONS_LATENCY.measure_latency();
        let mut heights_by_recipient = BTreeMap::<_, Vec<_>>::new();
        let targets = self.chain.nonempty_outbox_chain_ids();
        if let Some(tracked) = tracked {
            if let Some(target) = targets.iter().find(|target| !tracked.contains(*target)) {
                return Err(ChainError::CorruptedChainState(format!(
                    "outbox index contains untracked target {target}"
                ))
                .into());
            }
        }
        let outboxes = self.chain.load_outboxes(&targets).await?;
        for (target, outbox) in targets.into_iter().zip(outboxes) {
            let heights = outbox.queue.elements().await?;
            heights_by_recipient.insert(target, heights);
        }
        let cross_chain_requests = self
            .create_cross_chain_requests(heights_by_recipient)
            .await?;
        let mut notifications = Vec::new();
        if let Some(old_round) = old_round {
            let round = self.chain.manager.current_round();
            if round > old_round {
                let height = self.chain.tip_state.get().next_block_height;
                notifications.push(Notification {
                    chain_id: self.chain_id(),
                    reason: Reason::NewRound { height, round },
                });
            }
        }
        Ok(NetworkActions {
            cross_chain_requests,
            notifications,
        })
    }

    /// Returns confirmed blocks by hash, checking the cache first and batch-loading the rest
    /// from storage. The order of the returned blocks matches the order of the input hashes.
    async fn read_confirmed_blocks(
        &self,
        hashes: Vec<CryptoHash>,
    ) -> Result<Vec<Option<CacheArc<ConfirmedBlock>>>, WorkerError> {
        let mut blocks = Vec::with_capacity(hashes.len());
        let mut uncached_indices = Vec::new();
        let mut uncached_hashes = Vec::new();

        for (i, hash) in hashes.iter().enumerate() {
            if let Some(block) = self.block_values.get(hash) {
                blocks.push(Some(block));
            } else {
                blocks.push(None);
                uncached_indices.push(i);
                uncached_hashes.push(*hash);
            }
        }

        if !uncached_hashes.is_empty() {
            let from_storage = self.storage.read_confirmed_blocks(uncached_hashes).await?;
            for (i, maybe_block) in uncached_indices.into_iter().zip(from_storage) {
                blocks[i] = maybe_block;
            }
        }

        Ok(blocks)
    }

    #[instrument(skip_all, fields(
        chain_id = %self.chain_id(),
        num_recipients = %heights_by_recipient.len()
    ))]
    async fn create_cross_chain_requests(
        &self,
        heights_by_recipient: BTreeMap<ChainId, Vec<BlockHeight>>,
    ) -> Result<Vec<CrossChainRequest>, WorkerError> {
        // Load all the certificates we will need, regardless of the medium.
        let heights = heights_by_recipient
            .values()
            .flatten()
            .copied()
            .collect::<BTreeSet<_>>();
        let hashes = self.chain.block_hashes(heights.iter().copied()).await?;

        let blocks = self.read_confirmed_blocks(hashes.clone()).await?;

        let mut height_to_blocks = HashMap::new();
        for (block, hash) in blocks.into_iter().zip(hashes) {
            let block = block.ok_or_else(|| WorkerError::ReadCertificatesError(vec![hash]))?;
            let hashed_block = CacheArc::unwrap_or_clone(block).into_inner();
            height_to_blocks.insert(hashed_block.inner().header.height, hashed_block);
        }

        let sender = self.chain.chain_id();
        let mut cross_chain_requests = Vec::new();
        for (recipient, heights) in heights_by_recipient {
            // Extract the predecessor height for this recipient from the first
            // block's `previous_message_blocks`. This lets the recipient detect
            // gaps even before it consumes the missing message.
            let previous_height = heights.first().and_then(|first_height| {
                let block = height_to_blocks.get(first_height)?;
                let (_, prev_height) =
                    block.inner().body.previous_message_blocks.get(&recipient)?;
                Some(*prev_height)
            });
            let mut bundles = Vec::new();
            let mut bundles_size = 0;
            for height in heights {
                let Some(hashed_block) = height_to_blocks.get(&height) else {
                    tracing::warn!(
                        %height,
                        %recipient,
                        "spurious entry in outbox; skipping this and higher sender blocks"
                    );
                    break;
                };
                let new_bundles = hashed_block
                    .inner()
                    .message_bundles_for(recipient, hashed_block.hash())
                    .collect::<Vec<_>>();
                let new_size = new_bundles
                    .iter()
                    .map(|(_epoch, bundle)| bundle.estimated_size())
                    .sum::<usize>();
                // If adding this block's bundles would exceed the chunk limit,
                // stop here. Always include at least one block's bundles.
                if bundles_size + new_size > self.config.cross_chain_message_chunk_limit {
                    if bundles.is_empty() {
                        warn!(
                            "Single block at height {height} produces an UpdateRecipient \
                            of ~{new_size} bytes, exceeding the chunk limit of {}",
                            self.config.cross_chain_message_chunk_limit
                        );
                    } else {
                        debug!(
                            "Stopping cross-chain batch for {recipient} at height {height}: \
                            adding ~{new_size} bytes would exceed chunk limit of {} \
                            (current batch ~{bundles_size} bytes)",
                            self.config.cross_chain_message_chunk_limit
                        );
                        break;
                    }
                }
                bundles.extend(new_bundles);
                bundles_size += new_size;
            }
            if !bundles.is_empty() {
                cross_chain_requests.push(CrossChainRequest::UpdateRecipient {
                    sender,
                    recipient,
                    bundles,
                    previous_height,
                });
            }
        }
        Ok(cross_chain_requests)
    }

    /// Processes a leader timeout issued for this multi-owner chain.
    #[instrument(skip_all, fields(
        chain_id = %self.chain_id(),
        height = %certificate.inner().height()
    ))]
    pub(crate) async fn process_timeout(
        &mut self,
        certificate: TimeoutCertificate,
    ) -> Result<(ChainInfoResponse, NetworkActions), WorkerError> {
        // Check that the chain is active and ready for this timeout.
        // Verify the certificate. Returns a catch-all error to make client code more robust.
        self.initialize_and_save_if_needed().await?;
        let (chain_epoch, committee) = self.chain.current_committee().await?;
        certificate.check(&committee)?;
        if self
            .chain
            .tip_state
            .get()
            .already_validated_block(certificate.inner().height())?
        {
            return Ok((self.chain_info_response().await?, NetworkActions::default()));
        }
        ensure!(
            certificate.inner().epoch() == chain_epoch,
            WorkerError::InvalidEpoch {
                chain_id: certificate.inner().chain_id(),
                chain_epoch,
                epoch: certificate.inner().epoch()
            }
        );
        let old_round = self.chain.manager.current_round();
        self.chain
            .manager
            .handle_timeout_certificate(certificate, self.storage.clock().current_time());
        self.save().await?;
        let actions = self.create_network_actions(Some(old_round)).await?;
        Ok((self.chain_info_response().await?, actions))
    }

    /// Tries to load all blobs published in this proposal.
    ///
    /// If they cannot be found, it creates an entry in `pending_proposed_blobs` so they can be
    /// submitted one by one.
    #[instrument(skip_all, fields(
        chain_id = %self.chain_id(),
        block_height = %proposal.content.block.height
    ))]
    async fn load_proposal_blobs(
        &mut self,
        proposal: &BlockProposal,
    ) -> Result<Vec<Blob>, WorkerError> {
        let owner = proposal.owner();
        let BlockProposal {
            content:
                ProposalContent {
                    block,
                    round,
                    outcome: _,
                },
            original_proposal,
            signature: _,
        } = proposal;

        let mut maybe_blobs = self
            .maybe_get_required_blobs(proposal.required_blob_ids(), None)
            .await?;
        let missing_blob_ids = missing_blob_ids(&maybe_blobs);
        if !missing_blob_ids.is_empty() {
            let chain = &mut self.chain;
            if chain.ownership().await?.open_multi_leader_rounds {
                // TODO(#3203): Allow multiple pending proposals on permissionless chains.
                chain.pending_proposed_blobs.clear();
            }
            let validated = matches!(original_proposal, Some(OriginalProposal::Regular { .. }));
            chain
                .pending_proposed_blobs
                .try_load_entry_mut(&owner)
                .await?
                .update(*round, validated, maybe_blobs)?;
            self.save().await?;
            return Err(WorkerError::BlobsNotFound(missing_blob_ids));
        }
        let published_blobs = block
            .published_blob_ids()
            .iter()
            .filter_map(|blob_id| maybe_blobs.remove(blob_id).flatten())
            .collect::<Vec<_>>();
        Ok(published_blobs)
    }

    /// Processes a validated block issued for this multi-owner chain.
    #[instrument(skip_all, fields(
        chain_id = %self.chain_id(),
        block_height = %certificate.block().header.height
    ))]
    pub(crate) async fn process_validated_block(
        &mut self,
        certificate: ValidatedBlockCertificate,
    ) -> Result<(ChainInfoResponse, NetworkActions, BlockOutcome), WorkerError> {
        let block = certificate.block();

        let header = &block.header;
        let height = header.height;
        // Check that the chain is active and ready for this validated block.
        // Verify the certificate. Returns a catch-all error to make client code more robust.
        self.initialize_and_save_if_needed().await?;
        let tip_state = self.chain.tip_state.get();
        ensure!(
            header.height == tip_state.next_block_height,
            ChainError::UnexpectedBlockHeight {
                expected_block_height: tip_state.next_block_height,
                found_block_height: header.height,
            }
        );
        let (epoch, committee) = self.chain.current_committee().await?;
        check_block_epoch(epoch, header.chain_id, header.epoch)?;
        certificate.check(&committee)?;
        let already_committed_block = self.chain.tip_state.get().already_validated_block(height)?;
        let should_skip_validated_block = || {
            self.chain
                .manager
                .check_validated_block(&certificate)
                .map(|outcome| outcome == manager::Outcome::Skip)
        };
        if already_committed_block || should_skip_validated_block()? {
            // If we just processed the same pending block, return the chain info unchanged.
            return Ok((
                self.chain_info_response().await?,
                NetworkActions::default(),
                BlockOutcome::Skipped,
            ));
        }

        self.block_values
            .insert_hashed(Cow::Borrowed(certificate.inner().inner()));
        let required_blob_ids = block.required_blob_ids();
        let maybe_blobs = self
            .maybe_get_required_blobs(required_blob_ids, Some(block.created_blobs()))
            .await?;
        let missing_blob_ids = missing_blob_ids(&maybe_blobs);
        if !missing_blob_ids.is_empty() {
            self.chain
                .pending_validated_blobs
                .update(certificate.round, true, maybe_blobs)?;
            self.save().await?;
            return Err(WorkerError::BlobsNotFound(missing_blob_ids));
        }
        let blobs = maybe_blobs
            .into_iter()
            .filter_map(|(blob_id, maybe_blob)| Some((blob_id, maybe_blob?)))
            .collect();
        let old_round = self.chain.manager.current_round();
        self.chain.manager.create_final_vote(
            certificate,
            self.config.key_pair(),
            self.storage.clock().current_time(),
            blobs,
        )?;
        self.save().await?;
        let actions = self.create_network_actions(Some(old_round)).await?;
        Ok((
            self.chain_info_response().await?,
            actions,
            BlockOutcome::Processed,
        ))
    }

    /// Initializes `next_expected_events` from `stream_event_counts` (which reflects
    /// all executed blocks), then replays any preprocessed-but-not-yet-executed blocks to
    /// advance the indices further.
    ///
    /// This handles the migration case where the `next_expected_events` field was added to
    /// `ChainStateView` after blocks had already been processed.
    async fn initialize_next_expected_events(&mut self) -> Result<(), WorkerError> {
        if self.chain.next_expected_events.count().await? > 0 {
            return Ok(()); // Already initialized.
        }
        for (stream_id, index) in self
            .chain
            .execution_state
            .stream_event_counts
            .index_values()
            .await?
        {
            self.chain.next_expected_events.insert(&stream_id, index)?;
        }
        let chain_id = self.chain_id();
        let index_values = self.chain.preprocessed_blocks.index_values().await?;
        let hashes = index_values.iter().map(|(_, hash)| *hash).collect();
        let blocks = self.read_confirmed_blocks(hashes).await?;
        let tracked = self.reconcile_tracked_outboxes().await?;
        for ((height, _), maybe_block) in index_values.into_iter().zip(blocks) {
            let block =
                maybe_block.ok_or_else(|| WorkerError::LocalBlockNotFound { height, chain_id })?;
            self.chain
                .preprocess_block(&block, tracked.as_deref().map(|h| h.inner()))
                .await?;
        }
        Ok(())
    }

    /// Processes a confirmed block (aka a commit).
    #[instrument(skip_all, fields(
        chain_id = %certificate.block().header.chain_id,
        height = %certificate.block().header.height,
        block_hash = %certificate.hash(),
    ))]
    pub(crate) async fn process_confirmed_block(
        &mut self,
        certificate: ConfirmedBlockCertificate,
        mode: ProcessConfirmedBlockMode,
        notify_when_messages_are_delivered: Option<oneshot::Sender<()>>,
    ) -> Result<(ChainInfoResponse, NetworkActions, BlockOutcome), WorkerError> {
        let block = certificate.block();
        let height = block.header.height;
        let chain_id = block.header.chain_id;

        // Check if we already processed this block.
        let tip = self.chain.tip_state.get().clone();
        if tip.next_block_height > height {
            let actions = self.create_network_actions(None).await?;
            self.register_delivery_notifier(height, &actions, notify_when_messages_are_delivered)
                .await;
            return Ok((
                self.chain_info_response().await?,
                actions,
                BlockOutcome::Skipped,
            ));
        }

        // We haven't processed the block - verify the certificate first. A miss produces
        // `ExecutionError::EventsNotFound`, which the client-side retry path watches for
        // to trigger an admin-chain sync.
        let epoch = block.header.epoch;
        let committee = self
            .chain
            .execution_state
            .context()
            .extra()
            .get_committees(epoch..=epoch)
            .await
            .with_execution_context(ChainExecutionContext::Block)?
            .remove(&epoch)
            .ok_or_else(|| {
                ChainError::InternalError(format!(
                    "missing committee for epoch {epoch}; this is a bug"
                ))
            })?;
        certificate.check(&committee)?;

        // Certificate check passed - which means the blobs the block requires are legitimate and
        // we can take note of it, so that if any are missing, we will accept them when the client
        // sends them.
        let required_blob_ids = block.required_blob_ids();
        let blobs_result = self
            .get_required_blobs(required_blob_ids.iter().copied(), block.created_blobs())
            .await
            .map(|blobs| blobs.into_values().collect::<Vec<_>>());

        if let Ok(blobs) = &blobs_result {
            self.storage
                .write_blobs_and_certificate(blobs, &certificate)
                .await?;
            let events = block
                .body
                .events
                .iter()
                .flatten()
                .map(|event| (event.id(chain_id), event.value.clone()));
            self.storage.write_events(events).await?;
        }

        // Update the blob state with last used certificate hash.
        let blob_state = certificate.value().to_blob_state(blobs_result.is_ok());
        let blob_ids = required_blob_ids.into_iter().collect::<Vec<_>>();
        self.storage
            .maybe_write_blob_states(&blob_ids, blob_state)
            .await?;

        let blobs = blobs_result?
            .into_iter()
            .map(|blob| (blob.id(), blob))
            .collect::<BTreeMap<_, _>>();

        // Dispatch on mode + whether there's a gap:
        //  - `Preprocess` mode, or `Auto` with a gap: preprocess.
        //  - `Execute` mode with a gap: error.
        //  - `Auto`/`Execute` mode + contiguous: execute directly.
        use ProcessConfirmedBlockMode::{Auto, Execute, Preprocess};
        let gap = tip.next_block_height < height;
        match (mode, gap) {
            (Preprocess, _) | (Auto, true) => {
                self.preprocess_certified_block(certificate, notify_when_messages_are_delivered)
                    .await
            }
            (Execute, true) => Err(WorkerError::InvalidBlockChaining),
            (Auto | Execute, false) => {
                self.execute_contiguous_block(
                    certificate,
                    blobs,
                    tip,
                    notify_when_messages_are_delivered,
                )
                .await
            }
        }
    }

    /// Preprocesses a confirmed block: updates outboxes and event streams without
    /// executing it, and does not advance the chain tip.
    async fn preprocess_certified_block(
        &mut self,
        certificate: ConfirmedBlockCertificate,
        notify_when_messages_are_delivered: Option<oneshot::Sender<()>>,
    ) -> Result<(ChainInfoResponse, NetworkActions, BlockOutcome), WorkerError> {
        let block_hash = certificate.hash();
        let block = certificate.block();
        let chain_id = block.header.chain_id;
        let height = block.header.height;

        if block.body.events.iter().any(|events| !events.is_empty()) {
            self.initialize_next_expected_events().await?;
        }
        let tracked = self.reconcile_tracked_outboxes().await?;
        let updated_event_streams = self
            .chain
            .preprocess_block(certificate.value(), tracked.as_deref().map(|h| h.inner()))
            .await?;
        self.save().await?;
        let mut actions = self.create_network_actions(None).await?;
        if !updated_event_streams.is_empty() {
            actions.notifications.push(Notification {
                chain_id,
                reason: Reason::NewEvents {
                    height,
                    hash: block_hash,
                    event_streams: updated_event_streams,
                },
            });
        }
        trace!("Preprocessed confirmed block {height}");
        self.register_delivery_notifier(height, &actions, notify_when_messages_are_delivered)
            .await;
        Ok((
            self.chain_info_response().await?,
            actions,
            BlockOutcome::Preprocessed,
        ))
    }

    /// Executes a confirmed block whose height equals `tip.next_block_height`,
    /// updating inboxes, applying the block, and persisting the chain.
    async fn execute_contiguous_block(
        &mut self,
        certificate: ConfirmedBlockCertificate,
        mut blobs: BTreeMap<BlobId, Blob>,
        tip: ChainTipState,
        notify_when_messages_are_delivered: Option<oneshot::Sender<()>>,
    ) -> Result<(ChainInfoResponse, NetworkActions, BlockOutcome), WorkerError> {
        let block_hash = certificate.hash();
        let block = certificate.block();
        let chain_id = block.header.chain_id;
        let height = block.header.height;

        // This should always be true for valid certificates.
        ensure!(
            tip.block_hash == block.header.previous_block_hash,
            WorkerError::InvalidBlockChaining
        );

        // Verify that the chain is active and that the epoch we used for verifying
        // the certificate is actually the active one on the chain.
        self.initialize_and_save_if_needed().await?;
        let (epoch, _) = self.chain.current_committee().await?;
        check_block_epoch(epoch, chain_id, block.header.epoch)?;

        let published_blobs = block
            .published_blob_ids()
            .iter()
            .filter_map(|blob_id| blobs.remove(blob_id))
            .collect::<Vec<_>>();

        if block.body.events.iter().any(|events| !events.is_empty()) {
            // Initialize next_expected_events for any streams that don't have entries yet.
            self.initialize_next_expected_events().await?;
        }

        let local_time = self.storage.clock().current_time();
        if block.header.timestamp.duration_since(local_time) > self.config.block_time_grace_period {
            warn!(
                block_timestamp = %block.header.timestamp,
                %local_time,
                "Confirmed block has a timestamp in the future beyond the block time grace period"
            );
        }
        let tracked = self.reconcile_tracked_outboxes().await?;
        let chain = &mut self.chain;
        chain
            .remove_bundles_from_inboxes(
                block.header.timestamp,
                false,
                block.body.incoming_bundles(),
            )
            .await?;
        let confirmed_block = if let Some(mut execution_state) = self
            .execution_state_cache
            .as_ref()
            .and_then(|cache| cache.remove(&block_hash))
        {
            chain.execution_state = execution_state
                .with_context(|ctx| {
                    chain
                        .execution_state
                        .context()
                        .clone_with_base_key(ctx.base_key().bytes.clone())
                })
                .await;
            certificate.into_value()
        } else {
            let (proposed_block, outcome) = block.clone().into_proposal();
            let (proposed_block, verified, _resource_tracker, _) = chain
                .execute_block(
                    proposed_block,
                    local_time,
                    None,
                    &published_blobs,
                    BlockExecution::HandleConfirmed {
                        oracle_responses: outcome.oracle_responses.clone(),
                    },
                )
                .await?;
            // We should always agree on the messages and state hash.
            if outcome != verified {
                return Err(ChainError::CorruptedChainState(format!(
                    "computed block outcome differs from the certificate.\n\
                    Computed: {verified:#?}\n\
                    Submitted: {outcome:#?}"
                ))
                .into());
            }
            ConfirmedBlock::new(Block::new(proposed_block, verified))
        };

        let event_streams = chain
            .apply_confirmed_block(
                &confirmed_block,
                local_time,
                tracked.as_deref().map(|h| h.inner()),
            )
            .await?;
        let mut actions = self.create_network_actions(None).await?;
        trace!("Processed confirmed block {height}");
        let hash = confirmed_block.inner().hash();
        actions.notifications.push(Notification {
            chain_id,
            reason: Reason::NewBlock {
                height,
                hash,
                event_streams: event_streams.clone(),
            },
        });
        if !event_streams.is_empty() {
            actions.notifications.push(Notification {
                chain_id,
                reason: Reason::NewEvents {
                    height,
                    hash,
                    event_streams,
                },
            });
        }
        self.save().await?;

        self.block_values
            .insert_hashed(Cow::Owned(confirmed_block.into_inner()));

        self.register_delivery_notifier(height, &actions, notify_when_messages_are_delivered)
            .await;

        Ok((
            self.chain_info_response().await?,
            actions,
            BlockOutcome::Processed,
        ))
    }

    /// Schedules a notification for when cross-chain messages are delivered up to the given
    /// `height`.
    #[instrument(level = "trace", skip(self, notify_when_messages_are_delivered))]
    async fn register_delivery_notifier(
        &self,
        height: BlockHeight,
        actions: &NetworkActions,
        notify_when_messages_are_delivered: Option<oneshot::Sender<()>>,
    ) {
        if let Some(notifier) = notify_when_messages_are_delivered {
            if actions
                .cross_chain_requests
                .iter()
                .any(|request| request.has_messages_lower_or_equal_than(height))
            {
                self.delivery_notifier.register(height, notifier);
            } else {
                // No need to wait. Also, cross-chain requests may not trigger the
                // notifier later, even if we register it.
                if let Err(()) = notifier.send(()) {
                    debug!("Failed to notify message delivery to caller (early case)");
                }
            }
        }
    }

    /// Updates the chain's inboxes, receiving messages from a cross-chain update.
    #[instrument(level = "debug", skip(self, bundles), fields(chain_id = %self.chain_id()))]
    pub(crate) async fn process_cross_chain_update(
        &mut self,
        origin: ChainId,
        bundles: Vec<(Epoch, MessageBundle)>,
        previous_height: Option<BlockHeight>,
    ) -> Result<CrossChainUpdateResult, WorkerError> {
        // Only process certificates with relevant heights and epochs.
        let mut inbox = self.chain.inboxes.try_load_entry_mut(&origin).await?;
        let next_height_to_receive = inbox.next_block_height_to_receive()?;
        let last_anticipated_block_height = match inbox.removed_bundles.back().await? {
            Some(bundle) => Some(bundle.height),
            None => None,
        };

        // Proactive gap detection: if the sender declares a predecessor height that
        // we haven't received yet, the inbox has a gap.
        if let Some(prev) = previous_height {
            if prev >= next_height_to_receive {
                let chain_id = self.chain_id();
                if self.config.allow_revert_confirm && self.config.recovery_allowed_for(&chain_id) {
                    warn!(
                        %chain_id,
                        "Inbox gap detected from {origin}: \
                        sender declares previous height {prev} but we only have up to \
                        {next_height_to_receive}; requesting resend",
                    );
                    return Ok(CrossChainUpdateResult::GapDetected {
                        origin,
                        retransmit_from: next_height_to_receive,
                    });
                }
                return Err(ChainError::InboxGapDetected {
                    chain_id,
                    origin,
                    expected_height: prev,
                    actual_height: bundles.first().map(|(_, b)| b.height).unwrap_or_default(),
                }
                .into());
            }
        }

        let helper = CrossChainUpdateHelper::new(&self.config, &self.chain);
        let recipient = self.chain_id();
        let bundles = helper
            .select_message_bundles(
                &origin,
                recipient,
                next_height_to_receive,
                last_anticipated_block_height,
                bundles,
                &self.storage,
            )
            .await?;
        let Some(last_updated_height) = bundles.last().map(|bundle| bundle.height) else {
            return Ok(CrossChainUpdateResult::NothingToDo);
        };
        // Process the received messages in certificates.
        let local_time = self.storage.clock().current_time();
        let mut previous_height = None;
        for bundle in bundles {
            let add_to_received_log = previous_height != Some(bundle.height);
            previous_height = Some(bundle.height);
            // Update the staged chain state with the received block.
            self.chain
                .receive_message_bundle_with_inbox(
                    &mut inbox,
                    &origin,
                    bundle,
                    local_time,
                    add_to_received_log,
                )
                .await?;
        }
        inbox.observe_size_metric();
        drop(inbox);
        if !self.config.allow_inactive_chains && !self.chain.is_active().await? {
            // Refuse to create a chain state if the chain is still inactive by
            // now. Accordingly, do not send a confirmation, so that the
            // cross-chain update is retried later.
            warn!(
                chain_id = %self.chain_id(),
                "Refusing to deliver messages from {origin} \
                at height {last_updated_height} because the recipient is still inactive",
            );
            return Ok(CrossChainUpdateResult::NothingToDo);
        }
        Ok(CrossChainUpdateResult::Updated(last_updated_height))
    }

    /// Handles the cross-chain request confirming that the recipient was updated.
    #[instrument(skip_all, fields(
        chain_id = %self.chain_id(),
        %recipient,
        %latest_height
    ))]
    pub(crate) async fn confirm_updated_recipient(
        &mut self,
        recipient: ChainId,
        latest_height: BlockHeight,
    ) -> Result<bool, WorkerError> {
        // Reconcile the outbox indices with the *current* tracked set before draining the counter
        // and checking delivery.
        let tracked = self.reconcile_tracked_outboxes().await?;
        // The indices are now reconciled to the tracked set, so `all_messages_delivered_up_to`
        // (the `outbox_counters` fast path) is already the complete answer for tracked chains.
        Ok(self
            .chain
            .mark_messages_as_received(
                &recipient,
                latest_height,
                tracked.as_deref().map(|h| h.inner()),
            )
            .await?
            && self.chain.all_messages_delivered_up_to(latest_height))
    }

    /// Notifies delivery waiters that all messages up to `height` have been delivered.
    pub(crate) fn notify_delivery(&self, height: BlockHeight) {
        self.delivery_notifier.notify(height);
    }

    /// Processes a batch of cross-chain requests, performing at most one `save()`.
    ///
    /// Both update and confirmation requests are handled together so that a
    /// single write-lock acquisition covers all pending work for the chain.
    pub(crate) async fn process_batch(
        &mut self,
        requests: Vec<BatchRequest>,
    ) -> Result<(), WorkerError> {
        let mut update_results = Vec::new();
        let mut confirm_results = Vec::new();
        let mut need_save = false;
        let mut need_rollback = false;
        let mut recovery_error = None;
        let mut max_delivered_height: Option<BlockHeight> = None;

        for request in requests {
            match request {
                BatchRequest::Update {
                    origin,
                    bundles,
                    previous_height,
                    result_sender,
                } => {
                    if need_rollback {
                        send_result(result_sender, Err(WorkerError::BatchRolledBack));
                        continue;
                    }
                    let result = self
                        .process_cross_chain_update(origin, bundles, previous_height)
                        .await;
                    let update_result = match result {
                        Ok(update_result) => update_result,
                        Err(error) => {
                            need_rollback = true;
                            let (recovery, to_send) = classify_processing_error(error);
                            recovery_error = recovery_error.or(recovery);
                            send_result(result_sender, Err(to_send));
                            continue;
                        }
                    };
                    match &update_result {
                        CrossChainUpdateResult::Updated(_) => need_save = true,
                        CrossChainUpdateResult::GapDetected { .. }
                        | CrossChainUpdateResult::NothingToDo => {}
                    }
                    update_results.push((result_sender, update_result));
                }
                BatchRequest::Confirm {
                    recipient,
                    latest_height,
                    result_sender,
                } => {
                    if need_rollback {
                        send_result(result_sender, Err(WorkerError::BatchRolledBack));
                        continue;
                    }
                    match self
                        .confirm_updated_recipient(recipient, latest_height)
                        .await
                    {
                        Ok(fully_delivered) => {
                            need_save = true;
                            if fully_delivered {
                                max_delivered_height = Some(
                                    max_delivered_height
                                        .map_or(latest_height, |h| h.max(latest_height)),
                                );
                            }
                            confirm_results.push((result_sender, recipient));
                        }
                        Err(error) => {
                            need_rollback = true;
                            let (recovery, to_send) = classify_processing_error(error);
                            recovery_error = recovery_error.or(recovery);
                            send_result(result_sender, Err(to_send));
                        }
                    }
                }
            }
        }
        let mut save_error = None;
        if !need_rollback && need_save {
            if let Err(error) = self.save().await {
                tracing::error!(%error, "failed to save batch; rolling back");
                need_rollback = true;
                save_error = Some(error);
            }
        }
        if need_rollback {
            for (result_sender, _) in update_results {
                send_result(result_sender, Err(WorkerError::BatchRolledBack));
            }
            for (result_sender, _) in confirm_results {
                send_result(result_sender, Err(WorkerError::BatchRolledBack));
            }
            // Surface an error that left the worker poisoned or the chain state
            // corrupted so `chain_write` can evict or reset it. A `must_reload_view`
            // error only reaches us from a failed `save`, since resolving the journal
            // (the operation that raises it) happens only in `write_batch`. A
            // processing step instead can raise `CorruptedChainState` while reconciling
            // outboxes or message counters (see `confirm_updated_recipient`), which
            // calls for a reset rather than eviction. Ordinary processing errors (bad
            // height, validation failures) leave the worker healthy after rollback, so
            // they return `Ok` and were already reported to their individual senders.
            return match save_error.or(recovery_error) {
                Some(error) => Err(error),
                None => Ok(()),
            };
        }

        if let Some(height) = max_delivered_height {
            self.notify_delivery(height);
        }

        for (result_sender, update_result) in update_results {
            send_result(result_sender, Ok(update_result));
        }
        for (result_sender, recipient) in confirm_results {
            let result = self
                .create_cross_chain_actions_for_recipient(recipient)
                .await;
            send_result(result_sender, result);
        }
        Ok(())
    }

    /// Handles a `RevertConfirm` request: walks backward through
    /// `previous_message_blocks` to find all block heights that sent messages to
    /// `recipient` starting from the latest down to `retransmit_from`, re-adds them
    /// to the outbox, and creates cross-chain update actions to resend the bundles.
    #[instrument(skip_all, fields(
        chain_id = %self.chain_id(),
        %recipient,
        %retransmit_from,
    ))]
    pub(crate) async fn handle_revert_confirm(
        &mut self,
        recipient: ChainId,
        retransmit_from: BlockHeight,
    ) -> Result<NetworkActions, WorkerError> {
        self.reconcile_tracked_outboxes().await?;
        // 1. Walk backward through previous_message_blocks to collect all heights
        //    that sent messages to this recipient, from the latest down to retransmit_from.
        let Some(latest_height) = self.chain.previous_message_blocks.get(&recipient).await? else {
            warn!("RevertConfirm: no record of sending to {recipient}");
            return Ok(NetworkActions::default());
        };

        let mut heights_to_re_add = Vec::new();
        let mut current_height = latest_height;
        while current_height >= retransmit_from {
            // We arrived at current_height via previous_message_blocks links, starting from the
            // chain state and following the links downwards. So these blocks should all be in
            // confirmed_log or preprocessed_blocks already.
            heights_to_re_add.push(current_height);
            // Load the block at current_height to find the previous message block
            let hash = match &*self.chain.block_hashes([current_height]).await? {
                [hash] => *hash,
                _ => {
                    return Err(WorkerError::ConfirmedBlockHashNotFound {
                        height: current_height,
                        chain_id: self.chain_id(),
                    })
                }
            };
            let block = self
                .read_confirmed_blocks(vec![hash])
                .await?
                .pop()
                .flatten()
                .ok_or_else(|| WorkerError::LocalBlockNotFound {
                    height: current_height,
                    chain_id: self.chain_id(),
                })?;
            match block.block().body.previous_message_blocks.get(&recipient) {
                Some((_, prev_height)) if *prev_height >= retransmit_from => {
                    current_height = *prev_height;
                }
                _ => break,
            }
        }

        // 2. Re-add the heights to the outbox.
        let new_heights = self
            .chain
            .outboxes
            .try_load_entry_mut(&recipient)
            .await?
            .revert(&heights_to_re_add)
            .await?;

        if new_heights.is_empty() {
            debug!("RevertConfirm: all heights already in outbox for {recipient}");
            return Ok(NetworkActions::default());
        }

        // 3. Update the indices only for tracked recipients (mirroring `process_outgoing_messages`):
        //    an untracked recipient keeps its re-added outbox queue but is not counted or indexed.
        let new_heights_len = new_heights.len();
        if self.is_tracked(&recipient) {
            for h in new_heights {
                *self.chain.outbox_counters.get_mut().entry(h).or_default() += 1;
            }
            self.chain.nonempty_outboxes.get_mut().insert(recipient);
        }

        // 4. Create cross-chain requests for this recipient.
        let actions = self
            .create_cross_chain_actions_for_recipient(recipient)
            .await?;

        // 5. Save chain state.
        self.save().await?;

        warn!(
            "RevertConfirm: re-added {new_heights_len} heights to outbox for {recipient}, \
            starting from height {retransmit_from}"
        );

        Ok(actions)
    }

    /// If the config enables corruption recovery and the min-duration guard is
    /// satisfied, resets the chain state and re-executes all confirmed blocks.
    /// Returns `RevertConfirm` requests to dispatch, or `None` if no reset happened.
    pub(crate) async fn maybe_reset_corrupted_chain_state(
        &mut self,
    ) -> Result<Option<Vec<CrossChainRequest>>, WorkerError> {
        let Some(min_duration) = self.config.reset_on_corrupted_chain_state else {
            return Ok(None);
        };
        let chain_id = self.chain_id();
        if !self.config.recovery_allowed_for(&chain_id) {
            return Ok(None);
        }
        let local_time = self.storage.clock().current_time();
        let block_zero_time = *self.chain.block_zero_executed_at.get();
        let elapsed = local_time.duration_since(block_zero_time);
        if elapsed < min_duration {
            warn!(
                %chain_id, ?elapsed, ?min_duration,
                "Not resetting corrupted chain state; not enough time elapsed \
                since last block 0 execution"
            );
            return Ok(None);
        }
        warn!(%chain_id, "Corrupted chain state detected; resetting and re-executing");
        Ok(Some(self.reset_and_reexecute_chain().await?))
    }

    /// Resets the chain state completely and re-executes all confirmed blocks from storage.
    /// Returns a `RevertConfirm` request for every known sender so they resend cross-chain
    /// messages that may have been lost during the reset.
    #[instrument(skip_all, fields(
        chain_id = %self.chain_id(),
    ))]
    pub(crate) async fn reset_and_reexecute_chain(
        &mut self,
    ) -> Result<Vec<CrossChainRequest>, WorkerError> {
        let chain_id = self.chain_id();
        let tip_height = self.chain.tip_state.get().next_block_height;

        // 1. Collect all sender chain IDs and block hashes before clearing.
        let sender_ids = self.chain.inboxes.indices().await?;
        let hashes = self.chain.confirmed_log.read(..).await?;
        let preprocessed = self.chain.preprocessed_blocks.index_values().await?;

        // 2. Snapshot safety-critical manager state so that we cannot be tricked
        //    into double-signing if the reset wipes votes we already cast.
        let manager_snapshot = ManagerSafetySnapshot::capture(&self.chain.manager).await?;

        // 3. Clear the chain state entirely and save.
        self.chain.clear();
        self.knows_chain_is_active = false;
        self.save().await?;
        warn!(
            %chain_id,
            "Cleared chain state up to height {tip_height}; \
            re-executing all blocks"
        );

        // 4. Re-load certificates one at a time by hash and re-process them.
        let num_confirmed = hashes.len();
        let num_preprocessed = preprocessed.len();
        for (index, hash) in hashes.into_iter().enumerate() {
            let height = BlockHeight(index as u64);
            if index % 1000 == 0 {
                info!(
                    %chain_id, confirmed = index, total = num_confirmed,
                    "Re-executing confirmed blocks after reset"
                );
            }
            let cert = self
                .storage
                .read_certificate(hash)
                .await?
                .map(CacheArc::unwrap_or_clone)
                .ok_or_else(|| WorkerError::LocalBlockNotFound { height, chain_id })?;
            Box::pin(self.process_confirmed_block(cert, ProcessConfirmedBlockMode::Execute, None))
                .await?;
        }
        for (index, (height, hash)) in preprocessed.into_iter().enumerate() {
            if index % 1000 == 0 {
                info!(
                    %chain_id, preprocessed = index, total = num_preprocessed,
                    "Re-preprocessing blocks after reset"
                );
            }
            let cert = self
                .storage
                .read_certificate(hash)
                .await?
                .map(CacheArc::unwrap_or_clone)
                .ok_or_else(|| WorkerError::LocalBlockNotFound { height, chain_id })?;
            Box::pin(self.process_confirmed_block(
                cert,
                ProcessConfirmedBlockMode::Preprocess,
                None,
            ))
            .await?;
        }

        // 5. Restore any previously cast votes and locking block so we cannot be
        //    asked to sign a conflicting statement at the same height/round. Votes
        //    in the manager always belong to the pending height (one past the tip),
        //    so restoring is only meaningful if re-execution landed at the same
        //    tip. Otherwise the restored state would refer to a stale pending
        //    height and could only break the manager's invariants without any
        //    safety benefit — so we drop the snapshot in that case.
        let new_tip_height = self.chain.tip_state.get().next_block_height;
        if new_tip_height == tip_height {
            manager_snapshot.restore(&mut self.chain.manager)?;
            self.save().await?;
        } else {
            warn!(
                %tip_height, %new_tip_height,
                "Dropping manager snapshot: pre-reset tip differs from post-reset tip"
            );
        }

        // 6. Build RevertConfirm requests so each sender resends messages we may
        //    have lost during the reset.
        let revert_requests = sender_ids
            .into_iter()
            .map(|sender| CrossChainRequest::RevertConfirm {
                sender,
                recipient: chain_id,
                retransmit_from: BlockHeight::ZERO,
            })
            .collect::<Vec<_>>();

        warn!(
            tip_height = %self.chain.tip_state.get().next_block_height,
            num_revert_confirms = revert_requests.len(),
            "Chain reset and re-executed; sending RevertConfirm to senders"
        );

        Ok(revert_requests)
    }

    #[instrument(skip_all, fields(
        chain_id = %self.chain_id(),
        num_trackers = %new_trackers.len()
    ))]
    pub(crate) async fn update_received_certificate_trackers(
        &mut self,
        new_trackers: BTreeMap<ValidatorPublicKey, u64>,
    ) -> Result<(), WorkerError> {
        self.chain
            .update_received_certificate_trackers(new_trackers);
        self.save().await?;
        Ok(())
    }

    /// Returns the preprocessed block hashes in the given height range.
    #[instrument(skip_all, fields(
        chain_id = %self.chain_id(),
        start = %start,
        end = %end
    ))]
    pub(crate) async fn get_preprocessed_block_hashes(
        &self,
        start: BlockHeight,
        end: BlockHeight,
    ) -> Result<Vec<CryptoHash>, WorkerError> {
        let mut hashes = Vec::new();
        let mut height = start;
        while height < end {
            match self.chain.preprocessed_blocks.get(&height).await? {
                Some(hash) => hashes.push(hash),
                None => break,
            }
            height = height.try_add_one()?;
        }
        Ok(hashes)
    }

    /// Returns the next block height to receive from an inbox.
    #[instrument(skip_all, fields(
        chain_id = %self.chain_id(),
        origin = %origin
    ))]
    pub(crate) async fn get_inbox_next_height(
        &self,
        origin: ChainId,
    ) -> Result<BlockHeight, WorkerError> {
        Ok(match self.chain.inboxes.try_load_entry(&origin).await? {
            Some(inbox) => inbox.next_block_height_to_receive()?,
            None => BlockHeight::ZERO,
        })
    }

    /// Returns the locking blobs for the given blob IDs.
    /// Returns `Ok(None)` if any of the blobs is not found.
    #[instrument(skip_all, fields(
        chain_id = %self.chain_id(),
        num_blob_ids = %blob_ids.len()
    ))]
    pub(crate) async fn get_locking_blobs(
        &self,
        blob_ids: Vec<BlobId>,
    ) -> Result<Option<Vec<Blob>>, WorkerError> {
        let results = self
            .chain
            .manager
            .locking_blobs
            .multi_get(&blob_ids)
            .await?;
        Ok(results.into_iter().collect())
    }

    /// Gets block hashes for specified heights.
    pub(crate) async fn get_block_hashes(
        &self,
        heights: Vec<BlockHeight>,
    ) -> Result<Vec<CryptoHash>, WorkerError> {
        Ok(self.chain.block_hashes(heights).await?)
    }

    /// Gets proposed blobs from the manager for specified blob IDs.
    pub(crate) async fn get_proposed_blobs(
        &self,
        blob_ids: Vec<BlobId>,
    ) -> Result<Vec<Blob>, WorkerError> {
        let results = self
            .chain
            .manager
            .proposed_blobs
            .multi_get(&blob_ids)
            .await?;
        let mut blobs = Vec::with_capacity(blob_ids.len());
        let mut missing = Vec::new();
        for (blob_id, maybe_blob) in blob_ids.into_iter().zip(results) {
            match maybe_blob {
                Some(blob) => blobs.push(blob),
                None => missing.push(blob_id),
            }
        }
        if !missing.is_empty() {
            return Err(WorkerError::BlobsNotFound(missing));
        }
        Ok(blobs)
    }

    /// Gets the previous event blocks for specific streams.
    pub(crate) async fn get_previous_event_blocks(
        &self,
        stream_ids: Vec<StreamId>,
    ) -> Result<BTreeMap<StreamId, (BlockHeight, CryptoHash)>, WorkerError> {
        let heights = self
            .chain
            .previous_event_blocks
            .multi_get(&stream_ids)
            .await?;
        let mut result = BTreeMap::new();
        let mut indices = Vec::new();
        let mut streams_with_heights = Vec::new();
        for (stream_id, height) in stream_ids.into_iter().zip(heights) {
            if let Some(height) = height {
                let index = usize::try_from(height.0).map_err(|_| ArithmeticError::Overflow)?;
                indices.push(index);
                streams_with_heights.push((stream_id, height));
            }
        }
        let hashes = self.chain.confirmed_log.multi_get(indices).await?;
        for (hash, (stream_id, height)) in hashes.into_iter().zip(streams_with_heights) {
            if let Some(hash) = hash {
                result.insert(stream_id, (height, hash));
            }
        }
        Ok(result)
    }

    /// Gets the `next_expected_events` indices for the given streams.
    pub(crate) async fn get_next_expected_events(
        &self,
        stream_ids: Vec<StreamId>,
    ) -> Result<BTreeMap<StreamId, u32>, WorkerError> {
        let values = self
            .chain
            .next_expected_events
            .multi_get(&stream_ids)
            .await?;
        Ok(stream_ids
            .into_iter()
            .zip(values)
            .filter_map(|(id, val)| Some((id, val?)))
            .collect())
    }

    /// Gets event subscriptions.
    pub(crate) async fn get_event_subscriptions(
        &self,
    ) -> Result<EventSubscriptionsResult, WorkerError> {
        Ok(self
            .chain
            .execution_state
            .system
            .event_subscriptions
            .index_values()
            .await?)
    }

    /// Gets the stream event count for a stream, including preprocessed blocks.
    pub(crate) async fn get_stream_event_count(
        &self,
        stream_id: StreamId,
    ) -> Result<Option<u32>, WorkerError> {
        // Use next_expected_events which accounts for both fully executed and
        // preprocessed (sparsely downloaded) blocks, falling back to
        // stream_event_counts for the case where next_expected_events hasn't
        // been initialized yet.
        let next_expected = self.chain.next_expected_events.get(&stream_id).await?;
        if next_expected.is_some() {
            return Ok(next_expected);
        }
        Ok(self
            .chain
            .execution_state
            .stream_event_counts
            .get(&stream_id)
            .await?)
    }

    /// Gets received certificate trackers.
    pub(crate) async fn get_received_certificate_trackers(
        &self,
    ) -> Result<HashMap<ValidatorPublicKey, u64>, WorkerError> {
        Ok(self.chain.received_certificate_trackers.get().clone())
    }

    /// Gets tip state and outbox info for next_outbox_heights calculation.
    pub(crate) async fn get_tip_state_and_outbox_info(
        &self,
        receiver_id: ChainId,
    ) -> Result<(BlockHeight, Option<BlockHeight>), WorkerError> {
        let next_block_height = self.chain.tip_state.get().next_block_height;
        let next_height_to_schedule = self
            .chain
            .outboxes
            .try_load_entry(&receiver_id)
            .await?
            .map(|outbox| *outbox.next_height_to_schedule.get());
        Ok((next_block_height, next_height_to_schedule))
    }

    /// Gets the next height to preprocess.
    pub(crate) async fn get_next_height_to_preprocess(&self) -> Result<BlockHeight, WorkerError> {
        Ok(self.chain.next_height_to_preprocess().await?)
    }

    /// Gets the chain manager's seed for leader election.
    pub(crate) async fn get_manager_seed(&self) -> Result<u64, WorkerError> {
        Ok(*self.chain.manager.seed.get())
    }

    /// Attempts to vote for a leader timeout, if possible.
    #[instrument(skip_all, fields(
        chain_id = %self.chain_id(),
        height = %height,
        round = %round
    ))]
    async fn vote_for_leader_timeout(
        &mut self,
        height: BlockHeight,
        round: Round,
    ) -> Result<(), WorkerError> {
        let chain = &mut self.chain;
        ensure!(
            height == chain.tip_state.get().next_block_height,
            WorkerError::UnexpectedBlockHeight {
                expected_block_height: chain.tip_state.get().next_block_height,
                found_block_height: height
            }
        );
        let epoch = chain.execution_state.system.epoch.get();
        let chain_id = chain.chain_id();
        let key_pair = self.config.key_pair();
        let local_time = self.storage.clock().current_time();
        if chain
            .manager
            .create_timeout_vote(chain_id, height, round, *epoch, key_pair, local_time)?
        {
            self.save().await?;
        }
        Ok(())
    }

    /// Votes for falling back to a public chain.
    /// This is disabled on the testnet.
    #[instrument(skip_all, fields(
        chain_id = %self.chain_id()
    ))]
    async fn vote_for_fallback(&mut self) -> Result<(), WorkerError> {
        Err(WorkerError::NoFallbackMode)
    }

    #[instrument(skip_all, fields(
        chain_id = %self.chain_id(),
        blob_id = %blob.id()
    ))]
    pub(crate) async fn handle_pending_blob(
        &mut self,
        blob: Blob,
    ) -> Result<ChainInfoResponse, WorkerError> {
        let mut was_expected = self
            .chain
            .pending_validated_blobs
            .maybe_insert(&blob)
            .await?;
        for (_, mut pending_blobs) in self
            .chain
            .pending_proposed_blobs
            .try_load_all_entries_mut()
            .await?
        {
            if !pending_blobs.validated.get() {
                let (_, committee) = self.chain.current_committee().await?;
                let policy = committee.policy();
                policy
                    .check_blob_size(blob.content())
                    .with_execution_context(ChainExecutionContext::Block)?;
                ensure!(
                    u64::try_from(pending_blobs.pending_blobs.count().await?)
                        .is_ok_and(|count| count < policy.maximum_published_blobs),
                    WorkerError::TooManyPublishedBlobs(policy.maximum_published_blobs)
                );
            }
            was_expected = was_expected || pending_blobs.maybe_insert(&blob).await?;
        }
        ensure!(was_expected, WorkerError::UnexpectedBlob);
        self.save().await?;
        self.chain_info_response().await
    }

    /// Returns a stored [`Certificate`] for the chain's block at the requested [`BlockHeight`].
    #[cfg(with_testing)]
    #[instrument(skip_all, fields(
        chain_id = %self.chain_id(),
        height = %height
    ))]
    pub(crate) async fn read_certificate(
        &self,
        height: BlockHeight,
    ) -> Result<Option<CacheArc<ConfirmedBlockCertificate>>, WorkerError> {
        let certificate_hash = match self.chain.confirmed_log.get(height.try_into()?).await? {
            Some(hash) => hash,
            None => return Ok(None),
        };
        let certificate = self
            .storage
            .read_certificate(certificate_hash)
            .await?
            .ok_or_else(|| WorkerError::ReadCertificatesError(vec![certificate_hash]))?;
        Ok(Some(certificate))
    }

    /// Queries an application's state on the chain.
    #[instrument(skip_all, fields(
        chain_id = %self.chain_id(),
        query_application_id = %query.application_id()
    ))]
    pub(crate) async fn query_application(
        &mut self,
        query: Query,
        block_hash: Option<CryptoHash>,
    ) -> Result<(QueryOutcome, BlockHeight), WorkerError> {
        self.initialize_and_save_if_needed().await?;
        let next_block_height = self.chain.tip_state.get().next_block_height;
        let local_time = self.storage.clock().current_time();
        if let Some(requested_block) = block_hash {
            if let Some(mut state) = self
                .execution_state_cache
                .as_ref()
                .and_then(|cache| cache.remove(&requested_block))
            {
                // We try to use a cached execution state for the requested block.
                // We want to pretend that this block is committed, so we set the next block height.
                let next_block_height = next_block_height
                    .try_add_one()
                    .expect("block height to not overflow");
                let context = QueryContext {
                    chain_id: self.chain_id(),
                    next_block_height,
                    local_time,
                };
                let outcome = state
                    .with_context(|ctx| {
                        self.chain
                            .execution_state
                            .context()
                            .clone_with_base_key(ctx.base_key().bytes.clone())
                    })
                    .await
                    .query_application(context, query, self.service_runtime_endpoint.as_mut())
                    .await
                    .with_execution_context(ChainExecutionContext::Query)?;
                if let Some(cache) = &self.execution_state_cache {
                    cache.insert(&requested_block, state);
                }
                Ok((outcome, next_block_height))
            } else {
                tracing::debug!(requested_block = %requested_block, "requested block hash not found in cache, querying committed state");
                let outcome = self
                    .chain
                    .query_application(local_time, query, self.service_runtime_endpoint.as_mut())
                    .await?;
                Ok((outcome, next_block_height))
            }
        } else {
            let outcome = self
                .chain
                .query_application(local_time, query, self.service_runtime_endpoint.as_mut())
                .await?;
            Ok((outcome, next_block_height))
        }
    }

    /// Returns an application's description by reading the blob directly from storage.
    ///
    /// Does not track blob usage (which requires `&mut self`), making it safe for
    /// concurrent reads. Blob tracking is only relevant during block execution and is
    /// always rolled back for read-only queries.
    #[instrument(skip_all, fields(
        chain_id = %self.chain_id(),
        application_id = %application_id
    ))]
    pub(crate) async fn describe_application_readonly(
        &self,
        application_id: ApplicationId,
    ) -> Result<ApplicationDescription, WorkerError> {
        let blob_id = application_id.description_blob_id();
        let blob = self
            .storage
            .read_blob(blob_id)
            .await?
            .ok_or(WorkerError::BlobsNotFound(vec![blob_id]))?;
        Ok(bcs::from_bytes(blob.bytes())?)
    }

    /// Executes a block without persisting any changes to the state.
    ///
    /// The block may be modified to reflect the actual executed transactions
    /// (bundles may be rejected or removed based on the policy).
    #[instrument(skip_all, fields(
        chain_id = %self.chain_id(),
        block_height = %block.height
    ))]
    pub(crate) async fn stage_block_execution(
        &mut self,
        block: ProposedBlock,
        round: Option<u32>,
        published_blobs: &[Blob],
        policy: BundleExecutionPolicy,
    ) -> Result<
        (
            ProposedBlock,
            Block,
            ChainInfoResponse,
            ResourceTracker,
            HashSet<ChainId>,
        ),
        WorkerError,
    > {
        self.initialize_and_save_if_needed().await?;
        let local_time = self.storage.clock().current_time();
        let signer = block.authenticated_signer;
        let (_, committee) = self.chain.current_committee().await?;
        block.check_proposal_size(committee.policy().maximum_block_proposal_size)?;

        self.chain
            .remove_bundles_from_inboxes(block.timestamp, true, block.incoming_bundles())
            .await?;
        let (executed_block, resource_tracker, never_reject_origins) =
            Box::pin(self.execute_block(
                block,
                local_time,
                round,
                published_blobs,
                BlockExecution::StageProposal { policy },
            ))
            .await?;

        // No need to sign: only used internally.
        let info = ChainInfo::from_chain_view(&self.chain).await?;
        let mut response = ChainInfoResponse::new(info, None);
        if let Some(signer) = signer {
            response.info.requested_owner_balance = self
                .chain
                .execution_state
                .system
                .balances
                .get(&signer)
                .await?;
        }

        let (proposed_block, _) = executed_block.clone().into_proposal();
        Ok((
            proposed_block,
            executed_block,
            response,
            resource_tracker,
            never_reject_origins,
        ))
    }

    /// Validates and executes a block proposed to extend this chain.
    ///
    /// Returns network actions alongside the result so the caller can dispatch them
    /// even when the proposal is rejected: a `HasIncompatibleConfirmedVote` rejection
    /// can still advance `current_round` via `update_signed_proposal`, and subscribers
    /// need the resulting `NewRound` notification.
    #[instrument(skip_all, fields(
        chain_id = %self.chain_id(),
        block_height = %proposal.content.block.height
    ))]
    pub(crate) async fn handle_block_proposal(
        &mut self,
        proposal: BlockProposal,
    ) -> (Result<ChainInfoResponse, WorkerError>, NetworkActions) {
        #[cfg(with_metrics)]
        metrics::BLOCK_PROPOSALS_RECEIVED_TOTAL.inc();
        let chain_id = proposal.content.block.chain_id;
        let height = proposal.content.block.height;
        let old_round = self.chain.manager.current_round();
        match self.try_handle_block_proposal(proposal).await {
            Ok((response, actions)) => (Ok(response), actions),
            Err(err) => {
                let error_type = err.error_type();
                #[cfg(with_metrics)]
                metrics::BLOCK_PROPOSALS_REJECTED_TOTAL
                    .with_label_values(&[error_type.as_str()])
                    .inc();
                debug!(%chain_id, %height, %error_type, "Block proposal rejected");
                // Even on error, the manager's `current_round` may have advanced
                // (the `HasIncompatibleConfirmedVote` recovery path calls
                // `update_signed_proposal`). Surface the resulting `NewRound`
                // notification so subscribers can react.
                let actions = if self.chain.manager.current_round() != old_round {
                    self.create_network_actions(Some(old_round))
                        .await
                        .unwrap_or_default()
                } else {
                    NetworkActions::default()
                };
                (Err(err), actions)
            }
        }
    }

    async fn try_handle_block_proposal(
        &mut self,
        proposal: BlockProposal,
    ) -> Result<(ChainInfoResponse, NetworkActions), WorkerError> {
        self.initialize_and_save_if_needed().await?;
        proposal
            .check_invariants()
            .map_err(|msg| WorkerError::InvalidBlockProposal(msg.to_string()))?;
        proposal.check_signature()?;
        let owner = proposal.owner();
        let BlockProposal {
            content,
            original_proposal,
            signature: _,
        } = &proposal;
        let block = &content.block;
        let chain = &self.chain;
        // Check if the chain is ready for this new block proposal.
        chain.tip_state.get().verify_block_chaining(block)?;
        // Check the epoch.
        let (epoch, committee) = chain.current_committee().await?;
        check_block_epoch(epoch, block.chain_id, block.epoch)?;
        let policy = committee.policy().clone();
        block.check_proposal_size(policy.maximum_block_proposal_size)?;
        // Check the authentication of the block.
        ensure!(
            chain.manager.verify_owner(&owner, proposal.content.round)?,
            WorkerError::InvalidOwner
        );
        let old_round = self.chain.manager.current_round();
        match original_proposal {
            None => {
                if let Some(signer) = block.authenticated_signer {
                    // Check the authentication of the operations in the new block.
                    ensure!(signer == owner, WorkerError::InvalidSigner(owner));
                }
            }
            Some(OriginalProposal::Regular { certificate }) => {
                // Verify that this block has been validated by a quorum before.
                certificate.check(&committee)?;
            }
            Some(OriginalProposal::Fast(signature)) => {
                let original_proposal = BlockProposal {
                    content: ProposalContent {
                        block: content.block.clone(),
                        round: Round::Fast,
                        outcome: None,
                    },
                    signature: *signature,
                    original_proposal: None,
                };
                let super_owner = original_proposal.owner();
                ensure!(
                    chain
                        .manager
                        .ownership
                        .get()
                        .super_owners
                        .contains(&super_owner),
                    WorkerError::InvalidOwner
                );
                if let Some(signer) = block.authenticated_signer {
                    // Check the authentication of the operations in the new block.
                    ensure!(signer == super_owner, WorkerError::InvalidSigner(signer));
                }
                original_proposal.check_signature()?;
            }
        }
        let local_time = self.storage.clock().current_time();
        match chain.manager.check_proposed_block(&proposal) {
            Ok(manager::Outcome::Skip) => {
                // We already voted for this block.
                return Ok((self.chain_info_response().await?, NetworkActions::default()));
            }
            Ok(manager::Outcome::Accept) => {}
            Err(err) => {
                // A `HasIncompatibleConfirmedVote` rejection means the proposer is at a
                // round we'd otherwise be happy to sign at; only our prior confirmed vote
                // prevents us from voting. Record the proposal so `current_round` still
                // tracks the round the proposer is in — without it, the chain can wedge.
                // Other rejections (e.g. `WrongRound`, `InsufficientRound`) mean the
                // proposal is not actually valid for the chain's current state, so we
                // shouldn't let it advance our round.
                if matches!(err, ChainError::HasIncompatibleConfirmedVote(_, _))
                    && self
                        .chain
                        .manager
                        .update_signed_proposal(&proposal, local_time)
                {
                    self.save().await?;
                }
                return Err(err.into());
            }
        }

        // Make sure we remember that a proposal was signed, to determine the correct round to
        // propose in.
        if self
            .chain
            .manager
            .update_signed_proposal(&proposal, local_time)
        {
            self.save().await?;
        }

        let published_blobs = self.load_proposal_blobs(&proposal).await?;
        let ProposalContent {
            block,
            round,
            outcome,
        } = content;

        if self.config.key_pair().is_some()
            && block.timestamp.duration_since(local_time) > self.config.block_time_grace_period
        {
            return Err(WorkerError::InvalidTimestamp {
                local_time,
                block_timestamp: block.timestamp,
                block_time_grace_period: self.config.block_time_grace_period,
            });
        }
        // Note: WorkerState::handle_block_proposal delays processing proposals with future
        // timestamps (within the grace period) before acquiring the chain lock. By the time
        // we reach here, the block timestamp should be in the past or very close to current time.

        self.chain
            .remove_bundles_from_inboxes(block.timestamp, true, block.incoming_bundles())
            .await?;
        let block = if let Some(outcome) = outcome {
            outcome.clone().with(proposal.content.block.clone())
        } else {
            let (executed_block, _resource_tracker, _) = Box::pin(self.execute_block(
                block.clone(),
                local_time,
                round.multi_leader(),
                &published_blobs,
                BlockExecution::HandleProposal,
            ))
            .await?;
            executed_block
        };

        ensure!(
            !round.is_fast() || !block.has_oracle_responses(),
            WorkerError::FastBlockUsingOracles
        );
        let chain = &mut self.chain;
        // Check if the counters of tip_state would be valid.
        chain
            .tip_state
            .get_mut()
            .update_counters(&block.body.transactions, &block.body.messages)?;
        // Don't save the changes since the block is not confirmed yet.
        chain.rollback();

        // Create the vote and store it in the chain state.
        let blobs = self
            .get_required_blobs(proposal.expected_blob_ids(), block.created_blobs())
            .await?;
        let key_pair = self.config.key_pair();
        let manager = &mut self.chain.manager;
        match manager.create_vote(&proposal, block, key_pair, local_time, blobs)? {
            // Cache the value we voted on, so the client doesn't have to send it again.
            Some(Either::Left(vote)) => {
                self.block_values
                    .insert_hashed(Cow::Borrowed(vote.value.inner()));
            }
            Some(Either::Right(vote)) => {
                self.block_values
                    .insert_hashed(Cow::Borrowed(vote.value.inner()));
            }
            None => (),
        }
        self.save().await?;
        let actions = self.create_network_actions(Some(old_round)).await?;
        Ok((self.chain_info_response().await?, actions))
    }

    /// Prepares a [`ChainInfoResponse`] for a [`ChainInfoQuery`].
    #[instrument(skip_all, fields(
        chain_id = %self.chain_id()
    ))]
    async fn prepare_chain_info_response(
        &mut self,
        query: ChainInfoQuery,
    ) -> Result<ChainInfoResponse, WorkerError> {
        self.initialize_and_save_if_needed().await?;
        let mut info = ChainInfo::from_chain_view(&self.chain).await?;
        if query.request_committees {
            // Must reflect the chain's *own* view of which epochs it trusts:
            // `collect_epoch_changes` in the chain client uses `min(committees.keys())`
            // to decide where to start emitting `ProcessRemovedEpoch` ops. With a
            // process-wide snapshot we would re-process the admin chain's own
            // revocations. The load is transient — `committees` is evicted on the
            // next save.
            info.requested_committees = Some(
                self.chain
                    .execution_state
                    .system
                    .committees
                    .get()
                    .await?
                    .clone(),
            );
        }
        if query.request_owner_balance == AccountOwner::CHAIN {
            info.requested_owner_balance = Some(*self.chain.execution_state.system.balance.get());
        } else {
            info.requested_owner_balance = self
                .chain
                .execution_state
                .system
                .balances
                .get(&query.request_owner_balance)
                .await?;
        }
        if let Some(next_block_height) = query.test_next_block_height {
            // If not, send the same error as if a block with next_block_height was proposed.
            ensure!(
                self.chain.tip_state.get().next_block_height == next_block_height,
                WorkerError::UnexpectedBlockHeight {
                    expected_block_height: self.chain.tip_state.get().next_block_height,
                    found_block_height: next_block_height,
                }
            );
        }
        if query.request_pending_message_bundles {
            // Lazily initialize the nonempty_inboxes set for chains from older DB versions.
            let origins = if let Some(nonempty_origins) = self.chain.nonempty_inboxes.get().clone()
            {
                nonempty_origins.into_iter().collect::<Vec<_>>()
            } else {
                let pairs = self.chain.inboxes.try_load_all_entries().await?;
                let nonempty_origins = pairs
                    .into_iter()
                    .filter(|(_, inbox)| inbox.added_bundles.count() > 0)
                    .map(|(origin, _)| origin)
                    .collect::<BTreeSet<ChainId>>();
                let origins = nonempty_origins.iter().copied().collect::<Vec<_>>();
                *self.chain.nonempty_inboxes.get_mut() = Some(nonempty_origins);
                self.save().await?;
                origins
            };
            let mut bundles = Vec::new();
            let inboxes = self.chain.inboxes.try_load_entries(&origins).await?;
            let origins_and_inboxes = origins
                .into_iter()
                .zip(inboxes)
                .filter_map(|(origin, inbox)| Some((origin, inbox?)))
                .collect::<Vec<_>>();
            #[cfg(with_metrics)]
            metrics::NUM_INBOXES
                .with_label_values(&[])
                .observe(origins_and_inboxes.len() as f64);
            let is_closed = *self.chain.execution_state.system.closed.get();
            let action = if is_closed {
                MessageAction::Reject
            } else {
                MessageAction::Accept
            };
            for (origin, inbox) in origins_and_inboxes {
                for bundle in inbox.added_bundles.elements().await? {
                    bundles.push(IncomingBundle {
                        origin,
                        bundle,
                        action,
                    });
                }
            }
            if is_closed && !bundles.is_empty() {
                info!(
                    chain_id = %self.chain.chain_id(),
                    count = bundles.len(),
                    "Auto-rejecting all incoming message bundles because the chain is closed"
                );
            }
            info.requested_pending_message_bundles = bundles;
        }
        let hashes = self
            .chain
            .block_hashes(query.request_sent_certificate_hashes_by_heights)
            .await?;
        info.requested_sent_certificate_hashes = hashes;
        if let Some(start) = query.request_received_log_excluding_first_n {
            let start = usize::try_from(start).map_err(|_| ArithmeticError::Overflow)?;
            let max_received_log_entries = self.config.chain_info_max_received_log_entries;
            let end = start
                .saturating_add(max_received_log_entries)
                .min(self.chain.received_log.count());
            info.requested_received_log = self.chain.received_log.read(start..end).await?;
        }
        if query.request_manager_values {
            info.manager.add_values(&self.chain.manager);
        }
        Ok(ChainInfoResponse::new(info, self.config.key_pair()))
    }

    /// Executes a block with a specified policy for handling bundle failures.
    ///
    /// The block may be modified to reflect the actual executed transactions.
    #[instrument(skip_all, fields(
        chain_id = %self.chain_id(),
        block_height = %block.height
    ))]
    async fn execute_block(
        &mut self,
        block: ProposedBlock,
        local_time: Timestamp,
        round: Option<u32>,
        published_blobs: &[Blob],
        execution: BlockExecution,
    ) -> Result<(Block, ResourceTracker, HashSet<ChainId>), WorkerError> {
        let (proposed_block, outcome, resource_tracker, never_reject_origins) = Box::pin(
            self.chain
                .execute_block(block, local_time, round, published_blobs, execution),
        )
        .await?;
        let executed_block = Block::new(proposed_block, outcome);
        let block_hash = CryptoHash::new(&executed_block);
        if let Some(cache) = &self.execution_state_cache {
            cache.insert(
                &block_hash,
                Box::pin(
                    self.chain
                        .execution_state
                        .with_context(|ctx| InactiveContext(ctx.base_key().clone())),
                )
                .await,
            );
        }
        Ok((executed_block, resource_tracker, never_reject_origins))
    }

    /// Initializes and saves the current chain if it is not active yet.
    #[instrument(skip_all, fields(
        chain_id = %self.chain_id()
    ))]
    pub(crate) async fn initialize_and_save_if_needed(&mut self) -> Result<(), WorkerError> {
        if !self.knows_chain_is_active {
            let local_time = self.storage.clock().current_time();
            self.chain.initialize_if_needed(local_time).await?;
            self.save().await?;
            self.knows_chain_is_active = true;
        }
        Ok(())
    }

    pub(crate) async fn chain_info_response(&self) -> Result<ChainInfoResponse, WorkerError> {
        let info = ChainInfo::from_chain_view(&self.chain).await?;
        Ok(ChainInfoResponse::new(info, self.config.key_pair()))
    }

    /// Stores the chain state in persistent storage.
    ///
    /// If the save fails, the worker is marked as poisoned and must be reloaded.
    #[instrument(skip_all, fields(
        chain_id = %self.chain_id()
    ))]
    pub(crate) async fn save(&mut self) -> Result<(), WorkerError> {
        if let Err(error) = self.chain.save().await {
            if error.must_reload_view() {
                tracing::error!(
                    ?error,
                    chain_id = %self.chain_id(),
                    "Chain save failed with a nonrecoverable error; marking worker as poisoned"
                );
                self.poisoned = true;
            }
            return Err(WorkerError::ViewError(error));
        }
        // Committee lookups go through the process-global SharedCommittees cache, so the
        // chain-local `committees` map doesn't need to sit in memory across worker calls.
        self.chain.execution_state.system.committees.evict();
        Ok(())
    }
}

/// Classifies an error returned while processing a single cross-chain batch request.
///
/// If the error indicates a poisoned worker or corrupted chain state, it is returned
/// as the first element so `process_batch` can hand it to `chain_write` for recovery
/// (eviction or reset), and the originating caller is told `BatchRolledBack` so it
/// retries against a freshly loaded worker. Any other error leaves the worker healthy
/// after rollback and is reported to the caller unchanged.
fn classify_processing_error(error: WorkerError) -> (Option<WorkerError>, WorkerError) {
    if error.must_reload_view() || error.indicates_corrupted_chain_state() {
        (Some(error), WorkerError::BatchRolledBack)
    } else {
        (None, error)
    }
}

/// Sends a result through a oneshot channel, logging at `debug` level if the
/// receiver has been dropped.
pub(crate) fn send_result<T>(sender: oneshot::Sender<T>, value: T) {
    if sender.send(value).is_err() {
        tracing::debug!("cannot send cross-chain result; receiver dropped");
    }
}

/// Returns the missing indices and corresponding blob IDs.
fn missing_indices_blob_ids(maybe_blobs: &[(BlobId, Option<Blob>)]) -> (Vec<usize>, Vec<BlobId>) {
    let mut missing_indices = Vec::new();
    let mut missing_blob_ids = Vec::new();
    for (index, (blob_id, blob)) in maybe_blobs.iter().enumerate() {
        if blob.is_none() {
            missing_indices.push(index);
            missing_blob_ids.push(*blob_id);
        }
    }
    (missing_indices, missing_blob_ids)
}

/// Returns the blob IDs whose corresponding value is `None`.
fn missing_blob_ids<'a>(
    maybe_blobs: impl IntoIterator<Item = (&'a BlobId, &'a Option<Blob>)>,
) -> Vec<BlobId> {
    maybe_blobs
        .into_iter()
        .filter(|(_, maybe_blob)| maybe_blob.is_none())
        .map(|(blob_id, _)| *blob_id)
        .collect()
}

/// Returns an error if the block is not at the expected epoch.
fn check_block_epoch(
    chain_epoch: Epoch,
    block_chain: ChainId,
    block_epoch: Epoch,
) -> Result<(), WorkerError> {
    ensure!(
        block_epoch == chain_epoch,
        WorkerError::InvalidEpoch {
            chain_id: block_chain,
            epoch: block_epoch,
            chain_epoch
        }
    );
    Ok(())
}

/// Helper type for handling cross-chain updates.
pub(crate) struct CrossChainUpdateHelper {
    pub(crate) allow_messages_from_deprecated_epochs: bool,
    pub(crate) current_epoch: Epoch,
}

impl CrossChainUpdateHelper {
    /// Creates a new [`CrossChainUpdateHelper`].
    fn new<C>(config: &ChainWorkerConfig, chain: &ChainStateView<C>) -> Self
    where
        C: Context + Clone + 'static,
    {
        CrossChainUpdateHelper {
            allow_messages_from_deprecated_epochs: config.allow_messages_from_deprecated_epochs,
            current_epoch: *chain.execution_state.system.epoch.get(),
        }
    }

    /// Checks basic invariants and deals with repeated heights and deprecated epochs.
    /// * Returns a range of message bundles that are both new to us and not relying on
    ///   an untrusted set of validators.
    /// * In the case of validators, if the epoch(s) of the highest bundles are not
    ///   known to the process, we only accept bundles that contain messages that were
    ///   already executed by anticipation (i.e. received in certified blocks).
    /// * Basic invariants are checked for good measure. We still crucially trust
    ///   the worker of the sending chain to have verified and executed the blocks
    ///   correctly.
    pub(crate) async fn select_message_bundles<S: Storage>(
        &self,
        origin: &ChainId,
        recipient: ChainId,
        next_height_to_receive: BlockHeight,
        last_anticipated_block_height: Option<BlockHeight>,
        mut bundles: Vec<(Epoch, MessageBundle)>,
        storage: &S,
    ) -> Result<Vec<MessageBundle>, WorkerError> {
        let mut latest_height = None;
        let mut skipped_len = 0;
        let mut trusted_len = 0;
        for (i, (epoch, bundle)) in bundles.iter().enumerate() {
            // Make sure that heights are not decreasing.
            ensure!(
                latest_height <= Some(bundle.height),
                WorkerError::InvalidCrossChainRequest
            );
            latest_height = Some(bundle.height);
            // Check if the block has been received already.
            if bundle.height < next_height_to_receive {
                skipped_len = i + 1;
            }
            // Check if the height is trusted or the epoch is known to the process.
            // "Known" means we have the `NewCommittee` event (and therefore the committee
            // blob) locally — either in the shared cache or in storage.
            let epoch_is_known = self.allow_messages_from_deprecated_epochs
                || Some(bundle.height) <= last_anticipated_block_height
                || *epoch >= self.current_epoch
                || storage.get_or_load_committee(*epoch).await?.is_some();
            if epoch_is_known {
                trusted_len = i + 1;
            }
        }
        if skipped_len > 0 {
            let (_, sample_bundle) = &bundles[skipped_len - 1];
            debug!(
                "Ignoring repeated messages to {recipient:.8} from {origin:} at height {}",
                sample_bundle.height,
            );
        }
        if skipped_len < bundles.len() && trusted_len < bundles.len() {
            let (sample_epoch, sample_bundle) = &bundles[trusted_len];
            warn!(
                "Refusing messages to {recipient:.8} from {origin:} at height {} \
                 because the epoch {} is not known locally",
                sample_bundle.height, sample_epoch,
            );
        }
        let bundles = if skipped_len < trusted_len {
            bundles
                .drain(skipped_len..trusted_len)
                .map(|(_, bundle)| bundle)
                .collect()
        } else {
            vec![]
        };
        Ok(bundles)
    }
}