Api endpoint to add blob and receive attestation

nodes/nomos-node/Cargo.toml

@@ -15,11 +15,13 @@ chrono = "0.4"
 futures = "0.3"
 http = "0.2.9"
 hex = "0.4.3"
+kzgrs-backend = { path = "../../nomos-da/kzgrs-backend" }
 overwatch-rs = { git = "https://github.com/logos-co/Overwatch", rev = "2f70806" }
 overwatch-derive = { git = "https://github.com/logos-co/Overwatch", rev = "ac28d01" }
 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-network = { path = "../../nomos-services/network", features = ["libp2p"] }
 nomos-api = { path = "../../nomos-services/api" }
 nomos-log = { path = "../../nomos-services/log" }
@@ -52,4 +54,4 @@ time = "0.3"
 
 [features]
 mixnet = ["nomos-network/mixnet"]
-metrics = []
+metrics = []

nodes/nomos-node/src/api.rs

@@ -1,3 +1,4 @@
+use std::error::Error;
 use std::{fmt::Debug, hash::Hash};
 
 use axum::{
@@ -10,6 +11,23 @@ use hyper::{
     header::{CONTENT_TYPE, USER_AGENT},
     Body, StatusCode,
 };
+use nomos_api::{
+    http::{cl, consensus, da, libp2p, mempool, metrics, storage},
+    Backend,
+};
+use nomos_core::da::DaVerifier as CoreDaVerifier;
+use nomos_core::{
+    da::{attestation::Attestation, blob::Blob},
+    header::HeaderId,
+    tx::Transaction,
+};
+use nomos_da_verifier::backend::VerifierBackend;
+use nomos_mempool::{
+    network::adapters::libp2p::Libp2pAdapter as MempoolNetworkAdapter,
+    tx::service::openapi::Status, MempoolMetrics,
+};
+use nomos_network::backends::libp2p::Libp2p as NetworkBackend;
+use nomos_storage::backends::StorageSerde;
 use overwatch_rs::overwatch::handle::OverwatchHandle;
 use serde::{de::DeserializeOwned, Deserialize, Serialize};
 use tower_http::{
@@ -19,19 +37,6 @@ use tower_http::{
 use utoipa::OpenApi;
 use utoipa_swagger_ui::SwaggerUi;
 
-use nomos_core::{header::HeaderId, tx::Transaction};
-use nomos_mempool::{
-    network::adapters::libp2p::Libp2pAdapter as MempoolNetworkAdapter,
-    tx::service::openapi::Status, MempoolMetrics,
-};
-use nomos_network::backends::libp2p::Libp2p as NetworkBackend;
-use nomos_storage::backends::StorageSerde;
-
-use nomos_api::{
-    http::{cl, consensus, libp2p, mempool, metrics, storage},
-    Backend,
-};
-
 /// Configuration for the Http Server
 #[derive(Debug, Clone, serde::Deserialize, serde::Serialize)]
 pub struct AxumBackendSettings {
@@ -41,8 +46,11 @@ pub struct AxumBackendSettings {
     pub cors_origins: Vec<String>,
 }
 
-pub struct AxumBackend<T, S, const SIZE: usize> {
+pub struct AxumBackend<A, B, VB, T, S, const SIZE: usize> {
     settings: AxumBackendSettings,
+    _attestation: core::marker::PhantomData<A>,
+    _blob: core::marker::PhantomData<B>,
+    _verifier_backend: core::marker::PhantomData<VB>,
     _tx: core::marker::PhantomData<T>,
     _storage_serde: core::marker::PhantomData<S>,
 }
@@ -61,8 +69,14 @@ pub struct AxumBackend<T, S, const SIZE: usize> {
 struct ApiDoc;
 
 #[async_trait::async_trait]
-impl<T, S, const SIZE: usize> Backend for AxumBackend<T, S, SIZE>
+impl<A, B, VB, T, S, const SIZE: usize> Backend for AxumBackend<A, B, VB, T, S, SIZE>
 where
+    A: Attestation + Serialize + DeserializeOwned + Clone + Send + Sync + 'static,
+    B: Blob + Serialize + DeserializeOwned + Clone + Send + Sync + 'static,
+    <B as Blob>::BlobId: AsRef<[u8]> + Send + Sync + 'static,
+    VB: VerifierBackend + CoreDaVerifier<DaBlob = B, Attestation = A> + Send + Sync + 'static,
+    <VB as VerifierBackend>::Settings: Clone,
+    <VB as CoreDaVerifier>::Error: Error,
     T: Transaction
         + Clone
         + Debug
@@ -86,6 +100,9 @@ where
     {
         Ok(Self {
             settings,
+            _attestation: core::marker::PhantomData,
+            _blob: core::marker::PhantomData,
+            _verifier_backend: core::marker::PhantomData,
             _tx: core::marker::PhantomData,
             _storage_serde: core::marker::PhantomData,
         })
@@ -124,6 +141,7 @@ where
                 "/cryptarchia/headers",
                 routing::get(cryptarchia_headers::<T, S, SIZE>),
             )
+            .route("/da/add_blob", routing::post(add_blob::<A, B, VB, S>))
             .route("/network/info", routing::get(libp2p_info))
             .route("/storage/block", routing::post(block::<S, T>))
             .route("/mempool/add/tx", routing::post(add_tx::<T>))
@@ -261,6 +279,30 @@ where
     ))
 }
 
+#[utoipa::path(
+    post,
+    path = "/da/add_blob",
+    responses(
+        (status = 200, description = "Attestation for DA blob", body = Option<Attestation>),
+        (status = 500, description = "Internal server error", body = String),
+    )
+)]
+async fn add_blob<A, B, VB, SS>(
+    State(handle): State<OverwatchHandle>,
+    Json(blob): Json<B>,
+) -> Response
+where
+    A: Attestation + Serialize + DeserializeOwned + Clone + Send + Sync + 'static,
+    B: Blob + Serialize + DeserializeOwned + Clone + Send + Sync + 'static,
+    <B as Blob>::BlobId: AsRef<[u8]> + Send + Sync + 'static,
+    VB: VerifierBackend + CoreDaVerifier<DaBlob = B, Attestation = A>,
+    <VB as VerifierBackend>::Settings: Clone,
+    <VB as CoreDaVerifier>::Error: Error,
+    SS: StorageSerde + Send + Sync + 'static,
+{
+    make_request_and_return_response!(da::add_blob::<A, B, VB, SS>(&handle, blob))
+}
+
 #[utoipa::path(
     get,
     path = "/network/info",

nodes/nomos-node/src/config.rs

@@ -9,7 +9,9 @@ use clap::{Parser, ValueEnum};
 use color_eyre::eyre::{eyre, Result};
 use cryptarchia_ledger::Coin;
 use hex::FromHex;
+use kzgrs_backend::common::{attestation::Attestation, blob::DaBlob};
 use nomos_api::ApiService;
+use nomos_da_verifier::backend::kzgrs::KzgrsDaVerifier;
 use nomos_libp2p::{secp256k1::SecretKey, Multiaddr};
 use nomos_log::{Logger, LoggerBackend, LoggerFormat};
 use nomos_network::backends::libp2p::Libp2p as NetworkBackend;
@@ -116,7 +118,7 @@ pub struct MetricsArgs {
 pub struct Config {
     pub log: <Logger as ServiceData>::Settings,
     pub network: <NetworkService<NetworkBackend> as ServiceData>::Settings,
-    pub http: <ApiService<AxumBackend<Tx, Wire, MB16>> as ServiceData>::Settings,
+    pub http: <ApiService<AxumBackend<Attestation, DaBlob, KzgrsDaVerifier,Tx, Wire, MB16>> as ServiceData>::Settings,
     pub cryptarchia: <crate::Cryptarchia as ServiceData>::Settings,
 }
 

nodes/nomos-node/src/lib.rs

@@ -8,8 +8,11 @@ use full_replication::{Certificate, VidCertificate};
 use api::AxumBackend;
 use bytes::Bytes;
 pub use config::{Config, CryptarchiaArgs, HttpArgs, LogArgs, MetricsArgs, NetworkArgs};
+use kzgrs_backend::common::attestation::Attestation;
+use kzgrs_backend::common::blob::DaBlob;
 use nomos_api::ApiService;
 use nomos_core::{da::certificate, header::HeaderId, tx::Transaction, wire};
+use nomos_da_verifier::backend::kzgrs::KzgrsDaVerifier;
 use nomos_log::Logger;
 use nomos_mempool::da::verify::fullreplication::DaVerificationProvider as MempoolVerificationProvider;
 use nomos_mempool::network::adapters::libp2p::Libp2pAdapter as MempoolNetworkAdapter;
@@ -79,7 +82,9 @@ pub struct Nomos {
     cl_mempool: ServiceHandle<TxMempool>,
     da_mempool: ServiceHandle<DaMempool>,
     cryptarchia: ServiceHandle<Cryptarchia>,
-    http: ServiceHandle<ApiService<AxumBackend<Tx, Wire, MB16>>>,
+    http: ServiceHandle<
+        ApiService<AxumBackend<Attestation, DaBlob, KzgrsDaVerifier, Tx, Wire, MB16>>,
+    >,
     storage: ServiceHandle<StorageService<RocksBackend<Wire>>>,
     #[cfg(feature = "metrics")]
     metrics: ServiceHandle<Metrics>,

nomos-services/api/Cargo.toml

@@ -9,6 +9,7 @@ axum = ["dep:axum", "dep:hyper", "dep:tower-http", "utoipa-swagger-ui/axum"]
 
 [dependencies]
 async-trait = "0.1"
+bytes = "1.2"
 overwatch-rs = { git = "https://github.com/logos-co/Overwatch", rev = "2f70806" }
 overwatch-derive = { git = "https://github.com/logos-co/Overwatch", rev = "ac28d01" }
 tracing = "0.1"
@@ -21,6 +22,8 @@ nomos-mempool = { path = "../../nomos-services/mempool", features = [
   "openapi",
 ] }
 nomos-metrics = { path = "../../nomos-services/metrics" }
+nomos-da-indexer = { path = "../data-availability/indexer", features = ["rocksdb-backend"] }
+nomos-da-verifier = { path = "../data-availability/verifier", features = ["rocksdb-backend", "libp2p"] }
 nomos-storage = { path = "../../nomos-services/storage", features = ["rocksdb"] }
 nomos-libp2p = { path = "../../nomos-libp2p" }
 full-replication = { path = "../../nomos-da/full-replication" }

nomos-services/api/src/http/da.rs

@@ -0,0 +1,75 @@
+use bytes::Bytes;
+use cryptarchia_consensus::network::adapters::libp2p::LibP2pAdapter as ConsensusNetworkAdapter;
+use nomos_core::da::attestation::Attestation;
+use nomos_core::da::blob::Blob;
+use nomos_core::da::certificate::{self, select::FillSize as FillSizeWithBlobsCertificate};
+use nomos_core::da::DaVerifier as CoreDaVerifier;
+use nomos_core::header::HeaderId;
+use nomos_core::tx::{select::FillSize as FillSizeWithTx, Transaction};
+use nomos_da_indexer::storage::adapters::rocksdb::RocksAdapter as IndexerStorageAdapter;
+use nomos_da_indexer::{
+    consensus::adapters::cryptarchia::CryptarchiaConsensusAdapter, DataIndexerService,
+};
+use nomos_da_verifier::backend::VerifierBackend;
+use nomos_da_verifier::network::adapters::libp2p::Libp2pAdapter;
+use nomos_da_verifier::storage::adapters::rocksdb::RocksAdapter as VerifierStorageAdapter;
+use nomos_da_verifier::{DaVerifierMsg, DaVerifierService};
+use nomos_mempool::backend::mockpool::MockPool;
+use nomos_mempool::da::verify::fullreplication::DaVerificationProvider as MempoolVerificationProvider;
+use nomos_mempool::network::adapters::libp2p::Libp2pAdapter as MempoolNetworkAdapter;
+use nomos_storage::backends::rocksdb::RocksBackend;
+use nomos_storage::backends::StorageSerde;
+use overwatch_rs::overwatch::handle::OverwatchHandle;
+use overwatch_rs::DynError;
+use serde::de::DeserializeOwned;
+use serde::Serialize;
+use std::error::Error;
+use tokio::sync::oneshot;
+
+pub type DaIndexer<T, C, V, SS, const SIZE: usize> = DataIndexerService<
+    // Indexer specific.
+    Bytes,
+    IndexerStorageAdapter<SS, V>,
+    CryptarchiaConsensusAdapter<T, V>,
+    // Cryptarchia specific.
+    ConsensusNetworkAdapter<T, C>,
+    MockPool<HeaderId, T, <T as Transaction>::Hash>,
+    MempoolNetworkAdapter<T, <T as Transaction>::Hash>,
+    MockPool<HeaderId, V, <V as certificate::vid::VidCertificate>::CertificateId>,
+    MempoolNetworkAdapter<C, <C as certificate::Certificate>::Id>,
+    MempoolVerificationProvider,
+    FillSizeWithTx<SIZE, T>,
+    FillSizeWithBlobsCertificate<SIZE, V>,
+    RocksBackend<SS>,
+>;
+
+pub type DaVerifier<A, B, VB, SS> =
+    DaVerifierService<VB, Libp2pAdapter<B, A>, VerifierStorageAdapter<A, B, SS>>;
+
+pub async fn add_blob<A, B, VB, SS>(
+    handle: &OverwatchHandle,
+    blob: B,
+) -> Result<Option<A>, DynError>
+where
+    A: Attestation + Serialize + DeserializeOwned + Clone + Send + Sync + 'static,
+    B: Blob + Serialize + DeserializeOwned + Clone + Send + Sync + 'static,
+    <B as Blob>::BlobId: AsRef<[u8]> + Send + Sync + 'static,
+    VB: VerifierBackend + CoreDaVerifier<DaBlob = B, Attestation = A>,
+    <VB as VerifierBackend>::Settings: Clone,
+    <VB as CoreDaVerifier>::Error: Error,
+    SS: StorageSerde + Send + Sync + 'static,
+{
+    let relay = handle.relay::<DaVerifier<A, B, VB, SS>>().connect().await?;
+    let (sender, receiver) = oneshot::channel();
+    relay
+        .send(DaVerifierMsg::AddBlob {
+            blob,
+            reply_channel: sender,
+        })
+        .await
+        .map_err(|(e, _)| e)?;
+
+    Ok(receiver.await?)
+}
+
+pub async fn get_range() {}

nomos-services/api/src/http/mod.rs

@@ -1,6 +1,7 @@
 pub type DynError = Box<dyn std::error::Error + Send + Sync + 'static>;
 pub mod cl;
 pub mod consensus;
+pub mod da;
 pub mod libp2p;
 pub mod mempool;
 pub mod metrics;