linera_rpc/grpc/
relay.rs

1// Copyright (c) Zefchain Labs, Inc.
2// SPDX-License-Identifier: Apache-2.0
3
4//! Reaching other validators through this validator's own proxy.
5//!
6//! Shards hold the validator's secret key, so the proxy is the only component that should open
7//! outbound connections. A shard needing another validator — today, a chain worker exporting a
8//! block — sends its request to the proxy's existing internal port naming the intended validator,
9//! and the proxy performs it and returns the answer.
10//!
11//! [`RelayClient`] therefore implements [`ValidatorNode`] only for the operations block export
12//! uses; the rest are refused rather than silently doing something else. Retries are absent by
13//! design, since the export task already backs off per destination.
14
15use std::{
16    str::FromStr as _,
17    sync::{
18        atomic::{AtomicUsize, Ordering},
19        Arc,
20    },
21};
22
23use linera_base::{
24    crypto::CryptoHash,
25    data_types::{BlobContent, BlockHeight, NetworkDescription},
26    identifiers::{BlobId, ChainId, EventId},
27};
28use linera_chain::{data_types, types};
29use linera_core::node::{
30    BlobStream, CrossChainMessageDelivery, NodeError, NotificationStream, ValidatorNode,
31    ValidatorNodeProvider,
32};
33use linera_version::VersionInfo;
34use tonic::Request;
35use tracing::{debug, instrument, Level};
36
37use super::{
38    api::{self, validator_relay_client::ValidatorRelayClient},
39    pool::GrpcConnectionPool,
40    transport, GrpcError, GRPC_MAX_MESSAGE_SIZE,
41};
42use crate::{
43    config::ValidatorPublicNetworkConfig, node_provider::NodeOptions,
44    HandleConfirmedCertificateRequest, HandleLiteCertRequest,
45};
46
47/// Refuses an operation that the relay deliberately does not carry.
48fn unsupported(operation: &str) -> NodeError {
49    NodeError::GrpcError {
50        error: format!(
51            "{operation} is not available through the validator relay, which only carries the \
52             requests needed to export blocks"
53        ),
54    }
55}
56
57/// A validator reached through this validator's proxy.
58#[derive(Clone)]
59pub struct RelayClient {
60    /// The destination validator's address as the committee spells it, e.g. `grpc:host:port`.
61    /// Kept in the committee's own form so the proxy resolves it the way any other consumer
62    /// would.
63    destination: String,
64    /// The same validator as a URL, used only to name it in logs and metrics.
65    address: String,
66    client: ValidatorRelayClient<transport::Channel>,
67}
68
69impl RelayClient {
70    /// Turns the proxy's answer into the chain info the caller expects.
71    fn try_into_chain_info(
72        result: api::ChainInfoResult,
73    ) -> Result<linera_core::data_types::ChainInfoResponse, NodeError> {
74        let inner = result.inner.ok_or_else(|| NodeError::GrpcError {
75            error: "missing body from response".to_string(),
76        })?;
77        match inner {
78            api::chain_info_result::Inner::ChainInfoResponse(response) => {
79                Ok(response.try_into().map_err(|error| NodeError::GrpcError {
80                    error: format!("failed to unmarshal response: {error}"),
81                })?)
82            }
83            // bincode, matching what the validator used. Load-bearing: the recovery paths
84            // dispatch on the *variant* — `EventsNotFound` pushes the admin chain,
85            // `BlobsNotFound` uploads blobs — so a mangled error silently disables them.
86            api::chain_info_result::Inner::Error(error) => Err(bincode::deserialize(&error)
87                .map_err(|error| NodeError::GrpcError {
88                    error: format!("failed to unmarshal error message: {error}"),
89                })?),
90        }
91    }
92}
93
94impl ValidatorNode for RelayClient {
95    type NotificationStream = NotificationStream;
96
97    fn address(&self) -> String {
98        self.address.clone()
99    }
100
101    #[instrument(target = "relay_client", skip_all, err(level = Level::DEBUG), fields(destination = self.address))]
102    async fn handle_lite_certificate(
103        &self,
104        certificate: types::LiteCertificate<'_>,
105        delivery: CrossChainMessageDelivery,
106    ) -> Result<linera_core::data_types::ChainInfoResponse, NodeError> {
107        let inner = HandleLiteCertRequest {
108            certificate,
109            wait_for_outgoing_messages: delivery.wait_for_outgoing_messages(),
110        };
111        let request = api::RelayLiteCertificateRequest {
112            destination: self.destination.clone(),
113            inner: Some(inner.try_into()?),
114        };
115        debug!(handler = "relay_lite_certificate", "sending gRPC request");
116        let result = self
117            .client
118            .clone()
119            .relay_lite_certificate(Request::new(request))
120            .await?
121            .into_inner();
122        Self::try_into_chain_info(result)
123    }
124
125    #[instrument(target = "relay_client", skip_all, err(level = Level::DEBUG), fields(destination = self.address))]
126    async fn handle_confirmed_certificate(
127        &self,
128        certificate: linera_storage::Arc<types::ConfirmedBlockCertificate>,
129        delivery: CrossChainMessageDelivery,
130    ) -> Result<linera_core::data_types::ChainInfoResponse, NodeError> {
131        let inner = HandleConfirmedCertificateRequest {
132            certificate: linera_storage::Arc::unwrap_or_clone(certificate),
133            wait_for_outgoing_messages: delivery.wait_for_outgoing_messages(),
134        };
135        let request = api::RelayConfirmedCertificateRequest {
136            destination: self.destination.clone(),
137            inner: Some(inner.try_into()?),
138        };
139        debug!(
140            handler = "relay_confirmed_certificate",
141            "sending gRPC request"
142        );
143        let result = self
144            .client
145            .clone()
146            .relay_confirmed_certificate(Request::new(request))
147            .await?
148            .into_inner();
149        Self::try_into_chain_info(result)
150    }
151
152    #[instrument(target = "relay_client", skip_all, err(level = Level::DEBUG), fields(destination = self.address))]
153    async fn handle_chain_info_query(
154        &self,
155        query: linera_core::data_types::ChainInfoQuery,
156    ) -> Result<linera_core::data_types::ChainInfoResponse, NodeError> {
157        let request = api::RelayChainInfoQueryRequest {
158            destination: self.destination.clone(),
159            inner: Some(query.try_into()?),
160        };
161        debug!(handler = "relay_chain_info_query", "sending gRPC request");
162        let result = self
163            .client
164            .clone()
165            .relay_chain_info_query(Request::new(request))
166            .await?
167            .into_inner();
168        Self::try_into_chain_info(result)
169    }
170
171    #[instrument(target = "relay_client", skip(self), err(level = Level::DEBUG), fields(destination = self.address))]
172    async fn upload_blob(&self, content: BlobContent) -> Result<BlobId, NodeError> {
173        let request = api::RelayUploadBlobRequest {
174            destination: self.destination.clone(),
175            inner: Some(content.try_into()?),
176        };
177        debug!(handler = "relay_upload_blob", "sending gRPC request");
178        let blob_id = self
179            .client
180            .clone()
181            .relay_upload_blob(Request::new(request))
182            .await?
183            .into_inner();
184        Ok(blob_id.try_into()?)
185    }
186
187    async fn handle_block_proposal(
188        &self,
189        _proposal: data_types::BlockProposal,
190    ) -> Result<linera_core::data_types::ChainInfoResponse, NodeError> {
191        Err(unsupported("handle_block_proposal"))
192    }
193
194    async fn handle_validated_certificate(
195        &self,
196        _certificate: types::ValidatedBlockCertificate,
197    ) -> Result<linera_core::data_types::ChainInfoResponse, NodeError> {
198        Err(unsupported("handle_validated_certificate"))
199    }
200
201    async fn handle_timeout_certificate(
202        &self,
203        _certificate: types::GenericCertificate<types::Timeout>,
204    ) -> Result<linera_core::data_types::ChainInfoResponse, NodeError> {
205        Err(unsupported("handle_timeout_certificate"))
206    }
207
208    async fn get_version_info(&self) -> Result<VersionInfo, NodeError> {
209        Err(unsupported("get_version_info"))
210    }
211
212    async fn get_network_description(&self) -> Result<NetworkDescription, NodeError> {
213        Err(unsupported("get_network_description"))
214    }
215
216    async fn subscribe(
217        &self,
218        _chains: Vec<ChainId>,
219    ) -> Result<Self::NotificationStream, NodeError> {
220        Err(unsupported("subscribe"))
221    }
222
223    async fn download_blob(&self, _blob_id: BlobId) -> Result<BlobContent, NodeError> {
224        Err(unsupported("download_blob"))
225    }
226
227    async fn download_blobs(&self, _blob_ids: Vec<BlobId>) -> Result<BlobStream, NodeError> {
228        Err(unsupported("download_blobs"))
229    }
230
231    async fn download_pending_blob(
232        &self,
233        _chain_id: ChainId,
234        _blob_id: BlobId,
235    ) -> Result<BlobContent, NodeError> {
236        Err(unsupported("download_pending_blob"))
237    }
238
239    async fn handle_pending_blob(
240        &self,
241        _chain_id: ChainId,
242        _blob: BlobContent,
243    ) -> Result<linera_core::data_types::ChainInfoResponse, NodeError> {
244        Err(unsupported("handle_pending_blob"))
245    }
246
247    async fn download_certificate(
248        &self,
249        _hash: CryptoHash,
250    ) -> Result<types::ConfirmedBlockCertificate, NodeError> {
251        Err(unsupported("download_certificate"))
252    }
253
254    async fn download_certificates(
255        &self,
256        _hashes: Vec<CryptoHash>,
257    ) -> Result<Vec<types::ConfirmedBlockCertificate>, NodeError> {
258        Err(unsupported("download_certificates"))
259    }
260
261    async fn blob_last_used_by(&self, _blob_id: BlobId) -> Result<CryptoHash, NodeError> {
262        Err(unsupported("blob_last_used_by"))
263    }
264
265    async fn event_block_heights(
266        &self,
267        _event_ids: Vec<EventId>,
268    ) -> Result<Vec<Option<BlockHeight>>, NodeError> {
269        Err(unsupported("event_block_heights"))
270    }
271
272    async fn get_shard_info(
273        &self,
274        _chain_id: ChainId,
275    ) -> Result<linera_core::data_types::ShardInfo, NodeError> {
276        Err(unsupported("get_shard_info"))
277    }
278
279    async fn missing_blob_ids(&self, _blob_ids: Vec<BlobId>) -> Result<Vec<BlobId>, NodeError> {
280        Err(unsupported("missing_blob_ids"))
281    }
282
283    async fn blob_last_used_by_certificate(
284        &self,
285        _blob_id: BlobId,
286    ) -> Result<types::ConfirmedBlockCertificate, NodeError> {
287        Err(unsupported("blob_last_used_by_certificate"))
288    }
289
290    async fn download_certificates_by_heights(
291        &self,
292        _chain_id: ChainId,
293        _heights: Vec<BlockHeight>,
294    ) -> Result<Vec<types::ConfirmedBlockCertificate>, NodeError> {
295        Err(unsupported("download_certificates_by_heights"))
296    }
297}
298
299/// A node provider that reaches every validator through this validator's own proxies.
300#[derive(Clone)]
301pub struct RelayNodeProvider {
302    /// The internal addresses of this validator's proxies — the same ones the shards already
303    /// send notifications to.
304    relay_addresses: Vec<String>,
305    /// Round-robin cursor over `relay_addresses`. Each request goes through exactly one proxy,
306    /// but successive nodes draw different ones so egress is spread. Shared across clones, so
307    /// every chain worker in the process uses one rotation.
308    next: Arc<AtomicUsize>,
309    pool: GrpcConnectionPool,
310}
311
312impl RelayNodeProvider {
313    /// Creates a provider that relays through the given proxies, in rotation. Panics if
314    /// `relay_addresses` is empty: a shard must not open the connection itself.
315    pub fn new(relay_addresses: Vec<String>, options: NodeOptions) -> Self {
316        assert!(
317            !relay_addresses.is_empty(),
318            "relaying to other validators needs at least one proxy",
319        );
320        Self {
321            relay_addresses,
322            next: Arc::new(AtomicUsize::new(0)),
323            pool: GrpcConnectionPool::new(transport::Options::from(&options)),
324        }
325    }
326
327    /// Returns the next proxy in the rotation.
328    fn next_relay(&self) -> &str {
329        let index = self.next.fetch_add(1, Ordering::Relaxed) % self.relay_addresses.len();
330        &self.relay_addresses[index]
331    }
332}
333
334impl ValidatorNodeProvider for RelayNodeProvider {
335    type Node = RelayClient;
336
337    fn make_node(&self, address: &str) -> Result<Self::Node, NodeError> {
338        // Parsed even though the proxy is the one that dials it, so that a malformed committee
339        // entry is rejected here rather than on every request.
340        let network = ValidatorPublicNetworkConfig::from_str(address).map_err(|_| {
341            NodeError::CannotResolveValidatorAddress {
342                address: address.to_string(),
343            }
344        })?;
345        let relay = self.next_relay();
346        let channel = self
347            .pool
348            .channel(relay.to_owned())
349            .map_err(|error: GrpcError| NodeError::GrpcError {
350                error: format!("error creating channel to the relay {relay}: {error}"),
351            })?;
352        Ok(RelayClient {
353            destination: address.to_owned(),
354            address: network.http_address(),
355            client: ValidatorRelayClient::new(channel)
356                .max_encoding_message_size(GRPC_MAX_MESSAGE_SIZE)
357                .max_decoding_message_size(GRPC_MAX_MESSAGE_SIZE),
358        })
359    }
360}
linera_rpc/grpc/relay.rs

linera_rpc/grpc/
relay.rs
