feat(DA): request shares HTTP API

andrussal · andrussal · commit 5017f8661d4f · 2025-03-14T11:33:32.000+01:00
diff --git a/clients/common-http-client/Cargo.toml b/clients/common-http-client/Cargo.toml
@@ -5,6 +5,7 @@ name    = "common-http-client"
 version = "0.1.0"
 
 [dependencies]
+futures             = "0.3.31"
 nomos-core          = { workspace = true }
 nomos-da-messages   = { workspace = true }
 reqwest             = "0.12"
diff --git a/clients/common-http-client/src/lib.rs b/clients/common-http-client/src/lib.rs
@@ -1,14 +1,16 @@
 use std::{fmt::Debug, sync::Arc};
 
+use futures::{Stream, StreamExt};
 use nomos_core::da::blob::Blob;
-use nomos_da_messages::http::da::{DABlobCommitmentsRequest, DAGetLightBlobReq};
+use nomos_da_messages::http::da::{DABlobCommitmentsRequest, DAGetLightBlobReq, GetSharesRequest};
 use reqwest::{Client, ClientBuilder, RequestBuilder, StatusCode, Url};
 use serde::{de::DeserializeOwned, Serialize};
 
 // These could be moved into shared location, perhaps to upcoming `nomos-lib`
 const DA_GET_SHARED_COMMITMENTS: &str = "/da/get-commitments";
 
 const DA_GET_LIGHT_BLOB: &str = "/da/get-blob";
+const DA_GET_SHARES: &str = "/da/sampling/shares";
 
 #[derive(thiserror::Error, Debug)]
 pub enum Error {
@@ -131,4 +133,47 @@ impl CommonHttpClient {
         let request_url = base_url.join(path).map_err(Error::Url)?;
         self.get(request_url, &request).await
     }
+
+    pub async fn get_shares<B>(
+        &self,
+        base_url: Url,
+        blob_id: B::BlobId,
+        requested_shares: Vec<B::ColumnIndex>,
+        filter_shares: Vec<B::ColumnIndex>,
+        return_available: bool,
+    ) -> Result<impl Stream<Item = B::LightBlob>, Error>
+    where
+        B: Blob,
+        <B as Blob>::BlobId: serde::Serialize + Send + Sync,
+        <B as Blob>::ColumnIndex: serde::Serialize + DeserializeOwned + Send + Sync,
+        <B as Blob>::LightBlob: DeserializeOwned + Send + Sync,
+    {
+        let request: GetSharesRequest<B> = GetSharesRequest {
+            blob_id,
+            requested_shares,
+            filter_shares,
+            return_available,
+        };
+        let request_url = base_url
+            .join(DA_GET_SHARES.trim_start_matches('/'))
+            .map_err(Error::Url)?;
+        let mut request = self.client.get(request_url).json(&request);
+
+        if let Some(basic_auth) = &self.basic_auth {
+            request = request.basic_auth(&basic_auth.username, basic_auth.password.as_deref());
+        }
+
+        let response = request.send().await.map_err(Error::Request)?;
+        let status = response.status();
+
+        let shares_stream = response.bytes_stream().filter_map(|item| async move {
+            item.ok()
+                .and_then(|bytes| serde_json::from_slice::<B::LightBlob>(&bytes).ok())
+        });
+        match status {
+            StatusCode::OK => Ok(shares_stream),
+            StatusCode::INTERNAL_SERVER_ERROR => Err(Error::Server("Error".to_string())),
+            _ => Err(Error::Server(format!("Unexpected response [{status}]",))),
+        }
+    }
 }
diff --git a/clients/executor-http-client/Cargo.toml b/clients/executor-http-client/Cargo.toml
@@ -6,6 +6,7 @@ version = "0.1.0"
 
 [dependencies]
 common-http-client = { workspace = true }
+futures            = "0.3.31"
 nomos-core         = { workspace = true }
 nomos-executor     = { workspace = true }
 reqwest            = "0.12"
diff --git a/clients/executor-http-client/src/lib.rs b/clients/executor-http-client/src/lib.rs
@@ -1,5 +1,6 @@
 pub use common_http_client::BasicAuthCredentials;
 use common_http_client::{CommonHttpClient, Error};
+use futures::Stream;
 use nomos_core::da::blob::Blob;
 use nomos_executor::api::{handlers::DispersalRequest, paths};
 use reqwest::Url;
@@ -68,4 +69,29 @@ impl ExecutorHttpClient {
             .get_blob::<B, C>(base_url, blob_id, column_idx)
             .await
     }
+
+    pub async fn get_shares<B>(
+        &self,
+        base_url: Url,
+        blob_id: B::BlobId,
+        requested_shares: Vec<B::ColumnIndex>,
+        filter_shares: Vec<B::ColumnIndex>,
+        return_available: bool,
+    ) -> Result<impl Stream<Item = B::LightBlob>, Error>
+    where
+        B: Blob,
+        <B as Blob>::BlobId: serde::Serialize + Send + Sync,
+        <B as Blob>::ColumnIndex: serde::Serialize + DeserializeOwned + Send + Sync,
+        <B as Blob>::LightBlob: DeserializeOwned + Send + Sync,
+    {
+        self.client
+            .get_shares::<B>(
+                base_url,
+                blob_id,
+                requested_shares,
+                filter_shares,
+                return_available,
+            )
+            .await
+    }
 }
diff --git a/nodes/nomos-executor/src/api/backend.rs b/nodes/nomos-executor/src/api/backend.rs
@@ -5,7 +5,7 @@ use hyper::header::{CONTENT_TYPE, USER_AGENT};
 use nomos_api::Backend;
 use nomos_core::{
     da::{
-        blob::{info::DispersedBlobInfo, metadata, Blob},
+        blob::{info::DispersedBlobInfo, metadata, Blob, LightBlob},
         DaVerifier as CoreDaVerifier,
     },
     header::HeaderId,
@@ -20,8 +20,8 @@ use nomos_libp2p::PeerId;
 use nomos_mempool::{tx::service::openapi::Status, MempoolMetrics};
 use nomos_node::api::handlers::{
     add_blob, add_blob_info, add_tx, blacklisted_peers, block, block_peer, cl_metrics, cl_status,
-    cryptarchia_headers, cryptarchia_info, da_get_commitments, da_get_light_blob, get_range,
-    libp2p_info, unblock_peer,
+    cryptarchia_headers, cryptarchia_info, da_get_commitments, da_get_light_blob, da_get_shares,
+    get_range, libp2p_info, unblock_peer,
 };
 use nomos_storage::backends::StorageSerde;
 use overwatch::overwatch::handle::OverwatchHandle;
@@ -162,8 +162,15 @@ where
     DaBlob: Blob + Serialize + DeserializeOwned + Clone + Send + Sync + 'static,
     <DaBlob as Blob>::BlobId:
         AsRef<[u8]> + Clone + Serialize + DeserializeOwned + Send + Sync + 'static,
-    <DaBlob as Blob>::ColumnIndex: AsRef<[u8]> + DeserializeOwned + Send + Sync + 'static,
-    <DaBlob as Blob>::LightBlob: Serialize + DeserializeOwned + Clone + Send + Sync + 'static,
+    <DaBlob as Blob>::ColumnIndex:
+        AsRef<[u8]> + Serialize + DeserializeOwned + Hash + Eq + Send + Sync + 'static,
+    <DaBlob as Blob>::LightBlob: LightBlob<ColumnIndex = <DaBlob as Blob>::ColumnIndex>
+        + Serialize
+        + DeserializeOwned
+        + Clone
+        + Send
+        + Sync
+        + 'static,
     <DaBlob as Blob>::SharedCommitments:
         Serialize + DeserializeOwned + Clone + Send + Sync + 'static,
     DaBlobInfo: DispersedBlobInfo<BlobId = [u8; 32]>
@@ -412,6 +419,10 @@ where
                 paths::DA_GET_LIGHT_BLOB,
                 routing::get(da_get_light_blob::<DaStorageSerializer, DaBlob>),
             )
+            .route(
+                paths::DA_GET_SHARES,
+                routing::get(da_get_shares::<DaStorageSerializer, DaBlob>),
+            )
             .with_state(handle);
 
         Server::bind(&self.settings.address)
diff --git a/nodes/nomos-node/src/api/backend.rs b/nodes/nomos-node/src/api/backend.rs
@@ -5,7 +5,7 @@ use hyper::header::{CONTENT_TYPE, USER_AGENT};
 use nomos_api::Backend;
 use nomos_core::{
     da::{
-        blob::{info::DispersedBlobInfo, metadata::Metadata, Blob},
+        blob::{info::DispersedBlobInfo, metadata::Metadata, Blob, LightBlob},
         DaVerifier as CoreDaVerifier,
     },
     header::HeaderId,
@@ -31,8 +31,8 @@ use utoipa_swagger_ui::SwaggerUi;
 
 use super::handlers::{
     add_blob, add_blob_info, add_tx, blacklisted_peers, block, block_peer, cl_metrics, cl_status,
-    cryptarchia_headers, cryptarchia_info, da_get_commitments, da_get_light_blob, get_range,
-    libp2p_info, unblock_peer,
+    cryptarchia_headers, cryptarchia_info, da_get_commitments, da_get_light_blob, da_get_shares,
+    get_range, libp2p_info, unblock_peer,
 };
 use crate::api::paths;
 
@@ -140,8 +140,15 @@ where
     DaBlob: Blob + Serialize + DeserializeOwned + Clone + Send + Sync + 'static,
     <DaBlob as Blob>::BlobId:
         AsRef<[u8]> + Clone + Serialize + DeserializeOwned + Send + Sync + 'static,
-    <DaBlob as Blob>::ColumnIndex: AsRef<[u8]> + DeserializeOwned + Send + Sync + 'static,
-    DaBlob::LightBlob: Serialize + DeserializeOwned + Clone + Send + Sync + 'static,
+    <DaBlob as Blob>::ColumnIndex:
+        AsRef<[u8]> + Serialize + DeserializeOwned + Hash + Eq + Send + Sync + 'static,
+    DaBlob::LightBlob: LightBlob<ColumnIndex = <DaBlob as Blob>::ColumnIndex>
+        + Serialize
+        + DeserializeOwned
+        + Clone
+        + Send
+        + Sync
+        + 'static,
     DaBlob::SharedCommitments: Serialize + DeserializeOwned + Clone + Send + Sync + 'static,
     DaBlobInfo: DispersedBlobInfo<BlobId = [u8; 32]>
         + Clone
@@ -377,6 +384,10 @@ where
                 paths::DA_GET_LIGHT_BLOB,
                 routing::get(da_get_light_blob::<DaStorageSerializer, DaBlob>),
             )
+            .route(
+                paths::DA_GET_SHARES,
+                routing::get(da_get_shares::<DaStorageSerializer, DaBlob>),
+            )
             .with_state(handle);
 
         Server::bind(&self.settings.address)
diff --git a/nodes/nomos-node/src/api/handlers.rs b/nodes/nomos-node/src/api/handlers.rs
@@ -1,24 +1,28 @@
 use std::{error::Error, fmt::Debug, hash::Hash};
 
 use axum::{
+    body::StreamBody,
     extract::{Query, State},
-    response::Response,
+    response::{IntoResponse, Response},
     Json,
 };
+use http::{header, StatusCode};
 use nomos_api::http::{
     cl, consensus,
     da::{self, PeerMessagesFactory},
     libp2p, mempool, storage,
 };
 use nomos_core::{
     da::{
-        blob::{info::DispersedBlobInfo, metadata::Metadata, Blob},
+        blob::{info::DispersedBlobInfo, metadata::Metadata, Blob, LightBlob},
         BlobId, DaVerifier as CoreDaVerifier,
     },
     header::HeaderId,
     tx::Transaction,
 };
-use nomos_da_messages::http::da::{DABlobCommitmentsRequest, DAGetLightBlobReq, GetRangeReq};
+use nomos_da_messages::http::da::{
+    DABlobCommitmentsRequest, DAGetLightBlobReq, GetRangeReq, GetSharesRequest,
+};
 use nomos_da_network_core::SubnetworkId;
 use nomos_da_network_service::backends::NetworkBackend;
 use nomos_da_sampling::backend::DaSamplingServiceBackend;
@@ -259,7 +263,8 @@ where
     B: Blob + Serialize + DeserializeOwned + Clone + Send + Sync + 'static,
     <B as Blob>::BlobId: AsRef<[u8]> + Send + Sync + 'static,
     <B as Blob>::ColumnIndex: AsRef<[u8]> + Send + Sync + 'static,
-    <B as Blob>::LightBlob: Serialize + DeserializeOwned + Clone + Send + Sync + 'static,
+    <B as Blob>::LightBlob:
+        LightBlob + Serialize + DeserializeOwned + Clone + Send + Sync + 'static,
     <B as Blob>::SharedCommitments: Serialize + DeserializeOwned + Clone + Send + Sync + 'static,
     M: MembershipHandler<NetworkId = SubnetworkId, Id = PeerId>
         + Clone
@@ -505,7 +510,7 @@ where
     DaBlob: Blob,
     <DaBlob as Blob>::BlobId: AsRef<[u8]> + DeserializeOwned + Clone + Send + Sync + 'static,
     <DaBlob as Blob>::ColumnIndex: AsRef<[u8]> + DeserializeOwned + Send + Sync + 'static,
-    <DaBlob as Blob>::LightBlob: Serialize + DeserializeOwned + Send + Sync + 'static,
+    <DaBlob as Blob>::LightBlob: LightBlob + Serialize + DeserializeOwned + Send + Sync + 'static,
     StorageOp: StorageSerde + Send + Sync + 'static,
     <StorageOp as StorageSerde>::Error: Send + Sync,
 {
@@ -516,6 +521,50 @@ where
     ))
 }
 
+#[utoipa::path(
+    get,
+    path = paths::DA_GET_SHARES,
+    responses(
+        (status = 200, description = "Request shares for a blob", body = GetSharesRequest<DaBlob>),
+        (status = 500, description = "Internal server error", body = StreamBody),
+    )
+)]
+pub async fn da_get_shares<StorageOp, DaBlob>(
+    State(handle): State<OverwatchHandle>,
+    Json(request): Json<GetSharesRequest<DaBlob>>,
+) -> Response
+where
+    DaBlob: Blob,
+    <DaBlob as Blob>::BlobId: AsRef<[u8]> + DeserializeOwned + Clone + Send + Sync + 'static,
+    <DaBlob as Blob>::ColumnIndex: serde::Serialize + DeserializeOwned + Send + Sync + Eq + Hash,
+    <DaBlob as Blob>::LightBlob: LightBlob<ColumnIndex = <DaBlob as Blob>::ColumnIndex>
+        + Serialize
+        + DeserializeOwned
+        + Send
+        + Sync
+        + 'static,
+    StorageOp: StorageSerde + Send + Sync + 'static,
+    <StorageOp as StorageSerde>::Error: Send + Sync,
+    <DaBlob as Blob>::LightBlob: LightBlob,
+    <DaBlob as Blob>::ColumnIndex: 'static,
+{
+    match storage::get_shares::<StorageOp, DaBlob>(
+        &handle,
+        request.blob_id,
+        request.requested_shares,
+        request.filter_shares,
+        request.return_available,
+    )
+    .await
+    {
+        Ok(shares) => {
+            let body = StreamBody::new(shares);
+            IntoResponse::into_response(([(header::CONTENT_TYPE, "application/x-ndjson")], body))
+        }
+        Err(e) => IntoResponse::into_response((StatusCode::INTERNAL_SERVER_ERROR, e.to_string())),
+    }
+}
+
 #[utoipa::path(
     post,
     path = paths::MEMPOOL_ADD_TX,
diff --git a/nodes/nomos-node/src/api/paths.rs b/nodes/nomos-node/src/api/paths.rs
@@ -5,6 +5,7 @@ pub const CRYPTARCHIA_HEADERS: &str = "/cryptarchia/headers";
 pub const DA_ADD_BLOB: &str = "/da/add-blob";
 pub const DA_GET_RANGE: &str = "/da/get-range";
 pub const DA_GET_SHARED_COMMITMENTS: &str = "/da/get-commitments";
+pub const DA_GET_SHARES: &str = "/da/sampling/shares";
 pub const DA_GET_LIGHT_BLOB: &str = "/da/get-blob";
 pub const DA_BLOCK_PEER: &str = "/da/block-peer";
 pub const DA_UNBLOCK_PEER: &str = "/da/unblock-peer";
diff --git a/nomos-core/chain-defs/src/da/blob/mod.rs b/nomos-core/chain-defs/src/da/blob/mod.rs
@@ -2,6 +2,11 @@ pub mod info;
 pub mod metadata;
 pub mod select;
 
+pub trait LightBlob {
+    type ColumnIndex;
+    fn column_idx(&self) -> Self::ColumnIndex;
+}
+
 pub trait Blob {
     type BlobId;
     type ColumnIndex;
diff --git a/nomos-da/kzgrs-backend/src/common/blob.rs b/nomos-da/kzgrs-backend/src/common/blob.rs
@@ -126,6 +126,14 @@ pub struct DaLightBlob {
     pub rows_proofs: Vec<Proof>,
 }
 
+impl blob::LightBlob for DaLightBlob {
+    type ColumnIndex = [u8; 2];
+
+    fn column_idx(&self) -> Self::ColumnIndex {
+        self.column_idx.to_be_bytes()
+    }
+}
+
 #[derive(Debug, Clone, Default, Eq, PartialEq, Serialize, Deserialize)]
 pub struct DaBlobSharedCommitments {
     #[serde(
diff --git a/nomos-da/network/messages/src/http/da.rs b/nomos-da/network/messages/src/http/da.rs
@@ -25,3 +25,14 @@ pub struct DAGetLightBlobReq<B: Blob> {
     pub blob_id: B::BlobId,
     pub column_idx: B::ColumnIndex,
 }
+
+#[derive(Serialize, Deserialize)]
+pub struct GetSharesRequest<B: Blob>
+where
+    <B as Blob>::ColumnIndex: Serialize + DeserializeOwned,
+{
+    pub blob_id: B::BlobId,
+    pub requested_shares: Vec<B::ColumnIndex>,
+    pub filter_shares: Vec<B::ColumnIndex>,
+    pub return_available: bool,
+}
diff --git a/nomos-services/api/Cargo.toml b/nomos-services/api/Cargo.toml
@@ -11,7 +11,9 @@ instrumentation = []
 
 [dependencies]
 async-trait              = "0.1"
+bytes                    = "1.10.1"
 cryptarchia-consensus    = { workspace = true, features = ["libp2p"] }
+futures                  = "0.3.31"
 kzgrs-backend            = { workspace = true }
 nomos-blend-service      = { workspace = true, features = ["libp2p"] }
 nomos-core               = { workspace = true }
@@ -31,6 +33,7 @@ overwatch                = { workspace = true }
 overwatch-derive         = { workspace = true }
 rand                     = "0.8"
 serde                    = { version = "1", features = ["derive"] }
+serde_json               = "1.0.140"
 subnetworks-assignations = { workspace = true }
 tokio                    = { version = "1", default-features = false, features = ["sync"] }
 
diff --git a/nomos-services/api/src/http/da.rs b/nomos-services/api/src/http/da.rs
diff --git a/nomos-services/api/src/http/storage.rs b/nomos-services/api/src/http/storage.rs
diff --git a/tests/Cargo.toml b/tests/Cargo.toml
diff --git a/tests/src/tests/da/api.rs b/tests/src/tests/da/api.rs
