DA: Verifier and Indexer in Node (#713)

* Verifier and Indexer attached to node

* Serialization for kzgrs verifier settings

nodes/nomos-node/Cargo.toml

@@ -23,6 +23,7 @@ tracing = "0.1"
 multiaddr = "0.18"
 nomos-core = { path = "../../nomos-core" }
 nomos-da-verifier = { path = "../../nomos-services/data-availability/verifier", features = ["rocksdb-backend", "libp2p"] }
+nomos-da-indexer = { path = "../../nomos-services/data-availability/indexer", features = ["rocksdb-backend"] }
 nomos-da-network-service = { path = "../../nomos-services/data-availability/network" }
 nomos-network = { path = "../../nomos-services/network", features = ["libp2p"] }
 nomos-api = { path = "../../nomos-services/api" }

nodes/nomos-node/src/config.rs

@@ -121,6 +121,8 @@ pub struct Config {
     pub network: <NetworkService<NetworkBackend> as ServiceData>::Settings,
     pub da_network:
         <DaNetworkService<DaNetworkValidatorBackend<FillFromNodeList>> as ServiceData>::Settings,
+    pub da_indexer: <crate::DaIndexer as ServiceData>::Settings,
+    pub da_verifier: <crate::DaVerifier as ServiceData>::Settings,
     pub http: <NomosApiService as ServiceData>::Settings,
     pub cryptarchia: <crate::Cryptarchia as ServiceData>::Settings,
 }

nodes/nomos-node/src/lib.rs

@@ -16,9 +16,15 @@ pub use nomos_core::{
     da::blob::select::FillSize as FillSizeWithBlobs, tx::select::FillSize as FillSizeWithTx,
 };
 use nomos_core::{header::HeaderId, tx::Transaction, wire};
+use nomos_da_indexer::consensus::adapters::cryptarchia::CryptarchiaConsensusAdapter;
+use nomos_da_indexer::storage::adapters::rocksdb::RocksAdapter as IndexerStorageAdapter;
+use nomos_da_indexer::DataIndexerService;
 use nomos_da_network_service::backends::libp2p::validator::DaNetworkValidatorBackend;
 use nomos_da_network_service::NetworkService as DaNetworkService;
 use nomos_da_verifier::backend::kzgrs::KzgrsDaVerifier;
+use nomos_da_verifier::network::adapters::libp2p::Libp2pAdapter as VerifierNetworkAdapter;
+use nomos_da_verifier::storage::adapters::rocksdb::RocksAdapter as VerifierStorageAdapter;
+use nomos_da_verifier::DaVerifierService;
 #[cfg(feature = "tracing")]
 use nomos_log::Logger;
 use nomos_mempool::da::service::DaMempoolService;
@@ -68,11 +74,35 @@ pub type DaMempool = DaMempoolService<
     MockPool<HeaderId, BlobInfo, <BlobInfo as DispersedBlobInfo>::BlobId>,
 >;
 
+pub type DaIndexer = DataIndexerService<
+    // Indexer specific.
+    Bytes,
+    IndexerStorageAdapter<Wire, BlobInfo>,
+    CryptarchiaConsensusAdapter<Tx, BlobInfo>,
+    // Cryptarchia specific, should be the same as in `Cryptarchia` type above.
+    cryptarchia_consensus::network::adapters::libp2p::LibP2pAdapter<Tx, BlobInfo>,
+    MockPool<HeaderId, Tx, <Tx as Transaction>::Hash>,
+    MempoolNetworkAdapter<Tx, <Tx as Transaction>::Hash>,
+    MockPool<HeaderId, BlobInfo, <BlobInfo as DispersedBlobInfo>::BlobId>,
+    MempoolNetworkAdapter<BlobInfo, <BlobInfo as DispersedBlobInfo>::BlobId>,
+    FillSizeWithTx<MB16, Tx>,
+    FillSizeWithBlobs<MB16, BlobInfo>,
+    RocksBackend<Wire>,
+>;
+
+pub type DaVerifier = DaVerifierService<
+    KzgrsDaVerifier,
+    VerifierNetworkAdapter<DaBlob, ()>,
+    VerifierStorageAdapter<(), DaBlob, Wire>,
+>;
+
 #[derive(Services)]
 pub struct Nomos {
     #[cfg(feature = "tracing")]
     logging: ServiceHandle<Logger>,
     network: ServiceHandle<NetworkService<NetworkBackend>>,
+    da_indexer: ServiceHandle<DaIndexer>,
+    da_verifier: ServiceHandle<DaVerifier>,
     da_network: ServiceHandle<DaNetworkService<DaNetworkValidatorBackend<FillFromNodeList>>>,
     cl_mempool: ServiceHandle<TxMempool>,
     da_mempool: ServiceHandle<DaMempool>,

nodes/nomos-node/src/main.rs

@@ -91,6 +91,8 @@ fn main() -> Result<()> {
                 registry: registry.clone(),
             },
             da_network: config.da_network,
+            da_indexer: config.da_indexer,
+            da_verifier: config.da_verifier,
             cryptarchia: config.cryptarchia,
             #[cfg(feature = "metrics")]
             metrics: MetricsSettings { registry },

nomos-services/data-availability/indexer/src/lib.rs

@@ -23,7 +23,7 @@ use overwatch_rs::services::state::{NoOperator, NoState};
 use overwatch_rs::services::{ServiceCore, ServiceData, ServiceId};
 use overwatch_rs::DynError;
 use serde::de::DeserializeOwned;
-use serde::Serialize;
+use serde::{Deserialize, Serialize};
 use storage::DaStorageAdapter;
 use tokio::sync::oneshot::Sender;
 use tracing::error;
@@ -363,7 +363,7 @@ where
     }
 }
 
-#[derive(Debug, Clone)]
+#[derive(Debug, Clone, Serialize, Deserialize)]
 pub struct IndexerSettings<S> {
     pub storage: S,
 }

nomos-services/data-availability/indexer/src/storage/adapters/rocksdb.rs

@@ -1,6 +1,7 @@
+// std
 use std::path::PathBuf;
 use std::{marker::PhantomData, ops::Range};
-
+// crates
 use bytes::Bytes;
 use futures::{stream::FuturesUnordered, Stream};
 use nomos_core::da::blob::{
@@ -18,7 +19,8 @@ use overwatch_rs::{
     services::{relay::OutboundRelay, ServiceData},
     DynError,
 };
-
+use serde::{Deserialize, Serialize};
+// internal
 use crate::storage::DaStorageAdapter;
 
 pub struct RocksAdapter<S, B>
@@ -142,7 +144,7 @@ where
     }
 }
 
-#[derive(Debug, Clone)]
+#[derive(Debug, Clone, Serialize, Deserialize)]
 pub struct RocksAdapterSettings {
     pub blob_storage_directory: PathBuf,
 }

nomos-services/data-availability/tests/Cargo.toml

@@ -10,6 +10,7 @@ cryptarchia-consensus = { path = "../../../nomos-services/cryptarchia-consensus"
 cryptarchia-engine = { path = "../../../consensus/cryptarchia-engine", features = ["serde"] }
 cryptarchia-ledger = { path = "../../../ledger/cryptarchia-ledger", features = ["serde"] }
 full-replication = { path = "../../../nomos-da/full-replication" }
+hex = "0.4.3"
 kzgrs-backend = { path = "../../../nomos-da/kzgrs-backend" }
 nomos-core = { path = "../../../nomos-core" }
 nomos-da-indexer = { path = "../indexer", features = ["rocksdb-backend"] }

nomos-services/data-availability/tests/src/verifier_integration.rs

@@ -135,13 +135,13 @@ fn new_node(
     .unwrap()
 }
 
-fn generate_keys() -> (blst::min_sig::SecretKey, blst::min_sig::PublicKey) {
+fn generate_hex_keys() -> (String, String) {
     let mut rng = rand::thread_rng();
     let sk_bytes: [u8; 32] = rng.gen();
     let sk = blst::min_sig::SecretKey::key_gen(&sk_bytes, &[]).unwrap();
 
     let pk = sk.sk_to_pk();
-    (sk, pk)
+    (hex::encode(sk.to_bytes()), hex::encode(pk.to_bytes()))
 }
 
 pub fn rand_data(elements_count: usize) -> Vec<u8> {
@@ -198,8 +198,8 @@ fn test_verifier() {
 
     let blobs_dir = TempDir::new().unwrap().path().to_path_buf();
 
-    let (node1_sk, node1_pk) = generate_keys();
-    let (node2_sk, node2_pk) = generate_keys();
+    let (node1_sk, node1_pk) = generate_hex_keys();
+    let (node2_sk, node2_pk) = generate_hex_keys();
 
     let client_zone = new_client(NamedTempFile::new().unwrap().path().to_path_buf());
 
@@ -213,8 +213,8 @@ fn test_verifier() {
         &blobs_dir,
         vec![node_address(&swarm_config2)],
         KzgrsDaVerifierSettings {
-            sk: node1_sk,
-            nodes_public_keys: vec![node1_pk, node2_pk],
+            sk: node1_sk.clone(),
+            nodes_public_keys: vec![node1_pk.clone(), node2_pk.clone()],
         },
     );
 

nomos-services/data-availability/verifier/Cargo.toml

@@ -5,9 +5,10 @@ edition = "2021"
 
 [dependencies]
 async-trait = "0.1"
-blst = "0.3.11"
+blst = { version = "0.3.11", features = ["serde-secret"] }
 bytes = "1.2"
 futures = "0.3"
+hex = "0.4.3"
 kzgrs-backend = { path = "../../../nomos-da/kzgrs-backend" }
 nomos-core = { path = "../../../nomos-core" }
 nomos-da-storage = { path = "../../../nomos-da/storage" }

nomos-services/data-availability/verifier/src/backend/kzgrs.rs

@@ -4,6 +4,7 @@ use core::fmt;
 use blst::{min_sig::PublicKey, min_sig::SecretKey};
 use kzgrs_backend::{common::blob::DaBlob, verifier::DaVerifier as NomosKzgrsVerifier};
 use nomos_core::da::DaVerifier;
+use serde::{Deserialize, Serialize};
 // internal
 use super::VerifierBackend;
 
@@ -30,7 +31,22 @@ impl VerifierBackend for KzgrsDaVerifier {
     type Settings = KzgrsDaVerifierSettings;
 
     fn new(settings: Self::Settings) -> Self {
-        let verifier = NomosKzgrsVerifier::new(settings.sk, &settings.nodes_public_keys);
+        let bytes = hex::decode(settings.sk).expect("Secret key string should decode to bytes");
+        let secret_key =
+            SecretKey::from_bytes(&bytes).expect("Secret key should be reconstructed from bytes");
+
+        let nodes_public_keys = settings
+            .nodes_public_keys
+            .iter()
+            .map(|pk_hex| {
+                let pk_bytes =
+                    hex::decode(pk_hex).expect("Public key string should decode to bytes");
+                PublicKey::from_bytes(&pk_bytes)
+                    .expect("Public key should be reconstructed from bytes")
+            })
+            .collect::<Vec<PublicKey>>();
+
+        let verifier = NomosKzgrsVerifier::new(secret_key, &nodes_public_keys);
         Self { verifier }
     }
 }
@@ -52,8 +68,8 @@ impl DaVerifier for KzgrsDaVerifier {
 }
 
 // TODO: `sk` and `nodes_public_keys` need to be fetched from the params provider service.
-#[derive(Debug, Clone)]
+#[derive(Debug, Clone, Serialize, Deserialize)]
 pub struct KzgrsDaVerifierSettings {
-    pub sk: SecretKey,
-    pub nodes_public_keys: Vec<PublicKey>,
+    pub sk: String,
+    pub nodes_public_keys: Vec<String>,
 }

nomos-services/data-availability/verifier/src/lib.rs

@@ -6,6 +6,7 @@ use nomos_core::da::blob::Blob;
 // std
 use nomos_storage::StorageService;
 use overwatch_rs::services::life_cycle::LifecycleMessage;
+use serde::{Deserialize, Serialize};
 use std::error::Error;
 use std::fmt::{Debug, Formatter};
 use storage::DaStorageAdapter;
@@ -206,7 +207,7 @@ where
     }
 }
 
-#[derive(Debug, Clone)]
+#[derive(Debug, Clone, Serialize, Deserialize)]
 pub struct DaVerifierServiceSettings<BackendSettings, NetworkSettings, StorageSettings> {
     pub verifier_settings: BackendSettings,
     pub network_adapter_settings: NetworkSettings,

nomos-services/data-availability/verifier/src/storage/adapters/rocksdb.rs

@@ -1,5 +1,5 @@
 // std
-use serde::{de::DeserializeOwned, Serialize};
+use serde::{de::DeserializeOwned, Deserialize, Serialize};
 use std::{marker::PhantomData, path::PathBuf};
 // crates
 use nomos_core::da::blob::Blob;
@@ -117,7 +117,7 @@ fn create_blob_idx(blob_id: &[u8], column_idx: &[u8]) -> [u8; 34] {
     blob_idx
 }
 
-#[derive(Debug, Clone)]
+#[derive(Debug, Clone, Serialize, Deserialize)]
 pub struct RocksAdapterSettings {
     pub blob_storage_directory: PathBuf,
 }

tests/Cargo.toml

@@ -5,6 +5,7 @@ edition = "2021"
 publish = false
 
 [dependencies]
+blst = { version = "0.3.11" }
 nomos-node = { path = "../nodes/nomos-node", default-features = false }
 nomos-network = { path = "../nomos-services/network", features = ["libp2p"] }
 cryptarchia-consensus = { path = "../nomos-services/cryptarchia-consensus" }
@@ -17,8 +18,11 @@ cryptarchia-engine = { path = "../consensus/cryptarchia-engine", features = ["se
 cryptarchia-ledger = { path = "../ledger/cryptarchia-ledger", features = ["serde"] }
 nomos-mempool = { path = "../nomos-services/mempool", features = ["mock", "libp2p"] }
 nomos-da-network-service = { path = "../nomos-services/data-availability/network" }
+nomos-da-indexer = { path = "../nomos-services/data-availability/indexer" }
+nomos-da-verifier = { path = "../nomos-services/data-availability/verifier" }
 subnetworks-assignations = { path = "../nomos-da/network/subnetworks-assignations" }
 full-replication = { path = "../nomos-da/full-replication" }
+hex = "0.4.3"
 kzgrs-backend = { path = "../nomos-da/kzgrs-backend" }
 rand = "0.8"
 once_cell = "1"

tests/src/nodes/nomos.rs

@@ -6,6 +6,7 @@ use std::time::Duration;
 // internal
 use super::{create_tempdir, persist_tempdir, LOGS_PREFIX};
 use crate::{adjust_timeout, get_available_port, ConsensusConfig, Node};
+use blst::min_sig::SecretKey;
 use cryptarchia_consensus::{CryptarchiaInfo, CryptarchiaSettings, TimeConfig};
 use cryptarchia_ledger::{Coin, LedgerState};
 use kzgrs_backend::dispersal::BlobInfo;
@@ -17,8 +18,13 @@ use mixnet::{
     topology::{MixNodeInfo, MixnetTopology},
 };
 use nomos_core::{block::Block, header::HeaderId};
+use nomos_da_indexer::storage::adapters::rocksdb::RocksAdapterSettings as IndexerStorageAdapterSettings;
+use nomos_da_indexer::IndexerSettings;
 use nomos_da_network_service::backends::libp2p::validator::DaNetworkValidatorBackendSettings;
 use nomos_da_network_service::NetworkConfig as DaNetworkConfig;
+use nomos_da_verifier::backend::kzgrs::KzgrsDaVerifierSettings;
+use nomos_da_verifier::storage::adapters::rocksdb::RocksAdapterSettings as VerifierStorageAdapterSettings;
+use nomos_da_verifier::DaVerifierServiceSettings;
 use nomos_libp2p::{Multiaddr, SwarmConfig};
 use nomos_log::{LoggerBackend, LoggerFormat};
 use nomos_mempool::MempoolMetrics;
@@ -28,7 +34,7 @@ use nomos_network::{backends::libp2p::Libp2pConfig, NetworkConfig};
 use nomos_node::{api::AxumBackendSettings, Config, Tx};
 // crates
 use once_cell::sync::Lazy;
-use rand::{thread_rng, Rng};
+use rand::{thread_rng, Rng, RngCore};
 use reqwest::{Client, Url};
 use tempfile::NamedTempFile;
 use time::OffsetDateTime;
@@ -355,6 +361,15 @@ fn create_node_config(
     #[cfg(feature = "mixnet")] mixnet_config: MixnetConfig,
 ) -> Config {
     let swarm_config: SwarmConfig = Default::default();
+
+    let mut rng = rand::thread_rng();
+    let mut buff = [0u8; 32];
+    rng.fill_bytes(&mut buff);
+
+    let verifier_sk = SecretKey::key_gen(&buff, &[]).unwrap();
+    let verifier_pk_bytes = verifier_sk.sk_to_pk().to_bytes();
+    let verifier_sk_bytes = verifier_sk.to_bytes();
+
     let mut config = Config {
         network: NetworkConfig {
             backend: Libp2pConfig {
@@ -380,6 +395,21 @@ fn create_node_config(
                 membership: Default::default(),
             },
         },
+        da_indexer: IndexerSettings {
+            storage: IndexerStorageAdapterSettings {
+                blob_storage_directory: "./".into(),
+            },
+        },
+        da_verifier: DaVerifierServiceSettings {
+            verifier_settings: KzgrsDaVerifierSettings {
+                sk: hex::encode(verifier_sk_bytes),
+                nodes_public_keys: vec![hex::encode(verifier_pk_bytes)],
+            },
+            network_adapter_settings: (),
+            storage_adapter_settings: VerifierStorageAdapterSettings {
+                blob_storage_directory: "./".into(),
+            },
+        },
         log: Default::default(),
         http: nomos_api::ApiServiceSettings {
             backend_settings: AxumBackendSettings {