linera_core/client/chain_client/

mod.rs

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

mod state;
use std::{
    collections::{
        hash_map::{self, DefaultHasher},
        BTreeMap, BTreeSet, HashMap, HashSet,
    },
    convert::Infallible,
    hash::{Hash as _, Hasher as _},
    iter,
    sync::{
        atomic::{AtomicU64, Ordering},
        Arc,
    },
};

use custom_debug_derive::Debug;
use futures::{
    future::{self, Either, FusedFuture, Future},
    stream::{self, AbortHandle, FusedStream, FuturesUnordered, StreamExt, TryStreamExt},
};
#[cfg(with_metrics)]
use linera_base::prometheus_util::MeasureLatency as _;
use linera_base::{
    abi::Abi,
    crypto::{signer, CryptoHash, Signer, ValidatorPublicKey},
    data_types::{
        Amount, ApplicationDescription, ApplicationPermissions, ArithmeticError, Blob, BlobContent,
        BlockHeight, ChainDescription, Epoch, MessagePolicy, Round, TimeDelta, Timestamp,
    },
    ensure,
    identifiers::{
        Account, AccountOwner, ApplicationId, BlobId, BlobType, ChainId, EventId, IndexAndEvent,
        ModuleId, StreamId,
    },
    ownership::{ChainOwnership, TimeoutConfig},
    time::{Duration, Instant},
};
#[cfg(not(target_arch = "wasm32"))]
use linera_base::{data_types::Bytecode, vm::VmRuntime};
use linera_chain::{
    data_types::{
        BlockProposal, BundleExecutionPolicy, BundleFailurePolicy, ChainAndHeight, IncomingBundle,
        ProposedBlock, Transaction,
    },
    manager::LockingBlock,
    types::{
        Block, ConfirmedBlock, ConfirmedBlockCertificate, Timeout, TimeoutCertificate,
        ValidatedBlock,
    },
    ChainError, ChainExecutionContext,
};
use linera_execution::{
    committee::Committee,
    system::{
        AdminOperation, OpenChainConfig, SystemOperation, EPOCH_STREAM_NAME,
        REMOVED_EPOCH_STREAM_NAME,
    },
    ExecutionError, Operation, Query, QueryOutcome,
};
use linera_storage::{Arc as CacheArc, Clock as _, Storage as _};
use linera_views::ViewError;
use serde::Serialize;
pub use state::State;
use thiserror::Error;
use tokio::sync::mpsc;
use tokio_stream::wrappers::UnboundedReceiverStream;
use tracing::{debug, error, info, instrument, trace, warn, Instrument as _};

use super::{
    received_log::ReceivedLogs, validator_trackers::ValidatorTrackers, AbortOnDrop, Client,
    ListeningMode, PendingProposal, TimingType,
};
use crate::{
    data_types::{ChainInfo, ChainInfoQuery, ClientOutcome, RoundTimeout},
    environment::Environment,
    local_node::{LocalNodeClient, LocalNodeError},
    node::{
        CrossChainMessageDelivery, NodeError, NotificationStream, ValidatorNode,
        ValidatorNodeProvider as _,
    },
    remote_node::RemoteNode,
    updater::{communicate_with_quorum, CommunicateAction, CommunicationError},
    worker::{Notification, Reason, WorkerError},
    MaybeSendBoxFuture,
};

/// Number of certificates above which `find_received_certificates` reports its start,
/// progress, and completion at `info` level. Smaller syncs stay quiet outside `debug`/`trace`
/// logging. Counted in certificates, not in received-log entries: the entry total is summed
/// over validators, so the two differ and the log line reports both.
const LARGE_RECEIVED_CERTIFICATE_SYNC_THRESHOLD: usize = 50_000;

/// Number of certificates processed between two `info`-level progress messages during a
/// large sync of received certificates.
const RECEIVED_CERTIFICATE_SYNC_PROGRESS_INTERVAL: usize = 100_000;

/// Options that configure the behavior of a [`ChainClient`].
#[derive(Debug, Clone)]
pub struct Options {
    /// Maximum number of pending message bundles processed at a time in a block.
    pub max_pending_message_bundles: usize,
    /// Maximum number of message bundles to discard from a block proposal due to block limit
    /// errors before discarding all remaining bundles.
    ///
    /// Discarded bundles can be retried in the next block.
    pub max_block_limit_errors: u32,
    /// Time budget for staging message bundles. When set, limits bundle execution by
    /// wall-clock time, in addition to the count limit from `max_pending_message_bundles`.
    pub staging_bundles_time_budget: Option<Duration>,
    /// The policy for automatically handling incoming messages.
    pub message_policy: MessagePolicy,
    /// Chain IDs whose incoming bundles should be processed first when proposing a block.
    pub priority_bundle_origins: HashSet<ChainId>,
    /// Whether to block on cross-chain message delivery.
    pub cross_chain_message_delivery: CrossChainMessageDelivery,
    /// An additional delay, after reaching a quorum, to wait for additional validator signatures,
    /// as a fraction of time taken to reach quorum.
    pub quorum_grace_period: f64,
    /// The delay when downloading a blob, after which we try a second validator.
    pub blob_download_hedge_delay: Duration,
    /// The delay when downloading a batch of certificates, after which we try a second validator.
    pub certificate_batch_download_hedge_delay: Duration,
    /// Maximum number of certificates that we download at a time from one validator when
    /// synchronizing one of our chains.
    pub certificate_download_batch_size: u64,
    /// Maximum number of certificates read from local storage and uploaded to a validator
    /// at a time when synchronizing a chain.
    pub certificate_upload_batch_size: usize,
    /// Maximum number of sender certificates we try to download and receive in one go
    /// when syncing sender chains.
    pub sender_certificate_download_batch_size: usize,
    /// Maximum number of certificate batches downloaded concurrently during chain sync.
    pub max_concurrent_batch_downloads: usize,
    /// Maximum number of tasks that can be joined concurrently using buffer_unordered.
    pub max_joined_tasks: usize,
    /// Whether to allow creating blocks in the fast round. Fast blocks have lower latency but
    /// must be used carefully so that there are never any conflicting fast block proposals.
    pub allow_fast_blocks: bool,
    /// Initial probe interval for the notification circuit breaker. When a validator's
    /// notification stream exhausts retries, the circuit breaker waits this long before
    /// probing again. Doubles on each failed probe.
    pub notification_circuit_breaker_initial_probe_interval: Duration,
    /// Maximum probe interval for the notification circuit breaker. The probe interval
    /// doubles on each failure but is capped at this value.
    pub notification_circuit_breaker_max_probe_interval: Duration,
    /// Maximum number of event stream IDs to include in a single `PreviousEventBlocks`
    /// request. Larger sets are split into multiple requests.
    pub max_event_stream_queries: usize,
}

/// A validator's notification stream: its abort handle, plus when it began serving.
pub(super) struct StreamHandle {
    pub(super) abort: AbortHandle,
    /// Microseconds at which `subscribe` *and* the initial sync both returned, or
    /// [`NOT_SERVING`] while either is still in flight. Nothing polls the notification
    /// stream until the sync returns, so "subscribe resolved" is too weak to mean healthy.
    pub(super) serving_since: Arc<AtomicU64>,
}

impl StreamHandle {
    /// How long the stream has been serving notifications, or `None` if it never started.
    fn serving_for(&self, now: Timestamp) -> Option<Duration> {
        match self.serving_since.load(Ordering::Relaxed) {
            NOT_SERVING => None,
            micros => Some(now.duration_since(Timestamp::from(micros))),
        }
    }
}

pub(super) struct CircuitBreakerState {
    pub(super) next_probe_at: Timestamp,
    pub(super) probe_interval: Duration,
    /// Whether the stream has actually failed, as opposed to this being the deadline
    /// armed when it was launched. Only a breaker that tripped reports a recovery.
    pub(super) tripped: bool,
}

impl CircuitBreakerState {
    /// The deadline every launch arms, before anything is known about the stream.
    fn launched(now: Timestamp, probe_interval: Duration, spread: Duration) -> Self {
        Self::armed(now, probe_interval, spread, false)
    }

    /// A breaker for a stream that has failed.
    fn failed(now: Timestamp, probe_interval: Duration, spread: Duration) -> Self {
        Self::armed(now, probe_interval, spread, true)
    }

    /// A breaker for a long-lived stream that ended — routine, so it does not trip.
    ///
    /// Tripping here would report a recovery on the next successful probe for something
    /// this arm has just classified as healthy: one rolling validator restart would emit a
    /// spurious "recovered" line per chain per validator.
    fn churned(now: Timestamp, probe_interval: Duration, spread: Duration) -> Self {
        Self::armed(now, probe_interval, spread, false)
    }

    fn armed(now: Timestamp, probe_interval: Duration, spread: Duration, tripped: bool) -> Self {
        CircuitBreakerState {
            next_probe_at: now.saturating_add(TimeDelta::from_duration(probe_interval + spread)),
            probe_interval,
            tripped,
        }
    }

    /// Doubles the interval, up to `max`, and re-arms. Only a failure escalates, so this
    /// also trips the breaker.
    fn escalate(&mut self, now: Timestamp, max: Duration, spread: Duration) {
        self.probe_interval = self.probe_interval.saturating_mul(2).min(max);
        self.tripped = true;
        self.arm(now, spread);
    }

    /// Re-arms one interval out, plus a per-validator offset so that a fleet-wide stream
    /// loss does not re-converge every chain's probes on the same instant at every
    /// doubling.
    fn arm(&mut self, now: Timestamp, spread: Duration) {
        self.next_probe_at =
            now.saturating_add(TimeDelta::from_duration(self.probe_interval + spread));
    }
}

/// Sentinel in [`StreamHandle::serving_since`] for a stream that has not started serving.
pub(super) const NOT_SERVING: u64 = u64::MAX;

/// Floor for the circuit-breaker probe interval.
///
/// The flags are validated where they are parsed, so this is a backstop for programmatic
/// [`Options`], not a guard against arithmetic: a sub-second probe is a storm, since each
/// one is a `subscribe` plus a full `synchronize_chain_state_from`.
const MIN_PROBE_INTERVAL: Duration = Duration::from_secs(1);

/// Fraction of the probe interval used as the per-validator spread, as a divisor.
const PROBE_SPREAD_DIVISOR: u32 = 8;

/// How much longer a stream's FIRST launch gets before it is abandoned as stalled.
///
/// A re-probe and a first sync are different quantities. "How often to retry a validator"
/// is the probe interval; "how long an initial `synchronize_chain_state_from` may take" is
/// unbounded by nature — a chain tens of thousands of certificates behind legitimately
/// needs longer. Sharing one deadline aborts every stream at once on exactly the chains
/// furthest behind, throwing away the tail of each attempt just as the backoff doubles.
const FIRST_LAUNCH_GRACE: u32 = 4;

/// A fingerprint of the chain manager's observable consensus state, used to detect
/// whether a per-validator updater absorbed new info during a proposal attempt.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
struct ConsensusStateSnapshot {
    next_block_height: BlockHeight,
    current_round: Round,
    lock_round: Option<Round>,
    timeout_round: Option<Round>,
}

#[cfg(with_testing)]
impl Options {
    /// Returns a default set of options for use in tests.
    pub fn test_default() -> Self {
        use super::{
            DEFAULT_CERTIFICATE_DOWNLOAD_BATCH_SIZE, DEFAULT_CERTIFICATE_UPLOAD_BATCH_SIZE,
            DEFAULT_MAX_CONCURRENT_BATCH_DOWNLOADS, DEFAULT_MAX_EVENT_STREAM_QUERIES,
            DEFAULT_SENDER_CERTIFICATE_DOWNLOAD_BATCH_SIZE,
        };
        use crate::DEFAULT_QUORUM_GRACE_PERIOD;

        Options {
            max_pending_message_bundles: 10,
            max_block_limit_errors: 3,
            staging_bundles_time_budget: None,
            message_policy: MessagePolicy::default(),
            priority_bundle_origins: HashSet::new(),
            cross_chain_message_delivery: CrossChainMessageDelivery::NonBlocking,
            quorum_grace_period: DEFAULT_QUORUM_GRACE_PERIOD,
            blob_download_hedge_delay: Duration::from_secs(1),
            certificate_batch_download_hedge_delay: Duration::from_secs(1),
            certificate_download_batch_size: DEFAULT_CERTIFICATE_DOWNLOAD_BATCH_SIZE,
            certificate_upload_batch_size: DEFAULT_CERTIFICATE_UPLOAD_BATCH_SIZE,
            sender_certificate_download_batch_size: DEFAULT_SENDER_CERTIFICATE_DOWNLOAD_BATCH_SIZE,
            max_concurrent_batch_downloads: DEFAULT_MAX_CONCURRENT_BATCH_DOWNLOADS,
            max_joined_tasks: 100,
            allow_fast_blocks: false,
            notification_circuit_breaker_initial_probe_interval: Duration::from_secs(300),
            notification_circuit_breaker_max_probe_interval: Duration::from_secs(3600),
            max_event_stream_queries: DEFAULT_MAX_EVENT_STREAM_QUERIES,
        }
    }
}

impl Options {
    /// Builds the [`BundleExecutionPolicy`] based on the client options.
    pub fn bundle_execution_policy(&self) -> BundleExecutionPolicy {
        BundleExecutionPolicy {
            on_failure: BundleFailurePolicy::AutoRetry {
                max_failures: self.max_block_limit_errors,
                never_reject_application_ids: Arc::new(
                    self.message_policy.never_reject_application_ids.clone(),
                ),
            },
            time_budget: self.staging_bundles_time_budget,
        }
    }
}

/// Client to operate a chain by interacting with validators and the given local storage
/// implementation.
/// * The chain being operated is called the "local chain" or just the "chain".
/// * As a rule, operations are considered successful (and communication may stop) when
///   they succeeded in gathering a quorum of responses.
#[derive(Debug)]
pub struct ChainClient<Env: Environment> {
    /// The Linera [`Client`] that manages operations for this chain client.
    #[debug(skip)]
    pub(crate) client: Arc<Client<Env>>,
    /// The off-chain chain ID.
    chain_id: ChainId,
    /// The client options.
    #[debug(skip)]
    options: Options,
    /// The preferred owner of the chain used to sign proposals.
    /// `None` if we cannot propose on this chain.
    preferred_owner: Option<AccountOwner>,
    /// The next block height as read from the wallet.
    initial_next_block_height: BlockHeight,
    /// The last block hash as read from the wallet.
    initial_block_hash: Option<CryptoHash>,
    /// Optional timing sender for benchmarking.
    timing_sender: Option<mpsc::UnboundedSender<(u64, TimingType)>>,
    /// Sender chain IDs whose bundles were discarded due to the never-reject policy.
    /// These origins are excluded from `process_inbox` until the client is restarted.
    skipped_origins: Arc<papaya::HashSet<ChainId>>,
}

impl<Env: Environment> Clone for ChainClient<Env> {
    fn clone(&self) -> Self {
        Self {
            client: self.client.clone(),
            chain_id: self.chain_id,
            options: self.options.clone(),
            preferred_owner: self.preferred_owner,
            initial_next_block_height: self.initial_next_block_height,
            initial_block_hash: self.initial_block_hash,
            timing_sender: self.timing_sender.clone(),
            skipped_origins: self.skipped_origins.clone(),
        }
    }
}

/// Error type for [`ChainClient`].
#[derive(Debug, Error, strum::IntoStaticStr)]
#[allow(missing_docs)]
pub enum Error {
    #[error("Local node operation failed: {0}")]
    LocalNodeError(#[from] LocalNodeError),

    #[error("Remote node operation failed: {0}")]
    RemoteNodeError(#[from] NodeError),

    /// A validator reported state ahead of the local node for this chain. The caller may pull
    /// the missing state from that validator, then retry or surface the carried error.
    #[error("The local node is lagging behind a validator on chain {chain_id}: {error}")]
    LocalNodeLagging {
        chain_id: ChainId,
        error: Box<NodeError>,
    },

    #[error(transparent)]
    ArithmeticError(#[from] ArithmeticError),

    #[error("Missing certificates: {0:?}")]
    ReadCertificatesError(Vec<CryptoHash>),

    #[error("Missing confirmed block: {0:?}")]
    MissingConfirmedBlock(CryptoHash),

    #[error("JSON (de)serialization error: {0}")]
    JsonError(#[from] serde_json::Error),

    #[error("Chain operation failed: {0}")]
    ChainError(#[from] ChainError),

    #[error(transparent)]
    CommunicationError(#[from] CommunicationError<NodeError>),

    #[error("Internal error within chain client: {0}")]
    InternalError(&'static str),

    #[error(
        "Cannot accept a certificate from an unknown committee in the future. \
         Please synchronize the local view of the admin chain"
    )]
    CommitteeSynchronizationError,

    #[error("The local node is behind the trusted state in wallet and needs synchronization with validators")]
    WalletSynchronizationError,

    #[error("The state of the client is incompatible with the proposed block: {0}")]
    BlockProposalError(&'static str),

    #[error(
        "Cannot accept a certificate from a committee that was retired. \
         Try a newer certificate from the same origin"
    )]
    CommitteeDeprecationError,

    #[error("Protocol error within chain client: {0}")]
    ProtocolError(&'static str),

    #[error("Signer doesn't have key to sign for chain {0}")]
    CannotFindKeyForChain(ChainId),

    #[error("client is not configured to propose on chain {0}")]
    NoAccountKeyConfigured(ChainId),

    #[error("The chain client isn't owner on chain {0}")]
    NotAnOwner(ChainId),

    #[error(transparent)]
    ViewError(#[from] ViewError),

    #[error(
        "Failed to download certificates and update local node to the next height \
         {target_next_block_height} of chain {chain_id}"
    )]
    CannotDownloadCertificates {
        chain_id: ChainId,
        target_next_block_height: BlockHeight,
    },

    #[error("No validator provided a usable certificate registering blob {0}")]
    CannotDownloadBlob(BlobId),

    #[error(transparent)]
    BcsError(#[from] bcs::Error),

    #[error(
        "Unexpected quorum: validators voted for block hash {hash} in {round}, \
         expected block hash {expected_hash} in {expected_round}"
    )]
    UnexpectedQuorum {
        hash: CryptoHash,
        round: Round,
        expected_hash: CryptoHash,
        expected_round: Round,
    },

    #[error("signer error: {0}")]
    Signer(#[source] Box<dyn signer::Error>),

    #[error("Cannot revoke the current epoch {0}")]
    CannotRevokeCurrentEpoch(Epoch),

    #[error("Epoch is already revoked")]
    EpochAlreadyRevoked,

    #[error("Failed to download missing sender blocks from chain {chain_id} at height {height}")]
    CannotDownloadMissingSenderBlock {
        chain_id: ChainId,
        height: BlockHeight,
    },

    #[error(
        "A different block was already committed at this height. \
         The committed certificate hash is {0}"
    )]
    Conflict(CryptoHash),

    #[error(
        "Execution outcome mismatch: AutoRetry and committed execution produced \
         different outcomes for the same block"
    )]
    ExecutionOutcomeMismatch,
}

impl From<Infallible> for Error {
    fn from(infallible: Infallible) -> Self {
        match infallible {}
    }
}

impl Error {
    /// Wraps a signer error into a [`Error::Signer`] variant.
    pub fn signer_failure(err: impl signer::Error + 'static) -> Self {
        Self::Signer(Box::new(err))
    }

    /// Returns the qualified error variant name for the `error_type` metric label,
    /// delegating to the wrapped error's `error_type()` so the underlying worker or
    /// chain error name is surfaced rather than just the outer variant.
    pub fn error_type(&self) -> String {
        match self {
            Error::LocalNodeError(local_node_error) => local_node_error.error_type(),
            Error::ChainError(chain_error) => chain_error.error_type(),
            other => {
                let variant: &'static str = other.into();
                format!("ChainClientError::{variant}")
            }
        }
    }
}

impl<Env: Environment> ChainClient<Env> {
    /// Creates a new [`ChainClient`] for the given chain.
    pub fn new(
        client: Arc<Client<Env>>,
        chain_id: ChainId,
        options: Options,
        initial_block_hash: Option<CryptoHash>,
        initial_next_block_height: BlockHeight,
        preferred_owner: Option<AccountOwner>,
        timing_sender: Option<mpsc::UnboundedSender<(u64, TimingType)>>,
    ) -> Self {
        ChainClient {
            client,
            chain_id,
            options,
            preferred_owner,
            initial_block_hash,
            initial_next_block_height,
            timing_sender,
            skipped_origins: Arc::new(papaya::HashSet::new()),
        }
    }

    /// Returns whether this chain is in follow-only mode.
    pub fn is_follow_only(&self) -> bool {
        self.client.is_chain_follow_only(self.chain_id)
    }

    /// Returns the proposal mutex for this chain.
    ///
    /// The mutex serializes block proposals and holds the pending proposal (if any).
    #[instrument(level = "trace", skip(self))]
    fn proposal_mutex(&self) -> Arc<tokio::sync::Mutex<Option<PendingProposal>>> {
        self.client
            .chains
            .pin()
            .get(&self.chain_id)
            .expect("Chain client constructed for invalid chain")
            .proposal_mutex()
    }

    /// Returns the pending proposal, if any.
    #[instrument(level = "trace", skip(self))]
    pub async fn pending_proposal(&self) -> Option<PendingProposal> {
        self.proposal_mutex().lock().await.clone()
    }

    /// Gets a reference to the client's signer instance.
    #[instrument(level = "trace", skip(self))]
    pub fn signer(&self) -> &impl Signer {
        self.client.signer()
    }

    /// Returns whether the signer has a key for the given owner.
    pub async fn has_key_for(&self, owner: &AccountOwner) -> Result<bool, Error> {
        self.client.has_key_for(owner).await
    }

    /// Gets a mutable reference to the per-`ChainClient` options.
    #[instrument(level = "trace", skip(self))]
    pub fn options_mut(&mut self) -> &mut Options {
        &mut self.options
    }

    /// Gets a reference to the per-`ChainClient` options.
    #[instrument(level = "trace", skip(self))]
    pub fn options(&self) -> &Options {
        &self.options
    }

    /// Gets the ID of the associated chain.
    #[instrument(level = "trace", skip(self))]
    pub fn chain_id(&self) -> ChainId {
        self.chain_id
    }

    /// Gets a clone of the timing sender for benchmarking.
    pub fn timing_sender(&self) -> Option<mpsc::UnboundedSender<(u64, TimingType)>> {
        self.timing_sender.clone()
    }

    /// Gets the ID of the admin chain.
    #[instrument(level = "trace", skip(self))]
    pub fn admin_chain_id(&self) -> ChainId {
        self.client.admin_chain_id
    }

    /// Gets the currently preferred owner for signing the blocks.
    #[instrument(level = "trace", skip(self))]
    pub fn preferred_owner(&self) -> Option<AccountOwner> {
        self.preferred_owner
    }

    /// Sets the new, preferred owner for signing the blocks.
    #[instrument(level = "trace", skip(self))]
    pub fn set_preferred_owner(&mut self, preferred_owner: AccountOwner) {
        self.preferred_owner = Some(preferred_owner);
    }

    /// Obtains a `ChainStateView` for this client's chain.
    #[instrument(level = "trace")]
    pub async fn chain_state_view(
        &self,
    ) -> Result<crate::worker::ChainStateViewReadGuard<Env::Storage>, LocalNodeError> {
        self.client.local_node.chain_state_view(self.chain_id).await
    }

    /// Returns chain IDs that this chain subscribes to.
    #[instrument(level = "trace", skip(self))]
    pub async fn event_stream_publishers(
        &self,
    ) -> Result<BTreeMap<ChainId, BTreeSet<StreamId>>, LocalNodeError> {
        let subscriptions = self
            .client
            .local_node
            .get_event_subscriptions(self.chain_id)
            .await?;
        let mut publishers = subscriptions.into_iter().fold(
            BTreeMap::<ChainId, BTreeSet<StreamId>>::new(),
            |mut map, ((chain_id, stream_id), _)| {
                map.entry(chain_id).or_default().insert(stream_id);
                map
            },
        );
        if self.chain_id != self.client.admin_chain_id {
            publishers.insert(
                self.client.admin_chain_id,
                vec![
                    StreamId::system(EPOCH_STREAM_NAME),
                    StreamId::system(REMOVED_EPOCH_STREAM_NAME),
                ]
                .into_iter()
                .collect(),
            );
        }
        Ok(publishers)
    }

    /// Subscribes to notifications from this client's chain.
    #[instrument(level = "trace")]
    pub fn subscribe(&self) -> Result<NotificationStream, LocalNodeError> {
        self.subscribe_to(self.chain_id)
    }

    /// Subscribes to notifications from the specified chain.
    #[instrument(level = "trace")]
    pub fn subscribe_to(&self, chain_id: ChainId) -> Result<NotificationStream, LocalNodeError> {
        Ok(Box::pin(UnboundedReceiverStream::new(
            self.client.notifier.subscribe(vec![chain_id]),
        )))
    }

    /// Returns the storage client used by this client's local node.
    #[instrument(level = "trace")]
    pub fn storage_client(&self) -> &Env::Storage {
        self.client.storage_client()
    }

    /// Obtains the basic `ChainInfo` data for the local chain.
    #[instrument(level = "trace")]
    pub async fn chain_info(&self) -> Result<Box<ChainInfo>, LocalNodeError> {
        let query = ChainInfoQuery::new(self.chain_id);
        let response = self
            .client
            .local_node
            .handle_chain_info_query(query)
            .await?;
        Ok(response.info)
    }

    /// Obtains the basic `ChainInfo` data for the local chain, with chain manager values.
    #[instrument(level = "trace")]
    pub async fn chain_info_with_manager_values(&self) -> Result<Box<ChainInfo>, LocalNodeError> {
        let query = ChainInfoQuery::new(self.chain_id).with_manager_values();
        let response = self
            .client
            .local_node
            .handle_chain_info_query(query)
            .await?;
        Ok(response.info)
    }

    /// Returns a fingerprint of the chain manager's observable state: height,
    /// current round, locking block round, and timeout certificate round.
    ///
    /// [`Self::process_pending_block_without_prepare`] takes one snapshot just before
    /// each network call (re-taking it after our own local proposal handling) and
    /// compares against the post-call state. A change means a per-validator updater
    /// absorbed something we didn't have. The local node verifies signatures on
    /// anything it stores, so a single dishonest validator can't fake this.
    async fn consensus_state_snapshot(&self) -> Result<ConsensusStateSnapshot, Error> {
        let info = self.chain_info_with_manager_values().await?;
        let lock_round = info
            .manager
            .requested_locking
            .as_deref()
            .map(LockingBlock::round);
        let timeout_round = info.manager.timeout.as_deref().map(|cert| cert.round);
        Ok(ConsensusStateSnapshot {
            next_block_height: info.next_block_height,
            current_round: info.manager.current_round,
            lock_round,
            timeout_round,
        })
    }

    /// Returns the chain's description. Fetches it from the validators if necessary.
    pub async fn get_chain_description(&self) -> Result<ChainDescription, Error> {
        self.client.get_chain_description(self.chain_id).await
    }

    /// Returns the description of the given application. Fetches the description blob
    /// from the validators if necessary; the application need not be registered on
    /// this client's chain.
    pub async fn get_application_description(
        &self,
        application_id: ApplicationId,
    ) -> Result<ApplicationDescription, Error> {
        self.client
            .get_application_description(application_id)
            .await
    }

    /// Obtains up to `self.options.max_pending_message_bundles` pending message bundles for the
    /// local chain.
    #[instrument(level = "trace")]
    async fn pending_message_bundles(&self) -> Result<Vec<IncomingBundle>, Error> {
        if self.options.message_policy.is_ignore() {
            // Ignore all messages.
            return Ok(Vec::new());
        }

        let query = ChainInfoQuery::new(self.chain_id).with_pending_message_bundles();
        let info = self
            .client
            .local_node
            .handle_chain_info_query(query)
            .await?
            .info;
        if self.preferred_owner.is_some_and(|owner| {
            info.manager
                .ownership
                .is_super_owner_no_regular_owners(&owner)
        }) {
            // There are only super owners; they are expected to sync manually.
            ensure!(
                info.next_block_height >= self.initial_next_block_height,
                Error::WalletSynchronizationError
            );
        }

        let skipped = self.skipped_origins.pin();
        let mut bundles = info
            .requested_pending_message_bundles
            .into_iter()
            .filter_map(|bundle| bundle.apply_policy(&self.options.message_policy))
            .filter(|bundle| !skipped.contains(&bundle.origin))
            .collect::<Vec<_>>();
        let priority_origins = &self.options.priority_bundle_origins;
        bundles.sort_by(|a, b| {
            let a_priority = priority_origins.contains(&a.origin);
            let b_priority = priority_origins.contains(&b.origin);
            b_priority
                .cmp(&a_priority)
                .then(a.bundle.timestamp.cmp(&b.bundle.timestamp))
        });
        bundles.truncate(self.options.max_pending_message_bundles);
        Ok(bundles)
    }

    #[instrument(level = "trace")]
    async fn collect_stream_updates(&self) -> Result<Vec<Operation>, Error> {
        let subscription_map = self
            .client
            .local_node
            .get_event_subscriptions(self.chain_id)
            .await?;
        let futures = subscription_map
            .into_iter()
            .filter(|((chain_id, _), _)| {
                self.options
                    .message_policy
                    .restrict_chain_ids_to
                    .as_ref()
                    .is_none_or(|chain_set| chain_set.contains(chain_id))
            })
            .filter(|((_, stream_id), _)| {
                self.options
                    .message_policy
                    .process_events_from_application_ids
                    .as_ref()
                    .is_none_or(|app_set| app_set.contains(&stream_id.application_id))
            })
            .map(|((chain_id, stream_id), subscriptions)| {
                let client = self.client.clone();
                async move {
                    let counts = client
                        .local_node
                        .get_stream_indices(chain_id, stream_id.clone())
                        .await?;
                    let (first_index, next_index) = (counts.first_index, counts.next_index);
                    if next_index <= subscriptions.min_next_index {
                        return Ok::<_, Error>(Vec::new());
                    }
                    Ok(subscriptions
                        .applications
                        .into_iter()
                        .filter(|(_, app_index)| *app_index < next_index)
                        .map(|(application_id, _)| {
                            SystemOperation::UpdateStream {
                                application_id,
                                chain_id,
                                stream_id: stream_id.clone(),
                                first_index,
                                next_index,
                            }
                            .into()
                        })
                        .collect::<Vec<Operation>>())
                }
            });
        Ok(futures::stream::iter(futures)
            .buffer_unordered(self.options.max_joined_tasks)
            .try_collect::<Vec<_>>()
            .await?
            .into_iter()
            .flatten()
            .collect::<Vec<_>>())
    }

    /// Obtains the committee for the current epoch of the local chain.
    #[instrument(level = "trace")]
    pub async fn local_committee(&self) -> Result<Arc<Committee>, Error> {
        let info = match self.client.local_node.chain_info(self.chain_id).await {
            Ok(info) => info,
            Err(LocalNodeError::BlobsNotFound(_)) => {
                self.synchronize_chain_state(self.chain_id).await?;
                self.client.local_node.chain_info(self.chain_id).await?
            }
            Err(err) => return Err(err.into()),
        };
        let hash = info
            .committee_hash
            .ok_or(LocalNodeError::InactiveChain(self.chain_id))?;
        Ok(self
            .storage_client()
            .get_or_load_committee_by_hash(hash)
            .await
            .map_err(LocalNodeError::from)?)
    }

    /// Obtains the committee for the latest epoch on the admin chain.
    #[instrument(level = "trace")]
    pub async fn admin_committee(&self) -> Result<(Epoch, Arc<Committee>), LocalNodeError> {
        self.client.admin_committee().await
    }

    /// Obtains the identity of the current owner of the chain.
    ///
    /// Returns an error if we don't have the private key for the identity.
    #[instrument(level = "trace")]
    pub async fn identity(&self) -> Result<AccountOwner, Error> {
        let Some(preferred_owner) = self.preferred_owner else {
            return Err(Error::NoAccountKeyConfigured(self.chain_id));
        };
        let manager = self.chain_info().await?.manager;
        ensure!(
            manager.ownership.is_active(),
            LocalNodeError::InactiveChain(self.chain_id)
        );
        let fallback_owners = if manager.ownership.has_fallback() {
            self.local_committee()
                .await?
                .account_keys_and_weights()
                .map(|(key, _)| AccountOwner::from(key))
                .collect()
        } else {
            BTreeSet::new()
        };

        let is_owner = manager
            .ownership
            .can_propose_in_multi_leader_round(&preferred_owner)
            || fallback_owners.contains(&preferred_owner);

        if !is_owner {
            warn!(
                chain_id = %self.chain_id,
                ownership = ?manager.ownership,
                ?fallback_owners,
                ?preferred_owner,
                "The preferred owner is not configured as an owner of this chain",
            );
            return Err(Error::NotAnOwner(self.chain_id));
        }

        let has_signer = self.has_key_for(&preferred_owner).await?;

        if !has_signer {
            warn!(%self.chain_id, ?preferred_owner,
                "Chain is one of the owners but its Signer instance doesn't contain the key",
            );
            return Err(Error::CannotFindKeyForChain(self.chain_id));
        }

        Ok(preferred_owner)
    }

    /// Prepares the chain for a new owner by fetching the chain description and validating access.
    ///
    /// This is useful when assigning a chain to a client that may not have the owner key,
    /// e.g. when a faucet creates a chain with `open_multi_leader_rounds`.
    ///
    /// Returns the chain info if the owner can propose on this chain (either because they are
    /// an owner, or because `open_multi_leader_rounds` is enabled).
    #[instrument(level = "trace")]
    pub async fn prepare_for_owner(&self, owner: AccountOwner) -> Result<Box<ChainInfo>, Error> {
        ensure!(
            self.has_key_for(&owner).await?,
            Error::CannotFindKeyForChain(self.chain_id)
        );
        // Ensure we have the chain description blob.
        self.client
            .get_chain_description_blob(self.chain_id)
            .await?;

        // Get chain info.
        let info = self.chain_info().await?;

        // Validate that the owner can propose on this chain.
        ensure!(
            info.manager
                .ownership
                .can_propose_in_multi_leader_round(&owner),
            Error::NotAnOwner(self.chain_id)
        );

        Ok(info)
    }

    /// Prepares the chain for the next operation, i.e. makes sure we have synchronized it up to
    /// its current height.
    #[instrument(level = "trace")]
    pub async fn prepare_chain(&self) -> Result<Box<ChainInfo>, Error> {
        #[cfg(with_metrics)]
        let _latency = super::metrics::PREPARE_CHAIN_LATENCY.measure_latency();

        let mut info = self.synchronize_to_known_height().await?;

        if self.preferred_owner.is_none_or(|owner| {
            !info
                .manager
                .ownership
                .is_super_owner_no_regular_owners(&owner)
        }) {
            // If we are not a super owner or there are regular owners, we could be missing recent
            // certificates created by other clients. Further synchronize blocks from the network.
            // This is a best-effort that depends on network conditions.
            info = self.client.synchronize_chain_state(self.chain_id).await?;
        }

        if info.epoch > self.client.admin_committee().await?.0 {
            self.client
                .synchronize_chain_state(self.client.admin_chain_id)
                .await?;
        }

        Ok(info)
    }

    // Verifies that our local storage contains enough history compared to the
    // known block height. Otherwise, downloads the missing history from the
    // network.
    // The known height only differs if the wallet is ahead of storage.
    async fn synchronize_to_known_height(&self) -> Result<Box<ChainInfo>, Error> {
        let info = self
            .client
            .download_certificates(self.chain_id, self.initial_next_block_height)
            .await?;
        if info.next_block_height == self.initial_next_block_height {
            // Check that our local node has the expected block hash.
            ensure!(
                self.initial_block_hash == info.block_hash,
                Error::InternalError("Invalid chain of blocks in local node")
            );
        }
        Ok(info)
    }

    /// Attempts to update all validators about the local chain.
    #[instrument(level = "trace", skip(old_committee, latest_certificate))]
    pub async fn update_validators(
        &self,
        old_committee: Option<&Committee>,
        latest_certificate: Option<CacheArc<ConfirmedBlockCertificate>>,
    ) -> Result<(), Error> {
        let update_validators_start = linera_base::time::Instant::now();
        // Communicate the new certificate now.
        if let Some(old_committee) = old_committee {
            self.communicate_chain_updates(old_committee, latest_certificate.clone())
                .await?
        };
        if let Ok(new_committee) = self.local_committee().await {
            if old_committee.is_none_or(|old| *new_committee != *old) {
                // If the configuration just changed, communicate to the new committee as well.
                // (This is actually more important that updating the previous committee.)
                self.communicate_chain_updates(&new_committee, latest_certificate)
                    .await?;
            }
        }
        self.send_timing(update_validators_start, TimingType::UpdateValidators);
        Ok(())
    }

    /// Broadcasts certified blocks to validators.
    #[instrument(level = "trace", skip(committee))]
    pub async fn communicate_chain_updates(
        &self,
        committee: &Committee,
        latest_certificate: Option<CacheArc<ConfirmedBlockCertificate>>,
    ) -> Result<(), Error> {
        let delivery = self.options.cross_chain_message_delivery;
        let height = self.chain_info().await?.next_block_height;
        self.client
            .communicate_chain_updates(
                committee,
                self.chain_id,
                height,
                delivery,
                latest_certificate,
            )
            .await
    }

    /// Synchronizes all chains that any application on this chain subscribes to.
    /// We always consider the admin chain a relevant publishing chain, for new epochs.
    /// For event publisher chains, only event-bearing blocks are downloaded (partial sync).
    async fn synchronize_publisher_chains(&self) -> Result<(), Error> {
        let subscriptions = self
            .client
            .local_node
            .get_event_subscriptions(self.chain_id)
            .await?;
        // Group subscribed streams by publisher chain.
        let mut streams_by_chain = BTreeMap::<ChainId, BTreeSet<StreamId>>::new();
        for ((chain_id, stream_id), _) in &subscriptions {
            if *chain_id != self.chain_id {
                streams_by_chain
                    .entry(*chain_id)
                    .or_default()
                    .insert(stream_id.clone());
            }
        }
        // Always fully sync the admin chain for epoch changes.
        let admin_chain_id = self.client.admin_chain_id;
        if admin_chain_id != self.chain_id {
            self.client.synchronize_chain_state(admin_chain_id).await?;
        }
        // For event publisher chains, do a partial sync using previous_event_blocks.
        let (_, committee) = self.admin_committee().await?;
        let nodes = self.client.make_nodes(&committee)?;
        let tasks = streams_by_chain
            .into_iter()
            .filter(|(chain_id, _)| *chain_id != admin_chain_id)
            .map(|(chain_id, stream_ids)| {
                self.sync_publisher_chain_events(chain_id, stream_ids, &nodes, &committee)
            })
            .collect::<Vec<_>>();
        stream::iter(tasks)
            .buffer_unordered(self.options.max_joined_tasks)
            .collect::<Vec<_>>()
            .await
            .into_iter()
            .collect::<Result<Vec<_>, _>>()?;
        Ok(())
    }

    /// Downloads only event-bearing blocks for the given publisher chain and streams.
    ///
    /// Uses `communicate_with_quorum` so that each validator is queried for
    /// `previous_event_blocks` and the claimed blocks are downloaded from that same
    /// validator. Waiting for a quorum guarantees that any confirmed event is discovered,
    /// since at least one honest validator in the quorum must have it.
    async fn sync_publisher_chain_events(
        &self,
        publisher_chain_id: ChainId,
        stream_ids: BTreeSet<StreamId>,
        nodes: &[RemoteNode<Env::ValidatorNode>],
        committee: &Committee,
    ) -> Result<(), Error> {
        let stream_ids_ref = &stream_ids;
        communicate_with_quorum(
            nodes,
            committee,
            |_: &()| (),
            |remote_node| async move {
                self.client
                    .sync_events_from_node(publisher_chain_id, stream_ids_ref, &remote_node)
                    .await
            },
            self.options.quorum_grace_period,
        )
        .await?;
        Ok(())
    }

    /// Attempts to download new received certificates.
    ///
    /// This is a best effort: it will only find certificates that have been confirmed
    /// amongst sufficiently many validators of the current committee of the target
    /// chain.
    ///
    /// However, this should be the case whenever a sender's chain is still in use and
    /// is regularly upgraded to new committees.
    #[instrument(level = "debug", skip(self), fields(chain_id = %self.chain_id))]
    pub async fn find_received_certificates(&self) -> Result<(), Error> {
        debug!("starting find_received_certificates");
        let sync_start = Instant::now();
        #[cfg(with_metrics)]
        let _latency = super::metrics::FIND_RECEIVED_CERTIFICATES_LATENCY.measure_latency();
        // Use network information from the local chain.
        let chain_id = self.chain_id;
        let (_, committee) = self.admin_committee().await?;
        let nodes = self.client.make_nodes(&committee)?;

        let trackers = self
            .client
            .local_node
            .get_received_certificate_trackers(chain_id)
            .await?;

        trace!("find_received_certificates: read trackers");

        let received_log_batches = Arc::new(std::sync::Mutex::new(Vec::new()));
        // Proceed to downloading received logs.
        let result = communicate_with_quorum(
            &nodes,
            &committee,
            |_| (),
            |remote_node| {
                let client = &self.client;
                let tracker = trackers.get(&remote_node.public_key).copied().unwrap_or(0);
                let received_log_batches = Arc::clone(&received_log_batches);
                Box::pin(async move {
                    let batch = client
                        .get_received_log_from_validator(chain_id, &remote_node, tracker)
                        .await?;
                    let mut batches = received_log_batches.lock().unwrap();
                    batches.push((remote_node.public_key, batch));
                    Ok(())
                })
            },
            self.options.quorum_grace_period,
        )
        .await;

        if let Err(error) = result {
            error!(
                %error,
                "Failed to synchronize received_logs from at least a quorum of validators",
            );
        }

        let received_logs: Vec<_> = {
            let mut received_log_batches = received_log_batches.lock().unwrap();
            std::mem::take(received_log_batches.as_mut())
        };

        let received_logs_total = received_logs.iter().map(|x| x.1.len()).sum::<usize>();
        // How far behind each validator's tracker was. These differ by orders of magnitude in
        // practice — one validator can be millions of entries behind while the rest are a few
        // thousand — and it is the per-validator number that says which walk is expensive.
        let entries_per_validator = received_logs
            .iter()
            .map(|(public_key, batch)| (*public_key, batch.len()))
            .collect::<Vec<_>>();
        #[cfg(with_metrics)]
        super::metrics::FIND_RECEIVED_CERTIFICATES_LOG_ENTRIES
            .with_label_values(&[])
            .observe(received_logs_total as f64);

        debug!(
            received_logs_len = %received_logs.len(),
            %received_logs_total,
            "collected received logs"
        );

        let (received_logs, mut validator_trackers) = {
            (
                ReceivedLogs::from_received_result(received_logs.clone()),
                ValidatorTrackers::new(received_logs, &trackers),
            )
        };

        #[cfg(with_metrics)]
        {
            super::metrics::FIND_RECEIVED_CERTIFICATES_SENDER_CHAINS
                .with_label_values(&[])
                .observe(received_logs.num_chains() as f64);
            super::metrics::SENDER_CERTIFICATES_DISCOVERED_TOTAL
                .inc_by(received_logs.num_certs() as u64);
        }

        debug!(
            num_chains = %received_logs.num_chains(),
            num_certs = %received_logs.num_certs(),
            "find_received_certificates: total number of chains and certificates to sync",
        );

        let num_certs = received_logs.num_certs();
        let is_large_sync = num_certs >= LARGE_RECEIVED_CERTIFICATE_SYNC_THRESHOLD;
        if is_large_sync {
            // Resolved here rather than above so the common, quiet path allocates nothing.
            let entries_per_validator = entries_per_validator
                .iter()
                .map(|(public_key, entries)| {
                    let address = nodes
                        .iter()
                        .find(|node| node.public_key == *public_key)
                        .map_or_else(|| public_key.to_string(), |node| node.address());
                    (address, *entries)
                })
                .collect::<Vec<_>>();
            info!(
                %chain_id,
                num_chains = %received_logs.num_chains(),
                %num_certs,
                num_log_entries = %received_logs_total,
                ?entries_per_validator,
                "starting a large sync of received certificates",
            );
        }
        let max_blocks_per_chain =
            self.options.sender_certificate_download_batch_size / self.options.max_joined_tasks * 2;
        let mut num_synced_certs = 0;
        let mut next_progress_message = RECEIVED_CERTIFICATE_SYNC_PROGRESS_INTERVAL;
        for received_log in received_logs.into_batches(
            self.options.sender_certificate_download_batch_size,
            max_blocks_per_chain,
        ) {
            num_synced_certs += received_log.num_certs();
            validator_trackers = self
                .receive_sender_certificates(received_log, validator_trackers, &nodes)
                .await?;

            self.update_received_certificate_trackers(&validator_trackers)
                .await;

            if is_large_sync && num_synced_certs >= next_progress_message {
                info!(
                    %chain_id,
                    %num_synced_certs,
                    %num_certs,
                    "received-certificate sync in progress",
                );
                next_progress_message =
                    num_synced_certs + RECEIVED_CERTIFICATE_SYNC_PROGRESS_INTERVAL;
            }
        }

        if is_large_sync {
            info!(
                %chain_id,
                %num_certs,
                elapsed_secs = %sync_start.elapsed().as_secs(),
                "finished a large sync of received certificates",
            );
        }

        trace!("find_received_certificates finished");

        Ok(())
    }

    async fn update_received_certificate_trackers(&self, trackers: &ValidatorTrackers) {
        let updated_trackers = trackers.to_map();
        trace!(?updated_trackers, "updated tracker values");

        // Update the trackers.
        if let Err(error) = self
            .client
            .local_node
            .update_received_certificate_trackers(self.chain_id, updated_trackers)
            .await
        {
            error!(
                chain_id = %self.chain_id,
                %error,
                "Failed to update the certificate trackers",
            );
        }
    }

    /// Downloads and processes or preprocesses the certificates for blocks sending messages to
    /// this chain that we are still missing.
    async fn receive_sender_certificates(
        &self,
        mut received_logs: ReceivedLogs,
        mut validator_trackers: ValidatorTrackers,
        nodes: &[RemoteNode<Env::ValidatorNode>],
    ) -> Result<ValidatorTrackers, Error> {
        debug!(
            num_chains = %received_logs.num_chains(),
            num_certs = %received_logs.num_certs(),
            "receive_sender_certificates: number of chains and certificates to sync",
        );

        // Obtain the next block height we need in the local node, for each chain.
        let local_next_heights = self
            .client
            .local_node
            .next_outbox_heights(received_logs.chains(), self.chain_id)
            .await?;

        validator_trackers.filter_out_already_known(&mut received_logs, &local_next_heights);

        #[cfg(with_metrics)]
        super::metrics::SENDER_CERTIFICATES_MISSING_TOTAL.inc_by(received_logs.num_certs() as u64);

        debug!(
            remaining_total_certificates = %received_logs.num_certs(),
            "receive_sender_certificates: computed remote_heights"
        );

        let mut other_sender_chains = Vec::new();
        let (sender, mut receiver) = mpsc::unbounded_channel::<ChainAndHeight>();

        let cert_futures = received_logs.heights_per_chain().into_iter().filter_map({
            let received_logs = &received_logs;
            let other_sender_chains = &mut other_sender_chains;

            move |(sender_chain_id, remote_heights)| {
                if remote_heights.is_empty() {
                    // Our highest, locally executed block is higher than any block height
                    // from the current batch. Skip this batch, but remember to wait for
                    // the messages to be delivered to the inboxes.
                    other_sender_chains.push(sender_chain_id);
                    return None;
                };
                let remote_heights = remote_heights.into_iter().collect::<Vec<_>>();
                let sender = sender.clone();
                let client = self.client.clone();
                let nodes = nodes.to_vec();
                Some(async move {
                    client
                        .download_and_process_sender_chain(
                            sender_chain_id,
                            &nodes,
                            received_logs,
                            remote_heights,
                            sender,
                        )
                        .await
                })
            }
        });

        future::join(
            stream::iter(cert_futures)
                .buffer_unordered(self.options.max_joined_tasks)
                .collect::<()>(),
            async {
                while let Some(chain_and_height) = receiver.recv().await {
                    validator_trackers.downloaded_cert(chain_and_height);
                }
            },
        )
        .await;

        debug!(
            num_other_chains = %other_sender_chains.len(),
            "receive_sender_certificates: processing certificates finished"
        );

        // Certificates for these chains were omitted from `certificates` because they were
        // already processed locally. If they were processed in a concurrent task, it is not
        // guaranteed that their cross-chain messages were already handled.
        self.retry_pending_cross_chain_requests_from_sender_chains(nodes, other_sender_chains)
            .await;

        debug!("receive_sender_certificates: finished processing other_sender_chains");

        Ok(validator_trackers)
    }

    /// Retries cross chain requests on the chains which may have been processed on
    /// another task without the messages being correctly handled. Fetches missing blobs from
    /// the given nodes if necessary.
    async fn retry_pending_cross_chain_requests_from_sender_chains(
        &self,
        nodes: &[RemoteNode<Env::ValidatorNode>],
        other_sender_chains: Vec<ChainId>,
    ) {
        let stream = other_sender_chains
            .into_iter()
            .map(|chain_id| async move {
                if let Err(error) = match self
                    .client
                    .retry_pending_cross_chain_requests(chain_id)
                    .await
                {
                    Ok(()) => Ok(()),
                    Err(LocalNodeError::BlobsNotFound(blob_ids)) => {
                        if let Err(error) = self
                            .client
                            .update_local_node_with_blobs_from(blob_ids.clone(), nodes)
                            .await
                        {
                            error!(
                                ?blob_ids,
                                %error,
                                "Error while attempting to download blobs during retrying outgoing \
                                messages"
                            );
                        }
                        self.client
                            .retry_pending_cross_chain_requests(chain_id)
                            .await
                    }
                    err => err,
                } {
                    error!(
                        %chain_id,
                        %error,
                        "Failed to retry outgoing messages from chain"
                    );
                }
            })
            .collect::<FuturesUnordered<_>>();
        stream.for_each(future::ready).await;
    }

    /// Sends money.
    #[instrument(level = "trace")]
    pub async fn transfer(
        &self,
        owner: AccountOwner,
        amount: Amount,
        recipient: Account,
    ) -> Result<ClientOutcome<ConfirmedBlockCertificate>, Error> {
        // TODO(#467): check the balance of `owner` before signing any block proposal.
        self.execute_operation(SystemOperation::Transfer {
            owner,
            recipient,
            amount,
        })
        .await
    }

    /// Verify if a data blob is readable from storage.
    // TODO(#2490): Consider removing or renaming this.
    #[instrument(level = "trace")]
    pub async fn read_data_blob(
        &self,
        hash: CryptoHash,
    ) -> Result<ClientOutcome<ConfirmedBlockCertificate>, Error> {
        let blob_id = BlobId {
            hash,
            blob_type: BlobType::Data,
        };
        self.execute_operation(SystemOperation::VerifyBlob { blob_id })
            .await
    }

    /// Claims money in a remote chain.
    #[instrument(level = "trace")]
    pub async fn claim(
        &self,
        owner: AccountOwner,
        target_id: ChainId,
        recipient: Account,
        amount: Amount,
    ) -> Result<ClientOutcome<ConfirmedBlockCertificate>, Error> {
        self.execute_operation(SystemOperation::Claim {
            owner,
            target_id,
            recipient,
            amount,
        })
        .await
    }

    /// Requests a leader timeout vote from all validators. If a quorum signs it, creates a
    /// certificate and sends it to all validators, to make them enter the next round.
    #[instrument(level = "trace")]
    pub async fn request_leader_timeout(&self) -> Result<TimeoutCertificate, Error> {
        let chain_id = self.chain_id;
        let info = self.chain_info().await?;
        let committee = self.local_committee().await?;
        let height = info.next_block_height;
        let round = info.manager.current_round;
        let action = CommunicateAction::RequestTimeout {
            height,
            round,
            chain_id,
        };
        let value = Timeout::new(chain_id, height, info.epoch);
        let certificate = Box::new(
            self.client
                .communicate_chain_action(&committee, action, value)
                .await?,
        );
        self.client
            .handle_certificate::<Timeout>(*certificate.clone())
            .await?;
        // The block height didn't increase, but this will communicate the timeout as well.
        self.client
            .communicate_chain_updates(
                &committee,
                chain_id,
                height,
                CrossChainMessageDelivery::NonBlocking,
                None,
            )
            .await?;
        Ok(*certificate)
    }

    /// Downloads and processes any certificates we are missing for the given chain.
    #[instrument(level = "trace", skip_all)]
    pub async fn synchronize_chain_state(
        &self,
        chain_id: ChainId,
    ) -> Result<Box<ChainInfo>, Error> {
        self.client.synchronize_chain_state(chain_id).await
    }

    /// Downloads and processes any certificates we are missing for this chain, from the given
    /// committee.
    #[instrument(level = "trace", skip_all)]
    pub async fn synchronize_chain_state_from_committee(
        &self,
        committee: Arc<Committee>,
    ) -> Result<Box<ChainInfo>, Error> {
        self.client
            .synchronize_chain_from_committee(self.chain_id, committee)
            .await
    }

    /// Executes a list of operations.
    #[instrument(level = "trace", skip(operations, blobs))]
    pub async fn execute_operations(
        &self,
        operations: Vec<Operation>,
        blobs: Vec<Blob>,
    ) -> Result<ClientOutcome<ConfirmedBlockCertificate>, Error> {
        let timing_start = linera_base::time::Instant::now();

        let result = loop {
            let execute_block_start = linera_base::time::Instant::now();
            // TODO(#2066): Remove boxing once the call-stack is shallower
            match Box::pin(self.execute_block(operations.clone(), blobs.clone())).await {
                Ok(ClientOutcome::Committed(certificate)) => {
                    self.send_timing(execute_block_start, TimingType::ExecuteBlock);
                    break Ok(ClientOutcome::Committed(certificate));
                }
                Ok(ClientOutcome::WaitForTimeout(timeout)) => {
                    break Ok(ClientOutcome::WaitForTimeout(timeout));
                }
                Ok(ClientOutcome::Conflict(certificate)) => {
                    info!(
                        height = %certificate.block().header.height,
                        "Another block was committed."
                    );
                    break Ok(ClientOutcome::Conflict(certificate));
                }
                Err(Error::CommunicationError(CommunicationError::Trusted(
                    NodeError::UnexpectedBlockHeight {
                        expected_block_height,
                        found_block_height,
                    },
                ))) if expected_block_height > found_block_height => {
                    tracing::info!(
                        chain_id = %self.chain_id,
                        "Local state is outdated; synchronizing chain"
                    );
                    self.synchronize_chain_state(self.chain_id).await?;
                }
                Err(err) => return Err(err),
            };
        };

        self.send_timing(timing_start, TimingType::ExecuteOperations);

        result
    }

    /// Executes an operation.
    pub async fn execute_operation(
        &self,
        operation: impl Into<Operation>,
    ) -> Result<ClientOutcome<ConfirmedBlockCertificate>, Error> {
        self.execute_operations(vec![operation.into()], vec![])
            .await
    }

    /// Executes a new block.
    ///
    /// This must be preceded by a call to `prepare_chain()`.
    #[instrument(level = "trace", skip(operations, blobs))]
    async fn execute_block(
        &self,
        operations: Vec<Operation>,
        blobs: Vec<Blob>,
    ) -> Result<ClientOutcome<ConfirmedBlockCertificate>, Error> {
        #[cfg(with_metrics)]
        let _latency = super::metrics::EXECUTE_BLOCK_LATENCY.measure_latency();

        let result = self.try_execute_block(operations, blobs).await;
        if let Err(error) = &result {
            let error_type = error.error_type();
            #[cfg(with_metrics)]
            super::metrics::BLOCK_STAGING_FAILURES_TOTAL
                .with_label_values(&[error_type.as_str()])
                .inc();
            info!(chain_id = %self.chain_id, %error_type, "Block staging failed");
        }
        result
    }

    async fn try_execute_block(
        &self,
        operations: Vec<Operation>,
        blobs: Vec<Blob>,
    ) -> Result<ClientOutcome<ConfirmedBlockCertificate>, Error> {
        let mutex = self.proposal_mutex();
        let lock_start = linera_base::time::Instant::now();
        let mut proposal_guard = mutex.lock_owned().await;
        tracing::debug!(
            chain_id = %self.chain_id,
            lock_wait_ms = lock_start.elapsed().as_millis(),
            "acquired proposal_mutex in execute_block"
        );
        // TODO(#5092): We shouldn't need to call this explicitly.
        // Process any leftover pending proposal from a previous interrupted call.
        // Even if there is no pending proposal, this still calls
        // `request_leader_timeout_if_needed` which ensures the local chain state
        // is synchronized with the current consensus round.
        match self
            .process_pending_block_without_prepare(&mut proposal_guard)
            .await?
        {
            ClientOutcome::Committed(Some(certificate)) => {
                return Ok(self.classify_committed(certificate, &operations));
            }
            ClientOutcome::WaitForTimeout(timeout) => {
                return Ok(ClientOutcome::WaitForTimeout(timeout))
            }
            ClientOutcome::Conflict(certificate) => {
                return Ok(ClientOutcome::Conflict(certificate))
            }
            ClientOutcome::Committed(None) => {}
        }

        loop {
            // Collect pending messages and epoch changes after acquiring the lock to avoid
            // race conditions where messages valid for one block height are proposed at a
            // different height. This is recomputed on every retry because the set of
            // receivable messages and epoch changes can differ at a new height.
            let transactions = self
                .prepend_epochs_messages_and_events(operations.clone())
                .await?;

            if transactions.is_empty() {
                return Err(Error::LocalNodeError(LocalNodeError::WorkerError(
                    WorkerError::ChainError(Box::new(ChainError::EmptyBlock)),
                )));
            }

            self.new_pending_block(transactions, blobs.clone(), &mut proposal_guard)
                .await?;

            match self
                .process_pending_block_without_prepare(&mut proposal_guard)
                .await?
            {
                ClientOutcome::Committed(Some(certificate)) => {
                    return Ok(self.classify_committed(certificate, &operations));
                }
                // `process_pending_block_without_prepare` cleared the pending proposal
                // without committing ours, so our operations were not applied. This happens
                // in two ways:
                //   * the chain advanced past the height we staged at while we were
                //     proposing (e.g. a notification or background sync committed another
                //     block at our height) — the common #5664 case; or
                //   * the staged proposal's authenticated signer's key is no longer held
                //     (the preferred owner changed since we staged), so the stale proposal
                //     was discarded.
                // Either way, re-stage and retry: `new_pending_block` recomputes the block
                // at the current height and signs as the current identity, so the first
                // case advances to the new height (genuine external progress) and the
                // second adopts a signer we hold, so the retry converges. See #5664.
                ClientOutcome::Committed(None) => {
                    tracing::debug!(
                        chain_id = %self.chain_id,
                        "pending proposal cleared without committing ours; re-staging and retrying"
                    );
                    continue;
                }
                ClientOutcome::WaitForTimeout(timeout) => {
                    return Ok(ClientOutcome::WaitForTimeout(timeout));
                }
                ClientOutcome::Conflict(certificate) => {
                    return Ok(ClientOutcome::Conflict(certificate));
                }
            }
        }
    }

    /// Returns `Committed` if the committed block reflects our request — same
    /// authenticated owner, and our `operations`. All other transactions must be ones that
    /// are added automatically, to process messages, streams or new epochs.
    fn classify_committed(
        &self,
        certificate: ConfirmedBlockCertificate,
        operations: &[Operation],
    ) -> ClientOutcome<ConfirmedBlockCertificate> {
        let block = certificate.block();
        if self.preferred_owner.is_none()
            || block.header.authenticated_owner != self.preferred_owner
        {
            return ClientOutcome::Conflict(Box::new(certificate));
        }
        let mut operations_iter = operations.iter().peekable();
        for tx in &block.body.transactions {
            let is_expected = match tx {
                Transaction::ReceiveMessages(_) => true,
                Transaction::ExecuteOperation(op) if Some(&op) == operations_iter.peek() => {
                    operations_iter.next();
                    true
                }
                Transaction::ExecuteOperation(Operation::System(op)) => matches!(
                    **op,
                    SystemOperation::ProcessNewEpoch(_) | SystemOperation::UpdateStream { .. }
                ),
                Transaction::ExecuteOperation(Operation::User { .. }) => false,
            };
            if !is_expected {
                return ClientOutcome::Conflict(Box::new(certificate));
            }
        }
        if operations_iter.next().is_some() {
            ClientOutcome::Conflict(Box::new(certificate))
        } else {
            ClientOutcome::Committed(certificate)
        }
    }

    /// Creates a vector of transactions which, in addition to the provided operations,
    /// also contains the next pending epoch change, receiving message bundles and event
    /// stream updates (if there are any to be processed).
    /// This should be called when executing a block, in order to make sure that any pending
    /// messages or events are included in it.
    #[instrument(level = "trace", skip(operations))]
    async fn prepend_epochs_messages_and_events(
        &self,
        operations: Vec<Operation>,
    ) -> Result<Vec<Transaction>, Error> {
        let incoming_bundles = self.pending_message_bundles().await?;
        let stream_updates = self.collect_stream_updates().await?;
        Ok(self
            .next_epoch_change()
            .await?
            .into_iter()
            .map(Transaction::ExecuteOperation)
            .chain(
                incoming_bundles
                    .into_iter()
                    .map(Transaction::ReceiveMessages),
            )
            .chain(
                stream_updates
                    .into_iter()
                    .map(Transaction::ExecuteOperation),
            )
            .chain(operations.into_iter().map(Transaction::ExecuteOperation))
            .collect::<Vec<_>>())
    }

    /// Creates a new pending block and stores it in `proposal_guard`.
    ///
    /// The caller must hold the proposal mutex. The pending proposal is written directly
    /// into the guard so that it is always synchronized with the mutex.
    #[instrument(level = "trace", skip(transactions, blobs, proposal_guard))]
    async fn new_pending_block(
        &self,
        transactions: Vec<Transaction>,
        blobs: Vec<Blob>,
        proposal_guard: &mut Option<PendingProposal>,
    ) -> Result<Block, Error> {
        let identity = self.identity().await?;

        ensure!(
            proposal_guard.is_none(),
            Error::BlockProposalError(
                "Client state already has a pending block; \
                use the `linera retry-pending-block` command to commit that first"
            )
        );
        let info = self.chain_info().await?;
        let timestamp = self.next_timestamp(&transactions, info.timestamp);
        let proposed_block = ProposedBlock {
            epoch: info.epoch,
            chain_id: self.chain_id,
            transactions,
            previous_block_hash: info.block_hash,
            height: info.next_block_height,
            authenticated_owner: Some(identity),
            timestamp,
        };

        let round = self.round_for_oracle(&info, &identity).await?;
        // Make sure every incoming message succeeds and otherwise remove them.
        // Also, compute the final certified hash while we're at it.
        let (block, _, never_reject_origins) = self
            .client
            .stage_block_execution(
                proposed_block,
                round,
                blobs.clone(),
                self.options.bundle_execution_policy(),
            )
            .await?;
        // Record origins whose bundles were discarded due to the never-reject policy so
        // that `process_inbox` stops retrying them until the client is restarted.
        if !never_reject_origins.is_empty() {
            let skipped = self.skipped_origins.pin();
            for origin in never_reject_origins {
                skipped.insert(origin);
            }
        }
        let (proposed_block, auto_retry_outcome) = block.clone().into_proposal();
        *proposal_guard = Some(PendingProposal {
            block: proposed_block,
            blobs,
            auto_retry_outcome: Some(auto_retry_outcome),
            round: None,
        });
        Ok(block)
    }

    /// Returns a suitable timestamp for the next block.
    ///
    /// This will usually be the current time according to the local clock, but may be slightly
    /// ahead to make sure it's not earlier than the incoming messages or the previous block.
    #[instrument(level = "trace", skip(transactions))]
    fn next_timestamp(&self, transactions: &[Transaction], block_time: Timestamp) -> Timestamp {
        let local_time = self.storage_client().clock().current_time();
        transactions
            .iter()
            .filter_map(Transaction::incoming_bundle)
            .map(|msg| msg.bundle.timestamp)
            .max()
            .map_or(local_time, |timestamp| timestamp.max(local_time))
            .max(block_time)
    }

    /// Queries an application.
    #[instrument(level = "trace", skip(query))]
    pub async fn query_application(
        &self,
        query: Query,
        block_hash: Option<CryptoHash>,
    ) -> Result<(QueryOutcome, BlockHeight), Error> {
        let mut downloaded_blobs = HashSet::<BlobId>::new();
        let mut events = super::EventSetDownloader::new(&self.client);
        loop {
            let result = self
                .client
                .local_node
                .query_application(self.chain_id, query.clone(), block_hash)
                .await;
            if let Err(LocalNodeError::BlobsNotFound(blob_ids)) = &result {
                let new_blobs = super::filter_new(blob_ids, &downloaded_blobs);
                if !new_blobs.is_empty() {
                    let validators = self.client.validator_nodes().await?;
                    self.client
                        .update_local_node_with_blobs_from(new_blobs.clone(), &validators)
                        .await?;
                    downloaded_blobs.extend(new_blobs);
                    continue;
                }
            }
            if let Err(LocalNodeError::EventsNotFound(event_ids)) = &result {
                if events.download_new(event_ids).await? {
                    continue;
                }
            }
            return Ok(result?);
        }
    }

    /// Queries a system application.
    #[cfg(with_testing)]
    #[instrument(level = "trace", skip(query))]
    pub async fn query_system_application(
        &self,
        query: linera_execution::SystemQuery,
    ) -> Result<QueryOutcome<linera_execution::SystemResponse>, Error> {
        let (
            QueryOutcome {
                response,
                operations,
            },
            _,
        ) = self.query_application(Query::System(query), None).await?;
        match response {
            linera_execution::QueryResponse::System(response) => Ok(QueryOutcome {
                response,
                operations,
            }),
            _ => Err(Error::InternalError("Unexpected response for system query")),
        }
    }

    /// Queries a user application.
    #[instrument(level = "trace", skip(application_id, query))]
    #[cfg(with_testing)]
    pub async fn query_user_application<A: Abi>(
        &self,
        application_id: ApplicationId<A>,
        query: &A::Query,
    ) -> Result<QueryOutcome<A::QueryResponse>, Error> {
        let query = Query::user(application_id, query)?;
        let (
            QueryOutcome {
                response,
                operations,
            },
            _,
        ) = self.query_application(query, None).await?;
        match response {
            linera_execution::QueryResponse::User(response_bytes) => {
                let response = serde_json::from_slice(&response_bytes)?;
                Ok(QueryOutcome {
                    response,
                    operations,
                })
            }
            _ => Err(Error::InternalError("Unexpected response for user query")),
        }
    }

    /// Obtains the local balance of the chain account after staging the execution of
    /// incoming messages in a new block.
    ///
    /// Does not attempt to synchronize with validators. The result will reflect up to
    /// `max_pending_message_bundles` incoming message bundles and the execution fees for a single
    /// block.
    #[instrument(level = "trace")]
    pub async fn query_balance(&self) -> Result<Amount, Error> {
        let (balance, _) = self.query_balances_with_owner(AccountOwner::CHAIN).await?;
        Ok(balance)
    }

    /// Obtains the local balance of an account after staging the execution of incoming messages in
    /// a new block.
    ///
    /// Does not attempt to synchronize with validators. The result will reflect up to
    /// `max_pending_message_bundles` incoming message bundles and the execution fees for a single
    /// block.
    #[instrument(level = "trace", skip(owner))]
    pub async fn query_owner_balance(&self, owner: AccountOwner) -> Result<Amount, Error> {
        if owner.is_chain() {
            self.query_balance().await
        } else {
            Ok(self
                .query_balances_with_owner(owner)
                .await?
                .1
                .unwrap_or(Amount::ZERO))
        }
    }

    /// Obtains the local balance of an account and optionally another user after staging the
    /// execution of incoming messages in a new block.
    ///
    /// Does not attempt to synchronize with validators. The result will reflect up to
    /// `max_pending_message_bundles` incoming message bundles and the execution fees for a single
    /// block.
    #[instrument(level = "trace", skip(owner))]
    pub(crate) async fn query_balances_with_owner(
        &self,
        owner: AccountOwner,
    ) -> Result<(Amount, Option<Amount>), Error> {
        let incoming_bundles = self.pending_message_bundles().await?;
        // Since we disallow empty blocks, and there is no incoming messages,
        // that could change it, we query for the balance immediately.
        if incoming_bundles.is_empty() {
            let chain_balance = self.local_balance().await?;
            let owner_balance = self.local_owner_balance(owner).await?;
            return Ok((chain_balance, Some(owner_balance)));
        }
        let info = self.chain_info().await?;
        let transactions = incoming_bundles
            .into_iter()
            .map(Transaction::ReceiveMessages)
            .collect::<Vec<_>>();
        let timestamp = self.next_timestamp(&transactions, info.timestamp);
        let block = ProposedBlock {
            epoch: info.epoch,
            chain_id: self.chain_id,
            transactions,
            previous_block_hash: info.block_hash,
            height: info.next_block_height,
            authenticated_owner: if owner == AccountOwner::CHAIN {
                None
            } else {
                Some(owner)
            },
            timestamp,
        };
        match self
            .client
            .stage_block_execution(
                block,
                None,
                Vec::new(),
                self.options.bundle_execution_policy(),
            )
            .await
        {
            Ok((_, response, _)) => Ok((
                response.info.chain_balance,
                response.info.requested_owner_balance,
            )),
            Err(Error::LocalNodeError(LocalNodeError::WorkerError(WorkerError::ChainError(
                error,
            )))) if matches!(
                &*error,
                ChainError::ExecutionError(
                    execution_error,
                    ChainExecutionContext::Block
                ) if matches!(
                    **execution_error,
                    ExecutionError::FeesExceedFunding { .. }
                )
            ) =>
            {
                // We can't even pay for the execution of one empty block. Let's return zero.
                Ok((Amount::ZERO, Some(Amount::ZERO)))
            }
            Err(error) => Err(error),
        }
    }

    /// Reads the local balance of the chain account.
    ///
    /// Does not process the inbox or attempt to synchronize with validators.
    #[instrument(level = "trace")]
    pub async fn local_balance(&self) -> Result<Amount, Error> {
        let (balance, _) = self.local_balances_with_owner(AccountOwner::CHAIN).await?;
        Ok(balance)
    }

    /// Reads the local balance of a user account.
    ///
    /// Does not process the inbox or attempt to synchronize with validators.
    #[instrument(level = "trace", skip(owner))]
    pub async fn local_owner_balance(&self, owner: AccountOwner) -> Result<Amount, Error> {
        if owner.is_chain() {
            self.local_balance().await
        } else {
            Ok(self
                .local_balances_with_owner(owner)
                .await?
                .1
                .unwrap_or(Amount::ZERO))
        }
    }

    /// Reads the local balance of the chain account and optionally another user.
    ///
    /// Does not process the inbox or attempt to synchronize with validators.
    #[instrument(level = "trace", skip(owner))]
    pub(crate) async fn local_balances_with_owner(
        &self,
        owner: AccountOwner,
    ) -> Result<(Amount, Option<Amount>), Error> {
        ensure!(
            self.chain_info().await?.next_block_height >= self.initial_next_block_height,
            Error::WalletSynchronizationError
        );
        let mut query = ChainInfoQuery::new(self.chain_id);
        query.request_owner_balance = owner;
        let response = self
            .client
            .local_node
            .handle_chain_info_query(query)
            .await?;
        Ok((
            response.info.chain_balance,
            response.info.requested_owner_balance,
        ))
    }

    /// Sends tokens to a chain.
    #[instrument(level = "trace")]
    pub async fn transfer_to_account(
        &self,
        from: AccountOwner,
        amount: Amount,
        account: Account,
    ) -> Result<ClientOutcome<ConfirmedBlockCertificate>, Error> {
        self.transfer(from, amount, account).await
    }

    /// Burns tokens (transfer to a special address).
    #[cfg(with_testing)]
    #[instrument(level = "trace")]
    pub async fn burn(
        &self,
        owner: AccountOwner,
        amount: Amount,
    ) -> Result<ClientOutcome<ConfirmedBlockCertificate>, Error> {
        let recipient = Account::burn_address(self.chain_id);
        self.transfer(owner, amount, recipient).await
    }

    /// Fetches the latest chain info for this chain from the validators.
    #[instrument(level = "trace")]
    pub async fn fetch_chain_info(&self) -> Result<Box<ChainInfo>, Error> {
        let validators = self.client.validator_nodes().await?;
        self.client
            .fetch_chain_info(self.chain_id, &validators)
            .await
    }

    /// Attempts to synchronize chains that have sent us messages and populate our local
    /// inbox.
    ///
    /// To create a block that actually executes the messages in the inbox,
    /// `process_inbox` must be called separately.
    ///
    /// If the chain is in follow-only mode, this only downloads blocks for this chain without
    /// fetching manager values or sender/publisher chains.
    /// Synchronizes the chain state from validators, optionally stopping at a given
    /// block height or block timestamp.
    ///
    /// - If `next_height` is `Some`, downloads blocks up to (but not including) that height.
    /// - If `until_block_time` is `Some`, downloads blocks up to (but not including) the
    ///   first block with timestamp >= the given value.
    #[instrument(level = "trace")]
    pub async fn synchronize_up_to(
        &self,
        next_height: Option<BlockHeight>,
        until_block_time: Option<Timestamp>,
    ) -> Result<Box<ChainInfo>, Error> {
        let (_, committee) = self.client.admin_committee().await?;
        let validators = self.client.make_nodes(&committee)?;
        Box::pin(self.client.fetch_chain_info(self.chain_id, &validators)).await?;
        communicate_with_quorum(
            &validators,
            &committee,
            |_: &()| (),
            |remote_node| async move {
                self.client
                    .download_certificates_from(
                        &remote_node,
                        self.chain_id,
                        next_height.unwrap_or(BlockHeight::MAX),
                        until_block_time,
                    )
                    .await?;
                Ok(())
            },
            self.client.options.quorum_grace_period,
        )
        .await?;
        self.client
            .local_node
            .chain_info(self.chain_id)
            .await
            .map_err(Into::into)
    }

    /// Synchronizes this chain's state from the validators, including received messages.
    pub async fn synchronize_from_validators(&self) -> Result<Box<ChainInfo>, Error> {
        if self.is_follow_only() {
            return self.client.synchronize_chain_state(self.chain_id).await;
        }
        let info = self.prepare_chain().await?;
        self.synchronize_publisher_chains().await?;
        self.find_received_certificates().await?;
        Ok(info)
    }

    /// Processes the last pending block.
    #[instrument(level = "trace")]
    pub async fn process_pending_block(
        &self,
    ) -> Result<ClientOutcome<Option<ConfirmedBlockCertificate>>, Error> {
        self.prepare_chain().await?;
        let mutex = self.proposal_mutex();
        let mut proposal_guard = mutex.lock_owned().await;
        self.process_pending_block_without_prepare(&mut proposal_guard)
            .await
    }

    /// Processes the last pending block. Assumes that the local chain is up to date.
    ///
    /// The caller must hold the proposal mutex via `proposal_guard`. The pending proposal
    /// is read from and cleared through the guard, ensuring synchronization.
    ///
    /// On error, compares the chain manager's snapshot just before the failing network
    /// call against the current state. If a per-validator updater absorbed new info
    /// (e.g. a locking block we didn't have) while we were calling, retries with the
    /// refreshed state. The snapshot is updated after our own local proposal handling
    /// so that doesn't count as "new info from a validator".
    ///
    /// Retrying terminates without an explicit bound: each retry corresponds to local
    /// consensus state that actually advanced (verified by the local node, which checks
    /// signatures and quorums), and that state is monotone.
    #[instrument(level = "debug", skip(self, proposal_guard), fields(chain_id = %self.chain_id))]
    async fn process_pending_block_without_prepare(
        &self,
        proposal_guard: &mut Option<PendingProposal>,
    ) -> Result<ClientOutcome<Option<ConfirmedBlockCertificate>>, Error> {
        // The fallback sync below is done at most once, so a genuinely stuck proposal can't loop.
        let mut did_fallback_sync = false;
        loop {
            // `process_pending_block_inner` records the baseline snapshot itself, after
            // the initial timeout request, so that absorbing a timeout certificate isn't
            // counted as new info. It stays `None` only if the call fails before that
            // point, in which case there is no baseline to retry against.
            let mut snapshot = None;
            let err = match self
                .process_pending_block_inner(proposal_guard, &mut snapshot)
                .await
            {
                Ok(outcome) => return Ok(outcome),
                Err(err) => err,
            };
            let Some(snapshot) = snapshot else {
                return Err(err);
            };
            let Ok(current) = self.consensus_state_snapshot().await else {
                return Err(err);
            };
            if current == snapshot {
                // The post-quorum pull on rejection (`Client::process_lag_reports`) only covers
                // validators whose rejection was actually received: `communicate_with_quorum`
                // breaks as soon as a quorum is impossible and drops still-in-flight requests, so
                // a locking block held only by the slower-to-respond validators is never even
                // reported. Fall back once to an explicit quorum sync — guaranteed to reach a
                // lock-holder — and retry if it absorbed anything; otherwise the rejection was
                // genuine, so propagate it.
                //
                // TODO(#6453): this fallback path has no deterministic regression test yet —
                // forcing a lock-holder's response to be dropped needs a clock-driven
                // per-validator response delay in the test harness (building on #6448); until
                // then it is only covered indirectly by the (now non-flaky)
                // `test_lazy_pull_absorbs_locking_block_on_proposal_rejection`.
                if did_fallback_sync {
                    return Err(err);
                }
                did_fallback_sync = true;
                if let Err(error) = self.synchronize_chain_state(self.chain_id).await {
                    tracing::error!(%error, "fallback sync failed after rejected proposal");
                    return Err(err);
                }
                let Ok(after_sync) = self.consensus_state_snapshot().await else {
                    return Err(err);
                };
                if after_sync == snapshot {
                    return Err(err);
                }
                tracing::debug!(
                    chain_id = %self.chain_id,
                    ?snapshot,
                    ?after_sync,
                    %err,
                    "fallback sync absorbed new consensus state after rejected proposal; retrying"
                );
                continue;
            }
            tracing::debug!(
                chain_id = %self.chain_id,
                ?snapshot,
                ?current,
                %err,
                "local consensus state advanced during process_pending_block; retrying"
            );
        }
    }

    async fn process_pending_block_inner(
        &self,
        proposal_guard: &mut Option<PendingProposal>,
        snapshot: &mut Option<ConsensusStateSnapshot>,
    ) -> Result<ClientOutcome<Option<ConfirmedBlockCertificate>>, Error> {
        let process_start = linera_base::time::Instant::now();
        tracing::debug!("process_pending_block_without_prepare started");
        let info = self.request_leader_timeout_if_needed().await?;
        // After the timeout request (which may itself absorb a timeout cert from
        // validators), bake the resulting state into the snapshot. The rest of this
        // iteration will propose in the round the timeout produced, so a state diff
        // from just that absorption isn't a retry signal.
        *snapshot = Some(self.consensus_state_snapshot().await?);

        // Clear stale pending proposals whose height has already been committed.
        if let Some(pending) = &*proposal_guard {
            if pending.block.height < info.next_block_height {
                tracing::debug!(
                    "Clearing pending proposal: a block was committed at height {}",
                    pending.block.height
                );
                *proposal_guard = None;
            }
        }

        // If there is a validated block in the current round, finalize it.
        if info.manager.has_locking_block_in_current_round()
            && !info.manager.current_round.is_fast()
        {
            return self.finalize_locking_block(info).await;
        }
        let identity = self.identity().await?;

        let local_node = &self.client.local_node;
        // Otherwise we have to re-propose the highest validated block, if there is one.
        let (block, blobs, owner) = if let Some(locking) = &info.manager.requested_locking {
            let (block, blobs) = match &**locking {
                LockingBlock::Regular(certificate) => {
                    let blob_ids = certificate.block().required_blob_ids();
                    let blobs = local_node
                        .get_locking_blobs(&blob_ids, self.chain_id)
                        .await?
                        .ok_or_else(|| Error::InternalError("Missing local locking blobs"))?;
                    debug!("Retrying locking block from round {}", certificate.round);
                    (certificate.block().clone(), blobs)
                }
                LockingBlock::Fast(proposal) => {
                    let proposed_block = proposal.content.block.clone();
                    let blob_ids = proposed_block.published_blob_ids();
                    let blobs = local_node
                        .get_locking_blobs(&blob_ids, self.chain_id)
                        .await?
                        .ok_or_else(|| Error::InternalError("Missing local locking blobs"))?;
                    let (block, _, _) = self
                        .client
                        .stage_block_execution(
                            proposed_block,
                            None,
                            blobs.clone(),
                            BundleExecutionPolicy::committed(),
                        )
                        .await?;
                    debug!("Retrying locking block from fast round.");
                    (block, blobs)
                }
            };
            (block, blobs, identity)
        } else if let Some(pending) = proposal_guard.as_ref() {
            // Otherwise we are free to propose our own pending block. Sign it as the owner
            // that staged it: the block's operations are authenticated by that owner, and
            // validators reject the proposal with `WorkerError::InvalidSigner` if we re-sign
            // it as someone else (which would happen if `preferred_owner` changed since the
            // block was staged). The signer is global to the client, so we can sign as the
            // original author as long as we still hold their key.
            let owner = match pending.block.authenticated_owner {
                Some(staged_owner) if staged_owner != identity => {
                    if !self.has_key_for(&staged_owner).await? {
                        // If a fast-round proposal was already submitted, we can't safely
                        // drop it and propose a conflicting fast block: fast rounds skip
                        // the validation step and rely on the super owner not forking
                        // itself, so a second proposal could split votes between f+1 and
                        // 2f and wedge the round until it times out. Surface an error so
                        // the caller can recover the key or wait for the timeout.
                        if pending.round.is_some_and(|round| round.is_fast()) {
                            return Err(Error::BlockProposalError(
                                "pending fast block was signed by an owner whose key is no \
                                 longer available; recover the key or wait for the round to \
                                 time out before retrying",
                            ));
                        }
                        warn!(
                            ?staged_owner, %identity,
                            "Discarding pending block: no signer key for its authenticated owner",
                        );
                        *proposal_guard = None;
                        return Ok(ClientOutcome::Committed(None));
                    }
                    staged_owner
                }
                _ => identity,
            };
            let proposed_block = pending.block.clone();
            let blobs = pending.blobs.clone();
            let staging_outcome = pending.auto_retry_outcome.as_ref();
            let round = self.round_for_oracle(&info, &owner).await?;
            let (block, _, _) = self
                .client
                .stage_block_execution(
                    proposed_block,
                    round,
                    blobs.clone(),
                    BundleExecutionPolicy::committed(),
                )
                .await?;
            // Sanity check: the committed execution should produce the same outcome
            // as the initial AutoRetry execution. A mismatch indicates a divergence
            // between the two execution paths.
            if let Some(staging_outcome) = staging_outcome {
                ensure!(
                    block.outcome_matches(staging_outcome),
                    Error::ExecutionOutcomeMismatch
                );
            }
            debug!("Proposing the local pending block.");
            (block, blobs, owner)
        } else {
            return Ok(ClientOutcome::Committed(None)); // Nothing to do.
        };

        let has_oracle_responses = block.has_oracle_responses();
        let (proposed_block, outcome) = block.into_proposal();
        let round = match self
            .round_for_new_proposal(&info, &owner, has_oracle_responses)
            .await?
        {
            Either::Left(round) => round,
            Either::Right(timeout) => return Ok(ClientOutcome::WaitForTimeout(timeout)),
        };
        debug!("Proposing block for round {}", round);
        if let Some(pending) = proposal_guard.as_mut() {
            pending.round.get_or_insert(round);
        }

        let already_handled_locally = info
            .manager
            .already_handled_proposal(round, &proposed_block);
        // Create the final block proposal.
        let proposal = if let Some(locking) = info.manager.requested_locking {
            Box::new(match *locking {
                LockingBlock::Regular(cert) => {
                    BlockProposal::new_retry_regular(owner, round, cert, self.signer())
                        .await
                        .map_err(Error::signer_failure)?
                }
                LockingBlock::Fast(proposal) => {
                    BlockProposal::new_retry_fast(owner, round, proposal, self.signer())
                        .await
                        .map_err(Error::signer_failure)?
                }
            })
        } else {
            Box::new(
                BlockProposal::new_initial(owner, round, proposed_block.clone(), self.signer())
                    .await
                    .map_err(Error::signer_failure)?,
            )
        };
        if !already_handled_locally {
            // Check the final block proposal. This will be cheaper after #1401.
            if let Err(err) = local_node.handle_block_proposal(*proposal.clone()).await {
                match err {
                    LocalNodeError::BlobsNotFound(_) => {
                        local_node
                            .handle_pending_blobs(self.chain_id, blobs)
                            .await?;
                        local_node.handle_block_proposal(*proposal.clone()).await?;
                    }
                    err => return Err(err.into()),
                }
            }
        }
        // The local handle of our own proposal can set a Fast lock, advance
        // `proposed`/`signed_proposal`, and thereby bump `current_round`. None of that
        // is "new info from a validator", so fold it into the snapshot before the
        // network calls below.
        *snapshot = Some(self.consensus_state_snapshot().await?);
        let committee = self.local_committee().await?;
        let block = Block::new(proposed_block, outcome);
        // Send the query to validators.
        let submit_block_proposal_start = linera_base::time::Instant::now();
        let certificate = if round.is_fast() {
            let hashed_value = ConfirmedBlock::new(block);
            self.client
                .submit_block_proposal(committee.clone(), proposal, hashed_value)
                .await?
        } else {
            let hashed_value = ValidatedBlock::new(block);
            let certificate = self
                .client
                .submit_block_proposal(committee.clone(), proposal, hashed_value.clone())
                .await?;
            self.client.finalize_block(&committee, certificate).await?
        };
        self.send_timing(submit_block_proposal_start, TimingType::SubmitBlockProposal);
        tracing::debug!(
            total_process_ms = process_start.elapsed().as_millis(),
            "process_pending_block_without_prepare completing"
        );
        debug!(round = %certificate.round, "Sending confirmed block to validators");
        let certificate = self.client.storage_client().cache_certificate(certificate);
        self.update_validators(Some(&committee), Some(certificate.clone()))
            .await?;
        // Clear the pending proposal now that the block has been committed.
        *proposal_guard = None;
        Ok(ClientOutcome::Committed(Some(CacheArc::unwrap_or_clone(
            certificate,
        ))))
    }

    #[expect(
        clippy::cast_possible_truncation,
        reason = "elapsed millis fits in u64 for any realistic measurement window"
    )]
    fn send_timing(&self, start: Instant, timing_type: TimingType) {
        let Some(sender) = &self.timing_sender else {
            return;
        };
        if let Err(err) = sender.send((start.elapsed().as_millis() as u64, timing_type)) {
            tracing::warn!(%err, "Failed to send timing info");
        }
    }

    /// Requests a leader timeout certificate if the current round has timed out. Returns the
    /// chain info for the (possibly new) current round.
    async fn request_leader_timeout_if_needed(&self) -> Result<Box<ChainInfo>, Error> {
        let mut info = self.chain_info_with_manager_values().await?;
        // If the current round has timed out, we request a timeout certificate and retry in
        // the next round.
        if let Some(round_timeout) = info.manager.round_timeout {
            if round_timeout <= self.storage_client().clock().current_time() {
                if let Err(e) = self.request_leader_timeout().await {
                    debug!("Failed to obtain a timeout certificate: {}", e);
                } else {
                    info = self.chain_info_with_manager_values().await?;
                }
            }
        }
        Ok(info)
    }

    /// Finalizes the locking block.
    ///
    /// Panics if there is no locking block; fails if the locking block is not in the current round.
    async fn finalize_locking_block(
        &self,
        info: Box<ChainInfo>,
    ) -> Result<ClientOutcome<Option<ConfirmedBlockCertificate>>, Error> {
        let locking = info
            .manager
            .requested_locking
            .expect("Should have a locking block");
        let LockingBlock::Regular(certificate) = *locking else {
            panic!("Should have a locking validated block");
        };
        debug!(
            round = %certificate.round,
            "Finalizing locking block"
        );
        let committee = self.local_committee().await?;
        let certificate = self
            .client
            .finalize_block(&committee, certificate.clone())
            .await?;
        let certificate = self.client.storage_client().cache_certificate(certificate);
        self.update_validators(Some(&committee), Some(certificate.clone()))
            .await?;
        Ok(ClientOutcome::Committed(Some(CacheArc::unwrap_or_clone(
            certificate,
        ))))
    }

    /// Returns the number for the round number oracle to use when staging a block proposal.
    async fn round_for_oracle(
        &self,
        info: &ChainInfo,
        identity: &AccountOwner,
    ) -> Result<Option<u32>, Error> {
        // Pretend we do use oracles: If we don't, the round number is never read anyway.
        match self.round_for_new_proposal(info, identity, true).await {
            // If it is a multi-leader round, use its number for the oracle.
            Ok(Either::Left(round)) => Ok(round.multi_leader()),
            // If there is no suitable round with oracles, use None: If it works without oracles,
            // the block won't read the value. If it returns a timeout, it will be a single-leader
            // round, in which the oracle returns None.
            Err(Error::BlockProposalError(_)) | Ok(Either::Right(_)) => Ok(None),
            Err(err) => Err(err),
        }
    }

    /// Returns a round in which we can propose a new block or the given one, if possible.
    async fn round_for_new_proposal(
        &self,
        info: &ChainInfo,
        identity: &AccountOwner,
        has_oracle_responses: bool,
    ) -> Result<Either<Round, RoundTimeout>, Error> {
        let manager = &info.manager;
        let seed = manager.seed;
        // If there is a conflicting proposal in the current round, we can only propose if the
        // next round can be started without a timeout, i.e. if we are in a multi-leader round.
        // Similarly, we cannot propose a block that uses oracles in the fast round, and also
        // skip the fast round if fast blocks are not allowed.
        let skip_fast = manager.current_round.is_fast()
            && (has_oracle_responses || !self.options.allow_fast_blocks);
        let conflict = manager
            .requested_signed_proposal
            .as_ref()
            .into_iter()
            .chain(&manager.requested_proposed)
            .any(|proposal| proposal.content.round == manager.current_round)
            || skip_fast;
        let round = if !conflict {
            manager.current_round
        } else if let Some(round) = manager
            .ownership
            .next_round(manager.current_round)
            .filter(|_| manager.current_round.is_multi_leader() || manager.current_round.is_fast())
        {
            round
        } else if let Some(timeout) = info.round_timeout() {
            return Ok(Either::Right(timeout));
        } else {
            return Err(Error::BlockProposalError(
                "Conflicting proposal in the current round",
            ));
        };
        let current_committee = self
            .local_committee()
            .await?
            .validators
            .values()
            .map(|v| (AccountOwner::from(v.account_public_key), v.votes))
            .collect();
        if manager.should_propose(identity, round, seed, &current_committee) {
            return Ok(Either::Left(round));
        }
        if let Some(timeout) = info.round_timeout() {
            return Ok(Either::Right(timeout));
        }
        Err(Error::BlockProposalError(
            "Not a leader in the current round",
        ))
    }

    /// Discards the pending block proposal, if any, so that a fresh block can be
    /// proposed instead of retrying the queued one.
    ///
    /// Note that this does not update the copy persisted in the wallet (callers that
    /// persist pending proposals should refresh it afterwards). Also, this should never
    /// be used on a proposal already submitted in the fast round.
    #[instrument(level = "trace")]
    pub async fn clear_pending_proposal(&self) {
        *self.proposal_mutex().lock().await = None;
    }

    /// Rotates the key of the chain.
    ///
    /// Replaces current owners of the chain with the new key pair.
    #[cfg(with_testing)]
    #[instrument(level = "trace")]
    pub async fn rotate_key_pair(
        &self,
        public_key: linera_base::crypto::AccountPublicKey,
    ) -> Result<ClientOutcome<ConfirmedBlockCertificate>, Error> {
        self.transfer_ownership(public_key.into()).await
    }

    /// Transfers ownership of the chain to a single super owner.
    #[instrument(level = "trace")]
    pub async fn transfer_ownership(
        &self,
        new_owner: AccountOwner,
    ) -> Result<ClientOutcome<ConfirmedBlockCertificate>, Error> {
        self.execute_operation(SystemOperation::ChangeOwnership {
            super_owners: vec![new_owner],
            owners: Vec::new(),
            first_leader: None,
            multi_leader_rounds: 5,
            open_multi_leader_rounds: false,
            timeout_config: TimeoutConfig::default(),
        })
        .await
    }

    /// Adds another owner to the chain, and turns existing super owners into regular owners.
    #[instrument(level = "trace")]
    pub async fn share_ownership(
        &self,
        new_owner: AccountOwner,
        new_weight: u64,
    ) -> Result<ClientOutcome<ConfirmedBlockCertificate>, Error> {
        let ownership = self.prepare_chain().await?.manager.ownership;
        ensure!(
            ownership.is_active(),
            ChainError::InactiveChain(self.chain_id)
        );
        let mut owners = ownership.owners.into_iter().collect::<Vec<_>>();
        owners.extend(ownership.super_owners.into_iter().zip(iter::repeat(100)));
        owners.push((new_owner, new_weight));
        let operations = vec![Operation::system(SystemOperation::ChangeOwnership {
            super_owners: Vec::new(),
            owners,
            first_leader: ownership.first_leader,
            multi_leader_rounds: ownership.multi_leader_rounds,
            open_multi_leader_rounds: ownership.open_multi_leader_rounds,
            timeout_config: ownership.timeout_config,
        })];
        self.execute_block(operations, vec![]).await
    }

    /// Returns the current ownership settings on this chain.
    #[instrument(level = "trace")]
    pub async fn query_chain_ownership(&self) -> Result<ChainOwnership, Error> {
        Ok(self
            .client
            .local_node
            .chain_state_view(self.chain_id)
            .await?
            .execution_state
            .system
            .ownership
            .get()
            .await?
            .clone())
    }

    /// Changes the ownership of this chain. Fails if it would remove existing owners, unless
    /// `remove_owners` is `true`.
    #[instrument(level = "trace")]
    pub async fn change_ownership(
        &self,
        ownership: ChainOwnership,
    ) -> Result<ClientOutcome<ConfirmedBlockCertificate>, Error> {
        self.execute_operation(SystemOperation::ChangeOwnership {
            super_owners: ownership.super_owners.into_iter().collect(),
            owners: ownership.owners.into_iter().collect(),
            first_leader: ownership.first_leader,
            multi_leader_rounds: ownership.multi_leader_rounds,
            open_multi_leader_rounds: ownership.open_multi_leader_rounds,
            timeout_config: ownership.timeout_config.clone(),
        })
        .await
    }

    /// Returns the current application permissions on this chain.
    #[instrument(level = "trace")]
    pub async fn query_application_permissions(&self) -> Result<ApplicationPermissions, Error> {
        Ok(self
            .client
            .local_node
            .chain_state_view(self.chain_id)
            .await?
            .execution_state
            .system
            .application_permissions
            .get()
            .await?
            .clone())
    }

    /// Changes the application permissions configuration on this chain.
    #[instrument(level = "trace", skip(application_permissions))]
    pub async fn change_application_permissions(
        &self,
        application_permissions: ApplicationPermissions,
    ) -> Result<ClientOutcome<ConfirmedBlockCertificate>, Error> {
        self.execute_operation(SystemOperation::ChangeApplicationPermissions(
            application_permissions,
        ))
        .await
    }

    /// Opens a new chain with a derived UID, crediting `balance` to `account` on it.
    #[instrument(level = "trace", skip(self))]
    pub async fn open_chain(
        &self,
        ownership: ChainOwnership,
        application_permissions: ApplicationPermissions,
        account: AccountOwner,
        balance: Amount,
    ) -> Result<ClientOutcome<(ChainDescription, ConfirmedBlockCertificate)>, Error> {
        // Check if we have a key for any owner before consuming ownership.
        let mut has_key = false;
        for owner in ownership.all_owners() {
            if self.has_key_for(owner).await? {
                has_key = true;
                break;
            }
        }
        let config = OpenChainConfig {
            ownership,
            account,
            balance,
            application_permissions,
        };
        let operation = Operation::system(SystemOperation::OpenChain(config));
        let certificate = match self.execute_block(vec![operation], vec![]).await? {
            ClientOutcome::Committed(certificate) => certificate,
            ClientOutcome::Conflict(certificate) => {
                return Ok(ClientOutcome::Conflict(certificate));
            }
            ClientOutcome::WaitForTimeout(timeout) => {
                return Ok(ClientOutcome::WaitForTimeout(timeout));
            }
        };
        // The only operation, i.e. the last transaction, created the new chain.
        let chain_blob = certificate
            .block()
            .body
            .blobs
            .last()
            .and_then(|blobs| blobs.last())
            .ok_or_else(|| Error::InternalError("Failed to create a new chain"))?;
        let description = bcs::from_bytes::<ChainDescription>(chain_blob.bytes())?;
        // If we have a key for any owner, add it to the list of tracked chains.
        if has_key {
            self.client
                .extend_chain_mode(description.id(), ListeningMode::FullChain);
            self.client
                .retry_pending_cross_chain_requests(self.chain_id)
                .await?;
        }
        Ok(ClientOutcome::Committed((description, certificate)))
    }

    /// Publishes a checkpoint of the chain's execution state. The resulting block
    /// contains a single `SystemOperation::Checkpoint`; future nodes can bootstrap
    /// from the published blob instead of replaying the chain's history.
    #[instrument(level = "trace")]
    pub async fn checkpoint(&self) -> Result<ClientOutcome<ConfirmedBlockCertificate>, Error> {
        self.execute_operation(SystemOperation::Checkpoint).await
    }

    /// Closes the chain (and loses everything in it!!).
    /// Returns `None` if the chain was already closed.
    #[instrument(level = "trace")]
    pub async fn close_chain(
        &self,
    ) -> Result<ClientOutcome<Option<ConfirmedBlockCertificate>>, Error> {
        match self.execute_operation(SystemOperation::CloseChain).await {
            Ok(outcome) => Ok(outcome.map(Some)),
            Err(Error::LocalNodeError(LocalNodeError::WorkerError(WorkerError::ChainError(
                chain_error,
            )))) if matches!(*chain_error, ChainError::ClosedChain) => {
                Ok(ClientOutcome::Committed(None)) // Chain is already closed.
            }
            Err(error) => Err(error),
        }
    }

    /// Publishes some module, optionally along with a BCS-encoded `Formats`
    /// description that becomes a third blob alongside contract and service.
    #[cfg(not(target_arch = "wasm32"))]
    #[instrument(level = "trace", skip(contract, service, formats))]
    pub async fn publish_module(
        &self,
        contract: Bytecode,
        service: Bytecode,
        vm_runtime: VmRuntime,
        formats: Option<Vec<u8>>,
    ) -> Result<ClientOutcome<(ModuleId, ConfirmedBlockCertificate)>, Error> {
        let (blobs, module_id) =
            super::create_bytecode_blobs(contract, service, vm_runtime, formats).await;
        self.publish_module_blobs(blobs, module_id).await
    }

    /// Publishes some module.
    #[cfg(not(target_arch = "wasm32"))]
    #[instrument(level = "trace", skip(blobs, module_id))]
    pub async fn publish_module_blobs(
        &self,
        blobs: Vec<Blob>,
        module_id: ModuleId,
    ) -> Result<ClientOutcome<(ModuleId, ConfirmedBlockCertificate)>, Error> {
        self.execute_operations(
            vec![Operation::system(SystemOperation::PublishModule {
                module_id,
            })],
            blobs,
        )
        .await?
        .try_map(|certificate| Ok((module_id, certificate)))
    }

    /// Publishes some data blobs.
    #[instrument(level = "trace", skip(bytes))]
    pub async fn publish_data_blobs(
        &self,
        bytes: Vec<Vec<u8>>,
    ) -> Result<ClientOutcome<ConfirmedBlockCertificate>, Error> {
        let blobs = bytes.into_iter().map(Blob::new_data);
        let publish_blob_operations = blobs
            .clone()
            .map(|blob| {
                Operation::system(SystemOperation::PublishDataBlob {
                    blob_hash: blob.id().hash,
                })
            })
            .collect();
        self.execute_operations(publish_blob_operations, blobs.collect())
            .await
    }

    /// Publishes some data blob.
    #[instrument(level = "trace", skip(bytes))]
    pub async fn publish_data_blob(
        &self,
        bytes: Vec<u8>,
    ) -> Result<ClientOutcome<ConfirmedBlockCertificate>, Error> {
        self.publish_data_blobs(vec![bytes]).await
    }

    /// Creates an application by instantiating some bytecode.
    #[instrument(
        level = "trace",
        skip(self, parameters, instantiation_argument, required_application_ids)
    )]
    pub async fn create_application<
        A: Abi,
        Parameters: Serialize,
        InstantiationArgument: Serialize,
    >(
        &self,
        module_id: ModuleId<A, Parameters, InstantiationArgument>,
        parameters: &Parameters,
        instantiation_argument: &InstantiationArgument,
        required_application_ids: Vec<ApplicationId>,
    ) -> Result<ClientOutcome<(ApplicationId<A>, ConfirmedBlockCertificate)>, Error> {
        let instantiation_argument = serde_json::to_vec(instantiation_argument)?;
        let parameters = serde_json::to_vec(parameters)?;
        Ok(self
            .create_application_untyped(
                module_id.forget_abi(),
                parameters,
                instantiation_argument,
                required_application_ids,
            )
            .await?
            .map(|(app_id, cert)| (app_id.with_abi(), cert)))
    }

    /// Creates an application by instantiating some bytecode.
    #[instrument(
        level = "trace",
        skip(
            self,
            module_id,
            parameters,
            instantiation_argument,
            required_application_ids
        )
    )]
    pub async fn create_application_untyped(
        &self,
        module_id: ModuleId,
        parameters: Vec<u8>,
        instantiation_argument: Vec<u8>,
        required_application_ids: Vec<ApplicationId>,
    ) -> Result<ClientOutcome<(ApplicationId, ConfirmedBlockCertificate)>, Error> {
        self.execute_operation(SystemOperation::CreateApplication {
            module_id,
            parameters,
            instantiation_argument,
            required_application_ids,
        })
        .await?
        .try_map(|certificate| {
            // The first message of the only operation created the application.
            let mut creation = certificate
                .block()
                .created_blob_ids()
                .into_iter()
                .filter(|blob_id| blob_id.blob_type == BlobType::ApplicationDescription)
                .collect::<Vec<_>>();
            if creation.len() > 1 {
                return Err(Error::InternalError(
                    "Unexpected number of application descriptions published",
                ));
            }
            let blob_id = creation.pop().ok_or(Error::InternalError(
                "ApplicationDescription blob not found.",
            ))?;
            let id = ApplicationId::new(blob_id.hash);
            Ok((id, certificate))
        })
    }

    /// Creates a new committee and starts using it (admin chains only).
    #[instrument(level = "trace", skip(committee))]
    pub async fn stage_new_committee(
        &self,
        committee: Committee,
    ) -> Result<ClientOutcome<ConfirmedBlockCertificate>, Error> {
        let blob = Blob::new(BlobContent::new_committee(bcs::to_bytes(&committee)?));
        let blob_hash = blob.id().hash;
        match self
            .execute_operations(
                vec![Operation::system(SystemOperation::Admin(
                    AdminOperation::PublishCommitteeBlob { blob_hash },
                ))],
                vec![blob],
            )
            .await?
        {
            ClientOutcome::Committed(_) => {}
            outcome @ ClientOutcome::WaitForTimeout(_) | outcome @ ClientOutcome::Conflict(_) => {
                return Ok(outcome)
            }
        }
        let epoch = self.chain_info().await?.epoch.try_add_one()?;
        self.execute_operation(SystemOperation::Admin(AdminOperation::CreateCommittee {
            epoch,
            blob_hash,
        }))
        .await
    }

    /// Synchronizes the chain with the validators and creates blocks without any operations to
    /// process all incoming messages. This may require several blocks.
    ///
    /// If not all certificates could be processed due to a timeout, the timestamp for when to retry
    /// is returned, too.
    #[instrument(level = "trace")]
    pub async fn process_inbox(
        &self,
    ) -> Result<(Vec<ConfirmedBlockCertificate>, Option<RoundTimeout>), Error> {
        self.prepare_chain().await?;
        self.process_inbox_without_prepare().await
    }

    /// Creates blocks without any operations to process all incoming messages. This may require
    /// several blocks.
    ///
    /// If not all certificates could be processed due to a timeout, the timestamp for when to retry
    /// is returned, too.
    #[instrument(level = "trace")]
    pub async fn process_inbox_without_prepare(
        &self,
    ) -> Result<(Vec<ConfirmedBlockCertificate>, Option<RoundTimeout>), Error> {
        #[cfg(with_metrics)]
        let _latency = super::metrics::PROCESS_INBOX_WITHOUT_PREPARE_LATENCY.measure_latency();

        let mut certificates = Vec::new();
        loop {
            // We provide no operations - this means that the only operations executed
            // will be epoch changes, receiving messages and processing event stream
            // updates, if any are pending.
            match self.execute_block(vec![], vec![]).await {
                Ok(ClientOutcome::Committed(certificate)) => certificates.push(certificate),
                Ok(ClientOutcome::Conflict(certificate)) => certificates.push(*certificate),
                Ok(ClientOutcome::WaitForTimeout(timeout)) => {
                    return Ok((certificates, Some(timeout)));
                }
                // Nothing in the inbox and no stream updates to be processed.
                Err(Error::LocalNodeError(LocalNodeError::WorkerError(
                    WorkerError::ChainError(chain_error),
                ))) if matches!(*chain_error, ChainError::EmptyBlock) => {
                    return Ok((certificates, None));
                }
                Err(error) => return Err(error),
            };
        }
    }

    /// Returns an operation to process the next pending new epoch, if there is one.
    ///
    /// Later epochs are not included, since a block may advance the epoch at most once;
    /// they are processed by subsequent blocks.
    async fn next_epoch_change(&self) -> Result<Option<Operation>, Error> {
        let next_epoch = self.chain_info().await?.epoch.try_add_one()?;
        if !self
            .has_admin_event(EPOCH_STREAM_NAME, next_epoch.0)
            .await?
        {
            return Ok(None);
        }
        Ok(Some(Operation::system(SystemOperation::ProcessNewEpoch(
            next_epoch,
        ))))
    }

    /// Returns whether the system event on the admin chain with the given stream name and key
    /// exists in storage.
    async fn has_admin_event(&self, stream_name: &[u8], index: u32) -> Result<bool, Error> {
        let event_id = EventId {
            chain_id: self.client.admin_chain_id,
            stream_id: StreamId::system(stream_name),
            index,
        };
        Ok(self
            .client
            .storage_client()
            .read_event(event_id)
            .await?
            .is_some())
    }

    /// Returns the indices and events from the storage
    pub async fn events_from_index(
        &self,
        stream_id: StreamId,
        start_index: u32,
    ) -> Result<Vec<IndexAndEvent>, Error> {
        Ok(self
            .client
            .storage_client()
            .read_events_from_index(&self.chain_id, &stream_id, start_index)
            .await?)
    }

    /// Deprecates all configurations of voting rights up to the given one (admin chains only).
    /// Emits a `RemoveCommittee` event for every still-active epoch up to and including
    /// `revoked_epoch`.
    #[instrument(level = "trace")]
    pub async fn revoke_epochs(
        &self,
        revoked_epoch: Epoch,
    ) -> Result<ClientOutcome<ConfirmedBlockCertificate>, Error> {
        self.prepare_chain().await?;
        let current_epoch = self.chain_info().await?.epoch;
        ensure!(
            revoked_epoch < current_epoch,
            Error::CannotRevokeCurrentEpoch(current_epoch)
        );
        let mut operations = Vec::new();
        for epoch_index in 0..=revoked_epoch.0 {
            let epoch = Epoch(epoch_index);
            if self
                .has_admin_event(REMOVED_EPOCH_STREAM_NAME, epoch.0)
                .await?
            {
                continue;
            }
            operations.push(Operation::system(SystemOperation::Admin(
                AdminOperation::RemoveCommittee { epoch },
            )));
        }
        ensure!(!operations.is_empty(), Error::EpochAlreadyRevoked);
        self.execute_operations(operations, vec![]).await
    }

    /// Sends money to a chain.
    /// Do not check balance. (This may block the client)
    /// Do not confirm the transaction.
    #[cfg(with_testing)]
    #[instrument(level = "trace")]
    pub async fn transfer_to_account_unsafe_unconfirmed(
        &self,
        owner: AccountOwner,
        amount: Amount,
        recipient: Account,
    ) -> Result<ClientOutcome<ConfirmedBlockCertificate>, Error> {
        self.execute_operation(SystemOperation::Transfer {
            owner,
            recipient,
            amount,
        })
        .await
    }

    /// Reads the confirmed block with the given hash from storage.
    #[instrument(level = "trace", skip(hash))]
    pub async fn read_confirmed_block(
        &self,
        hash: CryptoHash,
    ) -> Result<Arc<ConfirmedBlock>, Error> {
        self.client
            .storage_client()
            .read_confirmed_block(hash)
            .await?
            .ok_or(Error::MissingConfirmedBlock(hash))
            .map(|b| b.into_std())
    }

    /// Reads the confirmed block certificate with the given hash from storage.
    #[instrument(level = "trace", skip(hash))]
    pub async fn read_certificate(
        &self,
        hash: CryptoHash,
    ) -> Result<CacheArc<ConfirmedBlockCertificate>, Error> {
        self.client
            .storage_client()
            .read_certificate(hash)
            .await?
            .ok_or(Error::ReadCertificatesError(vec![hash]))
    }

    /// Handles any cross-chain requests for any pending outgoing messages.
    #[instrument(level = "trace")]
    pub async fn retry_pending_outgoing_messages(&self) -> Result<(), Error> {
        self.client
            .retry_pending_cross_chain_requests(self.chain_id)
            .await?;
        Ok(())
    }

    #[instrument(level = "trace", skip(local_node))]
    async fn maybe_local_chain_info(
        &self,
        chain_id: ChainId,
        local_node: &LocalNodeClient<Env::Storage>,
    ) -> Result<Option<Box<ChainInfo>>, Error> {
        match local_node.chain_info(chain_id).await {
            Ok(info) => Ok(Some(info)),
            Err(LocalNodeError::BlobsNotFound(_) | LocalNodeError::InactiveChain(_)) => Ok(None),
            Err(err) => Err(err.into()),
        }
    }

    #[instrument(level = "trace", skip(chain_id, local_node))]
    async fn local_next_block_height(
        &self,
        chain_id: ChainId,
        local_node: &LocalNodeClient<Env::Storage>,
    ) -> Result<BlockHeight, Error> {
        Ok(self
            .maybe_local_chain_info(chain_id, local_node)
            .await?
            .map_or(BlockHeight::ZERO, |info| info.next_block_height))
    }

    /// Returns the next height we expect to receive from the given sender chain, according to the
    /// local inbox.
    #[instrument(level = "trace")]
    async fn local_next_height_to_receive(&self, origin: ChainId) -> Result<BlockHeight, Error> {
        Ok(self
            .client
            .local_node
            .get_inbox_next_height(self.chain_id, origin)
            .await?)
    }

    #[instrument(level = "trace", skip(remote_node, local_node, notification))]
    async fn process_notification(
        &self,
        remote_node: RemoteNode<Env::ValidatorNode>,
        local_node: LocalNodeClient<Env::Storage>,
        notification: Notification,
    ) -> Result<(), Error> {
        let listening_mode = self.client.chain_mode(notification.chain_id);
        let relevant = listening_mode
            .as_ref()
            .is_some_and(|mode| mode.is_relevant(&notification.reason));
        if !relevant {
            tracing::trace!(
                chain_id = %notification.chain_id,
                reason = ?notification.reason,
                ?listening_mode,
                "Ignoring notification due to listening mode"
            );
            return Ok(());
        }
        match notification.reason {
            Reason::NewIncomingBundle { origin, height } => {
                if self.options.message_policy.ignores_origin(&origin) {
                    trace!(
                        chain_id = %self.chain_id,
                        %origin,
                        %height,
                        "Skipping NewIncomingBundle notification: origin filtered by message_policy"
                    );
                    return Ok(());
                }
                if self.local_next_height_to_receive(origin).await? > height {
                    debug!(
                        chain_id = %self.chain_id,
                        "Accepting redundant notification for new message"
                    );
                    return Ok(());
                }
                self.client
                    .download_sender_block_with_sending_ancestors(
                        self.chain_id,
                        origin,
                        height,
                        &remote_node,
                    )
                    .await?;
                if self.local_next_height_to_receive(origin).await? <= height {
                    info!(
                        chain_id = %self.chain_id,
                        "NewIncomingBundle: Fail to synchronize new message after notification"
                    );
                }
            }
            Reason::NewBlock { height, .. } => {
                let chain_id = notification.chain_id;
                let local_height = self.local_next_block_height(chain_id, &local_node).await?;
                if local_height > height {
                    debug!(
                        chain_id = %self.chain_id,
                        "Accepting redundant notification for new block"
                    );
                    return Ok(());
                }
                // The below will only happen in modes other than EventsOnly, as the
                // NewBlock notification is only relevant to other modes.
                self.client
                    .synchronize_chain_state_from(&remote_node, chain_id)
                    .await?;
                if self.local_next_block_height(chain_id, &local_node).await? <= height {
                    error!("NewBlock: Fail to synchronize new block after notification");
                }
                trace!(
                    chain_id = %self.chain_id,
                    %height,
                    "NewBlock: processed notification",
                );
            }
            Reason::NewEvents {
                height, block_hash, ..
            } => {
                let chain_id = notification.chain_id;
                let local_height = self.local_next_block_height(chain_id, &local_node).await?;
                if local_height > height {
                    debug!(
                        chain_id = %self.chain_id,
                        "Accepting redundant notification for new events"
                    );
                    return Ok(());
                }
                trace!(
                    chain_id = %self.chain_id,
                    %height,
                    "NewEvents: processing notification"
                );
                // Use the subscribed streams from EventsOnly mode to figure out which
                // streams we are interested in.
                let relevant_streams = match self.listening_mode() {
                    Some(ListeningMode::EventsOnly(subscribed)) => subscribed,
                    // other cases should be unreachable, as the NewEvents notification is
                    // only relevant in the EventsOnly mode
                    _ => unreachable!(),
                };
                self.client
                    .download_event_bearing_blocks(
                        self.chain_id,
                        BTreeSet::from([(height, block_hash)]),
                        local_height,
                        &relevant_streams,
                        &remote_node,
                    )
                    .await?;
            }
            Reason::NewRound { height, round } => {
                let chain_id = notification.chain_id;
                if let Some(info) = self.maybe_local_chain_info(chain_id, &local_node).await? {
                    if (info.next_block_height, info.manager.current_round) >= (height, round) {
                        debug!(
                            chain_id = %self.chain_id,
                            "Accepting redundant notification for new round"
                        );
                        return Ok(());
                    }
                }
                self.client
                    .synchronize_chain_state_from(&remote_node, chain_id)
                    .await?;
                let Some(info) = self.maybe_local_chain_info(chain_id, &local_node).await? else {
                    error!(
                        chain_id = %self.chain_id,
                        "NewRound: Fail to read local chain info for {chain_id}"
                    );
                    return Ok(());
                };
                if (info.next_block_height, info.manager.current_round) < (height, round) {
                    info!(
                        chain_id = %self.chain_id,
                        "NewRound: Fail to synchronize new block after notification"
                    );
                }
            }
            Reason::BlockExecuted { .. } => {
                // No action needed.
            }
        }
        Ok(())
    }

    /// Returns whether this chain is tracked by the client, i.e. we are updating its inbox.
    pub fn is_tracked(&self) -> bool {
        self.client.is_tracked(self.chain_id)
    }

    /// Returns the listening mode for this chain, if it is tracked.
    pub fn listening_mode(&self) -> Option<ListeningMode> {
        self.client.chain_mode(self.chain_id)
    }

    /// Spawns a task that listens to notifications about the current chain from all validators,
    /// and synchronizes the local state accordingly.
    ///
    /// The listening mode must be set in `Client::chain_modes` before calling this method.
    #[instrument(level = "trace", fields(chain_id = ?self.chain_id))]
    pub async fn listen(
        &self,
    ) -> Result<(impl Future<Output = ()>, AbortOnDrop, NotificationStream), Error> {
        use future::FutureExt as _;

        async fn await_while_polling<F: FusedFuture>(
            future: F,
            background_work: impl FusedStream<Item = ()>,
        ) -> F::Output {
            tokio::pin!(future);
            tokio::pin!(background_work);
            loop {
                futures::select! {
                    _ = background_work.next() => (),
                    result = future => return result,
                }
            }
        }

        let mut senders: HashMap<ValidatorPublicKey, StreamHandle> = HashMap::new();
        let mut circuit_breakers: HashMap<ValidatorPublicKey, CircuitBreakerState> = HashMap::new();
        let notifications = self.subscribe()?;
        let (abortable_notifications, abort) = stream::abortable(self.subscribe()?);

        // Beware: if this future ceases to make progress, notification processing will
        // deadlock, because of the issue described in
        // https://github.com/linera-io/linera-protocol/pull/1173.

        // TODO(#2013): replace this lock with an asynchronous communication channel

        let mut process_notifications = FuturesUnordered::new();

        // A failed update here leaves no streams, no breakers and nothing to wake the
        // loop, and nothing retries `listen()` on failure — so the retry deadline has to
        // be armed from the start, exactly as the in-loop failure path does. (A chain
        // whose listening mode is later UPGRADED does get a fresh `listen()`, which
        // discards this listener's accumulated breaker state.)
        let mut retry_update_at: Option<Timestamp> = None;
        match self
            .update_notification_streams(&mut senders, &mut circuit_breakers)
            .await
        {
            Ok(tasks) => process_notifications.extend(tasks),
            Err(error) => {
                error!("Failed to update committee: {error}");
                retry_update_at = Some(self.retry_update_deadline());
            }
        };

        let this = self.clone();
        // Boxed so that nesting this loop's futures does not overflow the trait solver
        // proving `Send` for callers that spawn it.
        let update_streams: MaybeSendBoxFuture<'static, ()> = Box::pin(
            async move {
                let mut abortable_notifications = abortable_notifications.fuse();
                // The probe timer, held across iterations so it is not rebuilt on every
                // notification; `armed_probe_at` is the deadline it currently encodes.
                let mut armed_probe_at: Option<Timestamp> = None;
                let probe_due = Either::Right(future::pending()).fuse();
                tokio::pin!(probe_due);

                'listen: loop {
                    // Earliest deadline of any kind. Waking on it is what lets a chain
                    // with no live streams repair them: otherwise the update runs only on
                    // this chain's next `NewBlock`, which cannot arrive while every
                    // stream carrying it is dead.
                    let next_probe_at = circuit_breakers
                        .values()
                        .map(|state| state.next_probe_at)
                        .chain(retry_update_at)
                        .min();
                    // Rebuild the timer only when the deadline actually moves; a
                    // notification that does not trigger an update must not discard and
                    // recreate it, which under `TestClock` also strands a `oneshot` per
                    // iteration in the shared sleep map.
                    if next_probe_at != armed_probe_at {
                        armed_probe_at = next_probe_at;
                        probe_due.set(
                            match next_probe_at {
                                Some(deadline) => Either::Left(
                                    this.storage_client().clock().sleep_until(deadline),
                                ),
                                None => Either::Right(future::pending()),
                            }
                            .fuse(),
                        );
                    }

                    let update_now = futures::select! {
                        maybe_notification = abortable_notifications.next() => {
                            match maybe_notification {
                                // Re-subscribe to validators on new blocks to handle
                                // committee changes.
                                Some(notification) => {
                                    matches!(notification.reason, Reason::NewBlock { .. })
                                }
                                None => break 'listen,
                            }
                        }
                        // A validator notification stream task finished: update now, so
                        // that the ended stream is registered in the circuit breaker and
                        // a probe gets scheduled even if this chain never produces
                        // another block.
                        _ = process_notifications.select_next_some() => {
                            // Drain the completions that are already ready first. The tasks
                            // are individual, so a simultaneous loss of every stream on this
                            // chain queues one per stream, and each would otherwise drive a
                            // full update of its own — an N-times amplification per chain at
                            // exactly the moment a fleet-wide event has hit every chain on
                            // the worker, with each update able to become a quorum
                            // `synchronize_chain_state()` through the `BlobsNotFound`
                            // fallback. One pass registers all of them.
                            while process_notifications.next().now_or_never().flatten().is_some() {}
                            true
                        }
                        // A circuit-breaker probe is due. Clear the armed deadline so
                        // the timer is rebuilt below: it is a `Fuse` that has now
                        // completed, and `select!` skips terminated arms in silence, so
                        // leaving it in place when the recomputed deadline happens to be
                        // unchanged would cost the listener its only timer wake-up.
                        _ = probe_due => {
                            armed_probe_at = None;
                            true
                        }
                    };
                    if !update_now {
                        continue;
                    }
                    // `await_while_polling` keeps draining the stream tasks while the
                    // update runs, and both of the update's await points precede its
                    // transition scan, so a stream that ends mid-update is still seen.
                    match Box::pin(await_while_polling(
                        this.update_notification_streams(&mut senders, &mut circuit_breakers)
                            .fuse(),
                        &mut process_notifications,
                    ))
                    .await
                    {
                        Ok(tasks) => {
                            retry_update_at = None;
                            process_notifications.extend(tasks);
                        }
                        Err(error) => {
                            error!("Failed to update committee: {error}");
                            // The update returns before touching any breaker, so every
                            // deadline it would have advanced is still in the past and
                            // would keep winning `min()` below, re-firing the timer into
                            // the same failing call. Push out only those — clamping the
                            // rest up to `retry` would postpone a probe that is genuinely
                            // due sooner, which is what taking the `min` exists to avoid.
                            let retry = this.retry_update_deadline();
                            this.defer_due_probes(&mut circuit_breakers, retry);
                            retry_update_at = Some(retry);
                        }
                    }
                }

                for handle in senders.into_values() {
                    handle.abort.abort();
                }

                let () = process_notifications.collect().await;
            }
            .in_current_span(),
        );

        Ok((update_streams, AbortOnDrop(abort), notifications))
    }

    /// The configured initial probe interval, floored.
    fn initial_probe_interval(&self) -> Duration {
        self.options
            .notification_circuit_breaker_initial_probe_interval
            .max(MIN_PROBE_INTERVAL)
    }

    /// A stable per-validator offset, so that probes armed by one fleet-wide event stay
    /// spread out instead of re-converging at every doubling.
    ///
    /// Always a fraction of the INITIAL interval, never of the current escalated one — a
    /// parameter here would read as though the spread grows with the backoff, which would
    /// make a capped breaker's jitter larger than the gap it is jittering.
    fn probe_spread(&self, validator: &ValidatorPublicKey) -> Duration {
        let mut hasher = DefaultHasher::new();
        (self.chain_id, validator).hash(&mut hasher);
        (self.initial_probe_interval() / PROBE_SPREAD_DIVISOR)
            .mul_f64((hasher.finish() % 1_000) as f64 / 1_000.0)
    }

    /// Pushes the already-due probe deadlines out to `retry`, after an update failed
    /// before it could touch any of them.
    ///
    /// Only the due ones: they are what would otherwise keep winning `min()` and re-fire
    /// the timer straight back into the same failing call. Clamping the rest up to `retry`
    /// as well would postpone a probe that is legitimately due sooner, which is exactly
    /// what taking the `min` over both deadlines exists to avoid.
    pub(super) fn defer_due_probes(
        &self,
        circuit_breakers: &mut HashMap<ValidatorPublicKey, CircuitBreakerState>,
        retry: Timestamp,
    ) {
        let now = self.storage_client().clock().current_time();
        for state in circuit_breakers.values_mut() {
            if state.next_probe_at <= now {
                state.next_probe_at = retry;
            }
        }
    }

    /// When to retry an update that failed before it could touch the breaker state.
    fn retry_update_deadline(&self) -> Timestamp {
        self.storage_client()
            .clock()
            .current_time()
            .saturating_add(TimeDelta::from_duration(self.initial_probe_interval()))
    }

    /// Creates or repairs the per-validator notification stream tasks for this chain,
    /// and updates the circuit-breaker state for validators whose streams have ended.
    /// Returns the newly created stream tasks; the caller is responsible for polling
    /// them.
    #[instrument(level = "trace", skip(senders, circuit_breakers))]
    pub(super) async fn update_notification_streams(
        &self,
        senders: &mut HashMap<ValidatorPublicKey, StreamHandle>,
        circuit_breakers: &mut HashMap<ValidatorPublicKey, CircuitBreakerState>,
    ) -> Result<Vec<MaybeSendBoxFuture<'static, ()>>, Error> {
        let initial_probe_interval = self.initial_probe_interval();
        // Floored by the initial interval rather than by `MIN_PROBE_INTERVAL`: sharing the
        // floor would let a `0` max pin both ends together and delete escalation entirely.
        let max_probe_interval = self
            .options
            .notification_circuit_breaker_max_probe_interval
            .max(initial_probe_interval);
        let (nodes, local_node) = {
            // For EventsOnly chains we may not have the chain's own committee locally,
            // and attempting to fetch it would trigger a full sync. Use the admin
            // committee instead — we only need it to know which validators to connect to.
            let committee = if self
                .listening_mode()
                .is_some_and(|m| m.should_sync_chain_state())
            {
                self.local_committee().await?
            } else {
                self.client.admin_committee().await?.1
            };
            let nodes = self
                .client
                .validator_node_provider()
                .make_nodes(&committee)?
                .collect::<HashMap<_, _>>();
            (nodes, self.client.local_node.clone())
        };
        // Read the (possibly simulated) node clock once, so circuit-breaker probe scheduling
        // can be driven deterministically in tests instead of depending on the wall clock.
        //
        // AFTER the committee read above, which is this function's only await and can be a
        // quorum round trip via `local_committee()`'s `BlobsNotFound` fallback. Sampling it
        // before would arm every deadline below at a moment already past by the time the
        // update returns — an unpaced retry on the `Ok` path, which has no `defer_due_probes`
        // — and would make `serving_for` saturate to zero for a stream that began serving
        // during the await, charging a healthy stream as a persistent fault.
        let now = self.storage_client().clock().current_time();
        // Detect circuit breaker state transitions before cleaning up senders.
        //
        // Health is judged on whether the stream reached the point of serving
        // notifications, not on the task still running: the probe body awaits the initial
        // sync before anything polls the stream, so "alive" on its own also covers a probe
        // that has delivered nothing and may never deliver anything.
        let mut abort_stalled_probes = Vec::new();
        for (validator, handle) in senders.iter() {
            if !nodes.contains_key(validator) {
                continue;
            }
            let died = handle.abort.is_aborted();
            let serving_for = handle.serving_for(now);
            match (died, serving_for) {
                // Serving for at least one interval and still running: healthy.
                //
                // The same threshold the churn arm uses, and for the same reason. Dropping
                // the breaker after ANY amount of serving would make the ladder depend on
                // whether an unrelated update happened to land inside a brief serving
                // window — the initial sync emits `NewBlock`, so on an active chain one
                // usually does — and a validator that had escalated to the cap would be
                // reset to the base cadence by a stream that served for seconds. Below the
                // threshold the breaker simply stays, and its deadline is in the future.
                (false, Some(lifetime)) if lifetime >= initial_probe_interval => {
                    // Drop the breaker either way so the timer goes quiet, but only report
                    // a recovery for one that actually tripped: every launch arms a
                    // deadline, so gating on the removal alone announces a recovery for
                    // every healthy validator on the first update after it starts serving.
                    if circuit_breakers
                        .remove(validator)
                        .is_some_and(|state| state.tripped)
                    {
                        info!(
                            %validator,
                            chain_id = %self.chain_id,
                            "Validator recovered from circuit breaker"
                        );
                    }
                }
                // Serving, but not yet for long enough to clear the breaker.
                //
                // Its deadline is NOT necessarily in the future: a probe arms at
                // `launch + probe_interval + spread` while the stream only starts serving
                // after its sync returns, so whenever that sync outlasts this validator's
                // jitter offset the timer fires here with the deadline already elapsed.
                // Leaving it elapsed would keep it winning `min()` in `listen()`, and
                // `sleep_until` on a past instant returns immediately — a full
                // `update_notification_streams` per pass until the lifetime finally
                // crosses the threshold. Re-arm at exactly the qualifying instant; the
                // interval and the ladder are untouched.
                (false, Some(lifetime)) => {
                    if let Some(state) = circuit_breakers.get_mut(validator) {
                        if now >= state.next_probe_at {
                            state.next_probe_at = now.saturating_add(TimeDelta::from_duration(
                                initial_probe_interval.saturating_sub(lifetime),
                            ));
                        }
                    }
                }
                // Still connecting or syncing when the deadline passed. Abort it, or its
                // `senders` entry stays occupied and the validator is never probed again.
                (false, None) => {
                    if let Some(state) = circuit_breakers.get_mut(validator) {
                        if now >= state.next_probe_at {
                            // A breaker that has not tripped is the deadline armed at
                            // launch, so this is the stream's FIRST recorded failure and
                            // belongs at the configured interval, not at twice it.
                            if state.tripped {
                                state.escalate(
                                    now,
                                    max_probe_interval,
                                    self.probe_spread(validator),
                                );
                            } else {
                                *state = CircuitBreakerState::failed(
                                    now,
                                    initial_probe_interval,
                                    self.probe_spread(validator),
                                );
                            }
                            abort_stalled_probes.push(*validator);
                            warn!(
                                %validator,
                                chain_id = %self.chain_id,
                                next_probe_in = ?state.probe_interval,
                                "Stream did not start serving in time; aborting and retrying later"
                            );
                        }
                    }
                }
                // Died after serving for at least one interval: the stream did real work
                // before stopping, so re-arm at the initial interval instead of doubling.
                // Below that threshold it is not churn — the transport already absorbs
                // reconnects internally, so a short-lived stream means a persistent fault.
                (true, Some(lifetime)) if lifetime >= initial_probe_interval => {
                    circuit_breakers.insert(
                        *validator,
                        CircuitBreakerState::churned(
                            now,
                            initial_probe_interval,
                            self.probe_spread(validator),
                        ),
                    );
                    info!(
                        %validator,
                        chain_id = %self.chain_id,
                        ?lifetime,
                        next_probe_in = ?initial_probe_interval,
                        "Long-lived validator notification stream ended; re-probing"
                    );
                }
                // Died without serving, or after too short a life to count as churn.
                (true, _) => {
                    // Escalate only a breaker that already tripped. Every launch arms one,
                    // so keying on presence alone charges a validator's FIRST stream death
                    // as a failed probe: the ladder starts at twice the configured
                    // interval and the error-level record that a stream was lost at all
                    // becomes unreachable.
                    if let Some(state) = circuit_breakers
                        .get_mut(validator)
                        .filter(|state| state.tripped)
                    {
                        state.escalate(now, max_probe_interval, self.probe_spread(validator));
                        warn!(
                            %validator,
                            chain_id = %self.chain_id,
                            next_probe_in = ?state.probe_interval,
                            "Validator still unhealthy after probe; increasing probe interval"
                        );
                    } else {
                        circuit_breakers.insert(
                            *validator,
                            CircuitBreakerState::failed(
                                now,
                                initial_probe_interval,
                                self.probe_spread(validator),
                            ),
                        );
                        error!(
                            %validator,
                            chain_id = %self.chain_id,
                            next_probe_in = ?initial_probe_interval,
                            "Validator notification stream ended; entering circuit breaker"
                        );
                    }
                }
            }
        }
        for validator in abort_stalled_probes {
            if let Some(handle) = senders.get(&validator) {
                handle.abort.abort();
            }
        }

        senders.retain(|validator, handle| {
            if !nodes.contains_key(validator) {
                handle.abort.abort();
            }
            !handle.abort.is_aborted()
        });
        circuit_breakers.retain(|validator, _| nodes.contains_key(validator));

        let mut validator_tasks: Vec<MaybeSendBoxFuture<'static, ()>> = Vec::new();
        for (public_key, node) in nodes {
            let hash_map::Entry::Vacant(entry) = senders.entry(public_key) else {
                continue;
            };

            let spread = self.probe_spread(&public_key);
            match circuit_breakers.entry(public_key) {
                hash_map::Entry::Occupied(mut breaker) => {
                    let state = breaker.get_mut();
                    if now < state.next_probe_at {
                        continue;
                    }
                    // Re-arm before launching: the outcome is only known at a later
                    // update, and a deadline left in the past re-enters this loop at once.
                    state.arm(now, spread);
                    debug!(
                        validator = %public_key,
                        chain_id = %self.chain_id,
                        "Probing unhealthy validator"
                    );
                }
                // A validator's first launch. Arm a deadline anyway, or a `subscribe` or
                // sync that never returns leaves nothing to abort it and nothing to wake
                // the timer on its account. Recovery drops the breaker once it serves.
                hash_map::Entry::Vacant(breaker) => {
                    breaker.insert(CircuitBreakerState::launched(
                        now,
                        initial_probe_interval * FIRST_LAUNCH_GRACE,
                        spread,
                    ));
                }
            }

            let address = node.address();
            let this = self.clone();
            let listening_mode_for_sync = self.listening_mode();
            let serving_since = Arc::new(AtomicU64::new(NOT_SERVING));
            let serving_since_for_task = serving_since.clone();
            let stream = stream::once({
                let node = node.clone();
                async move {
                    let stream = node.subscribe(vec![this.chain_id]).await?;
                    // The stream is established but nothing polls it until this block
                    // returns, so the sync below still has to finish before the validator
                    // counts as serving.
                    let remote_node = RemoteNode { public_key, node };
                    if listening_mode_for_sync
                        .as_ref()
                        .is_some_and(|mode| mode.should_sync_chain_state())
                    {
                        this.client
                            .synchronize_chain_state_from(&remote_node, this.chain_id)
                            .await?;
                    } else {
                        // For EventsOnly chains, do a lightweight initial sync:
                        // query the validator for the latest event-bearing blocks
                        // for our subscribed streams, then download them sparsely.
                        if let Some(ListeningMode::EventsOnly(subscribed)) =
                            listening_mode_for_sync.as_ref()
                        {
                            if let Err(error) = this
                                .client
                                .sync_events_from_node(this.chain_id, subscribed, &remote_node)
                                .await
                            {
                                debug!(
                                    chain_id = %this.chain_id,
                                    %error,
                                    "Failed initial sparse sync for EventsOnly chain"
                                );
                            }
                        }
                    }
                    serving_since_for_task.store(
                        this.storage_client().clock().current_time().micros(),
                        Ordering::Relaxed,
                    );
                    Ok::<_, Error>(stream)
                }
            })
            .filter_map(move |result| {
                let address = address.clone();
                async move {
                    if let Err(error) = &result {
                        info!(?error, address, "could not connect to validator");
                    } else {
                        debug!(address, "connected to validator");
                    }
                    result.ok()
                }
            })
            .flatten();
            let (stream, abort) = stream::abortable(stream);
            let mut stream = Box::pin(stream);
            let abort_on_exit = abort.clone();
            let this = self.clone();
            let local_node = local_node.clone();
            let remote_node = RemoteNode { public_key, node };
            validator_tasks.push(Box::pin(async move {
                while let Some(notification) = stream.next().await {
                    if let Err(error) = this
                        .process_notification(
                            remote_node.clone(),
                            local_node.clone(),
                            notification.clone(),
                        )
                        .await
                    {
                        tracing::info!(
                            chain_id = %this.chain_id,
                            address = remote_node.address(),
                            ?notification,
                            %error,
                            "failed to process notification",
                        );
                    }
                }
                warn!(
                    chain_id = %this.chain_id,
                    address = remote_node.address(),
                    "Validator notification stream ended"
                );
                abort_on_exit.abort();
            }));
            entry.insert(StreamHandle {
                abort,
                serving_since,
            });
        }
        Ok(validator_tasks)
    }

    /// Attempts to update a validator with the local information: sends any confirmed
    /// block certificates the validator is missing, along with the blobs and other
    /// chains' blocks they depend on, and evidence for the current consensus round.
    ///
    /// The public key is used to verify the validator's responses.
    #[instrument(level = "trace", skip(node))]
    pub async fn sync_validator(
        &self,
        public_key: ValidatorPublicKey,
        node: Env::ValidatorNode,
    ) -> Result<(), Error> {
        let local_next_block_height = self.chain_info().await?.next_block_height;
        let mut updater = self
            .client
            .remote_node_updater(RemoteNode { public_key, node });
        updater
            .send_chain_information(
                self.chain_id,
                local_next_block_height,
                CrossChainMessageDelivery::NonBlocking,
                None,
            )
            .await
    }
}

#[cfg(with_testing)]
impl<Env: Environment> ChainClient<Env> {
    /// Processes a notification received from the given validator.
    pub async fn process_notification_from(
        &self,
        notification: Notification,
        validator: (ValidatorPublicKey, &str),
    ) {
        let mut node_list = self
            .client
            .validator_node_provider()
            .make_nodes_from_list(vec![validator])
            .unwrap();
        let (public_key, node) = node_list.next().unwrap();
        let remote_node = RemoteNode { node, public_key };
        let local_node = self.client.local_node.clone();
        self.process_notification(remote_node, local_node, notification)
            .await
            .unwrap();
    }
}

#[cfg(test)]
mod tests {
    use super::{Error, LocalNodeError};

    #[test]
    fn error_type_delegates_to_local_node_error() {
        assert_eq!(
            Error::LocalNodeError(LocalNodeError::InvalidChainInfoResponse).error_type(),
            "LocalNodeError::InvalidChainInfoResponse"
        );
    }

    #[test]
    fn error_type_falls_back_to_chain_client_variant() {
        assert_eq!(
            Error::WalletSynchronizationError.error_type(),
            "ChainClientError::WalletSynchronizationError"
        );
    }
}