feat(node-wasm, types)!: Add method to get blobs for wasm

grpc/tests/tonic.rs

@@ -1,4 +1,5 @@
 #![cfg(not(target_arch = "wasm32"))]
+//! gRPC tests
 
 use celestia_grpc::types::auth::Account;
 use celestia_grpc::types::tx::sign_tx;
@@ -7,7 +8,7 @@ use celestia_types::blob::MsgPayForBlobs;
 use celestia_types::nmt::Namespace;
 use celestia_types::{AppVersion, Blob};
 
-pub mod utils;
+mod utils;
 
 use crate::utils::{load_account, new_test_client};
 
@@ -101,7 +102,7 @@ async fn submit_blob() {
         .await
         .unwrap();
 
-    tokio::time::sleep(std::time::Duration::from_secs(3)).await;
+    tokio::time::sleep(std::time::Duration::from_secs(5)).await;
 
     let _submitted_tx = client
         .get_tx(response.txhash)

grpc/tests/utils/mod.rs

@@ -1,4 +1,5 @@
 #![cfg(not(target_arch = "wasm32"))]
+//! Utilities for grpc tests
 
 use std::{env, fs};
 
@@ -51,11 +52,11 @@ impl TestAuthInterceptor {
     }
 }
 
-pub fn env_or(var_name: &str, or_value: &str) -> String {
+fn env_or(var_name: &str, or_value: &str) -> String {
     env::var(var_name).unwrap_or_else(|_| or_value.to_owned())
 }
 
-pub async fn new_test_client() -> Result<GrpcClient<TestAuthInterceptor>> {
+pub(crate) async fn new_test_client() -> Result<GrpcClient<TestAuthInterceptor>> {
     let _ = dotenvy::dotenv();
     let url = env_or("CELESTIA_GRPC_URL", CELESTIA_GRPC_URL);
     let grpc_channel = Channel::from_shared(url)?.connect().await?;
@@ -64,7 +65,7 @@ pub async fn new_test_client() -> Result<GrpcClient<TestAuthInterceptor>> {
     Ok(GrpcClient::new(grpc_channel, auth_interceptor))
 }
 
-pub fn load_account(path: &str) -> TestAccount {
+pub(crate) fn load_account(path: &str) -> TestAccount {
     let account_file = format!("{path}.addr");
     let key_file = format!("{path}.plaintext-key");
 

node-wasm/Cargo.toml

@@ -24,7 +24,7 @@ crate-type = ["cdylib", "rlib"]
 
 [target.'cfg(target_arch = "wasm32")'.dependencies]
 blockstore.workspace = true
-celestia-types.workspace = true
+celestia-types = { workspace = true, features = ["wasm-bindgen"] }
 libp2p = { workspace = true, features = ["serde"] }
 lumina-node.workspace = true
 tendermint.workspace = true

node-wasm/src/client.rs

@@ -2,6 +2,8 @@
 
 use std::time::Duration;
 
+use celestia_types::nmt::Namespace;
+use celestia_types::Blob;
 use js_sys::Array;
 use libp2p::identity::Keypair;
 use serde::{Deserialize, Serialize};
@@ -257,6 +259,24 @@ impl NodeClient {
         headers.into()
     }
 
+    /// Request all blobs with provided namespace in the block corresponding to this header
+    /// using bitswap protocol.
+    #[wasm_bindgen(js_name = requestAllBlobs)]
+    pub async fn request_all_blobs(
+        &self,
+        header: JsValue,
+        namespace: Namespace,
+        timeout_secs: Option<f64>,
+    ) -> Result<Vec<Blob>> {
+        let command = NodeCommand::RequestAllBlobs {
+            header,
+            namespace,
+            timeout_secs,
+        };
+        let response = self.worker.exec(command).await?;
+        response.into_blobs().check_variant()?
+    }
+
     /// Get current header syncing info.
     #[wasm_bindgen(js_name = syncerInfo)]
     pub async fn syncer_info(&self) -> Result<SyncingInfoSnapshot> {
@@ -431,7 +451,7 @@ mod tests {
 
     use celestia_rpc::{prelude::*, Client};
     use celestia_types::p2p::PeerId;
-    use celestia_types::ExtendedHeader;
+    use celestia_types::{AppVersion, ExtendedHeader, TxConfig};
     use gloo_timers::future::sleep;
     use libp2p::{multiaddr::Protocol, Multiaddr};
     use rexie::Rexie;
@@ -493,6 +513,38 @@ mod tests {
             .unwrap();
     }
 
+    #[wasm_bindgen_test]
+    async fn get_blob() {
+        remove_database().await.expect("failed to clear db");
+        let rpc_client = Client::new(WS_URL).await.unwrap();
+        let namespace = Namespace::new_v0(&[0xCD, 0xDC, 0xCD, 0xDC, 0xCD, 0xDC]).unwrap();
+        let data = b"Hello, World";
+        let blobs = vec![Blob::new(namespace, data.to_vec(), AppVersion::V3).unwrap()];
+
+        let submitted_height = rpc_client
+            .blob_submit(&blobs, TxConfig::default())
+            .await
+            .expect("successful submission");
+
+        let header = rpc_client
+            .header_get_by_height(submitted_height)
+            .await
+            .expect("header for blob");
+
+        let bridge_ma = fetch_bridge_webtransport_multiaddr(&rpc_client).await;
+        let client = spawn_connected_node(vec![bridge_ma.to_string()]).await;
+
+        let mut blobs = client
+            .request_all_blobs(to_value(&header).unwrap(), namespace, None)
+            .await
+            .expect("to fetch blob");
+
+        assert_eq!(blobs.len(), 1);
+        let blob = blobs.pop().unwrap();
+        assert_eq!(blob.data, data);
+        assert_eq!(blob.namespace, namespace);
+    }
+
     async fn spawn_connected_node(bootnodes: Vec<String>) -> NodeClient {
         let message_channel = MessageChannel::new().unwrap();
         let mut worker = NodeWorker::new(message_channel.port1().into());

node-wasm/src/commands.rs

@@ -1,5 +1,7 @@
 use std::fmt::Debug;
 
+use celestia_types::nmt::Namespace;
+use celestia_types::Blob;
 use enum_as_inner::EnumAsInner;
 use js_sys::Array;
 use libp2p::Multiaddr;
@@ -53,6 +55,12 @@ pub(crate) enum NodeCommand {
     GetSamplingMetadata {
         height: u64,
     },
+    RequestAllBlobs {
+        #[serde(with = "serde_wasm_bindgen::preserve")]
+        header: JsValue,
+        namespace: Namespace,
+        timeout_secs: Option<f64>,
+    },
 }
 
 #[derive(Serialize, Deserialize, Debug)]
@@ -82,6 +90,7 @@ pub(crate) enum WorkerResponse {
     Headers(JsResult<Array, Error>),
     LastSeenNetworkHead(JsResult<JsValue, Error>),
     SamplingMetadata(Result<Option<SamplingMetadata>>),
+    Blobs(Result<Vec<Blob>>),
 }
 
 pub(crate) trait CheckableResponseExt {

node-wasm/src/worker.rs

@@ -1,5 +1,8 @@
 use std::fmt::Debug;
+use std::time::Duration;
 
+use celestia_types::nmt::Namespace;
+use celestia_types::Blob;
 use js_sys::Array;
 use libp2p::{Multiaddr, PeerId};
 use serde::{Deserialize, Serialize};
@@ -246,6 +249,20 @@ impl NodeWorkerInstance {
         Ok(self.node.get_sampling_metadata(height).await?)
     }
 
+    async fn request_all_blobs(
+        &mut self,
+        header: JsValue,
+        namespace: Namespace,
+        timeout_secs: Option<f64>,
+    ) -> Result<Vec<Blob>> {
+        let header = from_value(header)?;
+        let timeout = timeout_secs.map(Duration::from_secs_f64);
+        Ok(self
+            .node
+            .request_all_blobs(&header, namespace, timeout)
+            .await?)
+    }
+
     async fn process_command(&mut self, command: NodeCommand) -> WorkerResponse {
         match command {
             NodeCommand::IsRunning => WorkerResponse::IsRunning(true),
@@ -301,6 +318,14 @@ impl NodeWorkerInstance {
             NodeCommand::GetSamplingMetadata { height } => {
                 WorkerResponse::SamplingMetadata(self.get_sampling_metadata(height).await)
             }
+            NodeCommand::RequestAllBlobs {
+                header,
+                namespace,
+                timeout_secs,
+            } => WorkerResponse::Blobs(
+                self.request_all_blobs(header, namespace, timeout_secs)
+                    .await,
+            ),
             NodeCommand::InternalPing => WorkerResponse::InternalPong,
         }
     }

node/src/store/utils.rs

@@ -55,7 +55,7 @@ impl From<ExtendedHeader> for VerifiedExtendedHeaders {
     }
 }
 
-impl From<&'_ ExtendedHeader> for VerifiedExtendedHeaders {
+impl From<&ExtendedHeader> for VerifiedExtendedHeaders {
     fn from(value: &ExtendedHeader) -> Self {
         Self(vec![value.to_owned()])
     }

rpc/tests/blob.rs

@@ -6,7 +6,7 @@ use std::time::Duration;
 use celestia_rpc::blob::BlobsAtHeight;
 use celestia_rpc::prelude::*;
 use celestia_types::consts::appconsts::AppVersion;
-use celestia_types::{Blob, Commitment};
+use celestia_types::Blob;
 use jsonrpsee::core::client::Subscription;
 
 pub mod utils;
@@ -194,7 +194,7 @@ async fn blob_get_get_proof_wrong_commitment() {
     let namespace = random_ns();
     let data = random_bytes(5);
     let blob = Blob::new(namespace, data, AppVersion::V2).unwrap();
-    let commitment = Commitment(random_bytes_array());
+    let commitment = random_bytes_array().into();
 
     let submitted_height = blob_submit(&client, &[blob.clone()]).await.unwrap();
 

types/Cargo.toml

@@ -39,8 +39,14 @@ serde = { version = "1.0.203", features = ["derive"] }
 serde_repr = { version = "0.1.19", optional = true }
 sha2 = "0.10.6"
 thiserror = "1.0.61"
+
+# `time` is a dependency of a dependency but we need to specify it
+# for fixing rust-lang/rust#125319.
 time = { version = "0.3.36", default-features = false }
 
+[target.'cfg(target_arch = "wasm32")'.dependencies]
+wasm-bindgen = { version = "0.2.95", optional = true }
+
 [dev-dependencies]
 ed25519-consensus = "2.1.0"
 rand = "0.8.5"
@@ -57,7 +63,8 @@ wasm-bindgen-test = "0.3.42"
 default = ["p2p"]
 p2p = ["dep:libp2p-identity", "dep:multiaddr", "dep:serde_repr"]
 test-utils = ["dep:ed25519-consensus", "dep:rand"]
-wasm-bindgen = ["time/wasm-bindgen"]
+wasm-bindgen = ["dep:wasm-bindgen", "time/wasm-bindgen"]
+tonic = ["celestia-proto/tonic"]
 
 [package.metadata.docs.rs]
 features = ["p2p", "test-utils"]

types/src/blob.rs

@@ -17,12 +17,18 @@ pub use self::msg_pay_for_blobs::MsgPayForBlobs;
 pub use celestia_proto::celestia::blob::v1::MsgPayForBlobs as RawMsgPayForBlobs;
 pub use celestia_proto::proto::blob::v1::BlobProto as RawBlob;
 pub use celestia_proto::proto::blob::v1::BlobTx as RawBlobTx;
+#[cfg(all(feature = "wasm-bindgen", target_arch = "wasm32"))]
+use wasm_bindgen::prelude::*;
 
 /// Arbitrary data that can be stored in the network within certain [`Namespace`].
 // NOTE: We don't use the `serde(try_from)` pattern for this type
 // becase JSON representation needs to have `commitment` field but
 // Protobuf definition doesn't.
 #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+#[cfg_attr(
+    all(feature = "wasm-bindgen", target_arch = "wasm32"),
+    wasm_bindgen(getter_with_clone)
+)]
 pub struct Blob {
     /// A [`Namespace`] the [`Blob`] belongs to.
     pub namespace: Namespace,
@@ -395,7 +401,7 @@ mod tests {
     #[test]
     fn validate_blob_commitment_mismatch() {
         let mut blob = sample_blob();
-        blob.commitment.0.fill(7);
+        blob.commitment = [7; 32].into();
 
         blob.validate(AppVersion::V2).unwrap_err();
     }