#![allow(unknown_lints)]
use std::{
fmt::Debug,
marker::PhantomData,
net::SocketAddr,
sync::Arc,
task::{Context, Poll},
time::Duration,
};
use anyhow::Result;
use async_trait::async_trait;
use futures::{future::BoxFuture, FutureExt as _};
use linera_base::{data_types::BlockHeight, identifiers::ChainId};
use linera_chain::types::{ConfirmedBlock, LiteCertificate as ChainLiteCertificate};
use linera_core::{
data_types::{CertificatesByHeightRequest, ChainInfo, ChainInfoQuery},
node::NodeError,
notifier::ChannelNotifier,
JoinSetExt as _,
};
#[cfg(with_metrics)]
use linera_metrics::monitoring_server;
#[cfg(all(with_metrics, feature = "opentelemetry"))]
use linera_rpc::propagation::get_traffic_type_from_request;
#[cfg(feature = "opentelemetry")]
use linera_rpc::propagation::OtelContextLayer;
use linera_rpc::{
config::{ProxyConfig, ShardConfig, TlsConfig, ValidatorInternalNetworkConfig},
grpc::{
api::{
self,
notifier_service_server::{NotifierService, NotifierServiceServer},
validator_node_server::{ValidatorNode, ValidatorNodeServer},
validator_worker_client::ValidatorWorkerClient,
BlobContent, BlobId, BlobIds, BlockProposal, Certificate, CertificatesBatchRequest,
CertificatesBatchResponse, ChainInfoResult, CryptoHash, HandlePendingBlobRequest,
LiteCertificate, NetworkDescription, Notification, NotificationBatch,
PendingBlobRequest, PendingBlobResult, PreviousEventBlocksRequest,
PreviousEventBlocksResponse, RawCertificate, RawCertificatesBatch, SubscriptionRequest,
VersionInfo,
},
pool::GrpcConnectionPool,
GrpcProtoConversionError, GrpcProxyable, GRPC_CHUNKED_MESSAGE_FILL_LIMIT,
GRPC_MAX_MESSAGE_SIZE,
},
};
use linera_sdk::{linera_base_types::Blob, views::ViewError};
use linera_storage::{Arc as CacheArc, Storage};
use prost::Message;
use tokio::{select, task::JoinSet};
use tokio_stream::wrappers::UnboundedReceiverStream;
use tokio_util::sync::CancellationToken;
use tonic::{
transport::{Channel, Identity, Server, ServerTlsConfig},
Request, Response, Status,
};
use tonic_web::GrpcWebLayer;
use tower::{builder::ServiceBuilder, Layer, Service};
use tracing::{debug, info, instrument, Instrument as _, Level};
#[cfg(with_metrics)]
mod metrics {
use std::sync::LazyLock;
use linera_base::prometheus_util::{
linear_bucket_interval, register_histogram_vec, register_int_counter_vec,
};
use linera_rpc::grpc::{ERROR_TYPE_LABEL, METHOD_NAME_LABEL, TRAFFIC_TYPE_LABEL};
use prometheus::{HistogramVec, IntCounterVec};
pub static PROXY_REQUEST_LATENCY: LazyLock<HistogramVec> = LazyLock::new(|| {
register_histogram_vec(
"proxy_request_latency",
"Proxy request latency",
&[METHOD_NAME_LABEL, TRAFFIC_TYPE_LABEL],
linear_bucket_interval(1.0, 50.0, 5000.0),
)
});
pub static PROXY_REQUEST_COUNT: LazyLock<IntCounterVec> = LazyLock::new(|| {
register_int_counter_vec(
"proxy_request_count",
"Proxy request count",
&[METHOD_NAME_LABEL, TRAFFIC_TYPE_LABEL],
)
});
pub static PROXY_REQUEST_SUCCESS: LazyLock<IntCounterVec> = LazyLock::new(|| {
register_int_counter_vec(
"proxy_request_success",
"Proxy request success",
&[METHOD_NAME_LABEL, TRAFFIC_TYPE_LABEL],
)
});
pub static PROXY_REQUEST_ERROR: LazyLock<IntCounterVec> = LazyLock::new(|| {
register_int_counter_vec(
"proxy_request_error",
"Proxy request error",
&[METHOD_NAME_LABEL, TRAFFIC_TYPE_LABEL, ERROR_TYPE_LABEL],
)
});
}
#[cfg(with_metrics)]
fn grpc_status_name(status: Option<&str>) -> String {
match status {
Some(code) => format!("{:?}", tonic::Code::from_bytes(code.as_bytes())),
None => "HTTP_ERROR".to_owned(),
}
}
#[derive(Clone)]
pub struct PrometheusMetricsMiddlewareLayer;
#[derive(Clone)]
pub struct PrometheusMetricsMiddlewareService<T> {
service: T,
}
impl<S> Layer<S> for PrometheusMetricsMiddlewareLayer {
type Service = PrometheusMetricsMiddlewareService<S>;
fn layer(&self, service: S) -> Self::Service {
PrometheusMetricsMiddlewareService { service }
}
}
impl<S, B, ResponseBody> Service<http::Request<B>> for PrometheusMetricsMiddlewareService<S>
where
S: Service<http::Request<B>, Response = http::Response<ResponseBody>> + Send,
S::Future: Send + 'static,
B: Send + 'static,
{
type Response = S::Response;
type Error = S::Error;
type Future = BoxFuture<'static, Result<S::Response, S::Error>>;
fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
self.service.poll_ready(cx)
}
fn call(&mut self, request: http::Request<B>) -> Self::Future {
#[cfg(with_metrics)]
let start = linera_base::time::Instant::now();
#[cfg(with_metrics)]
let method_name =
linera_rpc::grpc::extract_grpc_method_name(request.uri().path()).to_owned();
#[cfg(all(with_metrics, feature = "opentelemetry"))]
let traffic_type: &'static str = get_traffic_type_from_request(&request);
#[cfg(all(with_metrics, not(feature = "opentelemetry")))]
let traffic_type: &'static str = "unknown";
let future = self.service.call(request);
async move {
let response = future.await?;
#[cfg(with_metrics)]
{
metrics::PROXY_REQUEST_LATENCY
.with_label_values(&[&method_name, traffic_type])
.observe(start.elapsed().as_secs_f64() * 1000.0);
metrics::PROXY_REQUEST_COUNT
.with_label_values(&[&method_name, traffic_type])
.inc();
let grpc_status = response
.headers()
.get("grpc-status")
.and_then(|v| v.to_str().ok());
let is_error =
!response.status().is_success() || grpc_status.is_some_and(|s| s != "0");
if is_error {
let error_type = grpc_status_name(grpc_status);
metrics::PROXY_REQUEST_ERROR
.with_label_values(&[&method_name, traffic_type, &error_type])
.inc();
} else {
metrics::PROXY_REQUEST_SUCCESS
.with_label_values(&[&method_name, traffic_type])
.inc();
}
}
Ok(response)
}
.boxed()
}
}
#[derive(Clone)]
pub struct GrpcProxy<S>(Arc<GrpcProxyInner<S>>);
struct GrpcProxyInner<S> {
internal_config: ValidatorInternalNetworkConfig,
worker_connection_pool: GrpcConnectionPool,
notifier: ChannelNotifier<Result<Notification, Status>>,
tls: TlsConfig,
storage: S,
id: usize,
}
impl<S> GrpcProxy<S>
where
S: Storage + Clone + Send + Sync + 'static,
{
pub fn new(
internal_config: ValidatorInternalNetworkConfig,
connect_timeout: Duration,
timeout: Duration,
tls: TlsConfig,
storage: S,
id: usize,
) -> Self {
Self(Arc::new(GrpcProxyInner {
internal_config,
worker_connection_pool: GrpcConnectionPool::default()
.with_connect_timeout(connect_timeout)
.with_timeout(timeout),
notifier: ChannelNotifier::default(),
tls,
storage,
id,
}))
}
fn as_validator_node(&self) -> ValidatorNodeServer<Self> {
ValidatorNodeServer::new(self.clone())
.max_encoding_message_size(GRPC_MAX_MESSAGE_SIZE)
.max_decoding_message_size(GRPC_MAX_MESSAGE_SIZE)
}
fn config(&self) -> &ProxyConfig {
self.0
.internal_config
.proxies
.get(self.0.id)
.expect("No proxy config provided.")
}
fn as_notifier_service(&self) -> NotifierServiceServer<Self> {
NotifierServiceServer::new(self.clone())
}
fn public_address(&self) -> SocketAddr {
SocketAddr::from(([0, 0, 0, 0], self.config().public_port))
}
fn metrics_address(&self) -> SocketAddr {
SocketAddr::from(([0, 0, 0, 0], self.config().metrics_port))
}
fn internal_address(&self) -> SocketAddr {
SocketAddr::from(([0, 0, 0, 0], self.config().private_port))
}
fn shard_for(&self, proxyable: &impl GrpcProxyable) -> Option<ShardConfig> {
Some(
self.0
.internal_config
.get_shard_for(proxyable.chain_id()?)
.clone(),
)
}
fn worker_client_for_shard(
&self,
shard: &ShardConfig,
) -> Result<ValidatorWorkerClient<Channel>> {
let address = shard.http_address();
let channel = self.0.worker_connection_pool.channel(address)?;
let client = ValidatorWorkerClient::new(channel)
.max_encoding_message_size(GRPC_MAX_MESSAGE_SIZE)
.max_decoding_message_size(GRPC_MAX_MESSAGE_SIZE);
Ok(client)
}
#[instrument(
name = "GrpcProxy::run",
skip_all,
fields(
public_address = %self.public_address(),
internal_address = %self.internal_address(),
metrics_address = %self.metrics_address(),
),
err,
)]
#[cfg_attr(not(with_metrics), allow(unused_variables))]
pub async fn run(
self,
shutdown_signal: CancellationToken,
enable_memory_profiling: bool,
) -> Result<()> {
info!("Starting proxy");
let mut join_set = JoinSet::new();
#[cfg(with_metrics)]
monitoring_server::start_metrics_with_profiling(
self.metrics_address(),
shutdown_signal.clone(),
enable_memory_profiling,
)
.await;
let (health_reporter, health_service) = tonic_health::server::health_reporter();
health_reporter
.set_serving::<ValidatorNodeServer<GrpcProxy<S>>>()
.await;
let internal_server = join_set.spawn_task(
Server::builder()
.add_service(self.as_notifier_service())
.serve(self.internal_address())
.in_current_span(),
);
let reflection_service = tonic_reflection::server::Builder::configure()
.register_encoded_file_descriptor_set(linera_rpc::FILE_DESCRIPTOR_SET)
.build_v1()?;
#[cfg(feature = "opentelemetry")]
let layers = ServiceBuilder::new()
.layer(OtelContextLayer)
.layer(PrometheusMetricsMiddlewareLayer)
.into_inner();
#[cfg(not(feature = "opentelemetry"))]
let layers = ServiceBuilder::new()
.layer(PrometheusMetricsMiddlewareLayer)
.into_inner();
let public_server = join_set.spawn_task(
self.public_server()?
.max_concurrent_streams(Some(u32::MAX - 1)) .layer(layers)
.layer(
tower_http::cors::CorsLayer::permissive(),
)
.layer(GrpcWebLayer::new())
.accept_http1(true)
.add_service(health_service)
.add_service(self.as_validator_node())
.add_service(reflection_service)
.serve_with_shutdown(self.public_address(), shutdown_signal.cancelled_owned())
.in_current_span(),
);
select! {
internal_res = internal_server => internal_res??,
public_res = public_server => public_res??,
}
Ok(())
}
fn public_server(&self) -> Result<Server> {
match self.0.tls {
TlsConfig::Tls => {
use linera_rpc::{CERT_PEM, KEY_PEM};
let identity = Identity::from_pem(CERT_PEM, KEY_PEM);
let tls_config = ServerTlsConfig::new().identity(identity);
Ok(Server::builder().tls_config(tls_config)?)
}
TlsConfig::ClearText => Ok(Server::builder()),
}
}
#[cfg(feature = "opentelemetry")]
fn create_forwarding_request<T>(inner: T) -> Request<T> {
linera_rpc::propagation::create_request_with_current_span_context(inner)
}
#[cfg(not(feature = "opentelemetry"))]
fn create_forwarding_request<T>(inner: T) -> Request<T> {
Request::new(inner)
}
#[allow(clippy::result_large_err)]
#[instrument(skip_all, fields(remote_addr = ?request.remote_addr(), chain_id = ?request.get_ref().chain_id()))]
fn worker_client<R>(
&self,
request: Request<R>,
) -> Result<(ValidatorWorkerClient<Channel>, R), Status>
where
R: Debug + GrpcProxyable,
{
debug!("proxying request from {:?}", request.remote_addr());
let inner = request.into_inner();
let shard = self
.shard_for(&inner)
.ok_or_else(|| Status::not_found("could not find shard for message"))?;
let client = self
.worker_client_for_shard(&shard)
.map_err(|_| Status::internal("could not connect to shard"))?;
Ok((client, inner))
}
fn view_error_to_status(err: ViewError) -> Status {
let mut status = match &err {
ViewError::BcsError(_) => Status::invalid_argument(err.to_string()),
ViewError::StoreError { .. }
| ViewError::TokioJoinError(_)
| ViewError::TryLockError(_)
| ViewError::InconsistentEntries
| ViewError::PostLoadValuesError
| ViewError::IoError(_) => Status::internal(err.to_string()),
ViewError::KeyTooLong | ViewError::ArithmeticError(_) => {
Status::out_of_range(err.to_string())
}
ViewError::NotFound(_) | ViewError::MissingEntries(_) => {
Status::not_found(err.to_string())
}
};
status.set_source(Arc::new(err));
status
}
async fn get_certificate_hashes_by_heights_fallback(
&self,
chain_id: ChainId,
heights: Vec<BlockHeight>,
) -> Result<Vec<linera_base::crypto::CryptoHash>, Status> {
let chain_info_request =
ChainInfoQuery::new(chain_id).with_sent_certificate_hashes_by_heights(heights.clone());
let chain_info_response = self
.handle_chain_info_query(Request::new(chain_info_request.try_into()?))
.await?;
let chain_info_result = chain_info_response.into_inner();
let hashes: Vec<linera_base::crypto::CryptoHash> = match chain_info_result.inner {
Some(api::chain_info_result::Inner::ChainInfoResponse(response)) => {
let chain_info: ChainInfo =
bincode::deserialize(&response.chain_info).map_err(|e| {
Status::internal(format!("Failed to deserialize ChainInfo: {e}"))
})?;
chain_info.requested_sent_certificate_hashes
}
Some(api::chain_info_result::Inner::Error(error)) => {
let error =
bincode::deserialize(&error).unwrap_or_else(|err| NodeError::GrpcError {
error: format!("failed to unmarshal error message: {err}"),
});
return Err(Status::internal(format!(
"Chain info query failed: {error}"
)));
}
None => {
return Err(Status::internal("Empty chain info result"));
}
};
let indices: Vec<(BlockHeight, linera_base::crypto::CryptoHash)> =
heights.into_iter().zip(hashes.iter().copied()).collect();
self.0
.storage
.write_certificate_height_indices(chain_id, &indices)
.await
.map_err(Self::view_error_to_status)?;
Ok(hashes)
}
async fn collect_raw_certificates_by_hashes(
&self,
hashes: Vec<linera_base::crypto::CryptoHash>,
) -> Result<Vec<RawCertificate>, Status> {
let mut limiter: GrpcMessageLimiter<linera_chain::types::Certificate> =
GrpcMessageLimiter::new(GRPC_CHUNKED_MESSAGE_FILL_LIMIT);
let mut result = vec![];
for batch in hashes.chunks(100) {
let certificates: Vec<(Vec<u8>, Vec<u8>)> = self
.0
.storage
.read_certificates_raw(batch)
.await
.map_err(Self::view_error_to_status)?
.into_iter()
.flatten()
.map(CacheArc::unwrap_or_clone)
.collect();
let batch_size = certificates.len();
let batch_result = limiter.take_if(
certificates,
|lim, (lite_cert_bytes, confirmed_block_bytes)| {
Ok(lim
.fits_raw(lite_cert_bytes.len() + confirmed_block_bytes.len())
.then_some(RawCertificate {
lite_certificate: lite_cert_bytes,
confirmed_block: confirmed_block_bytes,
}))
},
)?;
let took_all = batch_result.len() == batch_size;
result.extend(batch_result);
if !took_all {
break;
}
}
Ok(result)
}
}
#[async_trait]
impl<S> ValidatorNode for GrpcProxy<S>
where
S: Storage + Clone + Send + Sync + 'static,
{
type SubscribeStream = UnboundedReceiverStream<Result<Notification, Status>>;
type DownloadBlobsStream =
std::pin::Pin<Box<dyn futures::Stream<Item = Result<BlobContent, Status>> + Send>>;
#[instrument(skip_all, err(Display), fields(method = "handle_block_proposal"))]
async fn handle_block_proposal(
&self,
request: Request<BlockProposal>,
) -> Result<Response<ChainInfoResult>, Status> {
let (mut client, inner) = self.worker_client(request)?;
client
.handle_block_proposal(Self::create_forwarding_request(inner))
.await
}
#[instrument(skip_all, err(Display), fields(method = "handle_lite_certificate"))]
async fn handle_lite_certificate(
&self,
request: Request<LiteCertificate>,
) -> Result<Response<ChainInfoResult>, Status> {
let (mut client, inner) = self.worker_client(request)?;
client
.handle_lite_certificate(Self::create_forwarding_request(inner))
.await
}
#[instrument(
skip_all,
err(Display),
fields(method = "handle_confirmed_certificate")
)]
async fn handle_confirmed_certificate(
&self,
request: Request<api::HandleConfirmedCertificateRequest>,
) -> Result<Response<ChainInfoResult>, Status> {
let (mut client, inner) = self.worker_client(request)?;
client
.handle_confirmed_certificate(Self::create_forwarding_request(inner))
.await
}
#[instrument(
skip_all,
err(Display),
fields(method = "handle_validated_certificate")
)]
async fn handle_validated_certificate(
&self,
request: Request<api::HandleValidatedCertificateRequest>,
) -> Result<Response<ChainInfoResult>, Status> {
let (mut client, inner) = self.worker_client(request)?;
client
.handle_validated_certificate(Self::create_forwarding_request(inner))
.await
}
#[instrument(skip_all, err(Display), fields(method = "handle_timeout_certificate"))]
async fn handle_timeout_certificate(
&self,
request: Request<api::HandleTimeoutCertificateRequest>,
) -> Result<Response<ChainInfoResult>, Status> {
let (mut client, inner) = self.worker_client(request)?;
client
.handle_timeout_certificate(Self::create_forwarding_request(inner))
.await
}
#[instrument(skip_all, err(Display), fields(method = "handle_chain_info_query"))]
async fn handle_chain_info_query(
&self,
request: Request<api::ChainInfoQuery>,
) -> Result<Response<ChainInfoResult>, Status> {
let (mut client, inner) = self.worker_client(request)?;
client
.handle_chain_info_query(Self::create_forwarding_request(inner))
.await
}
#[instrument(skip_all, err(Display), fields(method = "subscribe"))]
async fn subscribe(
&self,
request: Request<SubscriptionRequest>,
) -> Result<Response<Self::SubscribeStream>, Status> {
let subscription_request = request.into_inner();
let chain_ids = subscription_request
.chain_ids
.into_iter()
.map(ChainId::try_from)
.collect::<Result<Vec<ChainId>, _>>()?;
let rx = self
.0
.notifier
.subscribe_with_ack(chain_ids, Ok(Notification::default()));
Ok(Response::new(UnboundedReceiverStream::new(rx)))
}
#[instrument(skip_all, err(Display))]
async fn get_version_info(
&self,
_request: Request<()>,
) -> Result<Response<VersionInfo>, Status> {
Ok(Response::new(linera_version::VersionInfo::default().into()))
}
#[instrument(skip_all, err(Display), fields(method = "get_network_description"))]
async fn get_network_description(
&self,
_request: Request<()>,
) -> Result<Response<NetworkDescription>, Status> {
let description = self
.0
.storage
.read_network_description()
.await
.map_err(Self::view_error_to_status)?
.ok_or_else(|| Status::not_found("Cannot find network description in the database"))?;
Ok(Response::new(description.into()))
}
#[instrument(skip_all, err(Display), fields(method = "upload_blob"))]
async fn upload_blob(&self, request: Request<BlobContent>) -> Result<Response<BlobId>, Status> {
let content: linera_sdk::linera_base_types::BlobContent =
request.into_inner().try_into()?;
let blob = Blob::new(content);
let id = blob.id();
let result = self.0.storage.maybe_write_blobs(&[blob]).await;
if !result.map_err(Self::view_error_to_status)?[0] {
return Err(Status::not_found("Blob not found"));
}
Ok(Response::new(id.try_into()?))
}
#[instrument(skip_all, err(Display), fields(method = "download_blob"))]
async fn download_blob(
&self,
request: Request<BlobId>,
) -> Result<Response<BlobContent>, Status> {
let blob_id = request.into_inner().try_into()?;
let blob = self
.0
.storage
.read_blob(blob_id)
.await
.map_err(Self::view_error_to_status)?;
let blob = blob
.map(CacheArc::unwrap_or_clone)
.ok_or_else(|| Status::not_found(format!("Blob not found {blob_id}")))?;
Ok(Response::new(blob.into_content().try_into()?))
}
#[instrument(skip_all, err(Display), fields(method = "download_blobs"))]
async fn download_blobs(
&self,
request: Request<BlobIds>,
) -> Result<Response<Self::DownloadBlobsStream>, Status> {
let blob_ids = Vec::<linera_base::identifiers::BlobId>::try_from(request.into_inner())?;
let blobs = self
.0
.storage
.read_blobs(&blob_ids)
.await
.map_err(Self::view_error_to_status)?;
let stream = futures::stream::iter(blobs.into_iter().filter_map(|maybe_blob| {
let blob = maybe_blob?;
Some(
BlobContent::try_from(blob.content().clone())
.map_err(|err| Status::internal(err.to_string())),
)
}));
Ok(Response::new(Box::pin(stream)))
}
#[instrument(skip_all, err(Display), fields(method = "download_pending_blob"))]
async fn download_pending_blob(
&self,
request: Request<PendingBlobRequest>,
) -> Result<Response<PendingBlobResult>, Status> {
let (mut client, inner) = self.worker_client(request)?;
client
.download_pending_blob(Self::create_forwarding_request(inner))
.await
}
#[instrument(skip_all, err(Display), fields(method = "handle_pending_blob"))]
async fn handle_pending_blob(
&self,
request: Request<HandlePendingBlobRequest>,
) -> Result<Response<ChainInfoResult>, Status> {
let (mut client, inner) = self.worker_client(request)?;
client
.handle_pending_blob(Self::create_forwarding_request(inner))
.await
}
#[instrument(skip_all, err(Display), fields(method = "previous_event_blocks"))]
async fn previous_event_blocks(
&self,
request: Request<PreviousEventBlocksRequest>,
) -> Result<Response<PreviousEventBlocksResponse>, Status> {
let (mut client, inner) = self.worker_client(request)?;
client
.previous_event_blocks(Self::create_forwarding_request(inner))
.await
}
#[instrument(skip_all, err(Display), fields(method = "download_certificate"))]
async fn download_certificate(
&self,
request: Request<CryptoHash>,
) -> Result<Response<Certificate>, Status> {
let hash = request.into_inner().try_into()?;
let certificate: linera_chain::types::Certificate = self
.0
.storage
.read_certificate(hash)
.await
.map_err(Self::view_error_to_status)?
.ok_or_else(|| Status::not_found(hash.to_string()))?
.into_std()
.as_ref()
.into();
Ok(Response::new(certificate.try_into()?))
}
#[instrument(skip_all, err(Display), fields(method = "download_certificates"))]
async fn download_certificates(
&self,
request: Request<CertificatesBatchRequest>,
) -> Result<Response<CertificatesBatchResponse>, Status> {
let hashes: Vec<linera_base::crypto::CryptoHash> = request
.into_inner()
.hashes
.into_iter()
.map(linera_base::crypto::CryptoHash::try_from)
.collect::<Result<Vec<linera_base::crypto::CryptoHash>, _>>()?;
let raw_certificates = self.collect_raw_certificates_by_hashes(hashes).await?;
let certificates: Vec<linera_chain::types::Certificate> = raw_certificates
.into_iter()
.map(|raw| {
let lite_cert = bcs::from_bytes::<ChainLiteCertificate>(&raw.lite_certificate)
.map_err(|e| {
Status::internal(format!("Failed to deserialize lite certificate: {e}"))
})?;
let confirmed_block = bcs::from_bytes::<ConfirmedBlock>(&raw.confirmed_block)
.map_err(|e| {
Status::internal(format!("Failed to deserialize confirmed block: {e}"))
})?;
lite_cert
.with_value(confirmed_block)
.ok_or_else(|| Status::internal("Invalid certificate"))
.map(Into::into)
})
.collect::<Result<_, _>>()?;
Ok(Response::new(CertificatesBatchResponse::try_from(
certificates,
)?))
}
#[instrument(
skip_all,
err(Display),
fields(method = "download_certificates_by_heights")
)]
async fn download_certificates_by_heights(
&self,
request: Request<api::DownloadCertificatesByHeightsRequest>,
) -> Result<Response<CertificatesBatchResponse>, Status> {
let original_request: CertificatesByHeightRequest = request.into_inner().try_into()?;
let chain_id = original_request.chain_id;
let heights = original_request.heights;
let certificates_by_height: Vec<_> = self
.0
.storage
.read_certificates_by_heights(chain_id, &heights)
.await
.map_err(Self::view_error_to_status)?
.into_iter()
.flatten()
.collect();
let all_found = certificates_by_height.len() == heights.len();
if all_found {
let mut limiter: GrpcMessageLimiter<linera_chain::types::Certificate> =
GrpcMessageLimiter::new(GRPC_CHUNKED_MESSAGE_FILL_LIMIT);
let returned_certificates =
limiter.take_if(certificates_by_height, |lim, certificate| {
let cert: linera_chain::types::Certificate = (&*certificate).into();
Ok(lim.fits::<Certificate>(cert.clone())?.then_some(cert))
})?;
return Ok(Response::new(CertificatesBatchResponse::try_from(
returned_certificates,
)?));
}
let hashes = self
.get_certificate_hashes_by_heights_fallback(chain_id, heights)
.await?;
let certificates_request = CertificatesBatchRequest {
hashes: hashes.into_iter().map(|h| h.into()).collect(),
};
self.download_certificates(Request::new(certificates_request))
.await
}
#[instrument(skip_all, err(Display))]
async fn download_raw_certificates_by_heights(
&self,
request: Request<api::DownloadCertificatesByHeightsRequest>,
) -> Result<Response<api::RawCertificatesBatch>, Status> {
let original_request: CertificatesByHeightRequest = request.into_inner().try_into()?;
let chain_id = original_request.chain_id;
let heights = original_request.heights;
let raw_certificates_by_height = self
.0
.storage
.read_certificates_by_heights_raw(chain_id, &heights)
.await
.map_err(Self::view_error_to_status)?
.into_iter()
.flatten()
.map(CacheArc::unwrap_or_clone)
.collect::<Vec<(Vec<u8>, Vec<u8>)>>();
let all_found = raw_certificates_by_height.len() == heights.len();
if all_found {
let mut limiter: GrpcMessageLimiter<linera_chain::types::Certificate> =
GrpcMessageLimiter::new(GRPC_CHUNKED_MESSAGE_FILL_LIMIT);
let certificates =
limiter.take_if(raw_certificates_by_height, |lim, (lite, block)| {
Ok(lim
.fits_raw(lite.len() + block.len())
.then_some(RawCertificate {
lite_certificate: lite,
confirmed_block: block,
}))
})?;
return Ok(Response::new(RawCertificatesBatch { certificates }));
}
let hashes = self
.get_certificate_hashes_by_heights_fallback(chain_id, heights)
.await?;
let certificates = self.collect_raw_certificates_by_hashes(hashes).await?;
Ok(Response::new(RawCertificatesBatch { certificates }))
}
#[instrument(skip_all, err(level = Level::WARN), fields(
method = "blob_last_used_by"
))]
async fn blob_last_used_by(
&self,
request: Request<BlobId>,
) -> Result<Response<CryptoHash>, Status> {
let blob_id = request.into_inner().try_into()?;
let blob_state = self
.0
.storage
.read_blob_state(blob_id)
.await
.map_err(Self::view_error_to_status)?;
let blob_state =
blob_state.ok_or_else(|| Status::not_found(format!("Blob not found {blob_id}")))?;
let last_used_by = blob_state
.last_used_by
.ok_or_else(|| Status::not_found(format!("Blob not found {blob_id}")))?;
Ok(Response::new(last_used_by.into()))
}
#[instrument(skip_all, err(level = Level::WARN))]
async fn missing_blob_ids(
&self,
request: Request<BlobIds>,
) -> Result<Response<BlobIds>, Status> {
let blob_ids: Vec<linera_base::identifiers::BlobId> = request.into_inner().try_into()?;
let missing_blob_ids = self
.0
.storage
.missing_blobs(&blob_ids)
.await
.map_err(Self::view_error_to_status)?;
Ok(Response::new(missing_blob_ids.try_into()?))
}
#[instrument(target = "telemetry_only", skip_all, err(level = Level::WARN), fields(
method = "blob_last_used_by_certificate"
))]
async fn blob_last_used_by_certificate(
&self,
request: Request<BlobId>,
) -> Result<Response<Certificate>, Status> {
let cert_hash = self.blob_last_used_by(request).await?;
let request = Request::new(cert_hash.into_inner());
self.download_certificate(request).await
}
}
#[async_trait]
impl<S> NotifierService for GrpcProxy<S>
where
S: Storage + Clone + Send + Sync + 'static,
{
#[instrument(skip_all, err(Display), fields(method = "notify_batch"))]
async fn notify_batch(
&self,
request: Request<NotificationBatch>,
) -> Result<Response<()>, Status> {
for notification in request.into_inner().notifications {
let chain_id = notification
.chain_id
.clone()
.ok_or_else(|| Status::invalid_argument("Missing field: chain_id."))?
.try_into()?;
self.0.notifier.notify_chain(&chain_id, &Ok(notification));
}
Ok(Response::new(()))
}
}
struct GrpcMessageLimiter<T> {
remaining: usize,
_phantom: PhantomData<T>,
}
impl<T> GrpcMessageLimiter<T> {
fn new(limit: usize) -> Self {
Self {
remaining: limit,
_phantom: PhantomData,
}
}
#[cfg(test)]
fn empty() -> Self {
Self::new(0)
}
fn fits<U>(&mut self, el: T) -> Result<bool, GrpcProtoConversionError>
where
U: TryFrom<T, Error = GrpcProtoConversionError> + Message,
{
let required = U::try_from(el).map(|proto| proto.encoded_len())?;
Ok(self.fits_raw(required))
}
fn fits_raw(&mut self, bytes_len: usize) -> bool {
if self.remaining < bytes_len {
return false;
}
self.remaining = self.remaining.saturating_sub(bytes_len);
true
}
fn take_if<I, O, F>(&mut self, items: I, mut try_take: F) -> Result<Vec<O>, Status>
where
I: IntoIterator,
F: FnMut(&mut Self, I::Item) -> Result<Option<O>, Status>,
{
let mut result = vec![];
for item in items {
match try_take(self, item)? {
Some(output) => result.push(output),
None => break,
}
}
Ok(result)
}
}
#[cfg(test)]
mod proto_message_cap {
use linera_base::crypto::CryptoHash;
use linera_chain::{
data_types::BlockExecutionOutcome,
types::{Block, Certificate, ConfirmedBlock, ConfirmedBlockCertificate},
};
use linera_sdk::linera_base_types::{
ChainId, TestString, ValidatorKeypair, ValidatorSignature,
};
use super::{CertificatesBatchResponse, GrpcMessageLimiter};
fn test_certificate() -> Certificate {
let keypair = ValidatorKeypair::generate();
let validator = keypair.public_key;
let signature = ValidatorSignature::new(&TestString::new("Test"), &keypair.secret_key);
let block = Block::new(
linera_chain::test::make_first_block(ChainId(CryptoHash::test_hash("root_chain"))),
BlockExecutionOutcome::default(),
);
let signatures = vec![(validator, signature)];
Certificate::Confirmed(ConfirmedBlockCertificate::new(
ConfirmedBlock::new(block),
Default::default(),
signatures,
))
}
#[test]
fn takes_up_to_limit() {
let certificate = test_certificate();
let single_cert_size = prost::Message::encoded_len(
&CertificatesBatchResponse::try_from(vec![certificate.clone()]).unwrap(),
);
let certificates = vec![certificate.clone(), certificate.clone()];
let mut empty_limiter = GrpcMessageLimiter::empty();
assert!(!empty_limiter
.fits::<super::Certificate>(certificate.clone())
.unwrap());
let mut single_message_limiter = GrpcMessageLimiter::new(single_cert_size);
assert_eq!(
certificates
.clone()
.into_iter()
.take_while(|cert| single_message_limiter
.fits::<super::Certificate>(cert.clone())
.unwrap())
.collect::<Vec<_>>(),
vec![certificate.clone()]
);
let mut double_message_limiter = GrpcMessageLimiter::new(single_cert_size * 2);
assert_eq!(
certificates
.into_iter()
.take_while(|cert| double_message_limiter
.fits::<super::Certificate>(cert.clone())
.unwrap())
.collect::<Vec<_>>(),
vec![certificate.clone(), certificate.clone()]
);
}
}