Add executor binary crate (#766)

* Create executor binary reusing validator lib

* Use executor in main

* Use storage from config

Cargo.toml

@@ -28,6 +28,7 @@ members = [
     "nomos-cli",
     "nomos-utils",
     "nodes/nomos-node",
+    "nodes/nomos-executor",
     "mixnet",
     "consensus/carnot-engine",
     "consensus/cryptarchia-engine",

nodes/nomos-executor/Cargo.toml

@@ -0,0 +1,20 @@
+[package]
+name = "nomos-executor"
+version = "0.1.0"
+edition = "2021"
+
+[dependencies]
+color-eyre = "0.6.0"
+nomos-node = { path = "../nomos-node" }
+overwatch-rs = { git = "https://github.com/logos-co/Overwatch", rev = "2f70806" }
+overwatch-derive = { git = "https://github.com/logos-co/Overwatch", rev = "ac28d01" }
+clap = { version = "4.5.13", features = ["derive"] }
+tracing = "0.1.40"
+uuid = { version = "1.10.0", features = ["v4"] }
+serde_yaml = "0.9.34+deprecated"
+
+[features]
+default = ["tracing"]
+mixnet = ["nomos-node/mixnet"]
+metrics = ["nomos-node/metrics"]
+tracing = ["nomos-node/tracing"]

nodes/nomos-executor/src/lib.rs

@@ -0,0 +1,22 @@
+use nomos_node::*;
+use overwatch_derive::Services;
+use overwatch_rs::services::handle::ServiceHandle;
+
+#[derive(Services)]
+pub struct NomosExecutor {
+    #[cfg(feature = "tracing")]
+    logging: ServiceHandle<Logger>,
+    network: ServiceHandle<NetworkService<NetworkBackend>>,
+    da_indexer: ServiceHandle<DaIndexer>,
+    da_verifier: ServiceHandle<DaVerifier>,
+    da_sampling: ServiceHandle<DaSampling>,
+    da_network: ServiceHandle<DaNetworkService<DaNetworkValidatorBackend<NomosDaMembership>>>,
+    cl_mempool: ServiceHandle<TxMempool>,
+    da_mempool: ServiceHandle<DaMempool>,
+    cryptarchia: ServiceHandle<Cryptarchia>,
+    http: ServiceHandle<NomosApiService>,
+    storage: ServiceHandle<StorageService<RocksBackend<Wire>>>,
+    #[cfg(feature = "metrics")]
+    metrics: ServiceHandle<Metrics>,
+    system_sig: ServiceHandle<SystemSig>,
+}

nodes/nomos-executor/src/main.rs

@@ -0,0 +1,96 @@
+use nomos_node::*;
+#[cfg(feature = "metrics")]
+use nomos_node::{MetricsSettings, NomosRegistry};
+
+use clap::Parser;
+use color_eyre::eyre::{eyre, Result};
+use nomos_executor::{NomosExecutor, NomosExecutorServiceSettings};
+use overwatch_rs::overwatch::*;
+use tracing::{span, Level};
+use uuid::Uuid;
+
+#[derive(Parser, Debug)]
+#[command(author, version, about, long_about = None)]
+struct Args {
+    /// Path for a yaml-encoded network config file
+    config: std::path::PathBuf,
+    /// Overrides log config.
+    #[clap(flatten)]
+    log_args: LogArgs,
+    /// Overrides network config.
+    #[clap(flatten)]
+    network_args: NetworkArgs,
+    /// Overrides http config.
+    #[clap(flatten)]
+    http_args: HttpArgs,
+    #[clap(flatten)]
+    cryptarchia_args: CryptarchiaArgs,
+    /// Overrides metrics config.
+    #[clap(flatten)]
+    metrics_args: MetricsArgs,
+}
+
+fn main() -> Result<()> {
+    let Args {
+        config,
+        log_args,
+        http_args,
+        network_args,
+        cryptarchia_args,
+        metrics_args,
+    } = Args::parse();
+    let config = serde_yaml::from_reader::<_, Config>(std::fs::File::open(config)?)?
+        .update_log(log_args)?
+        .update_http(http_args)?
+        .update_network(network_args)?
+        .update_cryptarchia_consensus(cryptarchia_args)?;
+
+    let registry = cfg!(feature = "metrics")
+        .then(|| metrics_args.with_metrics.then(NomosRegistry::default))
+        .flatten();
+
+    #[cfg(debug_assertions)]
+    let debug_span = {
+        let debug_id = Uuid::new_v4();
+        span!(Level::DEBUG, "Nomos", debug_id = debug_id.to_string())
+    };
+    #[cfg(debug_assertions)]
+    let _guard = debug_span.enter();
+    let app = OverwatchRunner::<NomosExecutor>::run(
+        NomosExecutorServiceSettings {
+            network: config.network,
+            #[cfg(feature = "tracing")]
+            logging: config.log,
+            http: config.http,
+            cl_mempool: TxMempoolSettings {
+                backend: (),
+                network: MempoolAdapterSettings {
+                    topic: String::from(CL_TOPIC),
+                    id: <Tx as Transaction>::hash,
+                },
+                registry: registry.clone(),
+            },
+            da_mempool: DaMempoolSettings {
+                backend: (),
+                network: MempoolAdapterSettings {
+                    topic: String::from(DA_TOPIC),
+                    id: <BlobInfo as DispersedBlobInfo>::blob_id,
+                },
+                registry: registry.clone(),
+            },
+            da_network: config.da_network,
+            da_indexer: config.da_indexer,
+            da_sampling: config.da_sampling,
+            da_verifier: config.da_verifier,
+            cryptarchia: config.cryptarchia,
+            #[cfg(feature = "metrics")]
+            metrics: MetricsSettings { registry },
+            storage: config.storage,
+            system_sig: (),
+        },
+        None,
+    )
+    .map_err(|e| eyre!("Error encountered: {}", e))?;
+    app.wait_finished();
+    Ok(())
+}

nodes/nomos-node/src/lib.rs

@@ -9,18 +9,18 @@ use bytes::Bytes;
 use color_eyre::eyre::Result;
 pub use config::{Config, CryptarchiaArgs, HttpArgs, LogArgs, MetricsArgs, NetworkArgs};
 use kzgrs_backend::common::blob::DaBlob;
-use kzgrs_backend::dispersal::BlobInfo;
+pub use kzgrs_backend::dispersal::BlobInfo;
 use nomos_api::ApiService;
-use nomos_core::da::blob::info::DispersedBlobInfo;
+pub use nomos_core::da::blob::info::DispersedBlobInfo;
 pub use nomos_core::{
     da::blob::select::FillSize as FillSizeWithBlobs, tx::select::FillSize as FillSizeWithTx,
 };
-use nomos_core::{header::HeaderId, tx::Transaction, wire};
+pub 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;
+pub use nomos_da_network_service::backends::libp2p::validator::DaNetworkValidatorBackend;
-use nomos_da_network_service::NetworkService as DaNetworkService;
+pub use nomos_da_network_service::NetworkService as DaNetworkService;
 use nomos_da_sampling::backend::kzgrs::KzgrsSamplingBackend;
 use nomos_da_sampling::network::adapters::libp2p::Libp2pAdapter as SamplingLibp2pAdapter;
 use nomos_da_sampling::storage::adapters::rocksdb::RocksAdapter as SamplingStorageAdapter;
@@ -30,19 +30,26 @@ use nomos_da_verifier::network::adapters::libp2p::Libp2pAdapter as VerifierNetwo
 use nomos_da_verifier::storage::adapters::rocksdb::RocksAdapter as VerifierStorageAdapter;
 use nomos_da_verifier::DaVerifierService;
 #[cfg(feature = "tracing")]
-use nomos_log::Logger;
+pub use nomos_log::Logger;
-use nomos_mempool::da::service::DaMempoolService;
+pub use nomos_mempool::da::service::{DaMempoolService, DaMempoolSettings};
-use nomos_mempool::network::adapters::libp2p::Libp2pAdapter as MempoolNetworkAdapter;
+pub use nomos_mempool::network::adapters::libp2p::{
+    Libp2pAdapter as MempoolNetworkAdapter, Settings as MempoolAdapterSettings,
+};
+pub use nomos_mempool::TxMempoolSettings;
 use nomos_mempool::{backend::mockpool::MockPool, TxMempoolService};
+pub use nomos_metrics::NomosRegistry;
 #[cfg(feature = "metrics")]
-use nomos_metrics::Metrics;
+pub use nomos_metrics::{Metrics, MetricsSettings};
-use nomos_network::backends::libp2p::Libp2p as NetworkBackend;
+pub use nomos_network::backends::libp2p::Libp2p as NetworkBackend;
-use nomos_network::NetworkService;
+pub use nomos_network::NetworkService;
-use nomos_storage::{
+pub use nomos_storage::{
-    backends::{rocksdb::RocksBackend, StorageSerde},
+    backends::{
+        rocksdb::{RocksBackend, RocksBackendSettings},
+        StorageSerde,
+    },
     StorageService,
 };
-use nomos_system_sig::SystemSig;
+pub use nomos_system_sig::SystemSig;
 use overwatch_derive::*;
 use overwatch_rs::services::handle::ServiceHandle;
 use rand_chacha::ChaCha20Rng;