linera_core/
local_node.rs

1// Copyright (c) Facebook, Inc. and its affiliates.
2// Copyright (c) Zefchain Labs, Inc.
3// SPDX-License-Identifier: Apache-2.0
4
5use std::{
6    collections::{BTreeMap, VecDeque},
7    sync::Arc,
8};
9
10use futures::{stream::FuturesUnordered, TryStreamExt as _};
11use linera_base::{
12    crypto::ValidatorPublicKey,
13    data_types::{ArithmeticError, Blob, BlockHeight, Epoch},
14    identifiers::{BlobId, ChainId},
15};
16use linera_chain::{
17    data_types::{BlockProposal, ProposedBlock},
18    types::{Block, GenericCertificate},
19    ChainStateView,
20};
21use linera_execution::{committee::Committee, BlobState, Query, QueryOutcome};
22use linera_storage::Storage;
23use linera_views::ViewError;
24use thiserror::Error;
25use tokio::sync::OwnedRwLockReadGuard;
26use tracing::{instrument, warn};
27
28use crate::{
29    data_types::{ChainInfo, ChainInfoQuery, ChainInfoResponse},
30    notifier::Notifier,
31    worker::{ProcessableCertificate, WorkerError, WorkerState},
32};
33
34/// A local node with a single worker, typically used by clients.
35pub struct LocalNode<S>
36where
37    S: Storage,
38{
39    state: WorkerState<S>,
40}
41
42/// A client to a local node.
43#[derive(Clone)]
44pub struct LocalNodeClient<S>
45where
46    S: Storage,
47{
48    node: Arc<LocalNode<S>>,
49}
50
51/// Error type for the operations on a local node.
52#[derive(Debug, Error)]
53pub enum LocalNodeError {
54    #[error(transparent)]
55    ArithmeticError(#[from] ArithmeticError),
56
57    #[error(transparent)]
58    ViewError(#[from] ViewError),
59
60    #[error("Worker operation failed: {0}")]
61    WorkerError(WorkerError),
62
63    #[error("The local node doesn't have an active chain {0}")]
64    InactiveChain(ChainId),
65
66    #[error("The chain info response received from the local node is invalid")]
67    InvalidChainInfoResponse,
68
69    #[error("Blobs not found: {0:?}")]
70    BlobsNotFound(Vec<BlobId>),
71}
72
73impl From<WorkerError> for LocalNodeError {
74    fn from(error: WorkerError) -> Self {
75        match error {
76            WorkerError::BlobsNotFound(blob_ids) => LocalNodeError::BlobsNotFound(blob_ids),
77            error => LocalNodeError::WorkerError(error),
78        }
79    }
80}
81
82impl<S> LocalNodeClient<S>
83where
84    S: Storage + Clone + Send + Sync + 'static,
85{
86    #[instrument(level = "trace", skip_all)]
87    pub async fn handle_block_proposal(
88        &self,
89        proposal: BlockProposal,
90    ) -> Result<ChainInfoResponse, LocalNodeError> {
91        // In local nodes, we can trust fully_handle_certificate to carry all actions eventually.
92        let (response, _actions) = self.node.state.handle_block_proposal(proposal).await?;
93        Ok(response)
94    }
95
96    #[instrument(level = "trace", skip_all)]
97    pub async fn handle_certificate<T>(
98        &self,
99        certificate: GenericCertificate<T>,
100        notifier: &impl Notifier,
101    ) -> Result<ChainInfoResponse, LocalNodeError>
102    where
103        T: ProcessableCertificate,
104    {
105        Ok(Box::pin(
106            self.node
107                .state
108                .fully_handle_certificate_with_notifications(certificate, notifier),
109        )
110        .await?)
111    }
112
113    #[instrument(level = "trace", skip_all)]
114    pub async fn handle_chain_info_query(
115        &self,
116        query: ChainInfoQuery,
117    ) -> Result<ChainInfoResponse, LocalNodeError> {
118        // In local nodes, we can trust fully_handle_certificate to carry all actions eventually.
119        let (response, _actions) = self.node.state.handle_chain_info_query(query).await?;
120        Ok(response)
121    }
122
123    #[instrument(level = "trace", skip_all)]
124    pub fn new(state: WorkerState<S>) -> Self {
125        Self {
126            node: Arc::new(LocalNode { state }),
127        }
128    }
129
130    #[instrument(level = "trace", skip_all)]
131    pub(crate) fn storage_client(&self) -> S {
132        self.node.state.storage_client().clone()
133    }
134
135    #[instrument(level = "trace", skip_all)]
136    pub async fn stage_block_execution(
137        &self,
138        block: ProposedBlock,
139        round: Option<u32>,
140        published_blobs: Vec<Blob>,
141    ) -> Result<(Block, ChainInfoResponse), LocalNodeError> {
142        Ok(self
143            .node
144            .state
145            .stage_block_execution(block, round, published_blobs)
146            .await?)
147    }
148
149    /// Reads blobs from storage.
150    pub async fn read_blobs_from_storage(
151        &self,
152        blob_ids: &[BlobId],
153    ) -> Result<Option<Vec<Blob>>, LocalNodeError> {
154        let storage = self.storage_client();
155        Ok(storage.read_blobs(blob_ids).await?.into_iter().collect())
156    }
157
158    /// Reads blob states from storage.
159    pub async fn read_blob_states_from_storage(
160        &self,
161        blob_ids: &[BlobId],
162    ) -> Result<Vec<BlobState>, LocalNodeError> {
163        let storage = self.storage_client();
164        let mut blobs_not_found = Vec::new();
165        let mut blob_states = Vec::new();
166        for (blob_state, blob_id) in storage
167            .read_blob_states(blob_ids)
168            .await?
169            .into_iter()
170            .zip(blob_ids)
171        {
172            match blob_state {
173                None => blobs_not_found.push(*blob_id),
174                Some(blob_state) => blob_states.push(blob_state),
175            }
176        }
177        if !blobs_not_found.is_empty() {
178            return Err(LocalNodeError::BlobsNotFound(blobs_not_found));
179        }
180        Ok(blob_states)
181    }
182
183    /// Looks for the specified blobs in the local chain manager's locking blobs.
184    /// Returns `Ok(None)` if any of the blobs is not found.
185    pub async fn get_locking_blobs(
186        &self,
187        blob_ids: impl IntoIterator<Item = &BlobId>,
188        chain_id: ChainId,
189    ) -> Result<Option<Vec<Blob>>, LocalNodeError> {
190        let chain = self.chain_state_view(chain_id).await?;
191        let mut blobs = Vec::new();
192        for blob_id in blob_ids {
193            match chain.manager.locking_blobs.get(blob_id).await? {
194                None => return Ok(None),
195                Some(blob) => blobs.push(blob),
196            }
197        }
198        Ok(Some(blobs))
199    }
200
201    /// Writes the given blobs to storage if there is an appropriate blob state.
202    pub async fn store_blobs(&self, blobs: &[Blob]) -> Result<(), LocalNodeError> {
203        let storage = self.storage_client();
204        storage.maybe_write_blobs(blobs).await?;
205        Ok(())
206    }
207
208    pub async fn handle_pending_blobs(
209        &self,
210        chain_id: ChainId,
211        blobs: Vec<Blob>,
212    ) -> Result<(), LocalNodeError> {
213        for blob in blobs {
214            self.node.state.handle_pending_blob(chain_id, blob).await?;
215        }
216        Ok(())
217    }
218
219    /// Returns a read-only view of the [`ChainStateView`] of a chain referenced by its
220    /// [`ChainId`].
221    ///
222    /// The returned view holds a lock on the chain state, which prevents the local node from
223    /// changing the state of that chain.
224    #[instrument(level = "trace", skip(self))]
225    pub async fn chain_state_view(
226        &self,
227        chain_id: ChainId,
228    ) -> Result<OwnedRwLockReadGuard<ChainStateView<S::Context>>, LocalNodeError> {
229        Ok(self.node.state.chain_state_view(chain_id).await?)
230    }
231
232    #[instrument(level = "trace", skip(self))]
233    pub(crate) async fn chain_info(
234        &self,
235        chain_id: ChainId,
236    ) -> Result<Box<ChainInfo>, LocalNodeError> {
237        let query = ChainInfoQuery::new(chain_id);
238        Ok(self.handle_chain_info_query(query).await?.info)
239    }
240
241    #[instrument(level = "trace", skip(self, query))]
242    pub async fn query_application(
243        &self,
244        chain_id: ChainId,
245        query: Query,
246    ) -> Result<QueryOutcome, LocalNodeError> {
247        let outcome = self.node.state.query_application(chain_id, query).await?;
248        Ok(outcome)
249    }
250
251    /// Handles any pending local cross-chain requests.
252    #[instrument(level = "trace", skip(self))]
253    pub async fn retry_pending_cross_chain_requests(
254        &self,
255        sender_chain: ChainId,
256    ) -> Result<(), LocalNodeError> {
257        let (_response, actions) = self
258            .node
259            .state
260            .handle_chain_info_query(ChainInfoQuery::new(sender_chain).with_network_actions())
261            .await?;
262        let mut requests = VecDeque::from_iter(actions.cross_chain_requests);
263        while let Some(request) = requests.pop_front() {
264            let new_actions = self.node.state.handle_cross_chain_request(request).await?;
265            requests.extend(new_actions.cross_chain_requests);
266        }
267        Ok(())
268    }
269
270    /// Given a list of chain IDs, returns a map that assigns to each of them the next block
271    /// height to schedule, i.e. the lowest block height for which we haven't added the messages
272    /// to `receiver_id` to the outbox yet.
273    pub async fn next_outbox_heights(
274        &self,
275        chain_ids: impl IntoIterator<Item = &ChainId>,
276        receiver_id: ChainId,
277    ) -> Result<BTreeMap<ChainId, BlockHeight>, LocalNodeError> {
278        let futures =
279            FuturesUnordered::from_iter(chain_ids.into_iter().map(|chain_id| async move {
280                let chain = self.chain_state_view(*chain_id).await?;
281                let mut next_height = chain.tip_state.get().next_block_height;
282                if let Some(outbox) = chain.outboxes.try_load_entry(&receiver_id).await? {
283                    next_height = next_height.max(*outbox.next_height_to_schedule.get());
284                }
285                Ok::<_, LocalNodeError>((*chain_id, next_height))
286            }));
287        futures.try_collect().await
288    }
289
290    pub async fn update_received_certificate_trackers(
291        &self,
292        chain_id: ChainId,
293        new_trackers: BTreeMap<ValidatorPublicKey, u64>,
294    ) -> Result<(), LocalNodeError> {
295        self.node
296            .state
297            .update_received_certificate_trackers(chain_id, new_trackers)
298            .await?;
299        Ok(())
300    }
301}
302
303/// Extension trait for [`ChainInfo`]s from our local node. These should always be valid and
304/// contain the requested information.
305pub trait LocalChainInfoExt {
306    /// Returns the requested map of committees.
307    fn into_committees(self) -> Result<BTreeMap<Epoch, Committee>, LocalNodeError>;
308
309    /// Returns the current committee.
310    fn into_current_committee(self) -> Result<Committee, LocalNodeError>;
311
312    /// Returns a reference to the current committee.
313    fn current_committee(&self) -> Result<&Committee, LocalNodeError>;
314}
315
316impl LocalChainInfoExt for ChainInfo {
317    fn into_committees(self) -> Result<BTreeMap<Epoch, Committee>, LocalNodeError> {
318        self.requested_committees
319            .ok_or(LocalNodeError::InvalidChainInfoResponse)
320    }
321
322    fn into_current_committee(self) -> Result<Committee, LocalNodeError> {
323        self.requested_committees
324            .ok_or(LocalNodeError::InvalidChainInfoResponse)?
325            .remove(&self.epoch)
326            .ok_or(LocalNodeError::InactiveChain(self.chain_id))
327    }
328
329    fn current_committee(&self) -> Result<&Committee, LocalNodeError> {
330        self.requested_committees
331            .as_ref()
332            .ok_or(LocalNodeError::InvalidChainInfoResponse)?
333            .get(&self.epoch)
334            .ok_or(LocalNodeError::InactiveChain(self.chain_id))
335    }
336}