Indexer destination (#3949)

## Motivation

Generic indexer destination for the block exporter.
To be rebased on top of #3896 .

## Proposal

- Integration with the local net.
- Global indexing for the blobs for each destination kind.
- Also Closes #3669 

## Test Plan

Unit + integration tests

## Release Plan

- Nothing to do / These changes follow the usual release cycle.

Author: usagi32

Cargo.lock

@@ -246,17 +246,19 @@ pub struct Destination {
     pub endpoint: String,
     /// The port number of the target destination.
     pub port: u16,
+    /// The description for the gRPC based destination.
+    /// Discriminates the export mode and the client to use.
+    pub kind: DestinationKind,
 }
 
-impl Destination {
-    pub fn address(&self) -> String {
-        let tls = match self.tls {
-            TlsConfig::ClearText => "http",
-            TlsConfig::Tls => "https",
-        };
-
-        format!("{}://{}:{}", tls, self.endpoint, self.port)
-    }
+/// The description for the gRPC based destination.
+/// Discriminates the export mode and the client to use.
+#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, Eq, Copy)]
+pub enum DestinationKind {
+    /// The indexer description.
+    Indexer,
+    /// The validator description.
+    Validator,
 }
 
 /// The configuration file to impose various limits
@@ -295,3 +297,22 @@ impl Default for LimitsConfig {
         }
     }
 }
+
+impl Destination {
+    pub fn address(&self) -> String {
+        match self.kind {
+            DestinationKind::Indexer => {
+                let tls = match self.tls {
+                    TlsConfig::ClearText => "http",
+                    TlsConfig::Tls => "https",
+                };
+
+                format!("{}://{}:{}", tls, self.endpoint, self.port)
+            }
+
+            DestinationKind::Validator => {
+                format!("{}:{}:{}", "grpc", self.endpoint, self.port)
+            }
+        }
+    }
+}

linera-client/src/config.rs

@@ -126,6 +126,7 @@ tracing.workspace = true
 
 [build-dependencies]
 cfg_aliases.workspace = true
+tonic-build = { workspace = true, features = ["prost", "transport"] }
 
 [dev-dependencies]
 amm.workspace = true
@@ -159,6 +160,7 @@ test-case.workspace = true
 test-log = { workspace = true, features = ["trace"] }
 test-strategy.workspace = true
 tokio = { workspace = true, features = ["full", "test-util"] }
+tracing-subscriber = { workspace = true, features = ["fmt"] }
 
 [[bin]]
 name = "linera"

linera-service/Cargo.toml

@@ -1,10 +1,14 @@
 // Copyright (c) Zefchain Labs, Inc.
 // SPDX-License-Identifier: Apache-2.0
 
-fn main() {
+fn main() -> Result<(), Box<dyn std::error::Error>> {
     cfg_aliases::cfg_aliases! {
         with_revm: { feature = "revm" },
         with_testing: { any(test, feature = "test") },
         with_metrics: { all(not(target_arch = "wasm32"), feature = "metrics") },
     };
+
+    tonic_build::compile_protos("src/exporter/proto/indexer.proto")?;
+
+    Ok(())
 }

linera-service/build.rs

@@ -1936,7 +1936,6 @@ async fn run(options: &ClientOptions) -> Result<i32, Error> {
                 faucet_chain,
                 faucet_port,
                 faucet_amount,
-                block_exporters,
                 ..
             } => {
                 net_up_utils::handle_net_up_service(
@@ -1955,7 +1954,6 @@ async fn run(options: &ClientOptions) -> Result<i32, Error> {
                     *faucet_chain,
                     *faucet_port,
                     *faucet_amount,
-                    *block_exporters,
                 )
                 .boxed()
                 .await?;

linera-service/src/cli/main.rs

@@ -183,7 +183,6 @@ pub async fn handle_net_up_service(
     faucet_chain: Option<u32>,
     faucet_port: NonZeroU16,
     faucet_amount: Amount,
-    num_block_exporters: u32,
 ) -> anyhow::Result<()> {
     if num_initial_validators < 1 {
         panic!("The local test network must have at least one validator.");
@@ -223,7 +222,7 @@ pub async fn handle_net_up_service(
         cross_chain_config,
         storage_config_builder,
         path_provider,
-        num_block_exporters,
+        block_exporters: vec![],
     };
     let (mut net, client) = config.instantiate().await?;
     let faucet_service = print_messages_and_create_faucet(

linera-service/src/cli/net_up_utils.rs

@@ -19,9 +19,12 @@ use linera_base::{
     command::{resolve_binary, CommandExt},
     data_types::Amount,
 };
-use linera_client::client_options::ResourceControlPolicyConfig;
+use linera_client::{
+    client_options::ResourceControlPolicyConfig,
+    config::{BlockExporterConfig, DestinationKind},
+};
 use linera_core::node::ValidatorNodeProvider;
-use linera_rpc::config::CrossChainConfig;
+use linera_rpc::config::{CrossChainConfig, TlsConfig};
 #[cfg(all(feature = "storage-service", with_testing))]
 use linera_storage_service::common::storage_service_test_endpoint;
 #[cfg(all(feature = "rocksdb", feature = "scylladb", with_testing))]
@@ -184,7 +187,7 @@ pub struct LocalNetConfig {
     pub cross_chain_config: CrossChainConfig,
     pub storage_config_builder: InnerStorageConfigBuilder,
     pub path_provider: PathProvider,
-    pub num_block_exporters: u32,
+    pub block_exporters: Vec<BlockExporterConfig>,
 }
 
 /// A set of Linera validators running locally as native processes.
@@ -202,7 +205,7 @@ pub struct LocalNet {
     common_storage_config: InnerStorageConfig,
     cross_chain_config: CrossChainConfig,
     path_provider: PathProvider,
-    num_block_exporters: u32,
+    block_exporters: Vec<BlockExporterConfig>,
 }
 
 /// The name of the environment variable that allows specifying additional arguments to be passed
@@ -305,7 +308,7 @@ impl LocalNetConfig {
             num_proxies,
             storage_config_builder,
             path_provider,
-            num_block_exporters: 0,
+            block_exporters: vec![],
         }
     }
 }
@@ -326,7 +329,7 @@ impl LineraNetConfig for LocalNetConfig {
             storage_config,
             self.cross_chain_config,
             self.path_provider,
-            self.num_block_exporters,
+            self.block_exporters,
         )?;
         let client = net.make_client().await;
         ensure!(
@@ -391,7 +394,7 @@ impl LocalNet {
         common_storage_config: InnerStorageConfig,
         cross_chain_config: CrossChainConfig,
         path_provider: PathProvider,
-        num_block_exporters: u32,
+        block_exporters: Vec<BlockExporterConfig>,
     ) -> Result<Self> {
         Ok(Self {
             network,
@@ -407,7 +410,7 @@ impl LocalNet {
             common_storage_config,
             cross_chain_config,
             path_provider,
-            num_block_exporters,
+            block_exporters,
         })
     }
 
@@ -501,9 +504,9 @@ impl LocalNet {
             ));
         }
 
-        for j in 0..self.num_block_exporters {
+        for j in 0..self.block_exporters.len() {
             let host = Network::Grpc.localhost();
-            let port = Self::block_exporter_port(n, j as usize);
+            let port = Self::block_exporter_port(n, j);
             let config_content = format!(
                 r#"
 
@@ -514,7 +517,7 @@ impl LocalNet {
             );
 
             content.push_str(&config_content);
-            let exporter_config = self.generate_block_exporter_config(n, j);
+            let exporter_config = self.generate_block_exporter_config(n, j as u32);
             let config_path = self
                 .path_provider
                 .path()
@@ -536,7 +539,7 @@ impl LocalNet {
         let n = validator;
         let host = Network::Grpc.localhost();
         let port = Self::block_exporter_port(n, exporter_id as usize);
-        let config = format!(
+        let mut config = format!(
             r#"
             id = {exporter_id}
 
@@ -546,6 +549,36 @@ impl LocalNet {
             "#
         );
 
+        for destination in &self.block_exporters[exporter_id as usize]
+            .destination_config
+            .destinations
+        {
+            let tls = match destination.tls {
+                TlsConfig::ClearText => "ClearText",
+                TlsConfig::Tls => "Tls",
+            };
+
+            let endpoint = &destination.endpoint;
+            let port = destination.port;
+            let kind = match destination.kind {
+                DestinationKind::Indexer => "Indexer",
+                DestinationKind::Validator => "Validator",
+            };
+
+            let destination_string_to_push = format!(
+                r#"
+
+                [[destination_config.destinations]]
+                tls = "{tls}"
+                endpoint = "{endpoint}"
+                port = {port}
+                kind = "{kind}"
+                "#
+            );
+
+            config.push_str(&destination_string_to_push);
+        }
+
         config
     }
 
@@ -799,8 +832,8 @@ impl LocalNet {
             let server = self.run_server(index, shard).await?;
             validator.add_server(server);
         }
-        for block_exporter in 0..self.num_block_exporters {
-            let exporter = self.run_exporter(index, block_exporter).await?;
+        for block_exporter in 0..self.block_exporters.len() {
+            let exporter = self.run_exporter(index, block_exporter as u32).await?;
             validator.add_block_exporter(exporter);
         }
         self.running_validators.insert(index, validator);

linera-service/src/cli_wrappers/local_net.rs

@@ -1,16 +1,21 @@
 // Copyright (c) Zefchain Labs, Inc.
 // SPDX-License-Identifier: Apache-2.0
 
-use std::future::IntoFuture;
+use std::{future::IntoFuture, net::SocketAddr};
 
 use bincode::ErrorKind;
 use custom_debug_derive::Debug;
-use linera_base::{crypto::CryptoHash, data_types::BlockHeight, identifiers::ChainId};
-use linera_chain::data_types::IncomingBundle;
-use linera_rpc::grpc::GrpcProtoConversionError;
+use linera_base::{
+    crypto::CryptoHash,
+    data_types::BlockHeight,
+    identifiers::{BlobId, ChainId},
+};
+use linera_chain::{data_types::IncomingBundle, types::ConfirmedBlock};
+use linera_rpc::grpc::{GrpcError, GrpcProtoConversionError};
 use linera_sdk::views::ViewError;
 use serde::{Deserialize, Serialize};
 use tokio_util::sync::{CancellationToken, WaitForCancellationFutureOwned};
+use tonic::Status;
 
 #[derive(thiserror::Error, Debug)]
 pub(crate) enum ExporterError {
@@ -38,6 +43,15 @@ pub(crate) enum ExporterError {
     #[error("trying to re-initialize the chain: {0}")]
     ChainAlreadyExists(ChainId),
 
+    #[error("unable to establish an underlying stream")]
+    SynchronizationFailed(Box<Status>),
+
+    #[error(transparent)]
+    GrpcError(#[from] GrpcError),
+
+    #[error("wrong destination")]
+    DestinationError,
+
     #[error("generic error: {0}")]
     GenericError(Box<dyn std::error::Error + Send + Sync + 'static>),
 }
@@ -61,6 +75,21 @@ impl From<BadNotificationKind> for ExporterError {
     }
 }
 
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub(crate) struct CanonicalBlock {
+    pub blobs: Box<[BlobId]>,
+    pub block_hash: CryptoHash,
+}
+
+impl CanonicalBlock {
+    pub(crate) fn new(hash: CryptoHash, blobs: &[BlobId]) -> CanonicalBlock {
+        CanonicalBlock {
+            block_hash: hash,
+            blobs: blobs.to_vec().into_boxed_slice(),
+        }
+    }
+}
+
 #[derive(Debug, Clone, PartialEq, Eq, Hash, Copy)]
 pub(crate) struct BlockId {
     pub hash: CryptoHash,
@@ -90,6 +119,10 @@ impl BlockId {
             height: incoming_bundle.bundle.height,
         }
     }
+
+    pub(crate) fn from_confirmed_block(block: &ConfirmedBlock) -> BlockId {
+        BlockId::new(block.chain_id(), block.inner().hash(), block.height())
+    }
 }
 
 impl LiteBlockId {
@@ -104,14 +137,6 @@ impl From<BlockId> for LiteBlockId {
     }
 }
 
-#[macro_export]
-macro_rules! dispatch {
-    ($func:expr, log = $log_value:expr $(, $args:expr)* ) => {{
-        tracing::debug!("dispatching batch: {:?} from linera exporter", $log_value);
-        $func($($args),*).await
-    }};
-}
-
 #[derive(Clone)]
 pub(crate) struct ExporterCancellationSignal {
     token: CancellationToken,
@@ -131,3 +156,7 @@ impl IntoFuture for ExporterCancellationSignal {
         self.token.cancelled_owned()
     }
 }
+
+pub(crate) fn get_address(port: u16) -> SocketAddr {
+    SocketAddr::from(([0, 0, 0, 0], port))
+}