Initial commit for the block exporter. (#3746)

## Motivation

Initial work for #3661.

## Proposal
- Created a new service implementing the `NotifierService` from
linera-rpc to receive redirected notifications about new blocks from the
chain workers, similar to linera-proxy(closes #3662).
- A new CLI parsable type `BlockExporterConfig` to define the
configuration options for the linera-exporter from a toml file(closes
#3664).
- Extended `linera net up` to optionally run a block exporter with a
optional specified configuration(closes #3663).
- Added a new view object `BlockExporterStateView` to track and update
the latest information about the blocks from the workers(closes #3665).
- This PR also partially addresses #3666, without the topological
sorting part.

## Test Plan

new unit and integration tests + CI

Author: usagi32

Diff for: CLI.md

@@ -995,6 +995,9 @@ Start a Local Linera Network
 * `--faucet-amount <FAUCET_AMOUNT>` — The number of tokens to send to each new chain created by the faucet
 
   Default value: `1000`
+* `--block-exporters <BLOCK_EXPORTERS>` — The number of block exporters per validator in the local test network. Default is 0
+
+  Default value: `0`
 
 
 

Diff for: Cargo.lock

@@ -19,7 +19,9 @@ use linera_execution::{
     committee::{Committee, ValidatorState},
     ResourceControlPolicy,
 };
-use linera_rpc::config::{ValidatorInternalNetworkConfig, ValidatorPublicNetworkConfig};
+use linera_rpc::config::{
+    ExporterServiceConfig, ValidatorInternalNetworkConfig, ValidatorPublicNetworkConfig,
+};
 use linera_storage::Storage;
 use serde::{Deserialize, Serialize};
 
@@ -244,3 +246,37 @@ impl GenesisConfig {
         CryptoHash::new(self)
     }
 }
+
+/// The configuration file for the linera-exporter.
+#[derive(Serialize, Deserialize, Debug, Clone)]
+pub struct BlockExporterConfig {
+    /// The server configuration for the linera-exporter.
+    pub service_config: ExporterServiceConfig,
+
+    /// The configuration file for the export destinations.
+    #[serde(default)]
+    pub destination_config: DestinationConfig,
+
+    /// Identity for the block exporter state.
+    pub id: u32,
+}
+
+/// Configuration file for the exports.
+#[derive(Serialize, Deserialize, Debug, Clone, Default)]
+pub struct DestinationConfig {
+    /// The destination URIs to export to.
+    pub destinations: Vec<Destination>,
+}
+
+// Each destination has an ID and a configuration.
+pub type DestinationId = u16;
+
+/// The uri to provide export services to.
+#[allow(dead_code)]
+#[derive(Serialize, Deserialize, Debug, Clone)]
+pub struct Destination {
+    /// The host name of the target destination (IP or hostname).
+    pub endpoint: String,
+    /// The port number of the target destination.
+    pub port: u16,
+}

Diff for: linera-client/src/config.rs

@@ -140,6 +140,10 @@ pub struct ValidatorInternalNetworkPreConfig<P> {
     pub host: String,
     /// The port the proxy listens on the internal network.
     pub port: u16,
+    /// The server configurations for the linera-exporter.
+    /// They can be used as optional locations to forward notifications to destinations other than
+    /// the proxy, by the workers.
+    pub block_exporters: Vec<ExporterServiceConfig>,
     /// The port of the proxy's metrics endpoint.
     pub metrics_port: u16,
 }
@@ -152,6 +156,7 @@ impl<P> ValidatorInternalNetworkPreConfig<P> {
             shards: self.shards.clone(),
             host: self.host.clone(),
             port: self.port,
+            block_exporters: self.block_exporters.clone(),
             metrics_port: self.metrics_port,
         }
     }
@@ -161,6 +166,15 @@ impl ValidatorInternalNetworkConfig {
     pub fn proxy_address(&self) -> String {
         format!("{}://{}:{}", self.protocol.scheme(), self.host, self.port)
     }
+
+    pub fn exporter_addresses(&self) -> Vec<String> {
+        self.block_exporters
+            .iter()
+            .map(|ExporterServiceConfig { host, port }| {
+                format!("{}://{}:{}", self.protocol.scheme(), host, port)
+            })
+            .collect::<Vec<_>>()
+    }
 }
 
 impl ValidatorPublicNetworkConfig {
@@ -273,6 +287,15 @@ impl<P> ValidatorInternalNetworkPreConfig<P> {
     }
 }
 
+#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
+/// The server configuration for the linera-exporter.
+pub struct ExporterServiceConfig {
+    /// The host name of the server (IP or hostname).
+    pub host: String,
+    /// The port for the server to listen on.
+    pub port: u16,
+}
+
 #[test]
 fn cross_chain_config_to_args() {
     let config = CrossChainConfig::default();

Diff for: linera-rpc/src/config.rs

@@ -18,14 +18,14 @@ use futures::{
 use linera_base::{data_types::Blob, identifiers::ChainId};
 use linera_core::{
     node::NodeError,
-    worker::{NetworkActions, Notification, WorkerError, WorkerState},
+    worker::{NetworkActions, Notification, Reason, WorkerError, WorkerState},
     JoinSetExt as _, TaskHandle,
 };
 use linera_storage::Storage;
 use rand::Rng;
 use tokio::{sync::oneshot, task::JoinSet};
 use tokio_util::sync::CancellationToken;
-use tonic::{Request, Response, Status};
+use tonic::{transport::Channel, Request, Response, Status};
 use tower::{builder::ServiceBuilder, Layer, Service};
 use tracing::{debug, error, info, instrument, trace, warn};
 #[cfg(with_metrics)]
@@ -240,6 +240,7 @@ where
             Self::forward_notifications(
                 state.nickname().to_string(),
                 internal_network.proxy_address(),
+                internal_network.exporter_addresses(),
                 notification_receiver,
             )
         });
@@ -293,6 +294,7 @@ where
     async fn forward_notifications(
         nickname: String,
         proxy_address: String,
+        exporter_addresses: Vec<String>,
         mut receiver: Receiver<Notification>,
     ) {
         let channel = tonic::transport::Channel::from_shared(proxy_address.clone())
@@ -302,7 +304,20 @@ where
             .max_encoding_message_size(GRPC_MAX_MESSAGE_SIZE)
             .max_decoding_message_size(GRPC_MAX_MESSAGE_SIZE);
 
+        let mut exporter_clients: Vec<NotifierServiceClient<Channel>> = exporter_addresses
+            .iter()
+            .map(|address| {
+                let channel = tonic::transport::Channel::from_shared(address.clone())
+                    .expect("Exporter URI should be valid")
+                    .connect_lazy();
+                NotifierServiceClient::new(channel)
+                    .max_encoding_message_size(GRPC_MAX_MESSAGE_SIZE)
+                    .max_decoding_message_size(GRPC_MAX_MESSAGE_SIZE)
+            })
+            .collect::<Vec<_>>();
+
         while let Some(notification) = receiver.next().await {
+            let reason = &notification.reason;
             let notification: api::Notification = match notification.clone().try_into() {
                 Ok(notification) => notification,
                 Err(error) => {
@@ -319,6 +334,20 @@ where
                     "could not send notification",
                 )
             }
+
+            if let Reason::NewBlock { height: _, hash: _ } = reason {
+                for exporter_client in &mut exporter_clients {
+                    let request = tonic::Request::new(notification.clone());
+                    if let Err(error) = exporter_client.notify(request).await {
+                        error!(
+                            %error,
+                            nickname,
+                            ?notification,
+                            "could not send notification",
+                        )
+                    }
+                }
+            }
         }
     }
 

Diff for: linera-rpc/src/grpc/server.rs

@@ -60,6 +60,8 @@ async-lock.workspace = true
 async-trait.workspace = true
 async-tungstenite.workspace = true
 axum = { workspace = true, features = ["ws"] }
+bcs.workspace = true
+bincode.workspace = true
 cargo_toml.workspace = true
 cfg-if.workspace = true
 chrono = { workspace = true, features = ["clock"] }
@@ -174,6 +176,10 @@ name = "linera-benchmark"
 path = "src/benchmark.rs"
 required-features = ["benchmark"]
 
+[[bin]]
+name = "linera-exporter"
+path = "src/linera-exporter/main.rs"
+
 [[bench]]
 name = "transfers"
 harness = false