linera_core/

local_node.rs

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

use std::{
    collections::{BTreeMap, HashMap, HashSet, VecDeque},
    sync::Arc,
};

use futures::{stream::FuturesUnordered, TryStreamExt as _};
use linera_base::{
    crypto::{CryptoHash, ValidatorPublicKey},
    data_types::{ArithmeticError, Blob, BlockHeight},
    identifiers::{BlobId, ChainId, EventId, StreamId},
};
use linera_chain::{
    data_types::{BlockProposal, BundleExecutionPolicy, ProposedBlock},
    types::{Block, ConfirmedBlockCertificate, GenericCertificate},
    ChainError, ChainExecutionContext, StreamCounts,
};
use linera_execution::{BlobState, ExecutionError, Query, QueryOutcome, ResourceTracker};
use linera_storage::{Arc as CacheArc, Storage};
use linera_views::ViewError;
use thiserror::Error;
use tracing::{instrument, warn};

use crate::{
    chain_worker::ProcessConfirmedBlockMode,
    data_types::{ChainInfo, ChainInfoQuery, ChainInfoResponse},
    notifier::Notifier,
    worker::{ProcessableCertificate, WorkerError, WorkerState},
};

/// A local node with a single worker, typically used by clients.
pub struct LocalNode<S>
where
    S: Storage,
{
    state: WorkerState<S>,
}

/// A client to a local node.
#[derive(Clone)]
pub struct LocalNodeClient<S>
where
    S: Storage,
{
    node: Arc<LocalNode<S>>,
}

/// Error type for the operations on a local node.
#[derive(Debug, Error, strum::IntoStaticStr)]
#[allow(missing_docs)]
pub enum LocalNodeError {
    #[error(transparent)]
    ArithmeticError(#[from] ArithmeticError),

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

    #[error("Worker operation failed: {0}")]
    WorkerError(WorkerError),

    #[error("The local node doesn't have an active chain {0}")]
    InactiveChain(ChainId),

    #[error("The chain info response received from the local node is invalid")]
    InvalidChainInfoResponse,

    #[error("Blobs not found: {0:?}")]
    BlobsNotFound(Vec<BlobId>),

    #[error("Blocks not found: {0:?}")]
    BlocksNotFound(Vec<CryptoHash>),

    #[error("Events not found: {0:?}")]
    EventsNotFound(Vec<EventId>),
}

impl LocalNodeError {
    /// Returns the qualified error variant name for the `error_type` metric label,
    /// delegating to [`WorkerError::error_type`] for wrapped worker errors.
    pub fn error_type(&self) -> String {
        match self {
            LocalNodeError::WorkerError(worker_error) => worker_error.error_type(),
            other => {
                let variant: &'static str = other.into();
                format!("LocalNodeError::{variant}")
            }
        }
    }
}

impl From<ExecutionError> for LocalNodeError {
    fn from(error: ExecutionError) -> Self {
        match error {
            ExecutionError::BlobsNotFound(blob_ids) => LocalNodeError::BlobsNotFound(blob_ids),
            ExecutionError::EventsNotFound(event_ids) => LocalNodeError::EventsNotFound(event_ids),
            ExecutionError::ViewError(view_error) => LocalNodeError::ViewError(view_error),
            error => LocalNodeError::WorkerError(WorkerError::from(ChainError::ExecutionError(
                Box::new(error),
                ChainExecutionContext::Block,
            ))),
        }
    }
}

impl From<WorkerError> for LocalNodeError {
    fn from(error: WorkerError) -> Self {
        match error {
            WorkerError::BlobsNotFound(blob_ids) => LocalNodeError::BlobsNotFound(blob_ids),
            WorkerError::BlocksNotFound(hashes) => LocalNodeError::BlocksNotFound(hashes),
            WorkerError::EventsNotFound(event_ids) => LocalNodeError::EventsNotFound(event_ids),
            error => LocalNodeError::WorkerError(error),
        }
    }
}

impl<S> LocalNodeClient<S>
where
    S: Storage + Clone + 'static,
{
    #[instrument(level = "trace", skip_all)]
    pub async fn handle_block_proposal(
        &self,
        proposal: BlockProposal,
    ) -> Result<ChainInfoResponse, LocalNodeError> {
        // In local nodes, cross-chain actions will be handled internally, so we discard them.
        let (response, _actions) = Box::pin(self.node.state.handle_block_proposal(proposal)).await;
        Ok(response?)
    }

    #[instrument(level = "trace", skip_all)]
    pub async fn handle_certificate<T>(
        &self,
        certificate: GenericCertificate<T>,
        notifier: &impl Notifier,
    ) -> Result<ChainInfoResponse, LocalNodeError>
    where
        T: ProcessableCertificate,
    {
        Ok(Box::pin(
            self.node
                .state
                .fully_handle_certificate_with_notifications(certificate, notifier),
        )
        .await?)
    }

    /// Same as [`Self::handle_certificate`] but for a confirmed block certificate
    /// and with an explicit [`ProcessConfirmedBlockMode`]. The generic variant
    /// always uses [`ProcessConfirmedBlockMode::Auto`].
    #[instrument(level = "trace", skip_all)]
    pub async fn handle_confirmed_certificate(
        &self,
        certificate: ConfirmedBlockCertificate,
        mode: ProcessConfirmedBlockMode,
        notifier: &impl Notifier,
    ) -> Result<ChainInfoResponse, LocalNodeError> {
        Ok(Box::pin(
            self.node
                .state
                .fully_handle_confirmed_certificate_with_notifications(certificate, mode, notifier),
        )
        .await?)
    }

    /// Test helper: returns the stored [`ConfirmedBlockCertificate`] for a
    /// chain's block at a given height (or [`WorkerError::BlocksNotFound`] if
    /// the height is referenced but the bytes aren't local).
    #[cfg(with_testing)]
    pub async fn read_certificate(
        &self,
        chain_id: linera_base::identifiers::ChainId,
        height: linera_base::data_types::BlockHeight,
    ) -> Result<Option<CacheArc<ConfirmedBlockCertificate>>, LocalNodeError> {
        Ok(self.node.state.read_certificate(chain_id, height).await?)
    }

    #[instrument(level = "trace", skip_all)]
    pub async fn handle_chain_info_query(
        &self,
        query: ChainInfoQuery,
    ) -> Result<ChainInfoResponse, LocalNodeError> {
        Ok(self.node.state.handle_chain_info_query(query).await?)
    }

    #[instrument(level = "trace", skip_all)]
    pub fn new(state: WorkerState<S>) -> Self {
        Self {
            node: Arc::new(LocalNode { state }),
        }
    }

    #[instrument(level = "trace", skip_all)]
    pub(crate) fn storage_client(&self) -> S {
        self.node.state.storage_client().clone()
    }

    /// Executes a block with a policy for handling bundle failures.
    ///
    /// Returns the modified block (bundles may be rejected/removed based on the policy),
    /// the executed block, chain info response, and resource tracker.
    #[instrument(level = "trace", skip_all)]
    pub async fn stage_block_execution(
        &self,
        block: ProposedBlock,
        round: Option<u32>,
        published_blobs: Vec<Blob>,
        policy: BundleExecutionPolicy,
    ) -> Result<
        (
            ProposedBlock,
            Block,
            ChainInfoResponse,
            ResourceTracker,
            HashSet<ChainId>,
        ),
        LocalNodeError,
    > {
        Ok(self
            .node
            .state
            .stage_block_execution(block, round, published_blobs, policy)
            .await?)
    }

    /// Reads blobs from storage.
    pub async fn read_blobs_from_storage(
        &self,
        blob_ids: &[BlobId],
    ) -> Result<Option<Vec<CacheArc<Blob>>>, LocalNodeError> {
        let storage = self.storage_client();
        Ok(storage.read_blobs(blob_ids).await?.into_iter().collect())
    }

    /// Reads blob states from storage.
    pub async fn read_blob_states_from_storage(
        &self,
        blob_ids: &[BlobId],
    ) -> Result<Vec<BlobState>, LocalNodeError> {
        let storage = self.storage_client();
        let mut blobs_not_found = Vec::new();
        let mut blob_states = Vec::new();
        for (blob_state, blob_id) in storage
            .read_blob_states(blob_ids)
            .await?
            .into_iter()
            .zip(blob_ids)
        {
            match blob_state {
                None => blobs_not_found.push(*blob_id),
                Some(blob_state) => blob_states.push(blob_state),
            }
        }
        if !blobs_not_found.is_empty() {
            return Err(LocalNodeError::BlobsNotFound(blobs_not_found));
        }
        Ok(blob_states)
    }

    /// Looks for the specified blobs in the local chain manager's locking blobs.
    /// Returns `Ok(None)` if any of the blobs is not found.
    pub async fn get_locking_blobs(
        &self,
        blob_ids: impl IntoIterator<Item = &BlobId>,
        chain_id: ChainId,
    ) -> Result<Option<Vec<Blob>>, LocalNodeError> {
        let blob_ids_vec: Vec<_> = blob_ids.into_iter().copied().collect();
        Ok(self
            .node
            .state
            .get_locking_blobs(chain_id, blob_ids_vec)
            .await?)
    }

    /// Writes the given blobs to storage if there is an appropriate blob state.
    pub async fn store_blobs(&self, blobs: &[Blob]) -> Result<(), LocalNodeError> {
        let storage = self.storage_client();
        storage.maybe_write_blobs(blobs).await?;
        Ok(())
    }

    pub async fn handle_pending_blobs(
        &self,
        chain_id: ChainId,
        blobs: Vec<Blob>,
    ) -> Result<(), LocalNodeError> {
        for blob in blobs {
            self.node.state.handle_pending_blob(chain_id, blob).await?;
        }
        Ok(())
    }

    /// Returns a read-only view of the [`ChainStateView`] of a chain referenced by its
    /// [`ChainId`].
    ///
    /// The returned view holds a lock on the chain state, which prevents the local node from
    /// changing the state of that chain.
    #[instrument(level = "trace", skip(self))]
    pub async fn chain_state_view(
        &self,
        chain_id: ChainId,
    ) -> Result<crate::worker::ChainStateViewReadGuard<S>, LocalNodeError> {
        Ok(self.node.state.chain_state_view(chain_id).await?)
    }

    #[instrument(level = "trace", skip(self))]
    pub(crate) async fn chain_info(
        &self,
        chain_id: ChainId,
    ) -> Result<Box<ChainInfo>, LocalNodeError> {
        let query = ChainInfoQuery::new(chain_id);
        Ok(self.handle_chain_info_query(query).await?.info)
    }

    #[instrument(level = "trace", skip(self, query))]
    pub async fn query_application(
        &self,
        chain_id: ChainId,
        query: Query,
        block_hash: Option<CryptoHash>,
    ) -> Result<(QueryOutcome, BlockHeight), LocalNodeError> {
        let result = self
            .node
            .state
            .query_application(chain_id, query, block_hash)
            .await?;
        Ok(result)
    }

    /// Handles any pending local cross-chain requests.
    ///
    /// Does not initialize the sender chain's execution state, so it is safe to
    /// call even when the sender's `ChainDescription` blob is not in local storage.
    /// Previously this went through `handle_chain_info_query`, which unconditionally
    /// initialized the worker and therefore forced a `ChainDescription` download on
    /// every call.
    #[instrument(level = "trace", skip(self, notifier))]
    pub async fn retry_pending_cross_chain_requests(
        &self,
        sender_chain: ChainId,
        notifier: &impl Notifier,
    ) -> Result<(), LocalNodeError> {
        let actions = self
            .node
            .state
            .cross_chain_network_actions(sender_chain)
            .await?;
        let mut requests = VecDeque::from_iter(actions.cross_chain_requests);
        while let Some(request) = requests.pop_front() {
            let new_actions = self.node.state.handle_cross_chain_request(request).await?;
            notifier.notify(&new_actions.notifications);
            requests.extend(new_actions.cross_chain_requests);
        }
        Ok(())
    }

    /// Given a list of chain IDs, returns a map that assigns to each of them the next block
    /// height to schedule, i.e. the lowest block height for which we haven't added the messages
    /// to `receiver_id` to the outbox yet.
    pub async fn next_outbox_heights(
        &self,
        chain_ids: impl IntoIterator<Item = &ChainId>,
        receiver_id: ChainId,
    ) -> Result<BTreeMap<ChainId, BlockHeight>, LocalNodeError> {
        let futures = chain_ids
            .into_iter()
            .map(|chain_id| async move {
                let (next_block_height, next_height_to_schedule) = match self
                    .get_tip_state_and_outbox_info(*chain_id, receiver_id)
                    .await
                {
                    Ok(info) => info,
                    Err(LocalNodeError::BlobsNotFound(_) | LocalNodeError::InactiveChain(_)) => {
                        return Ok((*chain_id, BlockHeight::ZERO))
                    }
                    Err(err) => Err(err)?,
                };
                let next_height = if let Some(scheduled_height) = next_height_to_schedule {
                    next_block_height.max(scheduled_height)
                } else {
                    next_block_height
                };
                Ok::<_, LocalNodeError>((*chain_id, next_height))
            })
            .collect::<FuturesUnordered<_>>();
        futures.try_collect().await
    }

    pub async fn update_received_certificate_trackers(
        &self,
        chain_id: ChainId,
        new_trackers: BTreeMap<ValidatorPublicKey, u64>,
    ) -> Result<(), LocalNodeError> {
        self.node
            .state
            .update_received_certificate_trackers(chain_id, new_trackers)
            .await?;
        Ok(())
    }

    pub async fn get_preprocessed_block_hashes(
        &self,
        chain_id: ChainId,
        start: BlockHeight,
        end: BlockHeight,
    ) -> Result<Vec<linera_base::crypto::CryptoHash>, LocalNodeError> {
        Ok(self
            .node
            .state
            .get_preprocessed_block_hashes(chain_id, start, end)
            .await?)
    }

    pub async fn get_inbox_next_height(
        &self,
        chain_id: ChainId,
        origin: ChainId,
    ) -> Result<BlockHeight, LocalNodeError> {
        Ok(self
            .node
            .state
            .get_inbox_next_height(chain_id, origin)
            .await?)
    }

    /// Gets block hashes for the given heights.
    pub async fn get_block_hashes(
        &self,
        chain_id: ChainId,
        heights: Vec<BlockHeight>,
    ) -> Result<Vec<CryptoHash>, LocalNodeError> {
        Ok(self.node.state.get_block_hashes(chain_id, heights).await?)
    }

    /// Gets proposed blobs from the manager for specified blob IDs.
    pub async fn get_proposed_blobs(
        &self,
        chain_id: ChainId,
        blob_ids: Vec<BlobId>,
    ) -> Result<Vec<Blob>, LocalNodeError> {
        Ok(self
            .node
            .state
            .get_proposed_blobs(chain_id, blob_ids)
            .await?)
    }

    /// Gets event subscriptions from the chain.
    pub async fn get_event_subscriptions(
        &self,
        chain_id: ChainId,
    ) -> Result<crate::worker::EventSubscriptionsResult, LocalNodeError> {
        Ok(self.node.state.get_event_subscriptions(chain_id).await?)
    }

    /// Gets a stream's [`StreamCounts`]: its next expected event index and its lowest readable
    /// index, read from a single chain state view so they are mutually consistent.
    pub async fn get_stream_indices(
        &self,
        chain_id: ChainId,
        stream_id: StreamId,
    ) -> Result<StreamCounts, LocalNodeError> {
        Ok(self
            .node
            .state
            .get_stream_indices(chain_id, stream_id)
            .await?)
    }

    /// Gets the `next_expected_events` indices for the given streams.
    pub async fn next_expected_events(
        &self,
        chain_id: ChainId,
        stream_ids: Vec<StreamId>,
    ) -> Result<BTreeMap<StreamId, u32>, LocalNodeError> {
        Ok(self
            .node
            .state
            .next_expected_events(chain_id, stream_ids)
            .await?)
    }

    /// Test helper: resets a chain and re-executes it from its latest checkpoint.
    #[cfg(with_testing)]
    pub async fn reset_and_reexecute_chain(
        &self,
        chain_id: ChainId,
    ) -> Result<Vec<crate::data_types::CrossChainRequest>, LocalNodeError> {
        Ok(self.node.state.reset_and_reexecute_chain(chain_id).await?)
    }

    /// Gets received certificate trackers.
    pub async fn get_received_certificate_trackers(
        &self,
        chain_id: ChainId,
    ) -> Result<HashMap<ValidatorPublicKey, u64>, LocalNodeError> {
        Ok(self
            .node
            .state
            .get_received_certificate_trackers(chain_id)
            .await?)
    }

    /// Gets tip state and outbox info for next_outbox_heights calculation.
    pub async fn get_tip_state_and_outbox_info(
        &self,
        chain_id: ChainId,
        receiver_id: ChainId,
    ) -> Result<(BlockHeight, Option<BlockHeight>), LocalNodeError> {
        Ok(self
            .node
            .state
            .get_tip_state_and_outbox_info(chain_id, receiver_id)
            .await?)
    }

    /// Gets the next height to preprocess.
    pub async fn get_next_height_to_preprocess(
        &self,
        chain_id: ChainId,
    ) -> Result<BlockHeight, LocalNodeError> {
        Ok(self
            .node
            .state
            .get_next_height_to_preprocess(chain_id)
            .await?)
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn error_type_delegates_to_worker_error() {
        assert_eq!(
            LocalNodeError::WorkerError(WorkerError::InvalidOwner).error_type(),
            "WorkerError::InvalidOwner"
        );
    }

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