linera-core 0.15.21

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

use std::{
    collections::{BTreeMap, BTreeSet, HashMap},
    fmt,
    hash::Hash,
    mem,
};

use futures::{future, Future, StreamExt};
use linera_base::{
    crypto::ValidatorPublicKey,
    data_types::{BlockHeight, Round, TimeDelta},
    ensure,
    identifiers::{BlobId, BlobType, ChainId, StreamId},
    time::{timer::timeout, Duration, Instant},
};
use linera_chain::{
    data_types::{BlockProposal, LiteVote},
    manager::LockingBlock,
    types::{ConfirmedBlock, GenericCertificate, ValidatedBlock, ValidatedBlockCertificate},
};
use linera_execution::{committee::Committee, system::EPOCH_STREAM_NAME};
use linera_storage::{Arc as CacheArc, Clock, ResultReadCertificates, Storage};
use thiserror::Error;
use tokio::sync::mpsc;
use tracing::{instrument, Level};

use crate::{
    client::chain_client,
    data_types::{ChainInfo, ChainInfoQuery},
    environment::Environment,
    local_node::LocalNodeClient,
    node::{CrossChainMessageDelivery, NodeError, ValidatorNode},
    remote_node::RemoteNode,
    LocalNodeError,
};

/// The default amount of time we wait for additional validators to contribute
/// to the result, as a fraction of how long it took to reach a quorum.
pub const DEFAULT_QUORUM_GRACE_PERIOD: f64 = 0.2;
/// The maximum timeout for requests to a stake-weighted quorum if no quorum is reached.
const MAX_TIMEOUT: Duration = Duration::from_secs(60 * 60 * 24); // 1 day.

/// A report of clock skew from a validator, sent before retrying due to `InvalidTimestamp`.
pub type ClockSkewReport = (ValidatorPublicKey, TimeDelta);

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

    use linera_base::prometheus_util::{
        exponential_bucket_latencies, register_histogram_vec, register_int_counter_vec,
    };
    use prometheus::{HistogramVec, IntCounterVec};

    /// Requests dispatched to each validator while communicating with a quorum.
    ///
    /// Incremented at dispatch rather than on completion, so the series exists even for a
    /// validator that never answers. Subtracting the responses below from this yields the
    /// share of requests a validator left unanswered.
    ///
    /// The `address` label carries the validator's gRPC URL so that dashboards and alerts can
    /// name a validator without an out-of-band lookup of its public key. It is functionally
    /// dependent on `validator`, so it adds no series.
    pub(super) static QUORUM_REQUESTS: LazyLock<IntCounterVec> = LazyLock::new(|| {
        register_int_counter_vec(
            "communicate_with_quorum_requests_total",
            "Requests dispatched to each validator while communicating with a quorum",
            &["validator", "address"],
        )
    });

    /// Responses received from each validator while communicating with a quorum.
    ///
    /// The `outcome` label is `before_quorum` when the response arrived while a quorum was
    /// still outstanding, and `after_quorum` when it only arrived during the grace period
    /// that follows a quorum being reached. A validator whose responses are consistently
    /// `after_quorum` is holding up nothing yet, but it is the one that will stall
    /// confirmation as soon as any other validator degrades.
    pub(super) static QUORUM_RESPONSES: LazyLock<IntCounterVec> = LazyLock::new(|| {
        register_int_counter_vec(
            "communicate_with_quorum_responses_total",
            "Responses from each validator, by whether a quorum had already been reached",
            &["validator", "address", "outcome"],
        )
    });

    /// Time each validator took to respond while communicating with a quorum.
    pub(super) static QUORUM_RESPONSE_TIME: LazyLock<HistogramVec> = LazyLock::new(|| {
        register_histogram_vec(
            "communicate_with_quorum_response_time_ms",
            "Time taken by each validator to respond while communicating with a quorum, \
             in milliseconds",
            &["validator", "address"],
            exponential_bucket_latencies(60_000.0),
        )
    });
}

/// Used for `communicate_chain_action`
#[derive(Clone)]
pub enum CommunicateAction {
    SubmitBlock {
        proposal: Box<BlockProposal>,
        blob_ids: Vec<BlobId>,
        /// Channel to report clock skew before sleeping, so the caller can aggregate reports.
        clock_skew_sender: mpsc::UnboundedSender<ClockSkewReport>,
    },
    FinalizeBlock {
        certificate: Box<ValidatedBlockCertificate>,
        delivery: CrossChainMessageDelivery,
    },
    RequestTimeout {
        chain_id: ChainId,
        height: BlockHeight,
        round: Round,
    },
}

impl CommunicateAction {
    /// The round to which this action pertains.
    pub fn round(&self) -> Round {
        match self {
            CommunicateAction::SubmitBlock { proposal, .. } => proposal.content.round,
            CommunicateAction::FinalizeBlock { certificate, .. } => certificate.round,
            CommunicateAction::RequestTimeout { round, .. } => *round,
        }
    }
}

/// Pushes data to a single validator to bring it up to date with the local node.
///
/// This deliberately holds a [`LocalNodeClient`] rather than a full client: updating another
/// node must not mutate the client's own state. When a validator turns out to be *ahead* of the
/// local node, the updater signals that with [`chain_client::Error::LocalNodeLagging`] and lets
/// the caller decide whether to pull the missing state.
pub struct RemoteNodeUpdater<Env>
where
    Env: Environment,
{
    pub remote_node: RemoteNode<Env::ValidatorNode>,
    pub local_node: LocalNodeClient<Env::Storage>,
    pub admin_chain_id: ChainId,
    pub certificate_upload_batch_size: u64,
}

impl<Env: Environment> Clone for RemoteNodeUpdater<Env> {
    fn clone(&self) -> Self {
        RemoteNodeUpdater {
            remote_node: self.remote_node.clone(),
            local_node: self.local_node.clone(),
            admin_chain_id: self.admin_chain_id,
            certificate_upload_batch_size: self.certificate_upload_batch_size,
        }
    }
}

/// An error result for requests to a stake-weighted quorum.
#[derive(Error, Debug)]
pub enum CommunicationError<E: fmt::Debug> {
    /// No consensus is possible since validators returned different possibilities
    /// for the next block
    #[error(
        "No error but failed to find a consensus block. Consensus threshold: {0}, Proposals: {1:?}"
    )]
    NoConsensus(u64, Vec<(u64, usize)>),
    /// A single error that was returned by a sufficient number of nodes to be trusted as
    /// valid.
    #[error("Failed to communicate with a quorum of validators: {0}")]
    Trusted(E),
    /// No single error reached the validity threshold so we're returning a sample of
    /// errors for debugging purposes, together with their weight.
    #[error("Failed to communicate with a quorum of validators:\n{:#?}", .0)]
    Sample(Vec<(E, u64)>),
}

/// Executes a sequence of actions in parallel for all validators.
///
/// Tries to stop early when a quorum is reached. If `quorum_grace_period` is specified, other
/// validators are given additional time to contribute to the result. The grace period is
/// calculated as a fraction (defaulting to `DEFAULT_QUORUM_GRACE_PERIOD`) of the time taken to
/// reach quorum.
pub async fn communicate_with_quorum<'a, A, V, K, F, R, G>(
    validator_clients: &'a [RemoteNode<A>],
    committee: &Committee,
    group_by: G,
    execute: F,
    // Grace period as a fraction of time taken to reach quorum.
    quorum_grace_period: f64,
) -> Result<(K, Vec<(ValidatorPublicKey, V)>), CommunicationError<NodeError>>
where
    A: ValidatorNode + Clone + 'static,
    F: Clone + Fn(RemoteNode<A>) -> R,
    R: Future<Output = Result<V, chain_client::Error>> + 'a,
    G: Fn(&V) -> K,
    K: Hash + PartialEq + Eq + Clone + 'static,
    V: 'static,
{
    let mut responses: futures::stream::FuturesUnordered<_> = validator_clients
        .iter()
        .filter_map(|remote_node| {
            if committee.weight(&remote_node.public_key) == 0 {
                // This should not happen but better prevent it because certificates
                // are not allowed to include votes with weight 0.
                return None;
            }
            let execute = execute.clone();
            let remote_node = remote_node.clone();
            #[cfg(with_metrics)]
            metrics::QUORUM_REQUESTS
                .with_label_values(&[&remote_node.public_key.to_string(), &remote_node.address()])
                .inc();
            Some(async move {
                let public_key = remote_node.public_key;
                #[cfg(with_metrics)]
                let address = remote_node.address();
                #[cfg(with_metrics)]
                let request_start = Instant::now();
                let result = execute(remote_node).await;
                #[cfg(with_metrics)]
                metrics::QUORUM_RESPONSE_TIME
                    .with_label_values(&[&public_key.to_string(), &address])
                    .observe(request_start.elapsed().as_secs_f64() * 1000.0);
                (public_key, result)
            })
        })
        .collect();
    let total_validators = responses.len();

    let start_time = Instant::now();
    tracing::debug!(total_validators, "starting communicate_with_quorum");
    let mut end_time: Option<Instant> = None;
    let mut remaining_votes = committee.total_votes();
    let mut highest_key_score = 0;
    let mut value_scores: HashMap<K, (u64, Vec<(ValidatorPublicKey, V)>)> = HashMap::new();
    let mut error_scores = HashMap::new();
    let mut responses_received = 0;
    #[cfg(with_metrics)]
    let addresses: HashMap<ValidatorPublicKey, String> = validator_clients
        .iter()
        .map(|remote_node| (remote_node.public_key, remote_node.address()))
        .collect();

    'vote_wait: while let Ok(Some((name, result))) = timeout(
        end_time.map_or(MAX_TIMEOUT, |t| t.saturating_duration_since(Instant::now())),
        responses.next(),
    )
    .await
    {
        responses_received += 1;
        remaining_votes -= committee.weight(&name);
        #[cfg(with_metrics)]
        metrics::QUORUM_RESPONSES
            .with_label_values(&[
                &name.to_string(),
                addresses.get(&name).map_or("", String::as_str),
                if end_time.is_none() {
                    "before_quorum"
                } else {
                    "after_quorum"
                },
            ])
            .inc();
        match result {
            Ok(value) => {
                let key = group_by(&value);
                let entry = value_scores.entry(key.clone()).or_insert((0, Vec::new()));
                entry.0 += committee.weight(&name);
                entry.1.push((name, value));
                highest_key_score = highest_key_score.max(entry.0);
            }
            Err(err) => {
                // TODO(#2857): Handle non-remote errors properly.
                let err = match err {
                    chain_client::Error::RemoteNodeError(err) => err,
                    err => NodeError::ResponseHandlingError {
                        error: err.to_string(),
                    },
                };
                let entry = error_scores.entry(err.clone()).or_insert(0);
                *entry += committee.weight(&name);
            }
        }
        // If it becomes clear that no key can reach a quorum, break early.
        if highest_key_score + remaining_votes < committee.quorum_threshold() {
            break 'vote_wait;
        }

        // If a key reaches a quorum, wait for the grace period to collect more values
        // or error information and then stop.
        if end_time.is_none() && highest_key_score >= committee.quorum_threshold() {
            let time_to_quorum = start_time.elapsed();
            let grace_duration = time_to_quorum.mul_f64(quorum_grace_period);
            end_time = Some(Instant::now() + grace_duration);
            tracing::debug!(
                time_to_quorum_ms = time_to_quorum.as_millis(),
                grace_period_ms = grace_duration.as_millis(),
                "quorum reached, setting grace period"
            );
        }
    }
    tracing::debug!(
        total_wait_ms = start_time.elapsed().as_millis(),
        responses_received,
        total_validators,
        "exiting communicate_with_quorum loop"
    );

    let scores = value_scores
        .values()
        .map(|(weight, values)| (*weight, values.len()))
        .collect();
    // If a key has a quorum, return it with its values.
    if let Some((key, (_, values))) = value_scores
        .into_iter()
        .find(|(_, (score, _))| *score >= committee.quorum_threshold())
    {
        return Ok((key, values));
    }

    let mut sample = error_scores.into_iter().collect::<Vec<_>>();
    sample.sort_by_key(|(_, score)| std::cmp::Reverse(*score));
    sample.truncate(4);
    Err(match sample.as_slice() {
        [] => CommunicationError::NoConsensus(committee.quorum_threshold(), scores),
        [(_, score), ..] if *score >= committee.validity_threshold() => {
            // At least one honest validator returned this error.
            CommunicationError::Trusted(sample.into_iter().next().unwrap().0)
        }
        // Otherwise no specific error is available to report reliably.}
        _ => CommunicationError::Sample(sample),
    })
}

impl<Env> RemoteNodeUpdater<Env>
where
    Env: Environment + 'static,
{
    /// Logs a warning if the error is not an expected part of the protocol flow.
    fn warn_if_unexpected(&self, err: &NodeError) {
        if !err.is_expected() {
            tracing::warn!(
                remote_node = self.remote_node.address(),
                %err,
                "unexpected error from validator",
            );
        }
    }

    #[instrument(
        level = "trace", skip_all, err(level = Level::DEBUG),
        fields(chain_id = %certificate.block().header.chain_id)
    )]
    async fn send_confirmed_certificate(
        &mut self,
        certificate: &CacheArc<GenericCertificate<ConfirmedBlock>>,
        delivery: CrossChainMessageDelivery,
    ) -> Result<Box<ChainInfo>, chain_client::Error> {
        let mut result = self
            .remote_node
            .handle_optimized_confirmed_certificate(certificate, delivery)
            .await;

        let mut sent_admin_chain = false;
        let mut sent_blobs = false;
        loop {
            match result {
                Err(NodeError::EventsNotFound(event_ids))
                    if !sent_admin_chain
                        && certificate.inner().chain_id() != self.admin_chain_id
                        && event_ids.iter().all(|event_id| {
                            event_id.stream_id == StreamId::system(EPOCH_STREAM_NAME)
                                && event_id.chain_id == self.admin_chain_id
                        }) =>
                {
                    // The validator doesn't have the committee that signed the certificate.
                    self.update_admin_chain().await?;
                    sent_admin_chain = true;
                }
                Err(NodeError::BlobsNotFound(blob_ids)) if !sent_blobs => {
                    // The validator is missing the blobs required by the certificate.
                    self.remote_node
                        .check_blobs_not_found(certificate, &blob_ids)?;
                    // The certificate is confirmed, so the blobs must be in storage.
                    let maybe_blobs = self.local_node.read_blobs_from_storage(&blob_ids).await?;
                    let blobs = maybe_blobs.ok_or(NodeError::BlobsNotFound(blob_ids))?;
                    self.remote_node
                        .node
                        .upload_blobs(blobs.into_iter().map(|b| b.into_std()).collect())
                        .await?;
                    sent_blobs = true;
                }
                result => {
                    if let Err(err) = &result {
                        self.warn_if_unexpected(err);
                    }
                    return Ok(result?);
                }
            }
            result = self
                .remote_node
                .handle_confirmed_certificate(certificate.clone(), delivery)
                .await;
        }
    }

    async fn send_validated_certificate(
        &mut self,
        certificate: GenericCertificate<ValidatedBlock>,
        delivery: CrossChainMessageDelivery,
    ) -> Result<Box<ChainInfo>, chain_client::Error> {
        let result = self
            .remote_node
            .handle_optimized_validated_certificate(&certificate, delivery)
            .await;

        let chain_id = certificate.inner().chain_id();
        match &result {
            Err(original_err @ NodeError::BlobsNotFound(blob_ids)) => {
                self.remote_node
                    .check_blobs_not_found(&certificate, blob_ids)?;
                // The certificate is for a validated block, i.e. for our locking block.
                // Take the missing blobs from our local chain manager.
                let blobs = self
                    .local_node
                    .get_locking_blobs(blob_ids, chain_id)
                    .await?
                    .ok_or_else(|| original_err.clone())?;
                self.remote_node.send_pending_blobs(chain_id, blobs).await?;
            }
            Err(error) => {
                self.sync_remote_if_needed(
                    chain_id,
                    certificate.round,
                    certificate.block().header.height,
                    error,
                )
                .await?;
            }
            _ => return Ok(result?),
        }
        let result = self
            .remote_node
            .handle_validated_certificate(certificate)
            .await;
        if let Err(err) = &result {
            self.warn_if_unexpected(err);
        }
        Ok(result?)
    }

    /// Requests a vote for a timeout certificate for the given round from the remote node.
    ///
    /// If the remote node is not in that round or at that height yet, sends the chain information
    /// to update it.
    async fn request_timeout(
        &mut self,
        chain_id: ChainId,
        round: Round,
        height: BlockHeight,
    ) -> Result<Box<ChainInfo>, chain_client::Error> {
        let query = ChainInfoQuery::new(chain_id).with_timeout(height, round);
        let result = self
            .remote_node
            .handle_chain_info_query(query.clone())
            .await;
        if let Err(err) = &result {
            self.sync_remote_if_needed(chain_id, round, height, err)
                .await?;
            self.warn_if_unexpected(err);
        }
        Ok(result?)
    }

    /// Sends chain information to the remote node if it is the one lagging behind.
    ///
    /// If the error reveals that the *local* node is behind instead, returns
    /// [`chain_client::Error::LocalNodeLagging`] carrying the original error: pulling remote
    /// state is a client-level decision, not something updating another node may do.
    async fn sync_remote_if_needed(
        &mut self,
        chain_id: ChainId,
        round: Round,
        height: BlockHeight,
        error: &NodeError,
    ) -> Result<(), chain_client::Error> {
        let address = &self.remote_node.address();
        match error {
            NodeError::WrongRound(validator_round) if *validator_round > round => {
                tracing::debug!(
                    address, %chain_id, %validator_round, %round,
                    "validator is at a higher round; local node needs to synchronize",
                );
                return Err(chain_client::Error::LocalNodeLagging {
                    chain_id,
                    error: Box::new(error.clone()),
                });
            }
            NodeError::UnexpectedBlockHeight {
                expected_block_height,
                found_block_height,
            } if expected_block_height > found_block_height => {
                tracing::debug!(
                    address,
                    %chain_id,
                    %expected_block_height,
                    %found_block_height,
                    "validator is at a higher height; local node needs to synchronize",
                );
                return Err(chain_client::Error::LocalNodeLagging {
                    chain_id,
                    error: Box::new(error.clone()),
                });
            }
            NodeError::WrongRound(validator_round) if *validator_round < round => {
                tracing::debug!(
                    address, %chain_id, %validator_round, %round,
                    "validator is at a lower round; sending chain info",
                );
                self.send_chain_information(
                    chain_id,
                    height,
                    CrossChainMessageDelivery::NonBlocking,
                    None,
                )
                .await?;
            }
            NodeError::UnexpectedBlockHeight {
                expected_block_height,
                found_block_height,
            } if expected_block_height < found_block_height => {
                tracing::debug!(
                    address,
                    %chain_id,
                    %expected_block_height,
                    %found_block_height,
                    "Validator is at a lower height; sending chain info.",
                );
                self.send_chain_information(
                    chain_id,
                    height,
                    CrossChainMessageDelivery::NonBlocking,
                    None,
                )
                .await?;
            }
            NodeError::InactiveChain(inactive_chain_id) => {
                tracing::debug!(
                    address,
                    chain_id = %inactive_chain_id,
                    "Validator has inactive chain; sending chain info.",
                );
                self.send_chain_information(
                    *inactive_chain_id,
                    height,
                    CrossChainMessageDelivery::NonBlocking,
                    None,
                )
                .await?;
            }
            _ => {}
        }
        Ok(())
    }

    async fn send_block_proposal(
        &mut self,
        proposal: Box<BlockProposal>,
        mut blob_ids: Vec<BlobId>,
        clock_skew_sender: mpsc::UnboundedSender<ClockSkewReport>,
    ) -> Result<Box<ChainInfo>, chain_client::Error> {
        let chain_id = proposal.content.block.chain_id;
        // `sent_cross_chain_updates` tracks per-origin progress for the legacy per-sender
        // `MissingCrossChainUpdate` path (a non-upgraded validator); `synced_cross_chain_updates`
        // is the one-shot guard for the aggregated `MissingCrossChainUpdates` path;
        // `synced_round_and_height` is the one-shot guard for the round/height mismatch path.
        let mut sent_cross_chain_updates = BTreeMap::new();
        let mut synced_cross_chain_updates = false;
        let mut synced_round_and_height = false;
        let mut publisher_chain_ids_sent = BTreeSet::new();
        let storage = self.local_node.storage_client();
        loop {
            let local_time = storage.clock().current_time();
            match self
                .remote_node
                .handle_block_proposal(proposal.clone())
                .await
            {
                Ok(info) => return Ok(info),
                Err(ref err) if err.parse_invalid_timestamp().is_some() => {
                    let invalid_ts = err.parse_invalid_timestamp().unwrap();
                    // The validator's clock is behind the block's timestamp. We need to
                    // wait for two things:
                    // 1. Our clock to reach block_timestamp (in case the block timestamp
                    //    is in the future from our perspective too).
                    // 2. The validator's clock to catch up (in case of clock skew between
                    //    us and the validator).
                    let clock_skew = local_time.delta_since(invalid_ts.validator_local_time);
                    tracing::debug!(
                        remote_node = self.remote_node.address(),
                        %chain_id,
                        block_timestamp = %invalid_ts.block_timestamp,
                        ?clock_skew,
                        "validator's clock is behind; waiting and retrying",
                    );
                    // Report the clock skew before sleeping so the caller can aggregate.
                    if clock_skew_sender
                        .send((self.remote_node.public_key, clock_skew))
                        .is_err()
                    {
                        tracing::debug!("clock skew receiver dropped; skipping report");
                    }
                    storage
                        .clock()
                        .sleep_until(invalid_ts.block_timestamp.saturating_add(clock_skew))
                        .await;
                }
                Err(err @ (NodeError::WrongRound(_) | NodeError::UnexpectedBlockHeight { .. }))
                    if !synced_round_and_height =>
                {
                    // The validator disagrees with the proposal's round or height. If it is
                    // behind, `sync_remote_if_needed` pushes the chain and we retry; if it is
                    // ahead, the `LocalNodeLagging` signal propagates so the caller can pull
                    // its state and rebuild the proposal. One-shot: if the validator still
                    // disagrees after a sync, retrying would not make progress.
                    synced_round_and_height = true;
                    tracing::debug!(
                        remote_node = self.remote_node.address(),
                        %chain_id,
                        %err,
                        "validator disagrees on round or height; synchronizing",
                    );
                    self.sync_remote_if_needed(
                        chain_id,
                        proposal.content.round,
                        proposal.content.block.height,
                        &err,
                    )
                    .await?;
                }
                // A validator that understands the aggregated error reports *every* missing
                // cross-chain bundle in a single `MissingCrossChainUpdates`, so we sync all of
                // them at once and retry. If it still reports missing bundles after we synced the
                // whole set, retrying would not make progress, so we surface the error instead of
                // looping.
                Err(NodeError::MissingCrossChainUpdates {
                    chain_id: dependencies_chain_id,
                    bundles,
                }) if dependencies_chain_id == proposal.content.block.chain_id => {
                    ensure!(
                        !synced_cross_chain_updates,
                        NodeError::ResponseHandlingError {
                            error: format!(
                                "validator still reports missing cross-chain updates for chain \
                                 {dependencies_chain_id} after they were all synced"
                            ),
                        }
                    );
                    synced_cross_chain_updates = true;
                    tracing::debug!(
                        remote_node = %self.remote_node.address(),
                        %chain_id,
                        bundles = bundles.len(),
                        "validator reported missing cross-chain updates; syncing them in one batch",
                    );
                    // Sync each reported origin chain up to the needed height, collapsing any
                    // duplicate origins to the highest height.
                    let mut origin_heights: BTreeMap<ChainId, BlockHeight> = BTreeMap::new();
                    for (origin, height) in bundles {
                        let target = height.try_add_one()?;
                        let entry = origin_heights.entry(origin).or_insert(target);
                        *entry = (*entry).max(target);
                    }
                    self.send_chain_info_up_to_heights(
                        origin_heights,
                        CrossChainMessageDelivery::Blocking,
                    )
                    .await?;
                }
                Err(NodeError::MissingCrossChainUpdate {
                    chain_id,
                    origin,
                    height,
                }) if chain_id == proposal.content.block.chain_id
                    && sent_cross_chain_updates
                        .get(&origin)
                        .is_none_or(|h| *h < height) =>
                {
                    tracing::debug!(
                        remote_node = %self.remote_node.address(),
                        chain_id = %origin,
                        "Missing cross-chain update; sending chain to validator.",
                    );
                    sent_cross_chain_updates.insert(origin, height);
                    // Some received certificates may be missing for this validator
                    // (e.g. to create the chain or make the balance sufficient) so we are going to
                    // synchronize them now and retry.
                    self.send_chain_information(
                        origin,
                        height.try_add_one()?,
                        CrossChainMessageDelivery::Blocking,
                        None,
                    )
                    .await?;
                }
                Err(NodeError::EventsNotFound(event_ids)) => {
                    let mut publisher_heights = BTreeMap::new();
                    let chain_ids = event_ids
                        .iter()
                        .map(|event_id| event_id.chain_id)
                        .filter(|chain_id| !publisher_chain_ids_sent.contains(chain_id))
                        .collect::<BTreeSet<_>>();
                    tracing::debug!(
                        remote_node = self.remote_node.address(),
                        ?chain_ids,
                        "missing events; sending chains to validator",
                    );
                    ensure!(!chain_ids.is_empty(), NodeError::EventsNotFound(event_ids));
                    for chain_id in chain_ids {
                        let height = self
                            .local_node
                            .get_next_height_to_preprocess(chain_id)
                            .await?;
                        publisher_heights.insert(chain_id, height);
                        publisher_chain_ids_sent.insert(chain_id);
                    }
                    self.send_chain_info_up_to_heights(
                        publisher_heights,
                        CrossChainMessageDelivery::NonBlocking,
                    )
                    .await?;
                }
                Err(error @ NodeError::ChainError { .. }) => {
                    // The validator rejected the proposal because of its local chain
                    // manager state — most commonly an incompatible confirmed vote tied
                    // to a locking block we don't yet have. The caller should pull
                    // manager values from this validator so the local node absorbs
                    // whatever justified the rejection; if the local state actually
                    // advances, `execute_operations` will rebuild and re-propose; if
                    // not, the error propagates as usual.
                    self.warn_if_unexpected(&error);
                    tracing::debug!(
                        remote_node = self.remote_node.address(),
                        %chain_id,
                        %error,
                        "validator rejected proposal; manager state needs to be pulled",
                    );
                    return Err(chain_client::Error::LocalNodeLagging {
                        chain_id,
                        error: Box::new(error),
                    });
                }
                Err(NodeError::BlobsNotFound(_) | NodeError::InactiveChain(_))
                    if !blob_ids.is_empty() =>
                {
                    tracing::debug!("Missing blobs");
                    // For `BlobsNotFound`, we assume that the local node should already be
                    // updated with the needed blobs, so sending the chain information about the
                    // certificates that last used the blobs to the validator node should be enough.
                    let published_blob_ids =
                        BTreeSet::from_iter(proposal.content.block.published_blob_ids());
                    blob_ids.retain(|blob_id| !published_blob_ids.contains(blob_id));
                    let published_blobs = self
                        .local_node
                        .get_proposed_blobs(chain_id, published_blob_ids.into_iter().collect())
                        .await?;
                    self.remote_node
                        .send_pending_blobs(chain_id, published_blobs)
                        .await?;
                    let missing_blob_ids = self
                        .remote_node
                        .node
                        .missing_blob_ids(mem::take(&mut blob_ids))
                        .await?;

                    tracing::debug!("Sending chains for missing blobs");
                    self.send_chain_info_for_blobs(
                        &missing_blob_ids,
                        CrossChainMessageDelivery::NonBlocking,
                    )
                    .await?;
                }
                // Fail immediately on other errors.
                Err(err) => {
                    self.warn_if_unexpected(&err);
                    return Err(err.into());
                }
            }
        }
    }

    async fn update_admin_chain(&mut self) -> Result<(), chain_client::Error> {
        let local_admin_info = self.local_node.chain_info(self.admin_chain_id).await?;
        Box::pin(self.send_chain_information(
            self.admin_chain_id,
            local_admin_info.next_block_height,
            CrossChainMessageDelivery::NonBlocking,
            None,
        ))
        .await
    }

    /// Sends chain information to bring a validator up to date with a specific chain.
    ///
    /// This method performs a two-phase synchronization:
    /// 1. **Height synchronization**: sends the certificates we hold locally for the range
    ///    `[validator_next_height, target_block_height)`, in order.
    /// 2. **Round synchronization**: If heights match, ensures the validator has proposals/certificates
    ///    for the current consensus round.
    ///
    /// Only certificates that are actually in our local storage are sent; heights we don't have are
    /// silently skipped (see [`Self::read_certificates_for_heights`]). This is deliberate and is what
    /// makes the "leave gaps on the validator side" behavior (#4181) work: a chain we merely *receive*
    /// from is stored only at its message-bearing heights, so we push exactly those. The validator
    /// executes the contiguous prefix and preprocesses any block that sits above a gap — enough to
    /// deliver that block's cross-chain bundles without our ever having to send the intervening
    /// non-message blocks (which we don't have anyway).
    ///
    /// Because our local storage is guaranteed to hold every block we needed to build a proposal (a
    /// bundle can only be consumed after its ordered message-bearing predecessors were downloaded),
    /// this is the reliable way to catch a validator up. Deriving the set to send from a
    /// `MissingCrossChainUpdates` error instead is *not* reliable: that error lists only the bundles
    /// the current proposal is missing and omits already-consumed ancestors, which the validator
    /// still needs executed before it can schedule a later gap block's bundle.
    ///
    /// # Height Sync Strategy
    /// - For existing chains (target_block_height > 0):
    ///   * Optimistically sends the last certificate first (often that's all that's missing).
    ///   * Falls back to a full chain query if the validator needs more context.
    ///   * Sends any additional locally-held certificates in order.
    /// - For new chains (target_block_height == 0):
    ///   * Sends the chain description and dependencies first.
    ///   * Then queries the validator's state.
    ///
    /// # Round Sync Strategy
    /// Once heights match, if the local node is at a higher round, sends the evidence
    /// (proposal, validated block, or timeout certificate) that proves the current round.
    ///
    /// # Parameters
    /// - `chain_id`: The chain to synchronize
    /// - `target_block_height`: The height the validator should reach
    /// - `delivery`: Message delivery mode (blocking or non-blocking)
    /// - `latest_certificate`: Optional certificate at target_block_height - 1 to avoid a storage lookup
    ///
    /// # Returns
    /// - `Ok(())` if synchronization completed successfully or the validator is already up to date
    /// - `Err` if there was a communication or storage error
    #[instrument(level = "debug", skip_all, fields(%chain_id))]
    pub async fn send_chain_information(
        &mut self,
        chain_id: ChainId,
        target_block_height: BlockHeight,
        delivery: CrossChainMessageDelivery,
        latest_certificate: Option<CacheArc<GenericCertificate<ConfirmedBlock>>>,
    ) -> Result<(), chain_client::Error> {
        // Phase 1: Height synchronization
        let info = if target_block_height.0 > 0 {
            self.sync_chain_height(chain_id, target_block_height, delivery, latest_certificate)
                .await?
        } else {
            self.initialize_new_chain_on_validator(chain_id).await?
        };

        // Phase 2: Round synchronization (if needed)
        // Height synchronization is complete. Now check if we need to synchronize
        // the consensus round at this height.
        let (remote_height, remote_round) = (info.next_block_height, info.manager.current_round);
        let query = ChainInfoQuery::new(chain_id).with_manager_values();
        let local_info = match self.local_node.handle_chain_info_query(query).await {
            Ok(response) => response.info,
            // If we don't have the full chain description locally, we can't help the
            // validator with round synchronization. This is not an error - the validator
            // should retry later once the chain is fully initialized locally.
            Err(LocalNodeError::BlobsNotFound(_)) => {
                tracing::debug!("local chain description not fully available, skipping round sync");
                return Ok(());
            }
            Err(error) => return Err(error.into()),
        };

        let manager = local_info.manager;
        if local_info.next_block_height != remote_height || manager.current_round <= remote_round {
            return Ok(());
        }

        // Validator is at our height but behind on consensus round
        self.sync_consensus_round(remote_round, &manager).await
    }

    /// Synchronizes a validator to a specific block height by sending the certificates we hold.
    ///
    /// Uses an optimistic approach: sends the last certificate first, then, based on the
    /// validator's reported height, sends the earlier certificates in the range. Only the heights
    /// we actually have in local storage are sent — any we're missing are silently skipped rather
    /// than treated as an error, which is what leaves genuine gaps on the validator (see
    /// [`Self::send_chain_information`] for why that is both safe and intended).
    ///
    /// Returns the [`ChainInfo`] from the validator after synchronization.
    async fn sync_chain_height(
        &mut self,
        chain_id: ChainId,
        target_block_height: BlockHeight,
        delivery: CrossChainMessageDelivery,
        latest_certificate: Option<CacheArc<GenericCertificate<ConfirmedBlock>>>,
    ) -> Result<Box<ChainInfo>, chain_client::Error> {
        let height = target_block_height.try_sub_one()?;

        // Get the certificate for the last block we want to send
        let certificate = if let Some(cert) = latest_certificate {
            cert
        } else {
            self.read_certificates_for_heights(chain_id, vec![height])
                .await?
                .into_iter()
                .next()
                .ok_or_else(|| {
                    chain_client::Error::InternalError(
                        "failed to read latest certificate for height sync",
                    )
                })?
        };

        // Optimistically try sending just the last certificate
        let info = match self
            .send_confirmed_certificate(&certificate, delivery)
            .await
        {
            Ok(info) => info,
            Err(error) => {
                tracing::debug!(
                    address = self.remote_node.address(), %error,
                    "validator failed to handle confirmed certificate; sending whole chain",
                );
                let query = ChainInfoQuery::new(chain_id);
                self.remote_node.handle_chain_info_query(query).await?
            }
        };

        // Calculate which block heights the validator is still missing
        let heights: Vec<_> = (info.next_block_height.0..target_block_height.0)
            .map(BlockHeight)
            .collect();

        if heights.is_empty() {
            return Ok(info);
        }

        let batch_size = self.certificate_upload_batch_size as usize;
        for chunk in heights.chunks(batch_size) {
            let certificates = self
                .read_certificates_for_heights(chain_id, chunk.to_vec())
                .await?;

            for certificate in certificates {
                self.send_confirmed_certificate(&certificate, delivery)
                    .await?;
            }
        }

        Ok(info)
    }

    /// Reads certificates for the given heights from local storage.
    ///
    /// First attempts to use the optimized `read_certificates_by_heights` method.
    /// Falls back to the traditional `get_block_hashes` + `read_certificates` approach
    /// if the direct height lookup doesn't return all certificates.
    ///
    /// When using the fallback, writes the discovered height->hash indices back to storage
    /// so subsequent lookups can use the optimized path.
    ///
    /// Heights we don't have are silently dropped: the returned vector contains only the
    /// certificates actually present, so callers naturally skip any block we never downloaded
    /// (e.g. a sender's non-message-bearing blocks). Callers must not assume the result covers
    /// every requested height.
    async fn read_certificates_for_heights(
        &self,
        chain_id: ChainId,
        heights: Vec<BlockHeight>,
    ) -> Result<Vec<CacheArc<GenericCertificate<ConfirmedBlock>>>, chain_client::Error> {
        let storage = self.local_node.storage_client();

        // First, try the direct height-based lookup
        let certificates_by_height = storage
            .read_certificates_by_heights(chain_id, &heights)
            .await?;

        // Check if we got all certificates (no None values)
        let all_found = certificates_by_height.len() == heights.len()
            && certificates_by_height.iter().all(|c| c.is_some());

        if all_found {
            return Ok(certificates_by_height.into_iter().flatten().collect());
        }

        // Fallback to the traditional approach
        let hashes = self
            .local_node
            .get_block_hashes(chain_id, heights.clone())
            .await?;

        let certificates = storage.read_certificates(&hashes).await?;

        match ResultReadCertificates::new(certificates, hashes.clone()) {
            ResultReadCertificates::Certificates(certs) => {
                // Write back the height->hash indices we learned from the fallback
                let indices: Vec<_> = heights.into_iter().zip(hashes.clone()).collect();
                storage
                    .write_certificate_height_indices(chain_id, &indices)
                    .await?;
                Ok(certs
                    .into_iter()
                    .map(|c| storage.cache_certificate(c))
                    .collect())
            }
            ResultReadCertificates::InvalidHashes(hashes) => {
                Err(chain_client::Error::ReadCertificatesError(hashes))
            }
        }
    }

    /// Initializes a new chain on the validator by sending the chain description and dependencies.
    ///
    /// This is called when the validator doesn't know about the chain yet.
    ///
    /// Returns the [`ChainInfo`] from the validator after initialization.
    async fn initialize_new_chain_on_validator(
        &self,
        chain_id: ChainId,
    ) -> Result<Box<ChainInfo>, chain_client::Error> {
        // Send chain description and all dependency chains
        self.send_chain_info_for_blobs(
            &[BlobId::new(chain_id.0, BlobType::ChainDescription)],
            CrossChainMessageDelivery::NonBlocking,
        )
        .await?;

        // Query the validator's state for this chain
        let query = ChainInfoQuery::new(chain_id);
        let info = self.remote_node.handle_chain_info_query(query).await?;
        Ok(info)
    }

    /// Synchronizes the consensus round state with the validator.
    ///
    /// If the validator is at the same height but an earlier round, sends the evidence
    /// (proposal, validated block, or timeout certificate) that justifies the current round.
    ///
    /// This is a best-effort operation - failures are logged but don't fail the entire sync.
    async fn sync_consensus_round(
        &self,
        remote_round: Round,
        manager: &linera_chain::manager::ChainManagerInfo,
    ) -> Result<(), chain_client::Error> {
        let target_round = manager.current_round;

        // First, push the locking certificate if it justifies our current round. A
        // locking block from an earlier round is not enough on its own to advance the
        // remote: the remote may still be ahead via a timeout or signed proposal, and
        // pushing a stale lock would not move them. Push only the current-round lock.
        if let Some(LockingBlock::Regular(validated)) = manager.requested_locking.as_deref() {
            if validated.round == target_round {
                match self
                    .remote_node
                    .handle_optimized_validated_certificate(
                        validated,
                        CrossChainMessageDelivery::NonBlocking,
                    )
                    .await
                {
                    Ok(info) => {
                        tracing::debug!("successfully sent validated block for round sync");
                        if info.manager.current_round >= target_round {
                            return Ok(());
                        }
                    }
                    Err(error) => {
                        tracing::debug!(%error, "failed to send validated block");
                    }
                }
            }
        }

        // Try to send a timeout certificate. The remote applies `next_round(cert.round)`
        // to its current round, which (for the cert we hold) lands at our current round.
        if let Some(cert) = &manager.timeout {
            if cert.round >= remote_round {
                match self
                    .remote_node
                    .handle_timeout_certificate(cert.as_ref().clone())
                    .await
                {
                    Ok(info) => {
                        tracing::debug!(round = %cert.round, "successfully sent timeout certificate");
                        if info.manager.current_round >= target_round {
                            return Ok(());
                        }
                    }
                    Err(error) => {
                        tracing::debug!(%error, round = %cert.round, "failed to send timeout certificate");
                    }
                }
            }
        }

        // Finally, try to push a proposal at the current round.
        for proposal in manager
            .requested_proposed
            .iter()
            .chain(manager.requested_signed_proposal.iter())
        {
            if proposal.content.round == target_round {
                match self
                    .remote_node
                    .handle_block_proposal(proposal.clone())
                    .await
                {
                    Ok(info) => {
                        tracing::debug!("successfully sent block proposal for round sync");
                        if info.manager.current_round >= target_round {
                            return Ok(());
                        }
                    }
                    Err(error) => {
                        tracing::debug!(%error, "failed to send block proposal");
                    }
                }
            }
        }

        // If we reach here, either we had no round sync data to send, or all attempts failed.
        // This is not a fatal error - height sync succeeded which is the primary goal.
        tracing::debug!("round sync not performed: no applicable data or all attempts failed");
        Ok(())
    }

    /// Sends chain information for all chains referenced by the given blobs.
    ///
    /// Reads blob states from storage, determines the specific chain heights needed,
    /// and sends chain information for those heights. With sparse chains, this only
    /// sends the specific blocks containing the blobs, not all blocks up to those heights.
    async fn send_chain_info_for_blobs(
        &self,
        blob_ids: &[BlobId],
        delivery: CrossChainMessageDelivery,
    ) -> Result<(), chain_client::Error> {
        let blob_states = self
            .local_node
            .read_blob_states_from_storage(blob_ids)
            .await?;

        let mut chain_heights: BTreeMap<ChainId, BTreeSet<BlockHeight>> = BTreeMap::new();
        for blob_state in blob_states {
            let block_chain_id = blob_state.chain_id;
            let block_height = blob_state.block_height;
            chain_heights
                .entry(block_chain_id)
                .or_default()
                .insert(block_height);
        }

        self.send_chain_info_at_heights(chain_heights, delivery)
            .await
    }

    /// Sends the blocks at exactly the specified heights on multiple chains.
    ///
    /// Unlike [`Self::send_chain_info_up_to_heights`], this sends *only* the blocks at the given
    /// heights, not the locally-held prefix leading up to them. Use it only when the required
    /// blocks are fully self-describing to the validator — e.g. bringing over the specific blocks
    /// that carry a set of blobs.
    async fn send_chain_info_at_heights(
        &self,
        chain_heights: impl IntoIterator<Item = (ChainId, BTreeSet<BlockHeight>)>,
        delivery: CrossChainMessageDelivery,
    ) -> Result<(), chain_client::Error> {
        future::try_join_all(chain_heights.into_iter().map(|(chain_id, heights)| {
            let mut updater = self.clone();
            async move {
                // Get all block hashes for this chain at the specified heights in one call
                let heights_vec = heights.into_iter().collect::<Vec<_>>();
                let certificates = updater
                    .local_node
                    .storage_client()
                    .read_certificates_by_heights(chain_id, &heights_vec)
                    .await?
                    .into_iter()
                    .flatten()
                    .collect::<Vec<_>>();

                // Send each certificate
                for certificate in certificates {
                    updater
                        .send_confirmed_certificate(&certificate, delivery)
                        .await?;
                }

                Ok::<_, chain_client::Error>(())
            }
        }))
        .await?;
        Ok(())
    }

    /// Brings a validator up to each given `(chain, height)` by pushing the locally-held prefix
    /// of that chain (via [`Self::send_chain_information`]), for all chains concurrently.
    async fn send_chain_info_up_to_heights(
        &self,
        chain_heights: impl IntoIterator<Item = (ChainId, BlockHeight)>,
        delivery: CrossChainMessageDelivery,
    ) -> Result<(), chain_client::Error> {
        future::try_join_all(chain_heights.into_iter().map(|(chain_id, height)| {
            let mut updater = self.clone();
            async move {
                updater
                    .send_chain_information(chain_id, height, delivery, None)
                    .await
            }
        }))
        .await?;
        Ok(())
    }

    pub async fn send_chain_update(
        &mut self,
        action: CommunicateAction,
    ) -> Result<LiteVote, chain_client::Error> {
        let chain_id = match &action {
            CommunicateAction::SubmitBlock { proposal, .. } => proposal.content.block.chain_id,
            CommunicateAction::FinalizeBlock { certificate, .. } => {
                certificate.inner().block().header.chain_id
            }
            CommunicateAction::RequestTimeout { chain_id, .. } => *chain_id,
        };
        // Send the block proposal, certificate or timeout request and return a vote.
        let vote = match action {
            CommunicateAction::SubmitBlock {
                proposal,
                blob_ids,
                clock_skew_sender,
            } => {
                let info = self
                    .send_block_proposal(proposal, blob_ids, clock_skew_sender)
                    .await?;
                info.manager.pending.ok_or_else(|| {
                    NodeError::MissingVoteInValidatorResponse("submit a block proposal".into())
                })?
            }
            CommunicateAction::FinalizeBlock {
                certificate,
                delivery,
            } => {
                let info = self
                    .send_validated_certificate(*certificate, delivery)
                    .await?;
                info.manager.pending.ok_or_else(|| {
                    NodeError::MissingVoteInValidatorResponse("finalize a block".into())
                })?
            }
            CommunicateAction::RequestTimeout { round, height, .. } => {
                let info = self.request_timeout(chain_id, round, height).await?;
                info.manager.timeout_vote.ok_or_else(|| {
                    NodeError::MissingVoteInValidatorResponse("request a timeout".into())
                })?
            }
        };
        vote.check(self.remote_node.public_key)?;
        Ok(vote)
    }
}