linera-client 0.15.20

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

use std::sync::Arc;

#[cfg(not(web))]
use futures::StreamExt as _;
use futures::{Future, TryStreamExt as _};
use linera_base::{
    crypto::{CryptoHash, ValidatorPublicKey},
    data_types::{ChainDescription, Epoch, Timestamp},
    identifiers::{Account, AccountOwner, ChainId},
    ownership::ChainOwnership,
    time::{Duration, Instant},
    util::future::FutureSyncExt as _,
};
use linera_chain::{manager::LockingBlock, types::ConfirmedBlockCertificate};
use linera_core::{
    client::{chain_client, ChainClient, Client, ListeningMode},
    data_types::{ChainInfo, ChainInfoQuery, ClientOutcome},
    join_set_ext::JoinSet,
    node::ValidatorNode,
    wallet, Environment, JoinSetExt as _, Wallet as _,
};
use linera_rpc::node_provider::{NodeOptions, NodeProvider};
use linera_storage::Storage as _;
use linera_version::VersionInfo;
use thiserror_context::Context;
use tracing::{debug, info, warn};
#[cfg(not(web))]
use {
    crate::{
        benchmark::{fungible_transfer, Benchmark, BenchmarkError},
        client_metrics::ClientMetrics,
    },
    futures::stream,
    linera_base::{
        crypto::AccountPublicKey,
        data_types::{Amount, BlockHeight},
        identifiers::{ApplicationId, BlobType},
    },
    linera_execution::{
        system::{OpenChainConfig, SystemOperation},
        Operation,
    },
    std::{collections::HashSet, path::Path},
    tokio::{sync::mpsc, task},
};
#[cfg(feature = "fs")]
use {
    linera_base::{
        data_types::{BlobContent, Bytecode},
        identifiers::ModuleId,
        vm::VmRuntime,
    },
    linera_core::client::create_bytecode_blobs,
    std::{fs, path::PathBuf},
};

use crate::{
    chain_listener::{self, ClientContext as _},
    client_options::{ChainOwnershipConfig, Options},
    config::GenesisConfig,
    error, util, Error,
};

/// Results from querying a validator about version, network description, and chain info.
pub struct ValidatorQueryResults {
    /// The validator's version information.
    pub version_info: Result<VersionInfo, Error>,
    /// The validator's genesis config hash.
    pub genesis_config_hash: Result<CryptoHash, Error>,
    /// The validator's chain info (if valid and signature check passed).
    pub chain_info: Result<ChainInfo, Error>,
}

impl ValidatorQueryResults {
    /// Returns a vector of references to all errors in the query results.
    pub fn errors(&self) -> Vec<&Error> {
        let mut errors = Vec::new();
        if let Err(e) = &self.version_info {
            errors.push(e);
        }
        if let Err(e) = &self.genesis_config_hash {
            errors.push(e);
        }
        if let Err(e) = &self.chain_info {
            errors.push(e);
        }
        errors
    }

    /// Prints validator information to stdout.
    ///
    /// Prints public key, address, and optionally weight, version info, and chain info.
    /// If `reference` is provided, only prints fields that differ from the reference.
    pub fn print(
        &self,
        public_key: Option<&ValidatorPublicKey>,
        address: Option<&str>,
        weight: Option<u64>,
        reference: Option<&ValidatorQueryResults>,
    ) {
        if let Some(key) = public_key {
            println!("Public key: {key}");
        }
        if let Some(address) = address {
            println!("Address: {address}");
        }
        if let Some(w) = weight {
            println!("Weight: {w}");
        }

        let ref_version = reference.and_then(|ref_results| ref_results.version_info.as_ref().ok());
        match &self.version_info {
            Ok(version_info) => {
                if ref_version.is_none_or(|ref_v| ref_v.crate_version != version_info.crate_version)
                {
                    println!("Linera protocol: v{}", version_info.crate_version);
                }
                if ref_version.is_none_or(|ref_v| ref_v.rpc_hash != version_info.rpc_hash) {
                    println!("RPC API hash: {}", version_info.rpc_hash);
                }
                if ref_version.is_none_or(|ref_v| ref_v.graphql_hash != version_info.graphql_hash) {
                    println!("GraphQL API hash: {}", version_info.graphql_hash);
                }
                if ref_version.is_none_or(|ref_v| ref_v.wit_hash != version_info.wit_hash) {
                    println!("WIT API hash: v{}", version_info.wit_hash);
                }
                if ref_version.is_none_or(|ref_v| {
                    (&ref_v.git_commit, ref_v.git_dirty)
                        != (&version_info.git_commit, version_info.git_dirty)
                }) {
                    println!(
                        "Source code: {}/tree/{}{}",
                        env!("CARGO_PKG_REPOSITORY"),
                        version_info.git_commit,
                        if version_info.git_dirty {
                            " (dirty)"
                        } else {
                            ""
                        }
                    );
                }
            }
            Err(err) => println!("Error getting version info: {err}"),
        }

        let ref_genesis_hash =
            reference.and_then(|ref_results| ref_results.genesis_config_hash.as_ref().ok());
        match &self.genesis_config_hash {
            Ok(hash) if ref_genesis_hash.is_some_and(|ref_hash| ref_hash == hash) => {}
            Ok(hash) => println!("Genesis config hash: {hash}"),
            Err(err) => println!("Error getting genesis config: {err}"),
        }

        let ref_info = reference.and_then(|ref_results| ref_results.chain_info.as_ref().ok());
        match &self.chain_info {
            Ok(info) => {
                if ref_info.is_none_or(|ref_info| info.block_hash != ref_info.block_hash) {
                    if let Some(hash) = info.block_hash {
                        println!("Block hash: {hash}");
                    } else {
                        println!("Block hash: None");
                    }
                }
                if ref_info
                    .is_none_or(|ref_info| info.next_block_height != ref_info.next_block_height)
                {
                    println!("Next height: {}", info.next_block_height);
                }
                if ref_info.is_none_or(|ref_info| info.timestamp != ref_info.timestamp) {
                    println!("Timestamp: {}", info.timestamp);
                }
                if ref_info.is_none_or(|ref_info| info.epoch != ref_info.epoch) {
                    println!("Epoch: {}", info.epoch);
                }
                if ref_info.is_none_or(|ref_info| {
                    info.manager.current_round != ref_info.manager.current_round
                }) {
                    println!("Round: {}", info.manager.current_round);
                }
                if let Some(leader) = info.manager.leader {
                    println!("Leader: {leader}");
                }
                if let Some(locking) = &info.manager.requested_locking {
                    match &**locking {
                        LockingBlock::Fast(proposal) => {
                            println!(
                                "Locking fast block from {}",
                                proposal.content.block.timestamp
                            );
                        }
                        LockingBlock::Regular(validated) => {
                            println!(
                                "Locking block {} in {} from {}",
                                validated.hash(),
                                validated.round,
                                validated.block().header.timestamp
                            );
                        }
                    }
                }
            }
            Err(err) => println!("Error getting chain info: {err}"),
        }
        println!();
    }
}

/// The state shared by the client commands: the core client, wallet configuration, and
/// network timeouts.
pub struct ClientContext<Env: Environment> {
    /// The core client used to interact with chains and validators.
    pub client: Arc<Client<Env>>,
    /// The genesis configuration of the network.
    // TODO(#5083): this doesn't really need to be stored
    pub genesis_config: crate::config::GenesisConfig,
    /// The timeout for sending requests to validators.
    pub send_timeout: Duration,
    /// The timeout for receiving responses from validators.
    pub recv_timeout: Duration,
    /// The delay before retrying a failed request to a validator.
    pub retry_delay: Duration,
    /// The maximum number of times to retry a failed request to a validator.
    pub max_retries: u32,
    /// The maximum backoff between retries of a failed request to a validator.
    pub max_backoff: Duration,
    /// The set of background tasks listening for chain notifications.
    pub chain_listeners: JoinSet,
    /// The default chain used when no chain is explicitly specified.
    // TODO(#5082): move this into the upstream UI layers (maybe just the CLI)
    pub default_chain: Option<ChainId>,
    /// The metrics collector, if metrics collection is enabled.
    #[cfg(not(web))]
    pub client_metrics: Option<ClientMetrics>,
}

impl<Env: Environment> chain_listener::ClientContext for ClientContext<Env> {
    type Environment = Env;

    fn wallet(&self) -> &Env::Wallet {
        self.client.wallet()
    }

    fn storage(&self) -> &Env::Storage {
        self.client.storage_client()
    }

    fn client(&self) -> &Arc<Client<Env>> {
        &self.client
    }

    #[cfg(not(web))]
    fn timing_sender(
        &self,
    ) -> Option<mpsc::UnboundedSender<(u64, linera_core::client::TimingType)>> {
        self.client_metrics
            .as_ref()
            .map(|metrics| metrics.timing_sender.clone())
    }

    async fn update_wallet_for_new_chain(
        &mut self,
        chain_id: ChainId,
        owner: Option<AccountOwner>,
        timestamp: Timestamp,
        epoch: Epoch,
    ) -> Result<(), Error> {
        self.update_wallet_for_new_chain(chain_id, owner, timestamp, epoch)
            .make_sync()
            .await
    }

    async fn update_wallet(&mut self, chain_client: &ChainClient<Env>) -> Result<(), Error> {
        self.update_wallet_from_client(chain_client)
            .make_sync()
            .await
    }
}

impl<S, Si, W> ClientContext<linera_core::environment::Impl<S, NodeProvider, Si, W>>
where
    S: linera_core::environment::Storage,
    Si: linera_core::environment::Signer,
    W: linera_core::environment::Wallet,
{
    // not worth refactoring this because
    // https://github.com/linera-io/linera-protocol/issues/5082
    // https://github.com/linera-io/linera-protocol/issues/5083
    /// Creates a new client context from the given storage, wallet, signer, and options.
    #[expect(clippy::too_many_arguments)]
    pub async fn new(
        storage: S,
        wallet: W,
        signer: Si,
        options: &Options,
        default_chain: Option<ChainId>,
        genesis_config: GenesisConfig,
        block_cache_size: usize,
        execution_state_cache_size: usize,
    ) -> Result<Self, Error> {
        #[cfg(not(web))]
        let timing_config = options.to_timing_config();
        let node_provider = NodeProvider::new(NodeOptions {
            send_timeout: options.send_timeout,
            recv_timeout: options.recv_timeout,
            retry_delay: options.retry_delay,
            max_retries: options.max_retries,
            max_backoff: options.max_backoff,
        });
        let chain_modes: Vec<_> = wallet
            .items()
            .map_ok(|(id, _chain)| (id, ListeningMode::FullChain))
            .try_collect()
            .await
            .map_err(error::Inner::wallet)?;
        let name = match chain_modes.len() {
            0 => "Client node".to_string(),
            1 => format!("Client node for {:.8}", chain_modes[0].0),
            n => format!(
                "Client node for {:.8} and {} others",
                chain_modes[0].0,
                n - 1
            ),
        };

        let client = Client::new(
            linera_core::environment::Impl {
                network: node_provider,
                storage,
                signer,
                wallet,
            },
            genesis_config.admin_chain_id(),
            options.long_lived_services,
            chain_modes,
            name,
            util::non_zero_duration(options.chain_worker_ttl),
            util::non_zero_duration(options.sender_chain_worker_ttl),
            options.cross_chain_batch_size_limit,
            options.to_chain_client_options(),
            &options.to_requests_scheduler_config(),
            block_cache_size,
            execution_state_cache_size,
        );

        #[cfg(not(web))]
        let client_metrics = if timing_config.enabled {
            Some(ClientMetrics::new(timing_config))
        } else {
            None
        };

        Ok(ClientContext {
            client: Arc::new(client),
            default_chain,
            genesis_config,
            send_timeout: options.send_timeout,
            recv_timeout: options.recv_timeout,
            retry_delay: options.retry_delay,
            max_retries: options.max_retries,
            max_backoff: options.max_backoff,
            chain_listeners: JoinSet::default(),
            #[cfg(not(web))]
            client_metrics,
        })
    }
}

impl<Env: Environment> ClientContext<Env> {
    // TODO(#5084) this (and other injected dependencies) should not be re-exposed by the
    // client interface
    /// Returns a reference to the wallet.
    pub fn wallet(&self) -> &Env::Wallet {
        self.client.wallet()
    }

    /// Returns the ID of the admin chain.
    pub fn admin_chain_id(&self) -> ChainId {
        self.client.admin_chain_id()
    }

    /// Retrieve the default account. Current this is the common account of the default
    /// chain.
    pub fn default_account(&self) -> Account {
        Account::chain(self.default_chain())
    }

    /// Retrieve the default chain.
    pub fn default_chain(&self) -> ChainId {
        self.default_chain
            .expect("default chain requested but none set")
    }

    /// Returns the lowest non-admin chain ID in the wallet.
    pub async fn first_non_admin_chain(&self) -> Result<ChainId, Error> {
        let admin_chain_id = self.admin_chain_id();
        let chain_ids = self
            .wallet()
            .chain_ids()
            .try_filter(|chain_id| futures::future::ready(*chain_id != admin_chain_id))
            .try_collect::<Vec<ChainId>>()
            .await
            .map_err(Error::wallet)?;
        Ok(chain_ids
            .into_iter()
            .min()
            .expect("No non-admin chain specified in wallet with no non-admin chain"))
    }

    /// Creates a node provider configured with this context's network options.
    // TODO(#5084) this should match the `NodeProvider` from the `Environment`
    pub fn make_node_provider(&self) -> NodeProvider {
        NodeProvider::new(self.make_node_options())
    }

    fn make_node_options(&self) -> NodeOptions {
        NodeOptions {
            send_timeout: self.send_timeout,
            recv_timeout: self.recv_timeout,
            retry_delay: self.retry_delay,
            max_retries: self.max_retries,
            max_backoff: self.max_backoff,
        }
    }

    /// Returns the client metrics, if metrics collection is enabled.
    #[cfg(not(web))]
    pub fn client_metrics(&self) -> Option<&ClientMetrics> {
        self.client_metrics.as_ref()
    }

    /// Updates the wallet entry for the client's chain from its current chain info.
    pub async fn update_wallet_from_client<Env_: Environment>(
        &self,
        chain_client: &ChainClient<Env_>,
    ) -> Result<(), Error> {
        let info = chain_client.chain_info().await?;
        let existing_owner = self
            .wallet()
            .get(info.chain_id)
            .await
            .map_err(error::Inner::wallet)?
            .and_then(|chain| chain.owner);

        // Only persist proposals that were made in the fast round: they need to be
        // remembered across sessions to make sure there are no conflicting fast proposals.
        let pending_proposal = chain_client
            .pending_proposal()
            .await
            .filter(|p| p.round.is_some_and(|r| r.is_fast()));
        self.wallet()
            .insert(
                info.chain_id,
                wallet::Chain {
                    pending_proposal,
                    owner: existing_owner,
                    ..info.as_ref().into()
                },
            )
            .await
            .map_err(error::Inner::wallet)?;

        Ok(())
    }

    /// Remembers the new chain and its owner (if any) in the wallet.
    pub async fn update_wallet_for_new_chain(
        &mut self,
        chain_id: ChainId,
        owner: Option<AccountOwner>,
        timestamp: Timestamp,
        epoch: Epoch,
    ) -> Result<(), Error> {
        self.wallet()
            .try_insert(
                chain_id,
                linera_core::wallet::Chain::new(owner, epoch, timestamp),
            )
            .await
            .map_err(error::Inner::wallet)?;
        Ok(())
    }

    /// Registers a chain from its description: initializes local storage, adds to
    /// wallet, and starts tracking it for cross-chain message delivery.
    pub async fn extend_with_chain(
        &mut self,
        description: ChainDescription,
        owner: Option<AccountOwner>,
    ) -> Result<(), Error> {
        let chain_id = description.id();
        self.client
            .storage_client()
            .create_chain(description.clone())
            .await?;
        self.wallet()
            .try_insert(
                chain_id,
                linera_core::wallet::Chain::new(
                    owner,
                    description.config().epoch,
                    description.timestamp(),
                ),
            )
            .await
            .map_err(error::Inner::wallet)?;
        self.client
            .extend_chain_mode(chain_id, ListeningMode::FullChain);
        Ok(())
    }

    /// Processes the chain's inbox, waiting for round timeouts, and updates the wallet.
    pub async fn process_inbox(
        &mut self,
        chain_client: &ChainClient<Env>,
    ) -> Result<Vec<ConfirmedBlockCertificate>, Error> {
        let mut certificates = Vec::new();
        // Try processing the inbox optimistically without waiting for validator notifications.
        let (new_certificates, maybe_timeout) = {
            chain_client.synchronize_from_validators().await?;
            let result = chain_client.process_inbox_without_prepare().await;
            self.update_wallet_from_client(chain_client).await?;
            result?
        };
        certificates.extend(new_certificates);
        if maybe_timeout.is_none() {
            return Ok(certificates);
        }

        // Start listening for notifications, so we learn about new rounds and blocks.
        let (listener, _listen_handle, mut notification_stream) = chain_client.listen().await?;
        self.chain_listeners.spawn_task(listener);

        loop {
            let (new_certificates, maybe_timeout) = {
                let result = chain_client.process_inbox().await;
                self.update_wallet_from_client(chain_client).await?;
                result?
            };
            certificates.extend(new_certificates);
            if let Some(timestamp) = maybe_timeout {
                util::wait_for_next_round(&mut notification_stream, timestamp).await
            } else {
                return Ok(certificates);
            }
        }
    }

    /// Assigns the given chain to the owner, tracking it and recording it in the wallet.
    pub async fn assign_new_chain_to_key(
        &mut self,
        chain_id: ChainId,
        owner: AccountOwner,
    ) -> Result<(), Error> {
        self.client
            .extend_chain_mode(chain_id, ListeningMode::FullChain);
        let chain_client = self.make_chain_client(chain_id).await?;

        // Ensure we have the chain description blob.
        chain_client.get_chain_description().await?;

        // Synchronize and get chain info.
        chain_client.synchronize_from_validators().await?;
        let info = chain_client.chain_info().await?;

        // Validate that the owner can propose on this chain (either as owner or via
        // open_multi_leader_rounds).
        if !info
            .manager
            .ownership
            .can_propose_in_multi_leader_round(&owner)
        {
            tracing::error!("Chain {chain_id} is not owned by {owner}.");
            return Err(error::Inner::ChainOwnership.into());
        }

        // Try to modify existing chain entry, setting the owner.
        let modified = self
            .wallet()
            .modify(chain_id, |chain| chain.owner = Some(owner))
            .await
            .map_err(error::Inner::wallet)?;
        // If the chain didn't exist, insert a new entry.
        if modified.is_none() {
            self.wallet()
                .insert(
                    chain_id,
                    wallet::Chain {
                        owner: Some(owner),
                        timestamp: info.timestamp,
                        epoch: Some(info.epoch),
                        ..Default::default()
                    },
                )
                .await
                .map_err(error::Inner::wallet)
                .context("assigning new chain")?;
        }
        Ok(())
    }

    /// Applies the given function to the chain client.
    ///
    /// Updates the wallet regardless of the outcome. As long as the function returns a round
    /// timeout, it will wait and retry.
    pub async fn apply_client_command<E, F, Fut, T>(
        &mut self,
        chain_client: &ChainClient<Env>,
        mut f: F,
    ) -> Result<T, Error>
    where
        F: FnMut(&ChainClient<Env>) -> Fut,
        Fut: Future<Output = Result<ClientOutcome<T>, E>>,
        Error: From<E>,
    {
        chain_client.prepare_chain().await?;
        // Try applying f optimistically without validator notifications. Return if committed.
        let result = f(chain_client).await;
        self.update_wallet_from_client(chain_client).await?;
        match result? {
            ClientOutcome::Committed(t) => return Ok(t),
            ClientOutcome::Conflict(certificate) => {
                return Err(chain_client::Error::Conflict(certificate.hash()).into());
            }
            ClientOutcome::WaitForTimeout(_) => {}
        }

        // Start listening for notifications, so we learn about new rounds and blocks.
        let (listener, _listen_handle, mut notification_stream) = chain_client.listen().await?;
        self.chain_listeners.spawn_task(listener);

        loop {
            // Try applying f. Return if committed.
            let result = f(chain_client).await;
            self.update_wallet_from_client(chain_client).await?;
            let timeout = match result? {
                ClientOutcome::Committed(t) => return Ok(t),
                ClientOutcome::Conflict(certificate) => {
                    return Err(chain_client::Error::Conflict(certificate.hash()).into());
                }
                ClientOutcome::WaitForTimeout(timeout) => timeout,
            };
            // Otherwise wait and try again in the next round.
            util::wait_for_next_round(&mut notification_stream, timeout).await;
        }
    }

    /// Returns the ownership configuration of the given chain.
    pub async fn ownership(&mut self, chain_id: Option<ChainId>) -> Result<ChainOwnership, Error> {
        let chain_id = chain_id.unwrap_or_else(|| self.default_chain());
        let chain_client = self.make_chain_client(chain_id).await?;
        let info = chain_client.chain_info().await?;
        Ok(info.manager.ownership)
    }

    /// Changes the ownership configuration of the given chain.
    pub async fn change_ownership(
        &mut self,
        chain_id: Option<ChainId>,
        ownership_config: ChainOwnershipConfig,
    ) -> Result<(), Error> {
        let chain_id = chain_id.unwrap_or_else(|| self.default_chain());
        let chain_client = self.make_chain_client(chain_id).await?;
        info!(
            ?ownership_config, %chain_id, preferred_owner=?chain_client.preferred_owner(),
            "Changing ownership of a chain"
        );
        let time_start = Instant::now();
        let mut ownership = chain_client.query_chain_ownership().await?;
        ownership_config.update(&mut ownership)?;

        if ownership.super_owners.is_empty() && ownership.owners.is_empty() {
            tracing::error!("At least one owner or super owner of the chain has to be set.");
            return Err(error::Inner::ChainOwnership.into());
        }

        let certificate = self
            .apply_client_command(&chain_client, |chain_client| {
                let ownership = ownership.clone();
                let chain_client = chain_client.clone();
                async move {
                    chain_client
                        .change_ownership(ownership)
                        .await
                        .map_err(Error::from)
                        .context("Failed to change ownership")
                }
            })
            .await?;
        let time_total = time_start.elapsed();
        info!("Operation confirmed after {} ms", time_total.as_millis());
        debug!("{:?}", certificate);
        Ok(())
    }

    /// Sets the preferred owner used to propose blocks on the given chain.
    pub async fn set_preferred_owner(
        &mut self,
        chain_id: Option<ChainId>,
        preferred_owner: AccountOwner,
    ) -> Result<(), Error> {
        let chain_id = chain_id.unwrap_or_else(|| self.default_chain());
        let mut chain_client = self.make_chain_client(chain_id).await?;
        let old_owner = chain_client.preferred_owner();
        info!(%chain_id, ?old_owner, %preferred_owner, "Changing preferred owner for chain");
        chain_client.set_preferred_owner(preferred_owner);
        self.update_wallet_from_client(&chain_client).await?;
        info!("New preferred owner set");
        Ok(())
    }

    /// Checks that the validator's version info is compatible with the local version.
    pub async fn check_compatible_version_info(
        &self,
        address: &str,
        node: &impl ValidatorNode,
    ) -> Result<VersionInfo, Error> {
        match node.get_version_info().await {
            Ok(version_info) if version_info.is_compatible_with(&linera_version::VERSION_INFO) => {
                debug!(
                    "Version information for validator {address}: {}",
                    version_info
                );
                Ok(version_info)
            }
            Ok(version_info) => Err(error::Inner::UnexpectedVersionInfo {
                remote: Box::new(version_info),
                local: Box::new(linera_version::VERSION_INFO.clone()),
            }
            .into()),
            Err(error) => Err(error::Inner::UnavailableVersionInfo {
                address: address.to_string(),
                error: Box::new(error),
            }
            .into()),
        }
    }

    /// Checks that the validator's network description matches the local genesis config.
    pub async fn check_matching_network_description(
        &self,
        address: &str,
        node: &impl ValidatorNode,
    ) -> Result<CryptoHash, Error> {
        let network_description = self.genesis_config.network_description();
        match node.get_network_description().await {
            Ok(description) => {
                if description == network_description {
                    Ok(description.genesis_config_hash)
                } else {
                    Err(error::Inner::UnexpectedNetworkDescription {
                        remote: Box::new(description),
                        local: Box::new(network_description),
                    }
                    .into())
                }
            }
            Err(error) => Err(error::Inner::UnavailableNetworkDescription {
                address: address.to_string(),
                error: Box::new(error),
            }
            .into()),
        }
    }

    /// Queries a validator for the given chain's info and verifies its signature.
    pub async fn check_validator_chain_info_response(
        &self,
        public_key: Option<&ValidatorPublicKey>,
        address: &str,
        node: &impl ValidatorNode,
        chain_id: ChainId,
    ) -> Result<ChainInfo, Error> {
        let query = ChainInfoQuery::new(chain_id).with_manager_values();
        match node.handle_chain_info_query(query).await {
            Ok(response) => {
                debug!(
                    "Validator {address} sees chain {chain_id} at block height {} and epoch {:?}",
                    response.info.next_block_height, response.info.epoch,
                );
                if let Some(public_key) = public_key {
                    if response.check(*public_key).is_ok() {
                        debug!("Signature for public key {public_key} is OK.");
                    } else {
                        return Err(error::Inner::InvalidSignature {
                            public_key: *public_key,
                        }
                        .into());
                    }
                } else {
                    warn!("Not checking signature as public key was not given");
                }
                Ok(*response.info)
            }
            Err(error) => Err(error::Inner::UnavailableChainInfo {
                address: address.to_string(),
                chain_id,
                error: Box::new(error),
            }
            .into()),
        }
    }

    /// Query a validator for version info, network description, and chain info.
    ///
    /// Returns a `ValidatorQueryResults` struct with the results of all three queries.
    pub async fn query_validator(
        &self,
        address: &str,
        node: &impl ValidatorNode,
        chain_id: ChainId,
        public_key: Option<&ValidatorPublicKey>,
    ) -> ValidatorQueryResults {
        let version_info = self.check_compatible_version_info(address, node).await;
        let genesis_config_hash = self.check_matching_network_description(address, node).await;
        let chain_info = self
            .check_validator_chain_info_response(public_key, address, node, chain_id)
            .await;

        ValidatorQueryResults {
            version_info,
            genesis_config_hash,
            chain_info,
        }
    }

    /// Query the local node for version info, network description, and chain info.
    ///
    /// Returns a `ValidatorQueryResults` struct with the local node's information.
    pub async fn query_local_node(
        &self,
        chain_id: ChainId,
    ) -> Result<ValidatorQueryResults, Error> {
        let version_info = Ok(linera_version::VERSION_INFO.clone());
        let genesis_config_hash = Ok(self
            .genesis_config
            .network_description()
            .genesis_config_hash);
        let chain_info = self
            .make_chain_client(chain_id)
            .await?
            .chain_info_with_manager_values()
            .await
            .map(|info| *info)
            .map_err(|e| e.into());

        Ok(ValidatorQueryResults {
            version_info,
            genesis_config_hash,
            chain_info,
        })
    }
}

#[cfg(feature = "fs")]
impl<Env: Environment> ClientContext<Env> {
    /// Publishes a module from its contract and service bytecode files.
    pub async fn publish_module(
        &mut self,
        chain_client: &ChainClient<Env>,
        contract: PathBuf,
        service: PathBuf,
        vm_runtime: VmRuntime,
    ) -> Result<ModuleId, Error> {
        let (blobs, module_id) = load_bytecode_blobs(&contract, &service, vm_runtime).await?;

        info!("Publishing module");
        let (module_id, _) = self
            .apply_client_command(chain_client, |chain_client| {
                let blobs = blobs.clone();
                let chain_client = chain_client.clone();
                async move {
                    chain_client
                        .publish_module_blobs(blobs, module_id)
                        .await
                        .context("Failed to publish module")
                }
            })
            .await?;

        info!("{}", "Module published successfully!");

        info!("Synchronizing client and processing inbox");
        self.process_inbox(chain_client).await?;
        Ok(module_id)
    }

    /// Publishes a data blob loaded from the given file.
    pub async fn publish_data_blob(
        &mut self,
        chain_client: &ChainClient<Env>,
        blob_path: PathBuf,
    ) -> Result<CryptoHash, Error> {
        info!("Loading data blob file");
        let blob_bytes = fs::read(&blob_path).map_err(|e| {
            std::io::Error::new(
                e.kind(),
                format!("failed to load data blob bytes from {blob_path:?}: {e}"),
            )
        })?;

        info!("Publishing data blob");
        self.apply_client_command(chain_client, |chain_client| {
            let blob_bytes = blob_bytes.clone();
            let chain_client = chain_client.clone();
            async move {
                chain_client
                    .publish_data_blob(blob_bytes)
                    .await
                    .context("Failed to publish data blob")
            }
        })
        .await?;

        info!("{}", "Data blob published successfully!");
        Ok(CryptoHash::new(&BlobContent::new_data(blob_bytes)))
    }

    // TODO(#2490): Consider removing or renaming this.
    /// Verifies that a data blob with the given hash is available.
    pub async fn read_data_blob(
        &mut self,
        chain_client: &ChainClient<Env>,
        hash: CryptoHash,
    ) -> Result<(), Error> {
        info!("Verifying data blob");
        self.apply_client_command(chain_client, |chain_client| {
            let chain_client = chain_client.clone();
            async move {
                chain_client
                    .read_data_blob(hash)
                    .await
                    .context("Failed to verify data blob")
            }
        })
        .await?;

        info!("{}", "Data blob verified successfully!");
        Ok(())
    }
}

#[cfg(all(feature = "fs", not(web)))]
impl<Env: Environment> ClientContext<Env> {
    /// Publishes a module along with the JSON-encoded `Formats` description loaded
    /// from `formats`. The module publication and the formats-registry write
    /// happen atomically in a single block.
    pub async fn publish_module_with_formats(
        &mut self,
        chain_client: &ChainClient<Env>,
        contract: PathBuf,
        service: PathBuf,
        vm_runtime: VmRuntime,
        formats: PathBuf,
        registry_application_id: ApplicationId,
    ) -> Result<ModuleId, Error> {
        let owner = chain_client
            .preferred_owner()
            .ok_or(error::Inner::ChainOwnership)?;
        let (blobs, formats_blob_bytes, module_id, registry_op_bytes) = self
            .prepare_bcs_publication(owner, &contract, &service, vm_runtime, &formats)
            .await?;

        // Publish the formats data blob in its own block first, so it is committed
        // before the registry `Write` operation asserts its existence.
        info!("Publishing the formats data blob");
        self.apply_client_command(chain_client, |chain_client| {
            let formats_blob_bytes = formats_blob_bytes.clone();
            let chain_client = chain_client.clone();
            async move {
                chain_client
                    .publish_data_blob(formats_blob_bytes)
                    .await
                    .context("Failed to publish the formats data blob")
            }
        })
        .await?;

        info!("Publishing module and registering its formats");
        self.apply_client_command(chain_client, |chain_client| {
            let blobs = blobs.clone();
            let registry_op_bytes = registry_op_bytes.clone();
            let chain_client = chain_client.clone();
            async move {
                chain_client
                    .execute_operations(
                        vec![
                            Operation::system(SystemOperation::PublishModule { module_id }),
                            Operation::User {
                                application_id: registry_application_id,
                                bytes: registry_op_bytes,
                            },
                        ],
                        blobs,
                    )
                    .await
                    .context("Failed to publish module and register formats")
            }
        })
        .await?;

        info!(
            "{}",
            "Module published and formats registered successfully!"
        );
        info!("Synchronizing client and processing inbox");
        self.process_inbox(chain_client).await?;
        Ok(module_id)
    }

    /// Publishes a module, registers its `Formats` description in the formats
    /// registry, and creates an application from that module — all atomically in
    /// a single block.
    #[allow(clippy::too_many_arguments)]
    pub async fn publish_bcs_application(
        &mut self,
        chain_client: &ChainClient<Env>,
        contract: PathBuf,
        service: PathBuf,
        vm_runtime: VmRuntime,
        formats: PathBuf,
        registry_application_id: ApplicationId,
        parameters: Vec<u8>,
        instantiation_argument: Vec<u8>,
        required_application_ids: Vec<ApplicationId>,
    ) -> Result<(ApplicationId, ModuleId), Error> {
        let owner = chain_client
            .preferred_owner()
            .ok_or(error::Inner::ChainOwnership)?;
        let (blobs, formats_blob_bytes, module_id, registry_op_bytes) = self
            .prepare_bcs_publication(owner, &contract, &service, vm_runtime, &formats)
            .await?;

        // Publish the formats data blob in its own block first, so it is committed
        // before the registry `Write` operation asserts its existence.
        info!("Publishing the formats data blob");
        self.apply_client_command(chain_client, |chain_client| {
            let formats_blob_bytes = formats_blob_bytes.clone();
            let chain_client = chain_client.clone();
            async move {
                chain_client
                    .publish_data_blob(formats_blob_bytes)
                    .await
                    .context("Failed to publish the formats data blob")
            }
        })
        .await?;

        info!("Publishing module, registering its formats and creating the application");
        let application_id = self
            .apply_client_command(chain_client, |chain_client| {
                let blobs = blobs.clone();
                let registry_op_bytes = registry_op_bytes.clone();
                let parameters = parameters.clone();
                let instantiation_argument = instantiation_argument.clone();
                let required_application_ids = required_application_ids.clone();
                let chain_client = chain_client.clone();
                async move {
                    let outcome: ClientOutcome<ConfirmedBlockCertificate> = chain_client
                        .execute_operations(
                            vec![
                                Operation::system(SystemOperation::PublishModule { module_id }),
                                Operation::User {
                                    application_id: registry_application_id,
                                    bytes: registry_op_bytes,
                                },
                                Operation::system(SystemOperation::CreateApplication {
                                    module_id,
                                    parameters,
                                    instantiation_argument,
                                    required_application_ids,
                                }),
                            ],
                            blobs,
                        )
                        .await?;
                    outcome.try_map(|certificate| {
                        let mut creation: Vec<_> = certificate
                            .block()
                            .created_blob_ids()
                            .into_iter()
                            .filter(|blob_id| blob_id.blob_type == BlobType::ApplicationDescription)
                            .collect();
                        if creation.len() != 1 {
                            return Err(chain_client::Error::InternalError(
                                "Unexpected number of application descriptions published",
                            ));
                        }
                        let blob_id = creation.pop().expect("checked length");
                        Ok(ApplicationId::new(blob_id.hash))
                    })
                }
            })
            .await?;

        info!(
            "{}",
            "Module published, formats registered and application created successfully!"
        );
        info!("Synchronizing client and processing inbox");
        self.process_inbox(chain_client).await?;
        Ok((application_id, module_id))
    }

    /// Loads the bytecode files and the SNAP file, building the bytecode blobs, the
    /// BCS-encoded formats data blob, and the formats-registry write operation
    /// authorized by `owner`. The formats blob is returned separately from the
    /// bytecode blobs because the caller publishes it in its own block first (so it
    /// is committed before the registry `Write` operation asserts its existence).
    async fn prepare_bcs_publication(
        &self,
        owner: AccountOwner,
        contract: &Path,
        service: &Path,
        vm_runtime: VmRuntime,
        formats: &Path,
    ) -> Result<
        (
            Vec<linera_base::data_types::Blob>,
            Vec<u8>,
            ModuleId,
            Vec<u8>,
        ),
        Error,
    > {
        let (blobs, module_id) = load_bytecode_blobs(contract, service, vm_runtime).await?;

        info!("Loading formats from {formats:?}");
        let parsed = read_formats_from_snap(formats)?;
        let formats_blob_bytes = bcs::to_bytes(&parsed)?;
        let formats_blob_hash = CryptoHash::new(&BlobContent::new_data(formats_blob_bytes.clone()));
        let registry_op = linera_sdk::abis::formats_registry::Operation::Write {
            owner,
            module_id,
            blob_hash: linera_base::identifiers::DataBlobHash(formats_blob_hash),
        };
        let registry_op_bytes = bcs::to_bytes(&registry_op)?;
        Ok((blobs, formats_blob_bytes, module_id, registry_op_bytes))
    }
}

/// Reads the contract and service Wasm bytecode files from disk and turns them
/// into the blobs needed for module publication. Shared between
/// [`ClientContext::publish_module`] and the formats-aware variants so the two
/// code paths can't drift on error messages or blob construction.
#[cfg(feature = "fs")]
async fn load_bytecode_blobs(
    contract: &Path,
    service: &Path,
    vm_runtime: VmRuntime,
) -> Result<(Vec<linera_base::data_types::Blob>, ModuleId), Error> {
    info!("Loading bytecode files");
    let contract_bytecode = Bytecode::load_from_file(contract).map_err(|e| {
        std::io::Error::new(
            e.kind(),
            format!("failed to load contract bytecode from {contract:?}: {e}"),
        )
    })?;
    let service_bytecode = Bytecode::load_from_file(service).map_err(|e| {
        std::io::Error::new(
            e.kind(),
            format!("failed to load service bytecode from {service:?}: {e}"),
        )
    })?;
    Ok(create_bytecode_blobs(contract_bytecode, service_bytecode, vm_runtime).await)
}

/// Parses the `Formats` description of an application from a SNAP file (the YAML
/// snapshot produced by the `format` test of the example applications). The body
/// between the `---` frontmatter delimiters is deserialized as
/// [`linera_sdk::formats::Formats`]. BCS-serializing the result yields the data-blob
/// payload the formats registry expects, so external tooling can produce a blob file
/// ready for `linera publish-data-blob` (see the `extract-formats` binary in the
/// `formats-registry` example).
#[cfg(all(feature = "fs", not(web)))]
pub fn read_formats_from_snap(path: &Path) -> Result<linera_sdk::formats::Formats, Error> {
    let content = fs::read_to_string(path).map_err(|e| {
        std::io::Error::new(e.kind(), format!("failed to read SNAP file {path:?}: {e}"))
    })?;
    let body = strip_snap_frontmatter(&content).ok_or_else(|| {
        std::io::Error::new(
            std::io::ErrorKind::InvalidData,
            format!("SNAP file {path:?} is missing the `---` frontmatter delimiters"),
        )
    })?;
    serde_yaml_08::from_str(body).map_err(|e| {
        std::io::Error::new(
            std::io::ErrorKind::InvalidData,
            format!("failed to parse SNAP body in {path:?} as Formats: {e}"),
        )
        .into()
    })
}

#[cfg(all(feature = "fs", not(web)))]
fn strip_snap_frontmatter(content: &str) -> Option<&str> {
    let rest = content.strip_prefix("---\n")?;
    let end = rest.find("\n---\n")?;
    Some(&rest[end + "\n---\n".len()..])
}

#[cfg(all(test, feature = "fs", not(web)))]
mod snap_loader_tests {
    use super::read_formats_from_snap;

    #[test]
    fn parses_fungible_snap() {
        let path = std::path::Path::new("../examples/fungible/tests/snapshots/format__format.snap");
        read_formats_from_snap(path).expect("fungible snap should parse");
    }

    #[test]
    fn parses_social_snap() {
        let path = std::path::Path::new("../examples/social/tests/snapshots/format__format.snap");
        read_formats_from_snap(path).expect("social snap should parse");
    }

    #[test]
    fn parses_counter_snap() {
        let path = std::path::Path::new("../examples/counter/tests/snapshots/format__format.snap");
        read_formats_from_snap(path).expect("counter snap should parse");
    }
}

#[cfg(not(web))]
impl<Env: Environment> ClientContext<Env> {
    /// Prepares the chains and fungible tokens needed to run a benchmark.
    pub async fn prepare_for_benchmark(
        &mut self,
        num_chains: usize,
        tokens_per_chain: Amount,
        fungible_application_id: Option<ApplicationId>,
        pub_keys: Vec<AccountPublicKey>,
        chains_config_path: Option<&Path>,
        close_chains: bool,
    ) -> Result<Vec<ChainClient<Env>>, Error> {
        let start = Instant::now();
        // Below all block proposals are supposed to succeed without retries, we
        // must make sure that all incoming payments have been accepted on-chain
        // and that no validator is missing user certificates.
        self.process_inboxes_and_force_validator_updates().await;
        info!(
            "Processed inboxes and forced validator updates in {} ms",
            start.elapsed().as_millis()
        );

        let start = Instant::now();
        let (benchmark_chains, chain_clients) = self
            .make_benchmark_chains(
                num_chains,
                tokens_per_chain,
                pub_keys,
                chains_config_path.is_some(),
                close_chains,
            )
            .await?;
        info!(
            "Got {} chains in {} ms",
            num_chains,
            start.elapsed().as_millis()
        );

        if let Some(id) = fungible_application_id {
            let start = Instant::now();
            self.supply_fungible_tokens(&benchmark_chains, id).await?;
            info!(
                "Supplied fungible tokens in {} ms",
                start.elapsed().as_millis()
            );
            // Need to process inboxes to make sure the chains receive the supplied tokens.
            let start = Instant::now();
            for chain_client in &chain_clients {
                chain_client.process_inbox().await?;
            }
            info!(
                "Processed inboxes after supplying fungible tokens in {} ms",
                start.elapsed().as_millis()
            );
        }

        let all_chains = Benchmark::<Env>::get_all_chains(chains_config_path, &benchmark_chains)?;
        let known_chain_ids: HashSet<_> = benchmark_chains.iter().map(|(id, _)| *id).collect();
        let unknown_chain_ids: Vec<_> = all_chains
            .iter()
            .filter(|id| !known_chain_ids.contains(id))
            .copied()
            .collect();
        if !unknown_chain_ids.is_empty() {
            // The current client won't have the blobs for the chains in the other wallets. Even
            // though it will eventually get those blobs, we're getting a head start here and
            // fetching those blobs in advance.
            for chain_id in &unknown_chain_ids {
                self.client.get_chain_description(*chain_id).await?;
            }
        }

        Ok(chain_clients)
    }

    /// Closes the benchmark chains, or processes their inboxes and updates the wallet.
    pub async fn wrap_up_benchmark(
        &mut self,
        chain_clients: Vec<ChainClient<Env>>,
        close_chains: bool,
        wrap_up_max_in_flight: usize,
    ) -> Result<(), Error> {
        if close_chains {
            info!("Closing chains...");
            let chain_ids: Vec<_> = chain_clients.iter().map(|c| c.chain_id()).collect();
            let stream = stream::iter(chain_clients)
                .map(|chain_client| async move {
                    Benchmark::<Env>::close_benchmark_chain(&chain_client).await?;
                    info!("Closed chain {:?}", chain_client.chain_id());
                    Ok::<(), BenchmarkError>(())
                })
                .buffer_unordered(wrap_up_max_in_flight);
            stream.try_collect::<Vec<_>>().await?;
            // Remove closed chains from wallet (the chain listener may have added them).
            for chain_id in chain_ids {
                if let Err(error) = self.wallet().remove(chain_id).await {
                    warn!(%chain_id, %error, "Failed to remove closed chain from wallet");
                }
            }
        } else {
            info!("Processing inbox for all chains...");
            let stream = stream::iter(chain_clients.clone())
                .map(|chain_client| async move {
                    chain_client.process_inbox().await?;
                    info!("Processed inbox for chain {:?}", chain_client.chain_id());
                    Ok::<(), chain_client::Error>(())
                })
                .buffer_unordered(wrap_up_max_in_flight);
            stream.try_collect::<Vec<_>>().await?;

            info!("Updating wallet from chain clients...");
            for chain_client in chain_clients {
                let info = chain_client.chain_info().await?;
                let client_owner = chain_client.preferred_owner();
                let pending_proposal = chain_client
                    .pending_proposal()
                    .await
                    .filter(|p| p.round.is_some_and(|r| r.is_fast()));
                self.wallet()
                    .insert(
                        info.chain_id,
                        wallet::Chain {
                            pending_proposal,
                            owner: client_owner,
                            ..info.as_ref().into()
                        },
                    )
                    .await
                    .map_err(error::Inner::wallet)?;
            }
        }

        Ok(())
    }

    async fn process_inboxes_and_force_validator_updates(&mut self) {
        let mut join_set = task::JoinSet::new();

        let chain_clients: Vec<_> = self
            .wallet()
            .owned_chain_ids()
            .map_err(|e| error::Inner::wallet(e).into())
            .and_then(|id| self.make_chain_client(id))
            .try_collect()
            .await
            .unwrap();

        for chain_client in chain_clients {
            join_set.spawn(async move {
                Self::process_inbox_without_updating_wallet(&chain_client)
                    .await
                    .expect("Processing inbox should not fail!");
                chain_client
            });
        }

        for chain_client in join_set.join_all().await {
            self.update_wallet_from_client(&chain_client).await.unwrap();
        }
    }

    async fn process_inbox_without_updating_wallet(
        chain_client: &ChainClient<Env>,
    ) -> Result<Vec<ConfirmedBlockCertificate>, Error> {
        // Try processing the inbox optimistically without waiting for validator notifications.
        chain_client.synchronize_from_validators().await?;
        let (certificates, maybe_timeout) = chain_client.process_inbox_without_prepare().await?;
        assert!(
            maybe_timeout.is_none(),
            "Should not timeout within benchmark!"
        );

        Ok(certificates)
    }

    /// Creates chains if necessary, and returns a map of exactly `num_chains` chain IDs
    /// with key pairs, as well as a map of the chain clients.
    ///
    /// If `close_chains` is true, chains are not looked up from or stored in the wallet,
    /// since they will be closed after the benchmark and shouldn't be reused.
    async fn make_benchmark_chains(
        &mut self,
        num_chains: usize,
        balance: Amount,
        pub_keys: Vec<AccountPublicKey>,
        wallet_only: bool,
        close_chains: bool,
    ) -> Result<(Vec<(ChainId, AccountOwner)>, Vec<ChainClient<Env>>), Error> {
        let mut chains_found_in_wallet = 0;
        let mut benchmark_chains = Vec::with_capacity(num_chains);
        let mut chain_clients = Vec::with_capacity(num_chains);
        let start = Instant::now();

        // When close_chains is true and we're creating our own chains (not wallet_only),
        // skip wallet lookup to avoid picking up existing chains that would then be closed.
        // When wallet_only is true, chains were pre-created by the parent process and must
        // be read from the wallet.
        if !close_chains || wallet_only {
            let mut owned_chain_ids = std::pin::pin!(self.wallet().owned_chain_ids());
            while let Some(chain_id) = owned_chain_ids.next().await {
                let chain_id = chain_id.map_err(error::Inner::wallet)?;
                if chains_found_in_wallet == num_chains {
                    break;
                }
                let chain_client = self.make_chain_client(chain_id).await?;
                let ownership = chain_client.chain_info().await?.manager.ownership;
                if !ownership.owners.is_empty() || ownership.super_owners.len() != 1 {
                    continue;
                }
                let owner = *ownership.super_owners.first().unwrap();
                chain_client.process_inbox().await?;
                benchmark_chains.push((chain_id, owner));
                chain_clients.push(chain_client);
                chains_found_in_wallet += 1;
            }
            info!(
                "Got {} chains from the wallet in {} ms",
                benchmark_chains.len(),
                start.elapsed().as_millis()
            );
        }

        let num_chains_to_create = num_chains - chains_found_in_wallet;

        let default_chain_client = self.make_chain_client(self.default_chain()).await?;

        if num_chains_to_create > 0 {
            if wallet_only {
                return Err(
                    error::Inner::Benchmark(BenchmarkError::NotEnoughChainsInWallet(
                        num_chains,
                        chains_found_in_wallet,
                    ))
                    .into(),
                );
            }
            let mut pub_keys_iter = pub_keys.into_iter().take(num_chains_to_create);
            let operations_per_block = 900; // Over this we seem to hit the block size limits.
            for i in (0..num_chains_to_create).step_by(operations_per_block) {
                let num_new_chains = operations_per_block.min(num_chains_to_create - i);
                // Each chain gets its own unique owner (previously all chains in a batch
                // shared one owner, which could cause conflicts during benchmarking).
                let owners: Vec<AccountOwner> = (&mut pub_keys_iter)
                    .take(num_new_chains)
                    .map(|pk| pk.into())
                    .collect();

                let certificate = Self::execute_open_chains_operations(
                    &default_chain_client,
                    balance,
                    owners.clone(),
                )
                .await?;
                info!("Block executed successfully");

                let block = certificate.block();
                for (i, owner) in owners.into_iter().enumerate() {
                    let chain_id = block.body.blobs[i]
                        .iter()
                        .find(|blob| blob.id().blob_type == BlobType::ChainDescription)
                        .map(|blob| ChainId(blob.id().hash))
                        .expect("failed to create a new chain");
                    self.client
                        .extend_chain_mode(chain_id, ListeningMode::FullChain);

                    let mut chain_client = self.client.create_chain_client(
                        chain_id,
                        None,
                        BlockHeight::ZERO,
                        &None,
                        Some(owner),
                        self.timing_sender(),
                    );
                    chain_client.set_preferred_owner(owner);
                    chain_client.process_inbox().await?;
                    benchmark_chains.push((chain_id, owner));
                    chain_clients.push(chain_client);
                }
            }

            info!(
                "Created {} chains in {} ms",
                num_chains_to_create,
                start.elapsed().as_millis()
            );
        }

        // Only update wallet if chains will be reused (not closed after benchmark)
        if !close_chains {
            info!("Updating wallet from client");
            self.update_wallet_from_client(&default_chain_client)
                .await?;
        }
        info!("Retrying pending outgoing messages");
        default_chain_client
            .retry_pending_outgoing_messages()
            .await
            .context("outgoing messages to create the new chains should be delivered")?;
        info!("Processing default chain inbox");
        default_chain_client.process_inbox().await?;

        Ok((benchmark_chains, chain_clients))
    }

    async fn execute_open_chains_operations(
        chain_client: &ChainClient<Env>,
        balance: Amount,
        owners: Vec<AccountOwner>,
    ) -> Result<ConfirmedBlockCertificate, Error> {
        let operations: Vec<_> = owners
            .iter()
            .map(|owner| {
                let config = OpenChainConfig {
                    ownership: ChainOwnership::single_super(*owner),
                    balance,
                    application_permissions: Default::default(),
                };
                Operation::system(SystemOperation::OpenChain(config))
            })
            .collect();
        info!("Executing {} OpenChain operations", operations.len());
        Ok(chain_client
            .execute_operations(operations, vec![])
            .await?
            .expect("should execute block with OpenChain operations"))
    }

    /// Supplies fungible tokens to the chains.
    async fn supply_fungible_tokens(
        &mut self,
        key_pairs: &[(ChainId, AccountOwner)],
        application_id: ApplicationId,
    ) -> Result<(), Error> {
        let default_chain_id = self.default_chain();
        let default_key = self
            .wallet()
            .get(default_chain_id)
            .await
            .unwrap()
            .unwrap()
            .owner
            .unwrap();
        // This should be enough to run the benchmark at 1M TPS for an hour.
        let amount = Amount::from_nanos(4);
        let operations: Vec<Operation> = key_pairs
            .iter()
            .map(|(chain_id, owner)| {
                fungible_transfer(application_id, *chain_id, default_key, *owner, amount)
            })
            .collect();
        let chain_client = self.make_chain_client(default_chain_id).await?;
        // Put at most 1000 fungible token operations in each block.
        for operation_chunk in operations.chunks(1000) {
            chain_client
                .execute_operations(operation_chunk.to_vec(), vec![])
                .await?
                .expect("should execute block with Transfer operations");
        }
        self.update_wallet_from_client(&chain_client).await?;

        Ok(())
    }
}