1use 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
34pub struct LocalNode<S>
36where
37 S: Storage,
38{
39 state: WorkerState<S>,
40}
41
42#[derive(Clone)]
44pub struct LocalNodeClient<S>
45where
46 S: Storage,
47{
48 node: Arc<LocalNode<S>>,
49}
50
51#[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 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 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 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 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 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 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 #[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 #[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 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
303pub trait LocalChainInfoExt {
306 fn into_committees(self) -> Result<BTreeMap<Epoch, Committee>, LocalNodeError>;
308
309 fn into_current_committee(self) -> Result<Committee, LocalNodeError>;
311
312 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}