DA Http Endpoints (#670)

* Api endpoint to add blob and receive attestation

* Nomos DA Certificate serialization

* Kzgrs verification params provider

* Indexer http api using kzgrs

* Plug da http endpoints into nomos api

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" }

nodes/nomos-node/src/api.rs

@@ -1,3 +1,5 @@
+use std::error::Error;
+use std::ops::Range;
 use std::{fmt::Debug, hash::Hash};
 
 use axum::{
@@ -10,6 +12,25 @@ 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::certificate::metadata::Metadata;
+use nomos_core::da::{certificate, DaVerifier as CoreDaVerifier};
+use nomos_core::{
+    da::{attestation::Attestation, blob::Blob},
+    header::HeaderId,
+    tx::Transaction,
+};
+use nomos_da_verifier::backend::VerifierBackend;
+use nomos_mempool::verify::MempoolVerificationProvider;
+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 +40,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 +49,14 @@ pub struct AxumBackendSettings {
     pub cors_origins: Vec<String>,
 }
 
-pub struct AxumBackend<T, S, const SIZE: usize> {
+pub struct AxumBackend<A, B, C, V, VP, VB, T, S, const SIZE: usize> {
     settings: AxumBackendSettings,
+    _attestation: core::marker::PhantomData<A>,
+    _blob: core::marker::PhantomData<B>,
+    _certificate: core::marker::PhantomData<C>,
+    _vid: core::marker::PhantomData<V>,
+    _params_provider: core::marker::PhantomData<VP>,
+    _verifier_backend: core::marker::PhantomData<VB>,
     _tx: core::marker::PhantomData<T>,
     _storage_serde: core::marker::PhantomData<S>,
 }
@@ -61,8 +75,46 @@ 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, C, V, VP, VB, T, S, const SIZE: usize> Backend
+    for AxumBackend<A, B, C, V, VP, 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,
+    C: certificate::Certificate<Id = [u8; 32]>
+        + Clone
+        + Debug
+        + Serialize
+        + DeserializeOwned
+        + Send
+        + Sync
+        + 'static,
+    <C as certificate::Certificate>::Id: Clone + Send + Sync,
+    V: certificate::vid::VidCertificate<CertificateId = [u8; 32]>
+        + From<C>
+        + Eq
+        + Debug
+        + Metadata
+        + Hash
+        + Clone
+        + Serialize
+        + DeserializeOwned
+        + Send
+        + Sync
+        + 'static,
+    <V as certificate::vid::VidCertificate>::CertificateId: Debug + Clone + Ord + Hash,
+    <V as Metadata>::AppId: AsRef<[u8]> + Clone + Serialize + DeserializeOwned + Send + Sync,
+    <V as Metadata>::Index:
+        AsRef<[u8]> + Clone + Serialize + DeserializeOwned + PartialOrd + Send + Sync,
+    VP: MempoolVerificationProvider<
+            Payload = C,
+            Parameters = <C as certificate::Certificate>::VerificationParameters,
+        > + 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 +138,12 @@ where
     {
         Ok(Self {
             settings,
+            _attestation: core::marker::PhantomData,
+            _blob: core::marker::PhantomData,
+            _certificate: core::marker::PhantomData,
+            _vid: core::marker::PhantomData,
+            _params_provider: core::marker::PhantomData,
+            _verifier_backend: core::marker::PhantomData,
             _tx: core::marker::PhantomData,
             _storage_serde: core::marker::PhantomData,
         })
@@ -124,6 +182,11 @@ where
                 "/cryptarchia/headers",
                 routing::get(cryptarchia_headers::<T, S, SIZE>),
             )
+            .route("/da/add_blob", routing::post(add_blob::<A, B, VB, S>))
+            .route(
+                "/da/get_range",
+                routing::post(get_range::<T, C, V, VP, S, SIZE>),
+            )
             .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 +324,100 @@ 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))
+}
+
+#[derive(Serialize, Deserialize)]
+struct GetRangeReq<V: Metadata>
+where
+    <V as Metadata>::AppId: Serialize + DeserializeOwned,
+    <V as Metadata>::Index: Serialize + DeserializeOwned,
+{
+    app_id: <V as Metadata>::AppId,
+    range: Range<<V as Metadata>::Index>,
+}
+
+#[utoipa::path(
+    post,
+    path = "/da/get_range",
+    responses(
+        (status = 200, description = "Range of blobs", body = Vec<([u8;8], Option<DaBlob>)>),
+        (status = 500, description = "Internal server error", body = String),
+    )
+)]
+async fn get_range<Tx, C, V, VP, SS, const SIZE: usize>(
+    State(handle): State<OverwatchHandle>,
+    Json(GetRangeReq { app_id, range }): Json<GetRangeReq<V>>,
+) -> Response
+where
+    Tx: Transaction
+        + Eq
+        + Clone
+        + Debug
+        + Hash
+        + Serialize
+        + DeserializeOwned
+        + Send
+        + Sync
+        + 'static,
+    <Tx as Transaction>::Hash: std::cmp::Ord + Debug + Send + Sync + 'static,
+    C: certificate::Certificate<Id = [u8; 32]>
+        + Clone
+        + Debug
+        + Serialize
+        + DeserializeOwned
+        + Send
+        + Sync
+        + 'static,
+    <C as certificate::Certificate>::Id: Clone + Send + Sync,
+    V: certificate::vid::VidCertificate<CertificateId = [u8; 32]>
+        + From<C>
+        + Eq
+        + Debug
+        + Metadata
+        + Hash
+        + Clone
+        + Serialize
+        + DeserializeOwned
+        + Send
+        + Sync
+        + 'static,
+    <V as certificate::vid::VidCertificate>::CertificateId: Debug + Clone + Ord + Hash,
+    <V as Metadata>::AppId: AsRef<[u8]> + Clone + Serialize + DeserializeOwned + Send + Sync,
+    <V as Metadata>::Index:
+        AsRef<[u8]> + Clone + Serialize + DeserializeOwned + PartialOrd + Send + Sync,
+    VP: MempoolVerificationProvider<
+        Payload = C,
+        Parameters = <C as certificate::Certificate>::VerificationParameters,
+    >,
+    SS: StorageSerde + Send + Sync + 'static,
+{
+    make_request_and_return_response!(da::get_range::<Tx, C, V, VP, SS, SIZE>(
+        &handle, app_id, range
+    ))
+}
+
 #[utoipa::path(
     get,
     path = "/network/info",
@@ -277,7 +434,7 @@ async fn libp2p_info(State(handle): State<OverwatchHandle>) -> Response {
     get,
     path = "/storage/block",
     responses(
-        (status = 200, description = "Get the block by block id", body = Block<Tx, full_replication::Certificate>),
+        (status = 200, description = "Get the block by block id", body = Block<Tx, kzgrs_backend::dispersal::Certificate>),
         (status = 500, description = "Internal server error", body = String),
     )
 )]

nodes/nomos-node/src/config.rs

@@ -3,13 +3,11 @@ use std::{
     path::PathBuf,
 };
 
-use crate::api::AxumBackend;
+use crate::NomosApiService;
-use crate::{Tx, Wire, MB16};
 use clap::{Parser, ValueEnum};
 use color_eyre::eyre::{eyre, Result};
 use cryptarchia_ledger::Coin;
 use hex::FromHex;
-use nomos_api::ApiService;
 use nomos_libp2p::{secp256k1::SecretKey, Multiaddr};
 use nomos_log::{Logger, LoggerBackend, LoggerFormat};
 use nomos_network::backends::libp2p::Libp2p as NetworkBackend;
@@ -116,7 +114,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: <NomosApiService as ServiceData>::Settings,
     pub cryptarchia: <crate::Cryptarchia as ServiceData>::Settings,
 }
 

nodes/nomos-node/src/lib.rs

@@ -3,16 +3,19 @@ mod config;
 mod tx;
 
 use color_eyre::eyre::Result;
-use full_replication::{Certificate, VidCertificate};
+use kzgrs_backend::dispersal::{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;
 #[cfg(feature = "tracing")]
 use nomos_log::Logger;
-use nomos_mempool::da::verify::fullreplication::DaVerificationProvider as MempoolVerificationProvider;
+use nomos_mempool::da::verify::kzgrs::DaVerificationProvider as MempoolVerificationProvider;
 use nomos_mempool::network::adapters::libp2p::Libp2pAdapter as MempoolNetworkAdapter;
 use nomos_mempool::{backend::mockpool::MockPool, TxMempoolService};
 #[cfg(feature = "metrics")]
@@ -38,6 +41,20 @@ use serde::{de::DeserializeOwned, Serialize};
 
 pub use tx::Tx;
 
+pub type NomosApiService = ApiService<
+    AxumBackend<
+        Attestation,
+        DaBlob,
+        Certificate,
+        VidCertificate,
+        MempoolVerificationProvider,
+        KzgrsDaVerifier,
+        Tx,
+        Wire,
+        MB16,
+    >,
+>;
+
 pub const CL_TOPIC: &str = "cl";
 pub const DA_TOPIC: &str = "da";
 const MB16: usize = 1024 * 1024 * 16;
@@ -81,7 +98,7 @@ pub struct Nomos {
     cl_mempool: ServiceHandle<TxMempool>,
     da_mempool: ServiceHandle<DaMempool>,
     cryptarchia: ServiceHandle<Cryptarchia>,
-    http: ServiceHandle<ApiService<AxumBackend<Tx, Wire, MB16>>>,
+    http: ServiceHandle<NomosApiService>,
     storage: ServiceHandle<StorageService<RocksBackend<Wire>>>,
     #[cfg(feature = "metrics")]
     metrics: ServiceHandle<Metrics>,

nodes/nomos-node/src/main.rs

@@ -1,4 +1,4 @@
-use full_replication::Certificate;
+use kzgrs_backend::dispersal::Certificate;
 #[cfg(feature = "metrics")]
 use nomos_metrics::MetricsSettings;
 use nomos_node::{
@@ -80,9 +80,10 @@ fn main() -> Result<()> {
                     topic: String::from(nomos_node::DA_TOPIC),
                     id: <Certificate as certificate::Certificate>::id,
                 },
-                verification_provider: full_replication::CertificateVerificationParameters {
+                verification_provider:
-                    threshold: 0,
+                    kzgrs_backend::dispersal::CertificateVerificationParameters {
-                },
+                        nodes_public_keys: Default::default(),
+                    },
                 registry: registry.clone(),
             },
             cryptarchia: config.cryptarchia,

nomos-da/kzgrs-backend/Cargo.toml

@@ -9,7 +9,7 @@ edition = "2021"
 ark-ff = "0.4"
 ark-serialize = "0.4.2"
 ark-poly = "0.4.2"
-bitvec = "1.0.1"
+bitvec = { version = "1.0.1", features = ["serde"] }
 blake2 = "0.10"
 blst = { version = "0.3.11", features = ["serde"] }
 itertools = "0.12"

nomos-da/kzgrs-backend/src/common/blob.rs

@@ -1,16 +1,16 @@
 // std
-use std::io::Cursor;
 // crates
-use ark_serialize::*;
 use kzgrs::Proof;
 use nomos_core::da::blob;
-use serde::ser::SerializeSeq;
+use serde::{Deserialize, Serialize};
-use serde::{Deserialize, Deserializer, Serialize, Serializer};
 use sha3::{Digest, Sha3_256};
 // internal
 use super::build_attestation_message;
 use crate::common::Column;
 use crate::common::Commitment;
+use crate::common::{
+    deserialize_canonical, deserialize_vec_canonical, serialize_canonical, serialize_vec_canonical,
+};
 
 #[derive(Debug, Clone, Serialize, Deserialize)]
 pub struct DaBlob {
@@ -44,7 +44,7 @@ pub struct DaBlob {
 
 impl DaBlob {
     pub fn id(&self) -> Vec<u8> {
-        build_attestation_message(&self.aggregated_column_commitment, &self.rows_commitments)
+        build_attestation_message(&self.aggregated_column_commitment, &self.rows_commitments).into()
     }
 
     pub fn column_id(&self) -> Vec<u8> {
@@ -58,59 +58,6 @@ impl blob::Blob for DaBlob {
     type BlobId = Vec<u8>;
 
     fn id(&self) -> Self::BlobId {
-        build_attestation_message(&self.aggregated_column_commitment, &self.rows_commitments)
+        build_attestation_message(&self.aggregated_column_commitment, &self.rows_commitments).into()
     }
 }
-
-fn serialize_canonical<S, T>(value: &T, serializer: S) -> Result<S::Ok, S::Error>
-where
-    S: Serializer,
-    T: CanonicalSerialize,
-{
-    let mut bytes = Vec::new();
-    value
-        .serialize_compressed(&mut bytes)
-        .map_err(serde::ser::Error::custom)?;
-    serializer.serialize_bytes(&bytes)
-}
-
-fn deserialize_canonical<'de, D, T>(deserializer: D) -> Result<T, D::Error>
-where
-    D: Deserializer<'de>,
-    T: CanonicalDeserialize,
-{
-    let bytes: Vec<u8> = serde::Deserialize::deserialize(deserializer)?;
-    let mut cursor = Cursor::new(bytes);
-    T::deserialize_compressed(&mut cursor).map_err(serde::de::Error::custom)
-}
-
-fn serialize_vec_canonical<S, T>(values: &[T], serializer: S) -> Result<S::Ok, S::Error>
-where
-    S: Serializer,
-    T: CanonicalSerialize,
-{
-    let mut container = serializer.serialize_seq(Some(values.len()))?;
-    for value in values {
-        let mut bytes = Vec::new();
-        value
-            .serialize_compressed(&mut bytes)
-            .map_err(serde::ser::Error::custom)?;
-        container.serialize_element(&bytes)?;
-    }
-    container.end()
-}
-
-fn deserialize_vec_canonical<'de, D, T>(deserializer: D) -> Result<Vec<T>, D::Error>
-where
-    D: Deserializer<'de>,
-    T: CanonicalDeserialize,
-{
-    let bytes_vecs: Vec<Vec<u8>> = Deserialize::deserialize(deserializer)?;
-    bytes_vecs
-        .iter()
-        .map(|bytes| {
-            let mut cursor = Cursor::new(bytes);
-            T::deserialize_compressed(&mut cursor).map_err(serde::de::Error::custom)
-        })
-        .collect()
-}

nomos-da/kzgrs-backend/src/common/mod.rs

@@ -2,8 +2,9 @@ pub mod attestation;
 pub mod blob;
 
 // std
-use ark_serialize::CanonicalSerialize;
+use ark_serialize::{CanonicalDeserialize, CanonicalSerialize};
-use serde::{Deserialize, Serialize};
+use serde::ser::SerializeSeq;
+use serde::{Deserialize, Deserializer, Serialize, Serializer};
 use std::io::Cursor;
 // crates
 use blake2::digest::{Update, VariableOutput};
@@ -156,7 +157,7 @@ pub fn hash_column_and_commitment<const HASH_SIZE: usize>(
 pub fn build_attestation_message(
     aggregated_column_commitment: &Commitment,
     rows_commitments: &[Commitment],
-) -> Vec<u8> {
+) -> [u8; 32] {
     let mut hasher = Sha3_256::new();
     Digest::update(
         &mut hasher,
@@ -165,7 +166,7 @@ pub fn build_attestation_message(
     for c in rows_commitments {
         Digest::update(&mut hasher, commitment_to_bytes(c));
     }
-    hasher.finalize().to_vec()
+    hasher.finalize().into()
 }
 
 pub fn commitment_to_bytes(commitment: &Commitment) -> Vec<u8> {
@@ -175,3 +176,56 @@ pub fn commitment_to_bytes(commitment: &Commitment) -> Vec<u8> {
         .expect("Serialization of commitment should work");
     buff.into_inner()
 }
+
+pub fn serialize_canonical<S, T>(value: &T, serializer: S) -> Result<S::Ok, S::Error>
+where
+    S: Serializer,
+    T: CanonicalSerialize,
+{
+    let mut bytes = Vec::new();
+    value
+        .serialize_compressed(&mut bytes)
+        .map_err(serde::ser::Error::custom)?;
+    serializer.serialize_bytes(&bytes)
+}
+
+pub fn deserialize_canonical<'de, D, T>(deserializer: D) -> Result<T, D::Error>
+where
+    D: Deserializer<'de>,
+    T: CanonicalDeserialize,
+{
+    let bytes: Vec<u8> = serde::Deserialize::deserialize(deserializer)?;
+    let mut cursor = Cursor::new(bytes);
+    T::deserialize_compressed(&mut cursor).map_err(serde::de::Error::custom)
+}
+
+pub fn serialize_vec_canonical<S, T>(values: &[T], serializer: S) -> Result<S::Ok, S::Error>
+where
+    S: Serializer,
+    T: CanonicalSerialize,
+{
+    let mut container = serializer.serialize_seq(Some(values.len()))?;
+    for value in values {
+        let mut bytes = Vec::new();
+        value
+            .serialize_compressed(&mut bytes)
+            .map_err(serde::ser::Error::custom)?;
+        container.serialize_element(&bytes)?;
+    }
+    container.end()
+}
+
+pub fn deserialize_vec_canonical<'de, D, T>(deserializer: D) -> Result<Vec<T>, D::Error>
+where
+    D: Deserializer<'de>,
+    T: CanonicalDeserialize,
+{
+    let bytes_vecs: Vec<Vec<u8>> = Deserialize::deserialize(deserializer)?;
+    bytes_vecs
+        .iter()
+        .map(|bytes| {
+            let mut cursor = Cursor::new(bytes);
+            T::deserialize_compressed(&mut cursor).map_err(serde::de::Error::custom)
+        })
+        .collect()
+}

nomos-da/kzgrs-backend/src/dispersal.rs

@@ -8,22 +8,34 @@ use blst::BLST_ERROR;
 use kzgrs::{Commitment, KzgRsError};
 use nomos_core::da::certificate::metadata::Next;
 use nomos_core::da::certificate::{self, metadata};
+use serde::{Deserialize, Serialize};
 
 // internal
 use crate::common::{attestation::Attestation, build_attestation_message, NOMOS_DA_DST};
+use crate::common::{
+    deserialize_canonical, deserialize_vec_canonical, serialize_canonical, serialize_vec_canonical,
+};
 use crate::encoder::EncodedData;
 
-#[derive(Debug, Clone, PartialEq)]
+#[derive(Debug, Clone, Eq, PartialEq, Serialize, Deserialize)]
 pub struct Certificate {
     aggregated_signatures: Signature,
     signers: BitVec<u8>,
+    #[serde(
+        serialize_with = "serialize_canonical",
+        deserialize_with = "deserialize_canonical"
+    )]
     aggregated_column_commitment: Commitment,
+    #[serde(
+        serialize_with = "serialize_vec_canonical",
+        deserialize_with = "deserialize_vec_canonical"
+    )]
     row_commitments: Vec<Commitment>,
     metadata: Metadata,
 }
 
 impl Certificate {
-    pub fn id(&self) -> Vec<u8> {
+    pub fn id(&self) -> [u8; 32] {
         build_attestation_message(&self.aggregated_column_commitment, &self.row_commitments)
     }
 
@@ -90,6 +102,12 @@ impl Certificate {
     }
 }
 
+impl Hash for Certificate {
+    fn hash<H: Hasher>(&self, state: &mut H) {
+        state.write(<Certificate as certificate::Certificate>::id(self).as_ref());
+    }
+}
+
 fn aggregate_signatures(signatures: Vec<Signature>) -> Result<Signature, BLST_ERROR> {
     let refs: Vec<&Signature> = signatures.iter().collect();
     AggregateSignature::aggregate(&refs, true).map(|agg_sig| agg_sig.to_signature())
@@ -111,7 +129,7 @@ pub struct CertificateVerificationParameters {
 
 impl certificate::Certificate for Certificate {
     type Signature = Signature;
-    type Id = Vec<u8>;
+    type Id = [u8; 32];
     type VerificationParameters = CertificateVerificationParameters;
 
     fn signers(&self) -> Vec<bool> {
@@ -131,10 +149,10 @@ impl certificate::Certificate for Certificate {
     }
 }
 
-#[derive(Copy, Clone, Default, Debug, Ord, PartialOrd, PartialEq, Eq)]
+#[derive(Copy, Clone, Default, Debug, PartialEq, PartialOrd, Ord, Eq, Serialize, Deserialize)]
 pub struct Index([u8; 8]);
 
-#[derive(Default, Debug, Copy, Clone, Eq, PartialEq)]
+#[derive(Default, Debug, Copy, Clone, Serialize, Deserialize, Eq, PartialEq)]
 pub struct Metadata {
     app_id: [u8; 32],
     index: Index,
@@ -146,17 +164,17 @@ impl Metadata {
     }
 }
 
-#[derive(Debug, Clone, Eq, PartialEq)]
+#[derive(Debug, Clone, Serialize, Deserialize, Eq, PartialEq)]
 pub struct VidCertificate {
-    id: Vec<u8>,
+    id: [u8; 32],
     metadata: Metadata,
 }
 
 impl certificate::vid::VidCertificate for VidCertificate {
-    type CertificateId = Vec<u8>;
+    type CertificateId = [u8; 32];
 
     fn certificate_id(&self) -> Self::CertificateId {
-        self.id.clone()
+        self.id
     }
 
     fn size(&self) -> usize {

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,9 +22,12 @@ 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" }
+kzgrs-backend = { path = "../../nomos-da/kzgrs-backend" }
 serde = { version = "1", features = ["derive"] }
 tokio = { version = "1.33", default-features = false, features = ["sync"] }
 

nomos-services/api/src/http/consensus/cryptarchia.rs

@@ -9,7 +9,7 @@ use cryptarchia_consensus::{
     network::adapters::libp2p::LibP2pAdapter as ConsensusNetworkAdapter, ConsensusMsg,
     CryptarchiaConsensus, CryptarchiaInfo,
 };
-use full_replication::{Certificate, VidCertificate};
+use kzgrs_backend::dispersal::{Certificate, VidCertificate};
 use nomos_core::{
     da::certificate::{self, select::FillSize as FillSizeWithBlobsCertificate},
     header::HeaderId,
@@ -17,7 +17,7 @@ use nomos_core::{
 };
 use nomos_mempool::{
     backend::mockpool::MockPool,
-    da::verify::fullreplication::DaVerificationProvider as MempoolVerificationProvider,
+    da::verify::kzgrs::DaVerificationProvider as MempoolVerificationProvider,
     network::adapters::libp2p::Libp2pAdapter as MempoolNetworkAdapter,
 };
 use nomos_storage::backends::{rocksdb::RocksBackend, StorageSerde};

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

@@ -0,0 +1,142 @@
+use bytes::Bytes;
+use core::ops::Range;
+use nomos_core::da::attestation::Attestation;
+use nomos_core::da::blob::Blob;
+use nomos_core::da::certificate::metadata::Metadata;
+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::DaMsg;
+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::network::adapters::libp2p::Libp2pAdapter as MempoolNetworkAdapter;
+use nomos_mempool::verify::MempoolVerificationProvider;
+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 std::fmt::Debug;
+use std::hash::Hash;
+use tokio::sync::oneshot;
+
+pub type DaIndexer<Tx, C, V, VP, SS, const SIZE: usize> = DataIndexerService<
+    // Indexer specific.
+    Bytes,
+    IndexerStorageAdapter<SS, V>,
+    CryptarchiaConsensusAdapter<Tx, V>,
+    // Cryptarchia specific, should be the same as in `Cryptarchia` type above.
+    cryptarchia_consensus::network::adapters::libp2p::LibP2pAdapter<Tx, V>,
+    MockPool<HeaderId, Tx, <Tx as Transaction>::Hash>,
+    MempoolNetworkAdapter<Tx, <Tx as Transaction>::Hash>,
+    MockPool<HeaderId, V, [u8; 32]>,
+    MempoolNetworkAdapter<C, <C as certificate::Certificate>::Id>,
+    VP,
+    FillSizeWithTx<SIZE, Tx>,
+    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<Tx, C, V, VP, SS, const SIZE: usize>(
+    handle: &OverwatchHandle,
+    app_id: <V as Metadata>::AppId,
+    range: Range<<V as Metadata>::Index>,
+) -> Result<Vec<(<V as Metadata>::Index, Option<Bytes>)>, DynError>
+where
+    Tx: Transaction
+        + Eq
+        + Clone
+        + Debug
+        + Hash
+        + Serialize
+        + DeserializeOwned
+        + Send
+        + Sync
+        + 'static,
+    <Tx as Transaction>::Hash: std::cmp::Ord + Debug + Send + Sync + 'static,
+    C: certificate::Certificate<Id = [u8; 32]>
+        + Clone
+        + Debug
+        + Serialize
+        + DeserializeOwned
+        + Send
+        + Sync
+        + 'static,
+    <C as certificate::Certificate>::Id: Clone + Send + Sync,
+    V: certificate::vid::VidCertificate<CertificateId = [u8; 32]>
+        + From<C>
+        + Eq
+        + Debug
+        + Metadata
+        + Hash
+        + Clone
+        + Serialize
+        + DeserializeOwned
+        + Send
+        + Sync
+        + 'static,
+    <V as certificate::vid::VidCertificate>::CertificateId: Debug + Clone + Ord + Hash,
+    <V as Metadata>::AppId: AsRef<[u8]> + Serialize + Clone + Send + Sync,
+    <V as Metadata>::Index:
+        AsRef<[u8]> + Serialize + DeserializeOwned + Clone + PartialOrd + Send + Sync,
+    VP: MempoolVerificationProvider<
+        Payload = C,
+        Parameters = <C as certificate::Certificate>::VerificationParameters,
+    >,
+    SS: StorageSerde + Send + Sync + 'static,
+{
+    let relay = handle
+        .relay::<DaIndexer<Tx, C, V, VP, SS, SIZE>>()
+        .connect()
+        .await?;
+    let (sender, receiver) = oneshot::channel();
+    relay
+        .send(DaMsg::GetRange {
+            app_id,
+            range,
+            reply_channel: sender,
+        })
+        .await
+        .map_err(|(e, _)| e)?;
+
+    Ok(receiver.await?)
+}

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;

nomos-services/mempool/Cargo.toml

@@ -14,6 +14,7 @@ nomos-metrics = { path = "../../nomos-services/metrics" }
 nomos-network = { path = "../network" }
 nomos-core = { path = "../../nomos-core" }
 full-replication = { path = "../../nomos-da/full-replication" }
+kzgrs-backend = { path = "../../nomos-da/kzgrs-backend" }
 overwatch-rs = { git = "https://github.com/logos-co/Overwatch", rev = "2f70806" }
 rand = { version = "0.8", optional = true }
 serde = { version = "1.0", features = ["derive"] }

nomos-services/mempool/src/da/verify/kzgrs.rs

@@ -0,0 +1,22 @@
+use kzgrs_backend::dispersal::{Certificate, CertificateVerificationParameters};
+
+use crate::verify::MempoolVerificationProvider;
+
+pub struct DaVerificationProvider {
+    settings: CertificateVerificationParameters,
+}
+
+#[async_trait::async_trait]
+impl MempoolVerificationProvider for DaVerificationProvider {
+    type Payload = Certificate;
+    type Parameters = CertificateVerificationParameters;
+    type Settings = CertificateVerificationParameters;
+
+    fn new(settings: Self::Settings) -> Self {
+        Self { settings }
+    }
+
+    async fn get_parameters(&self, _: &Self::Payload) -> Self::Parameters {
+        self.settings.clone()
+    }
+}

nomos-services/mempool/src/da/verify/mod.rs

@@ -1 +1,2 @@
 pub mod fullreplication;
+pub mod kzgrs;

tests/Cargo.toml

@@ -17,6 +17,7 @@ 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"] }
 full-replication = { path = "../nomos-da/full-replication" }
+kzgrs-backend = { path = "../nomos-da/kzgrs-backend" }
 rand = "0.8"
 once_cell = "1"
 secp256k1 = { version = "0.26", features = ["rand"] }

tests/src/nodes/nomos.rs

@@ -7,7 +7,7 @@ use super::{create_tempdir, persist_tempdir, LOGS_PREFIX};
 use crate::{adjust_timeout, get_available_port, ConsensusConfig, Node};
 use cryptarchia_consensus::{CryptarchiaInfo, CryptarchiaSettings, TimeConfig};
 use cryptarchia_ledger::{Coin, LedgerState};
-use full_replication::Certificate;
+use kzgrs_backend::dispersal::Certificate;
 #[cfg(feature = "mixnet")]
 use mixnet::{
     address::NodeAddress,