diff --git a/Cargo.lock b/Cargo.lock index 50c8c8504..d82ccfaf8 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -58,12 +58,6 @@ dependencies = [ "memchr", ] -[[package]] -name = "allocator-api2" -version = "0.2.21" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "683d7910e743518b0e34f1186f92494becacb047c7b6bf616c96772180fef923" - [[package]] name = "anstream" version = "0.6.20" @@ -483,32 +477,6 @@ version = "0.5.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "3a8241f3ebb85c056b509d4327ad0358fbbba6ffb340bf388f26350aeda225b1" -[[package]] -name = "beetswap" -version = "0.5.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "1a920844a4696d53e72190a8348f0048a6799647b3448d38bef90b85e0310df0" -dependencies = [ - "asynchronous-codec", - "blockstore", - "bytes", - "cid", - "fnv", - "futures-core", - "futures-timer", - "futures-util", - "libp2p-core", - "libp2p-identity", - "libp2p-swarm", - "multihash-codetable", - "quick-protobuf", - "smallvec", - "thiserror 2.0.16", - "tracing", - "unsigned-varint 0.8.0", - "web-time", -] - [[package]] name = "bindgen" version = "0.72.1" @@ -861,7 +829,6 @@ dependencies = [ "base64 0.22.1", "celestia-proto", "celestia-types", - "dotenvy", "futures-util", "getrandom 0.2.16", "gloo-net", @@ -874,7 +841,6 @@ dependencies = [ "multiaddr", "nmt-rs", "rand 0.8.5", - "rust-embed", "send_wrapper 0.6.0", "serde", "serde_json", @@ -2021,8 +1987,6 @@ version = "0.15.5" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "9229cfe53dfd69f0609a49f65461bd93001ea1ef889cd5529dd176593f5338a1" dependencies = [ - "allocator-api2", - "equivalent", "foldhash", ] @@ -2035,6 +1999,15 @@ dependencies = [ "hashbrown 0.14.5", ] +[[package]] +name = "hashlink" +version = "0.10.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7382cf6263419f2d8df38c55d7da83da5c18aef87fc7a7fc1fb1e344edfe14c1" +dependencies = [ + "hashbrown 0.15.5", +] + [[package]] name = "heapless" version = "0.7.17" @@ -2882,9 +2855,9 @@ dependencies = [ [[package]] name = "libp2p-core" -version = "0.43.1" +version = "0.43.2" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "4d28e2d2def7c344170f5c6450c0dbe3dfef655610dbfde2f6ac28a527abbe36" +checksum = "249128cd37a2199aff30a7675dffa51caf073b51aa612d2f544b19932b9aebca" dependencies = [ "either", "fnv", @@ -2937,7 +2910,7 @@ dependencies = [ "futures", "futures-timer", "getrandom 0.2.16", - "hashlink", + "hashlink 0.9.1", "hex_fmt", "libp2p-core", "libp2p-identity", @@ -2975,9 +2948,9 @@ dependencies = [ [[package]] name = "libp2p-identity" -version = "0.2.12" +version = "0.2.13" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "3104e13b51e4711ff5738caa1fb54467c8604c2e94d607e27745bcf709068774" +checksum = "f0c7892c221730ba55f7196e98b0b8ba5e04b4155651736036628e9f73ed6fc3" dependencies = [ "bs58", "ed25519-dalek", @@ -3152,19 +3125,19 @@ dependencies = [ [[package]] name = "libp2p-swarm" -version = "0.47.0" +version = "0.47.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "6aa762e5215919a34e31c35d4b18bf2e18566ecab7f8a3d39535f4a3068f8b62" +checksum = "ce88c6c4bf746c8482480345ea3edfd08301f49e026889d1cbccfa1808a9ed9e" dependencies = [ "either", "fnv", "futures", "futures-timer", "getrandom 0.2.16", + "hashlink 0.10.0", "libp2p-core", "libp2p-identity", "libp2p-swarm-derive", - "lru", "multistream-select", "rand 0.8.5", "smallvec", @@ -3187,16 +3160,16 @@ dependencies = [ [[package]] name = "libp2p-tcp" -version = "0.44.0" +version = "0.44.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "65b4e030c52c46c8d01559b2b8ca9b7c4185f10576016853129ca1fe5cd1a644" +checksum = "fb6585b9309699f58704ec9ab0bb102eca7a3777170fa91a8678d73ca9cafa93" dependencies = [ "futures", "futures-timer", "if-watch", "libc", "libp2p-core", - "socket2 0.5.10", + "socket2 0.6.0", "tokio", "tracing", ] @@ -3394,15 +3367,6 @@ dependencies = [ "tracing-subscriber", ] -[[package]] -name = "lru" -version = "0.12.5" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "234cf4f4a04dc1f57e24b96cc0cd600cf2af460d4161ac5ecdd0af8e1f3b2a38" -dependencies = [ - "hashbrown 0.15.5", -] - [[package]] name = "lru-slab" version = "0.1.2" @@ -3440,7 +3404,6 @@ version = "0.18.2" dependencies = [ "async-trait", "backoff", - "beetswap", "blockstore", "bytes", "celestia-proto", @@ -3482,6 +3445,8 @@ dependencies = [ "tokio-stream", "tokio-util", "tracing", + "tracing-subscriber", + "tracing-web", "uniffi", "void", "wasm-bindgen", @@ -3556,10 +3521,12 @@ dependencies = [ name = "lumina-utils" version = "0.5.2" dependencies = [ + "dotenvy", "futures", "gloo-timers 0.3.0", "js-sys", "pin-project", + "rust-embed", "send_wrapper 0.6.0", "tokio", "tokio-util", diff --git a/Cargo.toml b/Cargo.toml index eb3e73d1b..06935f06b 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -18,7 +18,6 @@ members = [ edition = "2024" [workspace.dependencies] -beetswap = "0.5" blockstore = "0.8" leopard-codec = "0.2" lumina-node = { version = "0.18.2", path = "node" } @@ -94,7 +93,6 @@ web-sys = "0.3.70" [patch.crates-io] # Uncomment to apply local changes -# beetswap = { path = "../beetswap" } # blockstore = { path = "../blockstore" } # nmt-rs = { path = "../nmt-rs" } # libp2p = { path = "../../rust-libp2p/libp2p" } diff --git a/ci/Dockerfile.node b/ci/Dockerfile.node index 705d6a2de..e0cdb5082 100644 --- a/ci/Dockerfile.node +++ b/ci/Dockerfile.node @@ -6,8 +6,8 @@ FROM docker.io/alpine:3.20 RUN apk update && apk add --no-cache bash jq dasel # Copy in the binary -COPY --from=ghcr.io/celestiaorg/celestia-node:v0.28.2 /bin/celestia /bin/celestia -COPY --from=ghcr.io/celestiaorg/celestia-node:v0.28.2 /bin/cel-key /bin/cel-key +COPY --from=celestia-node /bin/celestia /bin/celestia +COPY --from=celestia-node /bin/cel-key /bin/cel-key COPY ./run-node.sh /opt/entrypoint.sh diff --git a/ci/docker-compose.yml b/ci/docker-compose.yml index 425d239d0..5be19b360 100644 --- a/ci/docker-compose.yml +++ b/ci/docker-compose.yml @@ -38,7 +38,7 @@ services: node-0: image: node - platform: "linux/amd64" + # platform: "linux/amd64" build: context: . dockerfile: Dockerfile.node @@ -63,7 +63,7 @@ services: node-1: image: node - platform: "linux/amd64" + # platform: "linux/amd64" build: context: . dockerfile: Dockerfile.node @@ -76,7 +76,7 @@ services: node-2: image: node - platform: "linux/amd64" + # platform: "linux/amd64" build: context: . dockerfile: Dockerfile.node diff --git a/flake.lock b/flake.lock index 1937c4077..558673b56 100644 --- a/flake.lock +++ b/flake.lock @@ -8,11 +8,11 @@ "rust-analyzer-src": "rust-analyzer-src" }, "locked": { - "lastModified": 1758264155, - "narHash": "sha256-sgg1sd/pYO9C7ccY9tAvR392CDscx8sqXrHkxspjIH0=", + "lastModified": 1770447430, + "narHash": "sha256-smrRbWhvJF6BATB6pXbD8Cp04HRrVcYQkXqOhUF81nk=", "owner": "nix-community", "repo": "fenix", - "rev": "a7d9df0179fcc48259a68b358768024f8e5a6372", + "rev": "e1b28f6ca0d1722edceec1f2f3501558988d1aed", "type": "github" }, "original": { @@ -26,11 +26,11 @@ "nixpkgs-lib": "nixpkgs-lib" }, "locked": { - "lastModified": 1756770412, - "narHash": "sha256-+uWLQZccFHwqpGqr2Yt5VsW/PbeJVTn9Dk6SHWhNRPw=", + "lastModified": 1769996383, + "narHash": "sha256-AnYjnFWgS49RlqX7LrC4uA+sCCDBj0Ry/WOJ5XWAsa0=", "owner": "hercules-ci", "repo": "flake-parts", - "rev": "4524271976b625a4a605beefd893f270620fd751", + "rev": "57928607ea566b5db3ad13af0e57e921e6b12381", "type": "github" }, "original": { @@ -41,11 +41,11 @@ }, "nixpkgs": { "locked": { - "lastModified": 1758213207, - "narHash": "sha256-rqoqF0LEi+6ZT59tr+hTQlxVwrzQsET01U4uUdmqRtM=", + "lastModified": 1769461804, + "narHash": "sha256-msG8SU5WsBUfVVa/9RPLaymvi5bI8edTavbIq3vRlhI=", "owner": "nixos", "repo": "nixpkgs", - "rev": "f4b140d5b253f5e2a1ff4e5506edbf8267724bde", + "rev": "bfc1b8a4574108ceef22f02bafcf6611380c100d", "type": "github" }, "original": { @@ -57,11 +57,11 @@ }, "nixpkgs-lib": { "locked": { - "lastModified": 1754788789, - "narHash": "sha256-x2rJ+Ovzq0sCMpgfgGaaqgBSwY+LST+WbZ6TytnT9Rk=", + "lastModified": 1769909678, + "narHash": "sha256-cBEymOf4/o3FD5AZnzC3J9hLbiZ+QDT/KDuyHXVJOpM=", "owner": "nix-community", "repo": "nixpkgs.lib", - "rev": "a73b9c743612e4244d865a2fdee11865283c04e6", + "rev": "72716169fe93074c333e8d0173151350670b824c", "type": "github" }, "original": { @@ -80,11 +80,11 @@ "rust-analyzer-src": { "flake": false, "locked": { - "lastModified": 1758224093, - "narHash": "sha256-buZMH6NgzSLowTda+aArct5ISsMR/S888EdFaqUvbog=", + "lastModified": 1770290336, + "narHash": "sha256-rJ79U68ZLjCSg1Qq+63aBXi//W7blaKiYq9NnfeTboA=", "owner": "rust-lang", "repo": "rust-analyzer", - "rev": "958a8d06e3e5ba7dca7cc23b0639335071d65f2a", + "rev": "d2a00da09293267e5be2efb216698762929d7140", "type": "github" }, "original": { diff --git a/node-uniffi/src/types/event.rs b/node-uniffi/src/types/event.rs index ae839181f..570e15aaf 100644 --- a/node-uniffi/src/types/event.rs +++ b/node-uniffi/src/types/event.rs @@ -73,14 +73,14 @@ pub enum NodeEvent { /// The column of the share. column: u16, /// Share sampling timed out. - timed_out: bool, + failed: bool, }, /// Sampling result. SamplingResult { /// The block height that was sampled. height: u64, /// Sampling timed out. - timed_out: bool, + failed: bool, /// How much time sampling took in milliseconds. took_ms: u64, }, @@ -187,21 +187,21 @@ impl From for NodeEvent { square_width, row, column, - timed_out, + failed, } => NodeEvent::ShareSamplingResult { height, square_width, row, column, - timed_out, + failed, }, LuminaNodeEvent::SamplingResult { height, - timed_out, + failed, took, } => NodeEvent::SamplingResult { height, - timed_out, + failed, took_ms: took.as_millis() as u64, }, LuminaNodeEvent::FatalDaserError { error } => NodeEvent::FatalDaserError { error }, diff --git a/node/Cargo.toml b/node/Cargo.toml index ce2016843..c3d52a8b2 100644 --- a/node/Cargo.toml +++ b/node/Cargo.toml @@ -23,7 +23,6 @@ categories = [ crate-type = ["lib", "staticlib", "cdylib"] [dependencies] -beetswap.workspace = true celestia-proto.workspace = true celestia-types.workspace = true lumina-utils = { workspace = true, features = ["executor", "token", "time"] } @@ -84,7 +83,6 @@ rustls-pki-types = "1.11" [target.'cfg(target_arch = "wasm32")'.dependencies] backoff = { version = "0.4", features = ["wasm-bindgen"] } -beetswap = { workspace = true, features = ["wasm-bindgen"] } blockstore = { workspace = true, features = ["indexeddb"] } celestia-types = { workspace = true, features = ["wasm-bindgen"] } getrandom_03.workspace = true @@ -117,6 +115,9 @@ web-sys = { workspace = true, features = [ [target.'cfg(target_arch = "wasm32")'.dev-dependencies] wasm-bindgen-test.workspace = true celestia-rpc = { workspace = true, features = ["wasm-bindgen"] } +# TODO: added for debugging +tracing-subscriber = { workspace = true, features = ["time"] } +tracing-web = "*" [dev-dependencies] celestia-rpc = { workspace = true, features = ["p2p"] } diff --git a/node/src/blockstore.rs b/node/src/blockstore.rs index 5425c1b60..752adba1a 100644 --- a/node/src/blockstore.rs +++ b/node/src/blockstore.rs @@ -1,9 +1,5 @@ //! Blockstore types aliases with lumina specific constants. -use blockstore::{Blockstore, Result}; -use celestia_types::sample::SAMPLE_ID_CODEC; -use cid::CidGeneric; - use crate::p2p::MAX_MH_SIZE; /// An [`InMemoryBlockstore`] with maximum multihash size used by lumina. @@ -22,98 +18,3 @@ pub type RedbBlockstore = blockstore::RedbBlockstore; /// /// [`IndexedDbBlockstore`]: blockstore::IndexedDbBlockstore pub type IndexedDbBlockstore = blockstore::IndexedDbBlockstore; - -/// A blockstore which only stores samples and discards other CIDs. -pub(crate) struct SampleBlockstore { - blockstore: B, -} - -impl SampleBlockstore { - /// Wrap another blockstore with sample blockstore. - pub(crate) fn new(blockstore: B) -> Self { - Self { blockstore } - } -} - -impl Blockstore for SampleBlockstore -where - B: Blockstore, -{ - async fn get(&self, cid: &CidGeneric) -> Result>> { - self.blockstore.get(cid).await - } - - async fn put_keyed(&self, cid: &CidGeneric, data: &[u8]) -> Result<()> { - if cid.codec() == SAMPLE_ID_CODEC { - self.blockstore.put_keyed(cid, data).await - } else { - Ok(()) - } - } - - async fn remove(&self, cid: &CidGeneric) -> Result<()> { - self.blockstore.remove(cid).await - } - - async fn has(&self, cid: &CidGeneric) -> Result { - self.blockstore.has(cid).await - } - - async fn close(self) -> Result<()> { - self.blockstore.close().await - } -} - -#[cfg(test)] -mod tests { - use blockstore::Blockstore; - use celestia_types::{ - nmt::Namespace, row::RowId, row_namespace_data::RowNamespaceDataId, sample::SampleId, - }; - use lumina_utils::test_utils::async_test; - - use super::{InMemoryBlockstore, SampleBlockstore}; - - #[async_test] - async fn should_only_store_samples() { - macro_rules! cid { - ($bytes:expr) => { - ::cid::CidGeneric::try_from($bytes).unwrap() - }; - ($id:ty, $($args:expr),+ $(,)?) => { - $crate::p2p::shwap::convert_cid( - &<$id>::new($($args),+).unwrap().into() - ) - .unwrap() - }; - } - - let blockstore = SampleBlockstore::new(InMemoryBlockstore::new()); - - let sample_cids = [ - cid!(SampleId, 1, 2, 3), - cid!(SampleId, 1111, 232, 33), - cid!(SampleId, 123, 1, 888888888), - ]; - - let non_sample_cids = [ - cid!(RowId, 1, 1737), - cid!(RowId, 8812, 193139), - cid!(RowNamespaceDataId, Namespace::new_v0(b"a").unwrap(), 15, 12), - cid!(RowNamespaceDataId, Namespace::new_v0(b"z").unwrap(), 1, 1), - cid!([1; 64].as_ref()), - cid!([[1].as_ref(), &[18; 63]].concat()), - ]; - - for cid in sample_cids.iter().chain(non_sample_cids.iter()) { - blockstore.put_keyed(cid, &[10; 150]).await.unwrap(); - } - - for cid in &sample_cids { - assert!(blockstore.has(cid).await.unwrap()); - } - for cid in &non_sample_cids { - assert!(!blockstore.has(cid).await.unwrap()); - } - } -} diff --git a/node/src/daser.rs b/node/src/daser.rs index 17bf50ec2..b6a8576f9 100644 --- a/node/src/daser.rs +++ b/node/src/daser.rs @@ -5,6 +5,12 @@ use std::collections::HashSet; use std::sync::Arc; use std::time::Duration; +use blockstore::block::CidError; +use blockstore::{Blockstore, Error as BlockstoreError}; +use bytes::{Bytes, BytesMut}; +use celestia_types::sample::SampleId; +use cid::multihash::Multihash; +use cid::{Cid, CidGeneric}; use futures::future::BoxFuture; use futures::stream::FuturesUnordered; use futures::{FutureExt, StreamExt}; @@ -18,8 +24,7 @@ use tokio_util::sync::CancellationToken; use tracing::{debug, error, info, instrument, warn}; use crate::events::{EventPublisher, NodeEvent}; -use crate::p2p::shwap::sample_cid; -use crate::p2p::{P2p, P2pError}; +use crate::p2p::{P2p, P2pError, ShrExError}; use crate::store::{BlockRanges, Store, StoreError}; use crate::utils::{OneshotSenderExt, TimeExt}; @@ -39,6 +44,10 @@ pub enum DaserError { #[error("P2p: {0}")] P2p(#[from] P2pError), + /// An error propagated from the [`Store`] component. + #[error("Blockstore: {0}")] + Blockstore(#[from] BlockstoreError), + /// An error propagated from the [`Store`] component. #[error("Store: {0}")] Store(#[from] StoreError), @@ -66,14 +75,17 @@ pub(crate) struct Daser { } /// Arguments used to configure the [`Daser`]. -pub(crate) struct DaserArgs +pub(crate) struct DaserArgs where + B: Blockstore, S: Store, { /// Handler for the peer to peer messaging. pub(crate) p2p: Arc, /// Headers storage. pub(crate) store: Arc, + /// Blockstore. + pub(crate) blockstore: Arc, /// Event publisher. pub(crate) event_pub: EventPublisher, /// Size of the sampling window. @@ -99,17 +111,12 @@ pub(crate) enum DaserCmd { /// /// This is needed to avoid following race condition: /// - /// We have `Store` very tightly integrated with `beetswap::Multihasher` - /// and when Daser starts data sampling the header of that block must be - /// in the `Store` until the data sampling is finished. This can be fixed - /// only if we decouple `Store` from `beetswap::Multihasher`. - /// - /// However, even if we fix the above, a second problem arise: When Pruner - /// removes the header and samples of an ongoing data sampling, how are we - /// going to handle the incoming CIDs? We need somehow make sure that Pruner - /// will remove them after sampling is finished. - /// - /// After the above issues are fixed, this can be removed. + /// Assume some block is just about to fail out of a sampling window, and daser + /// is scheduling a request for it just before it happens. Pruner may wake up right + /// after the block falls outside of the sampling window as no longer needed (we + /// don't need to sample blocks outside the window). In such case, Pruner will prune + /// the header and when sampling finishes, daser will insert samples into the store. + /// Those samples would never be pruned. WantToPrune { height: u64, respond_to: oneshot::Sender, @@ -118,8 +125,9 @@ pub(crate) enum DaserCmd { impl Daser { /// Create and start the [`Daser`]. - pub(crate) fn start(args: DaserArgs) -> Result + pub(crate) fn start(args: DaserArgs) -> Result where + B: Blockstore + 'static, S: Store + 'static, { let cancellation_token = CancellationToken::new(); @@ -210,8 +218,11 @@ impl Drop for Daser { } } -struct Worker +type SamplingFut = BoxFuture<'static, Result<(u64, Vec<(Cid, Bytes)>, bool)>>; + +struct Worker where + B: Blockstore + 'static, S: Store + 'static, { cmd_rx: mpsc::Receiver, @@ -219,10 +230,11 @@ where event_pub: EventPublisher, p2p: Arc, store: Arc, + blockstore: Arc, max_samples_needed: usize, - sampling_futs: FuturesUnordered>>, + sampling_futs: FuturesUnordered, queue: BlockRanges, - timed_out: BlockRanges, + failed: BlockRanges, ongoing: BlockRanges, will_be_pruned: BlockRanges, sampling_window: Duration, @@ -233,25 +245,27 @@ where num_of_prunable_blocks: u64, } -impl Worker +impl Worker where + B: Blockstore, S: Store, { fn new( - args: DaserArgs, + args: DaserArgs, cancellation_token: CancellationToken, cmd_rx: mpsc::Receiver, - ) -> Result> { + ) -> Result> { Ok(Worker { cmd_rx, cancellation_token, event_pub: args.event_pub, p2p: args.p2p, store: args.store, + blockstore: args.blockstore, max_samples_needed: MAX_SAMPLES_NEEDED, sampling_futs: FuturesUnordered::new(), queue: BlockRanges::default(), - timed_out: BlockRanges::default(), + failed: BlockRanges::default(), ongoing: BlockRanges::default(), will_be_pruned: BlockRanges::default(), sampling_window: args.sampling_window, @@ -352,12 +366,16 @@ where _ = report_interval.tick() => self.report().await?, Some(cmd) = self.cmd_rx.recv() => self.on_cmd(cmd).await, Some(res) = self.sampling_futs.next() => { - // Beetswap only returns fatal errors that are not related + // The future only returns fatal errors that are not related // to P2P nor networking. - let (height, timed_out) = res?; + let (height, samples, failed) = res?; + + for (cid, sample) in samples { + self.blockstore.put_keyed(&cid, &sample).await?; + } - if timed_out { - self.timed_out.insert_relaxed(height..=height).expect("invalid height"); + if failed { + self.failed.insert_relaxed(height..=height).expect("invalid height"); } else { self.store.mark_as_sampled(height).await?; } @@ -374,7 +392,7 @@ where self.sampling_futs.clear(); self.queue = BlockRanges::default(); self.ongoing = BlockRanges::default(); - self.timed_out = BlockRanges::default(); + self.failed = BlockRanges::default(); self.head_height = None; Ok(()) @@ -475,7 +493,7 @@ where self.queue .remove_relaxed(1..=height) .expect("invalid height"); - self.timed_out + self.failed .insert_relaxed(1..=height) .expect("invalid height"); return Ok(false); @@ -522,22 +540,31 @@ where }) .collect::>(); - let mut sampling_timed_out = false; + let mut sampling_failed = false; + let mut samples = Vec::with_capacity(futs.len()); // Run futures to completion while let Some((row, column, res)) = futs.next().await { - let timed_out = match res { - Ok(_) => false, - // Validation is done at Bitswap level, through `ShwapMultihasher`. - // If the sample is not valid, it will never be delivered to us - // as the data of the CID. Because of that, the only signal - // that data sampling verification failed is query timing out. - Err(P2pError::RequestTimedOut) => true, + let failed = match res { + Ok(sample) => { + // put the sample into the blockstore + let cid = sample_cid(row, column, height).expect("Block height not 0"); + let mut bytes = BytesMut::new(); + sample.encode(&mut bytes); + samples.push((cid, bytes.freeze())); + + false + } + // Validation is done at DA protocol level. + // If we hit the timeout or maximum retries of the request + // we can mark it as failure + Err(P2pError::RequestTimedOut) + | Err(P2pError::ShrEx(ShrExError::MaxTriesReached)) => true, Err(e) => return Err(e.into()), }; - if timed_out { - sampling_timed_out = true; + if failed { + sampling_failed = true; } event_pub.send(NodeEvent::ShareSamplingResult { @@ -545,17 +572,17 @@ where square_width, row, column, - timed_out, + failed, }); } event_pub.send(NodeEvent::SamplingResult { height, - timed_out: sampling_timed_out, + failed: sampling_failed, took: now.elapsed(), }); - Ok((height, sampling_timed_out)) + Ok((height, samples, sampling_failed)) } .boxed(); @@ -573,7 +600,7 @@ where let sampled = self.store.get_sampled_ranges().await?; self.head_height = stored.head(); - self.queue = stored - &sampled - &self.timed_out - &self.ongoing - &self.will_be_pruned; + self.queue = stored - &sampled - &self.failed - &self.ongoing - &self.will_be_pruned; Ok(()) } @@ -628,26 +655,35 @@ fn random_indexes(square_width: u16, max_samples_needed: usize) -> HashSet<(u16, indexes } +fn sample_cid(row_index: u16, column_index: u16, block_height: u64) -> Result { + let sample_id = SampleId::new(row_index, column_index, block_height)?; + convert_cid(&sample_id.into()) +} + +fn convert_cid(cid: &CidGeneric) -> Result { + let multihash = Multihash::wrap(cid.hash().code(), cid.hash().digest()) + .map_err(|_| P2pError::Cid(CidError::InvalidMultihashLength(64)))?; + + Cid::new(cid.version(), cid.codec(), multihash) + .map_err(|_| P2pError::Cid(CidError::InvalidMultihashLength(64))) +} + #[cfg(test)] mod tests { use super::*; + use crate::blockstore::InMemoryBlockstore; use crate::events::{EventChannel, EventSubscriber}; use crate::node::SAMPLING_WINDOW; use crate::p2p::P2pCmd; - use crate::p2p::shwap::convert_cid; use crate::store::InMemoryStore; use crate::test_utils::{ExtendedHeaderGeneratorExt, MockP2pHandle}; use crate::utils::OneshotResultSender; - use bytes::BytesMut; - use celestia_proto::bitswap::Block; use celestia_types::consts::appconsts::AppVersion; - use celestia_types::sample::{Sample, SampleId}; + use celestia_types::sample::Sample; use celestia_types::test_utils::{ExtendedHeaderGenerator, generate_dummy_eds}; use celestia_types::{AxisType, DataAvailabilityHeader, ExtendedDataSquare}; - use cid::Cid; use lumina_utils::test_utils::async_test; use lumina_utils::time::sleep; - use prost::Message; use std::collections::HashMap; use std::time::Duration; @@ -718,6 +754,7 @@ mod tests { async fn received_valid_samples() { let (mock, mut handle) = P2p::mocked(); let store = Arc::new(InMemoryStore::new()); + let blockstore = Arc::new(InMemoryBlockstore::new()); let events = EventChannel::new(); let mut event_sub = events.subscribe(); @@ -725,6 +762,7 @@ mod tests { event_pub: events.publisher(), p2p: Arc::new(mock), store: store.clone(), + blockstore, sampling_window: SAMPLING_WINDOW, concurrency_limit: 1, additional_headersub_concurrency: DEFAULT_ADDITIONAL_HEADER_SUB_CONCURENCY, @@ -779,6 +817,7 @@ mod tests { async fn sampling_timeout() { let (mock, mut handle) = P2p::mocked(); let store = Arc::new(InMemoryStore::new()); + let blockstore = Arc::new(InMemoryBlockstore::new()); let events = EventChannel::new(); let mut event_sub = events.subscribe(); @@ -786,6 +825,7 @@ mod tests { event_pub: events.publisher(), p2p: Arc::new(mock), store: store.clone(), + blockstore, sampling_window: SAMPLING_WINDOW, concurrency_limit: 1, additional_headersub_concurrency: DEFAULT_ADDITIONAL_HEADER_SUB_CONCURENCY, @@ -823,12 +863,14 @@ mod tests { async fn backward_dasing() { let (mock, mut handle) = P2p::mocked(); let store = Arc::new(InMemoryStore::new()); + let blockstore = Arc::new(InMemoryBlockstore::new()); let events = EventChannel::new(); let _daser = Daser::start(DaserArgs { event_pub: events.publisher(), p2p: Arc::new(mock), store: store.clone(), + blockstore, sampling_window: SAMPLING_WINDOW, concurrency_limit: 1, additional_headersub_concurrency: DEFAULT_ADDITIONAL_HEADER_SUB_CONCURENCY, @@ -857,38 +899,40 @@ mod tests { store.insert(headers[4..=9].to_vec()).await.unwrap(); // Sample block 10 - handle_get_shwap_cid(&mut handle, 10, &edses[9], false).await; + handle_get_samples(&mut handle, 10, &edses[9], false).await; // Sample block 9 - handle_get_shwap_cid(&mut handle, 9, &edses[8], false).await; + handle_get_samples(&mut handle, 9, &edses[8], false).await; // To avoid race conditions we wait a bit for the block 8 to be scheduled - sleep(Duration::from_millis(10)).await; + sleep(Duration::from_millis(150)).await; // Insert 16-20 block headers store.insert(headers[15..=19].to_vec()).await.unwrap(); // To avoid race conditions we wait a bit for the new head (block 20) to be scheduled - sleep(Duration::from_millis(10)).await; + sleep(Duration::from_millis(150)).await; // Now daser runs two concurrent data sampling: block 8 and block 20 - handle_concurrent_get_shwap_cid( + handle_concurrent_get_sample( &mut handle, [(8, &edses[9], false), (20, &edses[19], false)], ) .await; // Sample and reject block 19 - handle_get_shwap_cid(&mut handle, 19, &edses[18], true).await; + handle_get_samples(&mut handle, 19, &edses[18], true).await; // Simulate disconnection handle.announce_all_peers_disconnected(); // Daser may scheduled Block 18 already, so we need to reply to that requests. - // For the sake of the test we reply with a bitswap timeout. - while let Some(cmd) = handle.try_recv_cmd().await { + // For the sake of the test we reply with a timeout. + // To avoid race conditions we wait a bit for all the potential requests to be scheduled + sleep(Duration::from_millis(150)).await; + while let Some(cmd) = dbg!(handle.try_recv_cmd().await) { match cmd { - P2pCmd::GetShwapCid { respond_to, .. } => { + P2pCmd::GetSample { respond_to, .. } => { let _ = respond_to.send(Err(P2pError::RequestTimedOut)); } cmd => panic!("Unexpected command: {cmd:?}"), @@ -902,18 +946,18 @@ mod tests { handle.announce_peer_connected(); // Because of disconnection and previous rejection of block 19, daser will resample it - handle_get_shwap_cid(&mut handle, 19, &edses[18], false).await; + handle_get_samples(&mut handle, 19, &edses[18], false).await; // Sample block 16 until 18 for height in (16..=18).rev() { let idx = height as usize - 1; - handle_get_shwap_cid(&mut handle, height, &edses[idx], false).await; + handle_get_samples(&mut handle, height, &edses[idx], false).await; } // Sample block 5 until 7 for height in (5..=7).rev() { let idx = height as usize - 1; - handle_get_shwap_cid(&mut handle, height, &edses[idx], false).await; + handle_get_samples(&mut handle, height, &edses[idx], false).await; } handle.expect_no_cmd().await; @@ -925,7 +969,7 @@ mod tests { store.insert(header).await.unwrap(); // Sample block 21 - handle_get_shwap_cid(&mut handle, 21, &eds, false).await; + handle_get_samples(&mut handle, 21, &eds, false).await; handle.expect_no_cmd().await; } @@ -934,6 +978,7 @@ mod tests { async fn concurrency_limits() { let (mock, mut handle) = P2p::mocked(); let store = Arc::new(InMemoryStore::new()); + let blockstore = Arc::new(InMemoryBlockstore::new()); let events = EventChannel::new(); // Concurrency limit @@ -955,6 +1000,7 @@ mod tests { event_pub: events.publisher(), p2p: Arc::new(mock), store: store.clone(), + blockstore, sampling_window: SAMPLING_WINDOW, concurrency_limit, additional_headersub_concurrency, @@ -967,8 +1013,8 @@ mod tests { let mut hold_respond_channels = Vec::new(); for _ in 0..(concurrency_limit * shares_per_block) { - let (cid, respond_to) = handle.expect_get_shwap_cid().await; - hold_respond_channels.push((cid, respond_to)); + let (_, _, block_height, respond_to) = handle.expect_get_sample().await; + hold_respond_channels.push((block_height, respond_to)); } // Concurrency limit reached @@ -981,8 +1027,8 @@ mod tests { .unwrap(); for _ in 0..shares_per_block { - let (cid, respond_to) = handle.expect_get_shwap_cid().await; - hold_respond_channels.push((cid, respond_to)); + let (_, _, block_height, respond_to) = handle.expect_get_sample().await; + hold_respond_channels.push((block_height, respond_to)); } handle.expect_no_cmd().await; @@ -994,8 +1040,8 @@ mod tests { // Now Daser will schedule the next block. for _ in 0..shares_per_block { - let (cid, respond_to) = handle.expect_get_shwap_cid().await; - hold_respond_channels.push((cid, respond_to)); + let (_, _, block_height, respond_to) = handle.expect_get_sample().await; + hold_respond_channels.push((block_height, respond_to)); } // And... concurrency limit is reached again. @@ -1008,12 +1054,12 @@ mod tests { .await .unwrap(); // Give some time for Daser to schedule it - sleep(Duration::from_millis(10)).await; + sleep(Duration::from_millis(150)).await; } for _ in 0..(additional_headersub_concurrency * shares_per_block) { - let (cid, respond_to) = handle.expect_get_shwap_cid().await; - hold_respond_channels.push((cid, respond_to)); + let (_, _, block_height, respond_to) = handle.expect_get_sample().await; + hold_respond_channels.push((block_height, respond_to)); } // Concurrency limit for heads is reached @@ -1029,8 +1075,8 @@ mod tests { stop_sampling_for(&mut hold_respond_channels, 28); for _ in 0..shares_per_block { - let (cid, respond_to) = handle.expect_get_shwap_cid().await; - hold_respond_channels.push((cid, respond_to)); + let (_, _, block_height, respond_to) = handle.expect_get_sample().await; + hold_respond_channels.push((block_height, respond_to)); } // Concurrency limit for heads is reached again @@ -1046,6 +1092,7 @@ mod tests { async fn ratelimit() { let (mock, mut handle) = P2p::mocked(); let store = Arc::new(InMemoryStore::new()); + let blockstore = Arc::new(InMemoryBlockstore::new()); let events = EventChannel::new(); let mut event_sub = events.subscribe(); @@ -1053,6 +1100,7 @@ mod tests { event_pub: events.publisher(), p2p: Arc::new(mock), store: store.clone(), + blockstore, sampling_window: Duration::from_secs(60), concurrency_limit: 1, additional_headersub_concurrency: DEFAULT_ADDITIONAL_HEADER_SUB_CONCURENCY, @@ -1160,20 +1208,19 @@ mod tests { } fn stop_sampling_for( - responders: &mut Vec<(Cid, OneshotResultSender, P2pError>)>, + responders: &mut Vec<(u64, OneshotResultSender)>, height: u64, ) { let mut indexes = Vec::new(); - for (idx, (cid, _)) in responders.iter().enumerate() { - let sample_id: SampleId = cid.try_into().unwrap(); - if sample_id.block_height() == height { + for (idx, (h, _)) in responders.iter().enumerate() { + if *h == height { indexes.push(idx) } } for idx in indexes.into_iter().rev() { - let (_cid, respond_to) = responders.remove(idx); + let (_, respond_to) = responders.remove(idx); respond_to.send(Err(P2pError::RequestTimedOut)).unwrap(); } } @@ -1217,10 +1264,10 @@ mod tests { height: u64, simulate_sampling_timeout: bool, ) { - let cids = handle_get_shwap_cid(handle, height, eds, simulate_sampling_timeout).await; + handle_get_samples(handle, height, eds, simulate_sampling_timeout).await; // Wait to be sampled - sleep(Duration::from_millis(100)).await; + sleep(Duration::from_millis(150)).await; // Check if block was sampled or timed-out. let sampled_ranges = store.get_sampled_ranges().await.unwrap(); @@ -1229,7 +1276,6 @@ mod tests { // Check if CIDs we requested successfully made it in the store let mut sampling_metadata = store.get_sampling_metadata(height).await.unwrap().unwrap(); sampling_metadata.cids.sort(); - assert_eq!(&sampling_metadata.cids, &cids); // Check if we received `SamplingStarted` event let mut remaining_shares = match event_sub.try_recv().unwrap().event { @@ -1241,11 +1287,11 @@ mod tests { assert_eq!(ev_height, height); assert_eq!(square_width, eds.square_width()); - // Make sure the share list matches the CIDs we received - let mut cids = shares + // Make sure the share list matches what's in metadata + let mut cids: Vec<_> = shares .iter() .map(|(row, col)| sample_cid(*row, *col, height).unwrap()) - .collect::>(); + .collect(); cids.sort(); assert_eq!(&sampling_metadata.cids, &cids); @@ -1262,12 +1308,12 @@ mod tests { square_width, row, column, - timed_out, + failed, } => { assert_eq!(ev_height, height); assert_eq!(square_width, eds.square_width()); assert_eq!( - timed_out, + failed, simulate_sampling_timeout && i == REQ_TIMEOUT_SHARE_NUM ); // Make sure it is in the list and remove it @@ -1283,11 +1329,11 @@ mod tests { match event_sub.try_recv().unwrap().event { NodeEvent::SamplingResult { height: ev_height, - timed_out, + failed, took, } => { assert_eq!(ev_height, height); - assert_eq!(timed_out, simulate_sampling_timeout); + assert_eq!(failed, simulate_sampling_timeout); assert_ne!(took, Duration::default()); } ev => panic!("Unexpected event: {ev}"), @@ -1295,10 +1341,10 @@ mod tests { } /// Responds to get_shwap_cid and returns all CIDs that were requested - async fn handle_concurrent_get_shwap_cid( + async fn handle_concurrent_get_sample( handle: &mut MockP2pHandle, handling_args: [(u64, &ExtendedDataSquare, bool); N], - ) -> Vec { + ) { struct Info<'a> { eds: &'a ExtendedDataSquare, simulate_sampling_timeout: bool, @@ -1325,16 +1371,14 @@ mod tests { .collect::>(); let needed_samples_sum = infos.values().map(|info| info.needed_samples).sum(); - let mut cids = Vec::with_capacity(needed_samples_sum); for _ in 0..needed_samples_sum { - let (cid, respond_to) = handle.expect_get_shwap_cid().await; - cids.push(cid); - - let sample_id: SampleId = cid.try_into().unwrap(); + let (row, col, height, respond_to) = handle.expect_get_sample().await; let info = infos - .get_mut(&sample_id.block_height()) - .unwrap_or_else(|| panic!("Unexpected height: {}", sample_id.block_height())); + .get_mut(&height) + .unwrap_or_else(|| panic!("Unexpected height: {}", height)); + + let sample = Sample::new(row, col, AxisType::Row, info.eds).unwrap(); info.requests_count += 1; @@ -1344,41 +1388,17 @@ mod tests { continue; } - let sample = gen_sample_of_cid(sample_id, info.eds).await; respond_to.send(Ok(sample)).unwrap(); } - - cids.sort(); - cids } /// Responds to get_shwap_cid and returns all CIDs that were requested - async fn handle_get_shwap_cid( + async fn handle_get_samples( handle: &mut MockP2pHandle, height: u64, eds: &ExtendedDataSquare, simulate_sampling_timeout: bool, - ) -> Vec { - handle_concurrent_get_shwap_cid(handle, [(height, eds, simulate_sampling_timeout)]).await - } - - async fn gen_sample_of_cid(sample_id: SampleId, eds: &ExtendedDataSquare) -> Vec { - let sample = Sample::new( - sample_id.row_index(), - sample_id.column_index(), - AxisType::Row, - eds, - ) - .unwrap(); - - let mut container = BytesMut::new(); - sample.encode(&mut container); - - let block = Block { - cid: convert_cid(&sample_id.into()).unwrap().to_bytes(), - container: container.to_vec(), - }; - - block.encode_to_vec() + ) { + handle_concurrent_get_sample(handle, [(height, eds, simulate_sampling_timeout)]).await } } diff --git a/node/src/events.rs b/node/src/events.rs index 8fa3962b2..247f491b2 100644 --- a/node/src/events.rs +++ b/node/src/events.rs @@ -203,16 +203,16 @@ pub enum NodeEvent { row: u16, /// The column of the share. column: u16, - /// Share sampling timed out. - timed_out: bool, + /// Share sampling failed. + failed: bool, }, /// Sampling result. SamplingResult { /// The block height that was sampled. height: u64, - /// Sampling timed out. - timed_out: bool, + /// Sampling failed. + failed: bool, /// How much time sampling took. took: Duration, }, @@ -362,10 +362,10 @@ impl fmt::Display for NodeEvent { height, row, column, - timed_out, + failed, .. } => { - let s = if *timed_out { "timed out" } else { "finished" }; + let s = if *failed { "failed" } else { "finished" }; write!( f, "Sampling for share [{row}, {column}] of block {height} {s}" @@ -373,10 +373,10 @@ impl fmt::Display for NodeEvent { } NodeEvent::SamplingResult { height, - timed_out, + failed, took, } => { - let s = if *timed_out { "timed out" } else { "finished" }; + let s = if *failed { "failed" } else { "finished" }; write!(f, "Sampling of block {height} {s}. Took: {took:?}") } NodeEvent::FatalDaserError { error } => { diff --git a/node/src/node.rs b/node/src/node.rs index 6ede8cb1a..21f73f959 100644 --- a/node/src/node.rs +++ b/node/src/node.rs @@ -21,18 +21,16 @@ use celestia_types::hash::Hash; use celestia_types::namespace_data::NamespaceData; use celestia_types::nmt::Namespace; use celestia_types::row::Row; -use celestia_types::row_namespace_data::RowNamespaceData; use celestia_types::sample::Sample; use celestia_types::{Blob, ExtendedDataSquare, ExtendedHeader, SharesAtHeight}; use lumina_utils::executor::{JoinHandle, spawn, spawn_cancellable}; -use crate::blockstore::{InMemoryBlockstore, SampleBlockstore}; +use crate::blockstore::InMemoryBlockstore; use crate::daser::{ DEFAULT_ADDITIONAL_HEADER_SUB_CONCURENCY, DEFAULT_CONCURENCY_LIMIT, Daser, DaserArgs, }; use crate::events::{EventChannel, EventSubscriber, NodeEvent}; use crate::node::subscriptions::{SubscriptionError, forward_new_blobs, forward_new_shares}; -use crate::p2p::shwap::sample_cid; use crate::p2p::{P2p, P2pArgs}; use crate::pruner::{Pruner, PrunerArgs}; use crate::store::{InMemoryStore, SamplingMetadata, Store, StoreError}; @@ -105,7 +103,7 @@ where { event_channel: EventChannel, p2p: Option>, - blockstore: Option>>, + blockstore: Option>, store: Option>, syncer: Option>>, daser: Option>, @@ -149,7 +147,7 @@ where let event_channel = EventChannel::new(); let event_sub = event_channel.subscribe(); let store = Arc::new(config.store); - let blockstore = Arc::new(SampleBlockstore::new(config.blockstore)); + let blockstore = Arc::new(config.blockstore); let p2p = Arc::new( P2p::start(P2pArgs { @@ -157,7 +155,6 @@ where local_keypair: config.p2p_local_keypair, bootnodes: config.p2p_bootnodes, listen_on: config.p2p_listen_on, - blockstore: blockstore.clone(), store: store.clone(), event_pub: event_channel.publisher(), }) @@ -176,6 +173,7 @@ where let daser = Arc::new(Daser::start(DaserArgs { p2p: p2p.clone(), store: store.clone(), + blockstore: blockstore.clone(), event_pub: event_channel.publisher(), sampling_window: config.sampling_window, concurrency_limit: DEFAULT_CONCURENCY_LIMIT, @@ -402,26 +400,6 @@ where .get_sample(row_index, column_index, block_height, timeout) .await?; - // We want to immediately remove the sample from blockstore - // but **only if** it wasn't chosen for DASing. Otherwise, we could - // accidentally remove samples needed for the block reconstruction. - // - // There's a small possibility of permanently storing this sample if - // persistent blockstore is used and user closes tab / kills process - // before the remove is called, but it is acceptable tradeoff to avoid complexity. - // - // TODO: It should be properly solved when we switch from bitswap to shrex. - if let Some(metadata) = self.get_sampling_metadata(block_height).await? { - let cid = sample_cid(row_index, column_index, block_height)?; - if !metadata.cids.contains(&cid) { - let blockstore = self - .blockstore - .as_ref() - .expect("Blockstore not initialized"); - let _ = blockstore.remove(&cid).await; - } - } - Ok(sample) } @@ -439,25 +417,6 @@ where Ok(self.p2p().get_eds(block_height, timeout).await?) } - /// Request a verified [`RowNamespaceData`] from the network. - /// - /// # Errors - /// - /// On failure to receive a verified [`RowNamespaceData`] within a certain time, the - /// `NodeError::P2p(P2pError::RequestTimedOut)` error will be returned. - pub async fn request_row_namespace_data( - &self, - namespace: Namespace, - row_index: u16, - block_height: u64, - timeout: Option, - ) -> Result { - Ok(self - .p2p() - .get_row_namespace_data(namespace, row_index, block_height, timeout) - .await?) - } - /// Request a verified [`NamespaceData`] from the network. /// /// # Errors diff --git a/node/src/p2p.rs b/node/src/p2p.rs index 19c6dc0b2..1386f3d20 100644 --- a/node/src/p2p.rs +++ b/node/src/p2p.rs @@ -13,24 +13,18 @@ //! - bitswap 1.2.0 //! - shwap - celestia's data availability protocol on top of bitswap -use std::collections::HashMap; -use std::future::poll_fn; use std::sync::Arc; -use std::task::Poll; use std::time::Duration; -use blockstore::Blockstore; use blockstore::block::CidError; use celestia_proto::p2p::pb::{HeaderRequest, header_request}; use celestia_types::fraud_proof::BadEncodingFraudProof; use celestia_types::hash::Hash; use celestia_types::namespace_data::NamespaceData; use celestia_types::nmt::Namespace; -use celestia_types::row::{Row, RowId}; -use celestia_types::row_namespace_data::{RowNamespaceData, RowNamespaceDataId}; -use celestia_types::sample::{Sample, SampleId}; +use celestia_types::row::Row; +use celestia_types::sample::Sample; use celestia_types::{Blob, ExtendedDataSquare, ExtendedHeader, FraudProof}; -use cid::Cid; use libp2p::gossipsub::TopicHash; use libp2p::identity::Keypair; use libp2p::swarm::{NetworkBehaviour, NetworkInfo}; @@ -38,7 +32,6 @@ use libp2p::{Multiaddr, PeerId, gossipsub}; use lumina_utils::executor::{JoinHandle, spawn}; use lumina_utils::time::{self, Interval}; use lumina_utils::token::Token; -use smallvec::SmallVec; use tendermint_proto::Protobuf; use tokio::select; use tokio::sync::{mpsc, oneshot, watch}; @@ -49,7 +42,6 @@ mod connection_control; mod header_ex; pub(crate) mod header_session; mod shrex; -pub(crate) mod shwap; mod swarm; mod swarm_manager; mod utils; @@ -57,14 +49,13 @@ mod utils; use crate::block_ranges::BlockRange; use crate::events::EventPublisher; use crate::p2p::header_session::HeaderSession; -use crate::p2p::shwap::{ShwapMultihasher, convert_cid, get_block_container}; use crate::p2p::swarm_manager::SwarmManager; use crate::peer_tracker::PeerTracker; use crate::peer_tracker::PeerTrackerInfo; use crate::store::{Store, StoreError}; use crate::utils::{ - MultiaddrExt, OneshotResultSender, OneshotResultSenderExt, OneshotSenderExt, - celestia_protocol_id, fraudsub_ident_topic, gossipsub_ident_topic, + MultiaddrExt, OneshotResultSender, OneshotSenderExt, fraudsub_ident_topic, + gossipsub_ident_topic, }; pub use crate::p2p::header_ex::HeaderExError; @@ -114,15 +105,11 @@ pub enum P2pError { #[error("Bootnode multiaddrs without peer ID: {0:?}")] BootnodeAddrsWithoutPeerId(Vec), - /// An error propagated from [`beetswap::Behaviour`]. - #[error("Bitswap: {0}")] - Bitswap(#[from] beetswap::Error), - /// Protobuf message failed to be decoded. #[error("ProtoBuf decoding error: {0}")] ProtoDecodeFailed(#[from] tendermint_proto::Error), - /// An error propagated from [`celestia_types`] that is related to [`Cid`]. + /// An error propagated from [`celestia_types`] that is related to `CID`. #[error("CID error: {0}")] Cid(CidError), @@ -165,7 +152,6 @@ impl P2pError { | P2pError::BootnodeAddrsWithoutPeerId(_) => true, P2pError::HeaderEx(_) | P2pError::ShrEx(_) - | P2pError::Bitswap(_) | P2pError::ProtoDecodeFailed(_) | P2pError::Cid(_) | P2pError::RequestTimedOut @@ -207,9 +193,8 @@ pub(crate) struct P2p { } /// Arguments used to configure the [`P2p`]. -pub struct P2pArgs +pub struct P2pArgs where - B: Blockstore, S: Store, { /// An id of the network to connect to. @@ -221,8 +206,6 @@ where /// List of the addresses on which to listen for incoming connections. pub listen_on: Vec, /// The store for headers. - pub blockstore: Arc, - /// The store for headers. pub store: Arc, /// Event publisher. pub event_pub: EventPublisher, @@ -256,10 +239,6 @@ pub(crate) enum P2pCmd { MarkAsArchival { peer_id: PeerId, }, - GetShwapCid { - cid: Cid, - respond_to: OneshotResultSender, P2pError>, - }, GetNetworkCompromisedToken { respond_to: oneshot::Sender, }, @@ -268,7 +247,6 @@ pub(crate) enum P2pCmd { }, // This is dead code because `get_row` still uses Bitswap. // We can use this when celestia-node#4288 is merged. - #[allow(dead_code)] GetRow { row_index: u16, block_height: u64, @@ -276,7 +254,6 @@ pub(crate) enum P2pCmd { }, // This is dead code because `get_sample` still uses Bitswap. // We can use this when celestia-node#4288 is merged. - #[allow(dead_code)] GetSample { row_index: u16, column_index: u16, @@ -296,9 +273,8 @@ pub(crate) enum P2pCmd { impl P2p { /// Creates and starts a new p2p handler. - pub async fn start(args: P2pArgs) -> Result + pub async fn start(args: P2pArgs) -> Result where - B: Blockstore + 'static, S: Store + 'static, { validate_bootnode_addrs(&args.bootnodes)?; @@ -525,43 +501,28 @@ impl P2p { Ok(headers) } - /// Request a [`Cid`] on bitswap protocol. - pub(crate) async fn get_shwap_cid( + /// Request a [`Row`] on bitswap protocol. + pub async fn get_row( &self, - cid: Cid, + row_index: u16, + block_height: u64, timeout: Option, - ) -> Result> { + ) -> Result { let (tx, rx) = oneshot::channel(); - self.send_command(P2pCmd::GetShwapCid { - cid, + self.send_command(P2pCmd::GetRow { + row_index, + block_height, respond_to: tx, }) .await?; - let data = match timeout { + match timeout { Some(dur) => time::timeout(dur, rx) .await - .map_err(|_| P2pError::RequestTimedOut)???, - None => rx.await??, - }; - - get_block_container(&cid, &data) - } - - /// Request a [`Row`] on bitswap protocol. - pub async fn get_row( - &self, - row_index: u16, - block_height: u64, - timeout: Option, - ) -> Result { - let id = RowId::new(row_index, block_height)?; - let cid = convert_cid(&id.into())?; - - let data = self.get_shwap_cid(cid, timeout).await?; - let row = Row::decode(id, &data[..]).map_err(|e| P2pError::Shwap(e.to_string()))?; - Ok(row) + .map_err(|_| P2pError::RequestTimedOut)??, + None => rx.await?, + } } /// Request a [`Sample`] on bitswap protocol. @@ -572,12 +533,22 @@ impl P2p { block_height: u64, timeout: Option, ) -> Result { - let id = SampleId::new(row_index, column_index, block_height)?; - let cid = convert_cid(&id.into())?; + let (tx, rx) = oneshot::channel(); - let data = self.get_shwap_cid(cid, timeout).await?; - let sample = Sample::decode(id, &data[..]).map_err(|e| P2pError::Shwap(e.to_string()))?; - Ok(sample) + self.send_command(P2pCmd::GetSample { + row_index, + column_index, + block_height, + respond_to: tx, + }) + .await?; + + match timeout { + Some(dur) => time::timeout(dur, rx) + .await + .map_err(|_| P2pError::RequestTimedOut)??, + None => rx.await?, + } } pub async fn get_eds( @@ -601,23 +572,6 @@ impl P2p { } } - /// Request a [`RowNamespaceData`] on bitswap protocol. - pub async fn get_row_namespace_data( - &self, - namespace: Namespace, - row_index: u16, - block_height: u64, - timeout: Option, - ) -> Result { - let id = RowNamespaceDataId::new(namespace, row_index, block_height)?; - let cid = convert_cid(&id.into())?; - - let data = self.get_shwap_cid(cid, timeout).await?; - let row_namespace_data = - RowNamespaceData::decode(id, &data[..]).map_err(|e| P2pError::Shwap(e.to_string()))?; - Ok(row_namespace_data) - } - pub async fn get_namespace_data( &self, namespace: Namespace, @@ -746,29 +700,25 @@ impl Drop for P2p { } #[derive(NetworkBehaviour)] -struct Behaviour +struct Behaviour where - B: Blockstore + 'static, S: Store + 'static, { - bitswap: beetswap::Behaviour, header_ex: header_ex::Behaviour, shr_ex: shrex::Behaviour, gossipsub: gossipsub::Behaviour, } -struct Worker +struct Worker where - B: Blockstore + 'static, S: Store + 'static, { cancellation_token: CancellationToken, - swarm: SwarmManager>, + swarm: SwarmManager>, header_sub_topic_hash: TopicHash, bad_encoding_fraud_sub_topic: TopicHash, cmd_rx: mpsc::Receiver, header_sub_state: Option, - bitswap_queries: HashMap, P2pError>>, network_compromised_token: Token, store: Arc, } @@ -778,13 +728,12 @@ struct HeaderSubState { channel: mpsc::Sender, } -impl Worker +impl Worker where - B: Blockstore, S: Store, { async fn new( - args: P2pArgs, + args: P2pArgs, cancellation_token: CancellationToken, cmd_rx: mpsc::Receiver, peer_tracker: PeerTracker, @@ -804,12 +753,6 @@ where fraudsub_ident_topic(BadEncodingFraudProof::TYPE, &args.network_id); let gossipsub = init_gossipsub(&args, [&header_sub_topic, &bad_encoding_fraud_sub_topic])?; - let bitswap = init_bitswap( - args.blockstore.clone(), - args.store.clone(), - &args.network_id, - )?; - let header_ex = header_ex::Behaviour::new(header_ex::Config { network_id: &args.network_id, header_store: args.store.clone(), @@ -823,7 +766,6 @@ where })?; swarm.attach_behaviour(Behaviour { - bitswap, gossipsub, header_ex, shr_ex, @@ -836,7 +778,6 @@ where bad_encoding_fraud_sub_topic: bad_encoding_fraud_sub_topic.hash(), header_sub_topic_hash: header_sub_topic.hash(), header_sub_state: None, - bitswap_queries: HashMap::new(), network_compromised_token: Token::new(), store: args.store, }) @@ -851,9 +792,6 @@ where _ = report_interval.tick() => { self.report(); } - _ = poll_closed(&mut self.bitswap_queries) => { - self.prune_canceled_bitswap_queries(); - } res = self.swarm.poll() => { match res { Ok(ev) => { @@ -877,25 +815,9 @@ where self.swarm.stop().await; } - fn prune_canceled_bitswap_queries(&mut self) { - let mut cancelled = SmallVec::<[_; 16]>::new(); - - for (query_id, chan) in &self.bitswap_queries { - if chan.is_closed() { - cancelled.push(*query_id); - } - } - - for query_id in cancelled { - self.bitswap_queries.remove(&query_id); - self.swarm.context().behaviour.bitswap.cancel(query_id); - } - } - - async fn on_behaviour_event(&mut self, ev: BehaviourEvent) -> Result<()> { + async fn on_behaviour_event(&mut self, ev: BehaviourEvent) -> Result<()> { match ev { BehaviourEvent::Gossipsub(ev) => self.on_gossip_sub_event(ev).await, - BehaviourEvent::Bitswap(ev) => self.on_bitswap_event(ev).await, BehaviourEvent::HeaderEx(ev) => self.on_header_ex_event(ev).await, BehaviourEvent::ShrEx(ev) => self.on_shrex_event(ev).await, } @@ -950,9 +872,6 @@ where P2pCmd::MarkAsArchival { peer_id } => { self.swarm.mark_as_archival(&peer_id); } - P2pCmd::GetShwapCid { cid, respond_to } => { - self.on_get_shwap_cid(cid, respond_to); - } P2pCmd::GetNetworkCompromisedToken { respond_to } => { respond_to.maybe_send(self.network_compromised_token.clone()) } @@ -1065,30 +984,6 @@ where } } - #[instrument(level = "trace", skip_all)] - fn on_get_shwap_cid(&mut self, cid: Cid, respond_to: OneshotResultSender, P2pError>) { - trace!("Requesting CID {cid} from bitswap"); - let query_id = self.swarm.context().behaviour.bitswap.get(&cid); - self.bitswap_queries.insert(query_id, respond_to); - } - - #[instrument(level = "trace", skip(self))] - async fn on_bitswap_event(&mut self, ev: beetswap::Event) { - match ev { - beetswap::Event::GetQueryResponse { query_id, data } => { - if let Some(respond_to) = self.bitswap_queries.remove(&query_id) { - respond_to.maybe_send_ok(data); - } - } - beetswap::Event::GetQueryError { query_id, error } => { - if let Some(respond_to) = self.bitswap_queries.remove(&query_id) { - let error: P2pError = error.into(); - respond_to.maybe_send_err(error); - } - } - } - } - #[instrument(level = "trace", skip(self))] async fn on_header_ex_event(&mut self, ev: header_ex::Event) { match ev { @@ -1235,23 +1130,6 @@ where } } -/// Awaits at least one channel from the `bitswap_queries` to close. -async fn poll_closed( - bitswap_queries: &mut HashMap, P2pError>>, -) { - poll_fn(|cx| { - for chan in bitswap_queries.values_mut() { - match chan.poll_closed(cx) { - Poll::Pending => continue, - Poll::Ready(_) => return Poll::Ready(()), - } - } - - Poll::Pending - }) - .await -} - fn validate_bootnode_addrs(addrs: &[Multiaddr]) -> Result<(), P2pError> { let mut invalid_addrs = Vec::new(); @@ -1268,12 +1146,11 @@ fn validate_bootnode_addrs(addrs: &[Multiaddr]) -> Result<(), P2pError> { } } -fn init_gossipsub<'a, B, S>( - args: &'a P2pArgs, +fn init_gossipsub<'a, S>( + args: &'a P2pArgs, topics: impl IntoIterator, ) -> Result where - B: Blockstore, S: Store, { // Set the message authenticity - How we expect to publish messages @@ -1299,21 +1176,3 @@ where Ok(gossipsub) } - -fn init_bitswap( - blockstore: Arc, - store: Arc, - network_id: &str, -) -> Result> -where - B: Blockstore + 'static, - S: Store + 'static, -{ - let protocol_prefix = celestia_protocol_id(network_id, "shwap"); - - Ok(beetswap::Behaviour::builder(blockstore) - .protocol_prefix(protocol_prefix.as_ref())? - .register_multihasher(ShwapMultihasher::new(store)) - .client_set_send_dont_have(false) - .build()) -} diff --git a/node/src/p2p/shwap.rs b/node/src/p2p/shwap.rs deleted file mode 100644 index e477c92ec..000000000 --- a/node/src/p2p/shwap.rs +++ /dev/null @@ -1,148 +0,0 @@ -use std::sync::Arc; - -use beetswap::multihasher::{Multihasher, MultihasherError}; -use blockstore::block::CidError; -use celestia_proto::bitswap::Block; -use celestia_types::row::{ROW_ID_MULTIHASH_CODE, Row, RowId}; -use celestia_types::row_namespace_data::{ - ROW_NAMESPACE_DATA_ID_MULTIHASH_CODE, RowNamespaceData, RowNamespaceDataId, -}; -use celestia_types::sample::{SAMPLE_ID_MULTIHASH_CODE, Sample, SampleId}; -use cid::{Cid, CidGeneric}; -use libp2p::multihash::Multihash; -use prost::Message; - -use crate::p2p::{MAX_MH_SIZE, P2pError, Result}; -use crate::store::Store; - -/// Multihasher for Shwap types. -pub(super) struct ShwapMultihasher -where - S: Store + 'static, -{ - header_store: Arc, -} - -impl ShwapMultihasher -where - S: Store + 'static, -{ - pub(super) fn new(header_store: Arc) -> Self { - ShwapMultihasher { header_store } - } -} - -impl Multihasher for ShwapMultihasher -where - S: Store + 'static, -{ - async fn hash( - &self, - multihash_code: u64, - input: &[u8], - ) -> Result, MultihasherError> { - macro_rules! hash_shwap_block { - ($id_type:ty, $container_type:ty) => {{ - let block = Block::decode(input).map_err(MultihasherError::custom_fatal)?; - let cid = CidGeneric::::read_bytes(block.cid.as_slice()) - .map_err(MultihasherError::custom_fatal)?; - - let id = <$id_type>::try_from(cid).map_err(MultihasherError::custom_fatal)?; - let container = <$container_type>::decode(id, block.container.as_slice()) - .map_err(MultihasherError::custom_fatal)?; - - let hash = convert_cid(&id.into()) - .map_err(MultihasherError::custom_fatal)? - .hash() - .to_owned(); - - let header = self - .header_store - .get_by_height(id.block_height()) - .await - .map_err(MultihasherError::custom_fatal)?; - - container - .verify(id, &header.dah) - .map_err(MultihasherError::custom_fatal)?; - - Ok(hash) - }}; - } - - match multihash_code { - ROW_ID_MULTIHASH_CODE => hash_shwap_block!(RowId, Row), - ROW_NAMESPACE_DATA_ID_MULTIHASH_CODE => { - hash_shwap_block!(RowNamespaceDataId, RowNamespaceData) - } - SAMPLE_ID_MULTIHASH_CODE => hash_shwap_block!(SampleId, Sample), - _ => Err(MultihasherError::UnknownMultihashCode), - } - } -} - -pub(crate) fn sample_cid(row_index: u16, column_index: u16, block_height: u64) -> Result { - let sample_id = SampleId::new(row_index, column_index, block_height)?; - convert_cid(&sample_id.into()) -} - -pub(crate) fn convert_cid(cid: &CidGeneric) -> Result { - beetswap::utils::convert_cid(cid).ok_or(P2pError::Cid(CidError::InvalidMultihashLength(64))) -} - -/// Extracts the `container` part from shwap's `Block` wrapper if its CID matches the expected one -pub(crate) fn get_block_container(expected_cid: &Cid, block: &[u8]) -> Result> { - let block = Block::decode(block)?; - let block_cid = Cid::read_bytes(block.cid.as_slice())?; - if block_cid != *expected_cid { - return Err(P2pError::Shwap(format!( - "cid in block ({block_cid}) different than expected ({expected_cid})", - ))); - } - - Ok(block.container) -} - -#[cfg(test)] -mod tests { - use super::*; - use crate::store::InMemoryStore; - use bytes::BytesMut; - use celestia_types::consts::appconsts::AppVersion; - use celestia_types::test_utils::{ExtendedHeaderGenerator, generate_dummy_eds}; - use celestia_types::{AxisType, DataAvailabilityHeader}; - use lumina_utils::test_utils::async_test; - - #[async_test] - async fn hash() { - let store = Arc::new(InMemoryStore::new()); - - let eds = generate_dummy_eds(4, AppVersion::V2); - let dah = DataAvailabilityHeader::from_eds(&eds); - - let mut generator = ExtendedHeaderGenerator::new(); - let header = generator.next_with_dah(dah.clone()); - - let sample = Sample::new(0, 0, AxisType::Row, &eds).unwrap(); - let mut sample_bytes = BytesMut::new(); - sample.encode(&mut sample_bytes); - - let cid = sample_cid(0, 0, 1).unwrap(); - let sample_id = SampleId::new(0, 0, 1).unwrap(); - - sample.verify(sample_id, &dah).unwrap(); - store.insert(header).await.unwrap(); - - let block = Block { - cid: cid.to_bytes(), - container: sample_bytes.to_vec(), - }; - - let hash = ShwapMultihasher::new(store) - .hash(SAMPLE_ID_MULTIHASH_CODE, &block.encode_to_vec()) - .await - .unwrap(); - - assert_eq!(hash, *cid.hash()); - } -} diff --git a/node/src/store.rs b/node/src/store.rs index bf8fb622e..c386fa37d 100644 --- a/node/src/store.rs +++ b/node/src/store.rs @@ -41,6 +41,9 @@ pub(crate) mod utils; /// Sampling metadata for a block. /// /// This struct persists DAS-ing information in a header store for future reference. +// TODO: after removal of bitswap, we no longer need to use blockstore to store samples. We could +// just store them directly in a store in some new table, indexed by height. This way we could +// remove sampling metadata and simplify pruner. #[derive(Debug, Default, Clone, Serialize, Deserialize)] #[cfg_attr(all(feature = "wasm-bindgen", target_arch = "wasm32"), wasm_bindgen)] pub struct SamplingMetadata { diff --git a/node/src/syncer.rs b/node/src/syncer.rs index d931c25a1..791098837 100644 --- a/node/src/syncer.rs +++ b/node/src/syncer.rs @@ -778,7 +778,6 @@ mod tests { use crate::p2p::header_session; use crate::store::InMemoryStore; use crate::test_utils::{MockP2pHandle, gen_filled_store}; - use crate::utils::OneshotResultSenderExt; use celestia_types::test_utils::ExtendedHeaderGenerator; use libp2p::request_response::OutboundFailure; use lumina_utils::test_utils::async_test; @@ -1509,9 +1508,9 @@ mod tests { sleep(Duration::from_secs(10)).await; for respond_chan in no_respond_chans { - respond_chan.maybe_send_err(P2pError::HeaderEx( + let _ = respond_chan.send(Err(P2pError::HeaderEx( HeaderExError::OutboundFailure(OutboundFailure::Timeout), - )); + ))); } }); } diff --git a/node/src/test_utils.rs b/node/src/test_utils.rs index a3692ba37..543fb18ba 100644 --- a/node/src/test_utils.rs +++ b/node/src/test_utils.rs @@ -5,8 +5,8 @@ use std::time::Duration; use celestia_proto::p2p::pb::{HeaderRequest, header_request::Data}; use celestia_types::ExtendedHeader; use celestia_types::hash::Hash; +use celestia_types::sample::Sample; use celestia_types::test_utils::ExtendedHeaderGenerator; -use cid::Cid; use lumina_utils::time::timeout; use tokio::sync::{mpsc, oneshot, watch}; @@ -211,13 +211,20 @@ impl MockP2pHandle { } } - /// Assert that a CID request was sent to the [`P2p`] worker and obtain a response channel. + /// Assert that a sample request was sent to the [`P2p`] worker and obtain a response channel. /// /// [`P2p`]: crate::p2p::P2p - pub async fn expect_get_shwap_cid(&mut self) -> (Cid, OneshotResultSender, P2pError>) { + pub async fn expect_get_sample( + &mut self, + ) -> (u16, u16, u64, OneshotResultSender) { match self.expect_cmd().await { - P2pCmd::GetShwapCid { cid, respond_to } => (cid, respond_to), - cmd => panic!("Expecting GetShwapCid, but received: {cmd:?}"), + P2pCmd::GetSample { + row_index, + column_index, + block_height, + respond_to, + } => (row_index, column_index, block_height, respond_to), + cmd => panic!("Expecting GetSample, but received: {cmd:?}"), } } } diff --git a/node/src/utils.rs b/node/src/utils.rs index 75ac8998e..d170aba24 100644 --- a/node/src/utils.rs +++ b/node/src/utils.rs @@ -67,29 +67,6 @@ where } } -pub(crate) trait OneshotResultSenderExt -where - T: Send + 'static, - E: Send + 'static, -{ - fn maybe_send_ok(self, val: T); - fn maybe_send_err(self, err: impl Into); -} - -impl OneshotResultSenderExt for oneshot::Sender> -where - T: Send + 'static, - E: Send + 'static, -{ - fn maybe_send_ok(self, val: T) { - let _ = self.send(Ok(val)); - } - - fn maybe_send_err(self, err: impl Into) { - let _ = self.send(Err(err.into())); - } -} - pub(crate) trait MultiaddrExt { fn peer_id(&self) -> Option; } diff --git a/node/tests/graceful_stop.rs b/node/tests/graceful_stop.rs index ba0f10462..c77e483ef 100644 --- a/node/tests/graceful_stop.rs +++ b/node/tests/graceful_stop.rs @@ -64,13 +64,13 @@ async fn new_node(path: impl AsRef) -> (Node, E let store = RedbStore::new(db.clone()).await.unwrap(); let blockstore = RedbBlockstore::new(db); - let (_, bridge_ma) = fetch_bridge_info().await; + let (_, bridge_addrs) = fetch_bridge_info().await; Node::builder() .store(store) .blockstore(blockstore) .network(Network::custom("private").unwrap()) - .bootnodes([bridge_ma]) + .bootnodes(bridge_addrs) .start_subscribed() .await .unwrap() diff --git a/node/tests/node.rs b/node/tests/node.rs index 3b3cb8bde..0a9631937 100644 --- a/node/tests/node.rs +++ b/node/tests/node.rs @@ -1,28 +1,14 @@ -#![cfg(not(target_arch = "wasm32"))] - -use std::time::Duration; - use celestia_types::consts::HASH_SIZE; -use celestia_types::consts::appconsts::AppVersion; -use celestia_types::fraud_proof::BadEncodingFraudProof; use celestia_types::hash::Hash; -use celestia_types::test_utils::{ExtendedHeaderGenerator, corrupt_eds, generate_dummy_eds}; -use futures::StreamExt; -use libp2p::swarm::NetworkBehaviour; -use libp2p::{Multiaddr, SwarmBuilder, gossipsub, noise, ping, tcp, yamux}; -use lumina_node::store::{InMemoryStore, Store}; -use lumina_node::test_utils::{ - ExtendedHeaderGeneratorExt, gen_filled_store, listening_test_node_builder, test_node_builder, -}; +use lumina_node::test_utils::{gen_filled_store, test_node_builder}; +use lumina_utils::test_utils::async_test; use rand::Rng; -use tendermint_proto::Protobuf; -use tokio::{select, spawn, sync::mpsc, time::sleep}; -use crate::utils::{fetch_bridge_info, new_connected_node}; +use crate::utils::new_connected_node; mod utils; -#[tokio::test] +#[async_test] async fn connects_to_the_go_bridge_node() { let (node, _) = new_connected_node().await; @@ -30,7 +16,7 @@ async fn connects_to_the_go_bridge_node() { assert!(info.num_peers() >= 1); } -#[tokio::test] +#[async_test] async fn header_store_access() { let (store, _) = gen_filled_store(100).await; let node = test_node_builder().store(store).start().await.unwrap(); @@ -84,17 +70,25 @@ async fn header_store_access() { } } +#[cfg(not(target_arch = "wasm32"))] #[tokio::test] async fn peer_discovery() { + use std::time::Duration; + + use lumina_node::test_utils::listening_test_node_builder; + use tokio::time::sleep; + + use crate::utils::fetch_bridge_info; + // Bridge node cannot connect to other nodes because it is behind Docker's NAT. // However Node2 and Node3 can discover its address via Node1. - let (bridge_peer_id, bridge_ma) = fetch_bridge_info().await; + let (bridge_peer_id, bridge_addrs) = fetch_bridge_info().await; // Node1 // // This node connects to Bridge node. let node1 = listening_test_node_builder() - .bootnodes([bridge_ma]) + .bootnodes(bridge_addrs) .start() .await .unwrap(); @@ -160,8 +154,18 @@ async fn peer_discovery() { assert_eq!(tracker_info.num_connected_trusted_peers, 1); } +#[cfg(not(target_arch = "wasm32"))] #[tokio::test] async fn stops_services_when_network_is_compromised() { + use std::time::Duration; + + use tokio::time::sleep; + + use celestia_types::AppVersion; + use celestia_types::test_utils::{ExtendedHeaderGenerator, corrupt_eds, generate_dummy_eds}; + use lumina_node::store::{InMemoryStore, Store}; + use lumina_node::test_utils::{ExtendedHeaderGeneratorExt, listening_test_node_builder}; + let mut generator = ExtendedHeaderGenerator::new(); let store = InMemoryStore::new(); @@ -205,7 +209,18 @@ async fn stops_services_when_network_is_compromised() { assert!(node.syncer_info().await.is_err()); } -fn spawn_befp_announcer(connect_to: Multiaddr) -> mpsc::Sender { +#[cfg(not(target_arch = "wasm32"))] +fn spawn_befp_announcer( + connect_to: libp2p::Multiaddr, +) -> tokio::sync::mpsc::Sender { + use celestia_types::fraud_proof::BadEncodingFraudProof; + use futures::StreamExt; + use libp2p::swarm::NetworkBehaviour; + use libp2p::{SwarmBuilder, gossipsub, noise, ping, tcp, yamux}; + use tendermint_proto::Protobuf; + use tokio::sync::mpsc; + use tokio::{select, spawn}; + #[derive(NetworkBehaviour)] struct Behaviour { ping: ping::Behaviour, diff --git a/node/tests/shwap.rs b/node/tests/shwap.rs index b45529e2e..f72f6b400 100644 --- a/node/tests/shwap.rs +++ b/node/tests/shwap.rs @@ -1,31 +1,50 @@ -#![cfg(not(target_arch = "wasm32"))] - +use std::cmp::Ordering; use std::collections::HashSet; use std::time::Duration; -use beetswap::utils::convert_cid; -use blockstore::Blockstore; -use celestia_rpc::{HeaderClient, ShareClient}; -use celestia_types::nmt::{Namespace, NamespacedSha2Hasher}; -use celestia_types::sample::SampleId; +use celestia_rpc::ShareClient; +use celestia_types::nmt::Namespace; use celestia_types::{AppVersion, Blob, ExtendedHeader}; -use cid::{Cid, CidGeneric}; -use lumina_node::NodeError; +use futures::stream::FuturesUnordered; +use futures::{FutureExt, StreamExt}; use lumina_node::blockstore::InMemoryBlockstore; use lumina_node::events::NodeEvent; use lumina_node::node::P2pError; +use lumina_node::store::InMemoryStore; use lumina_node::test_utils::test_node_builder; +use lumina_node::{Node, NodeError}; +use lumina_utils::test_utils::async_test; +use lumina_utils::time::timeout; use rand::RngCore; -use tokio::sync::mpsc; -use tokio::time::timeout; -use utils::new_connected_node_with_builder; -use crate::utils::{blob_submit, bridge_client, new_connected_node}; +use crate::utils::{ + blob_submit, bridge_client, new_connected_node, new_connected_node_with_builder, +}; mod utils; -#[tokio::test] +// TODO: running each of those tests in browser separately makes them pass in milliseconds, +// but running them all at once hangs after 2-3 tests. Probably an issue with opening connections +// to the same go node over and over. Couldn't dig deep enough to understand it fully tho. Would +// need to investigate the QLOG of chrome to see what happens on quic level, maybe it is +// https://issues.chromium.org/issues/400699540 +// ^- above link is now private, which means they think it is security issue, it was public before +// more info: https://github.com/celestiaorg/lumina/issues/287, it's the old bug + +#[async_test] async fn shwap_sampling_forward() { + #[cfg(target_arch = "wasm32")] + use tracing_subscriber::prelude::*; + + #[cfg(target_arch = "wasm32")] + let fmt_layer = tracing_subscriber::fmt::layer() + .with_ansi(false) + .with_timer(tracing_subscriber::fmt::time::UtcTime::rfc_3339()) + .with_writer(tracing_web::MakeConsoleWriter) // write events to the console + .with_filter(tracing_subscriber::filter::LevelFilter::DEBUG); + + #[cfg(target_arch = "wasm32")] + let _ = tracing_subscriber::registry().with(fmt_layer).try_init(); let (node, _) = new_connected_node().await; // create new events sub to ignore all previous events @@ -42,22 +61,19 @@ async fn shwap_sampling_forward() { break height; } }; - // timeout is double of the block time on CI + // Timeout is double of the block time on CI let new_head = timeout(Duration::from_secs(9), get_new_head).await.unwrap(); // wait for height to be sampled let wait_height_sampled = async { loop { let ev = events.recv().await.unwrap(); - let NodeEvent::SamplingResult { - height, timed_out, .. - } = ev.event - else { + let NodeEvent::SamplingResult { height, failed, .. } = ev.event else { continue; }; if height == new_head { - assert!(!timed_out); + assert!(!failed); break; } } @@ -66,10 +82,25 @@ async fn shwap_sampling_forward() { .await .unwrap(); } + + // shouldn't be needed but I'm trying anything + node.stop().await; } -#[tokio::test] +#[async_test] async fn shwap_sampling_backward() { + #[cfg(target_arch = "wasm32")] + use tracing_subscriber::prelude::*; + + #[cfg(target_arch = "wasm32")] + let fmt_layer = tracing_subscriber::fmt::layer() + .with_ansi(false) + .with_timer(tracing_subscriber::fmt::time::UtcTime::rfc_3339()) + .with_writer(tracing_web::MakeConsoleWriter) // write events to the console + .with_filter(tracing_subscriber::filter::LevelFilter::DEBUG); + + #[cfg(target_arch = "wasm32")] + let _ = tracing_subscriber::registry().with(fmt_layer).try_init(); let (node, mut events) = new_connected_node().await; let current_head = node.get_local_head_header().await.unwrap().height(); @@ -91,7 +122,7 @@ async fn shwap_sampling_backward() { } } }; - let (from_height, to_height) = timeout(Duration::from_secs(4), new_batch_synced) + let (from_height, to_height) = timeout(Duration::from_secs(5), new_batch_synced) .await .unwrap(); @@ -102,14 +133,11 @@ async fn shwap_sampling_backward() { timeout(Duration::from_secs(10), async { loop { let ev = events.recv().await.unwrap(); - let NodeEvent::SamplingResult { - height, timed_out, .. - } = ev.event - else { + let NodeEvent::SamplingResult { height, failed, .. } = ev.event else { continue; }; - assert!(!timed_out); + assert!(!failed); headers_to_sample.remove(&height); if headers_to_sample.is_empty() { @@ -119,10 +147,24 @@ async fn shwap_sampling_backward() { }) .await .unwrap(); + + node.stop().await; } -#[tokio::test] +#[async_test] async fn shwap_request_sample() { + #[cfg(target_arch = "wasm32")] + use tracing_subscriber::prelude::*; + + #[cfg(target_arch = "wasm32")] + let fmt_layer = tracing_subscriber::fmt::layer() + .with_ansi(false) + .with_timer(tracing_subscriber::fmt::time::UtcTime::rfc_3339()) + .with_writer(tracing_web::MakeConsoleWriter) // write events to the console + .with_filter(tracing_subscriber::filter::LevelFilter::DEBUG); + + #[cfg(target_arch = "wasm32")] + let _ = tracing_subscriber::registry().with(fmt_layer).try_init(); let (node, _) = new_connected_node().await; let client = bridge_client().await; @@ -131,7 +173,7 @@ async fn shwap_request_sample() { let blob = Blob::new(ns, random_bytes(blob_len), None, AppVersion::V2).unwrap(); let height = blob_submit(&client, &[blob]).await; - let header = node.get_header_by_height(height).await.unwrap(); + let header = wait_for_height(&node, height).await; let square_width = header.square_width(); // check existing sample @@ -146,7 +188,7 @@ async fn shwap_request_sample() { .await .unwrap(); let sample = node - .request_sample(0, 0, height, Some(Duration::from_millis(500))) + .request_sample(0, 0, height, Some(Duration::from_secs(1))) .await .unwrap(); assert_eq!(expected, sample.share); @@ -157,15 +199,29 @@ async fn shwap_request_sample() { square_width + 1, square_width + 1, height, - Some(Duration::from_millis(500)), + Some(Duration::from_secs(10)), ) .await .unwrap_err(); - assert!(matches!(err, NodeError::P2p(P2pError::RequestTimedOut))); + assert!(matches!(err, NodeError::P2p(P2pError::ShrEx(_)))); + + node.stop().await; } -#[tokio::test] +#[async_test] async fn shwap_request_row() { + #[cfg(target_arch = "wasm32")] + use tracing_subscriber::prelude::*; + + #[cfg(target_arch = "wasm32")] + let fmt_layer = tracing_subscriber::fmt::layer() + .with_ansi(false) + .with_timer(tracing_subscriber::fmt::time::UtcTime::rfc_3339()) + .with_writer(tracing_web::MakeConsoleWriter) // write events to the console + .with_filter(tracing_subscriber::filter::LevelFilter::DEBUG); + + #[cfg(target_arch = "wasm32")] + let _ = tracing_subscriber::registry().with(fmt_layer).try_init(); let (node, _) = new_connected_node().await; let client = bridge_client().await; @@ -174,7 +230,8 @@ async fn shwap_request_row() { let blob = Blob::new(ns, random_bytes(blob_len), None, AppVersion::V2).unwrap(); let height = blob_submit(&client, &[blob]).await; - let header = node.get_header_by_height(height).await.unwrap(); + + let header = wait_for_height(&node, height).await; let eds = client .share_get_eds(header.height(), header.app_version()) .await @@ -193,76 +250,25 @@ async fn shwap_request_row() { .request_row(square_width + 1, height, Some(Duration::from_secs(1))) .await .unwrap_err(); - assert!(matches!(err, NodeError::P2p(P2pError::RequestTimedOut))); -} - -#[tokio::test] -async fn shwap_request_row_namespace_data() { - let (node, _) = new_connected_node().await; - let client = bridge_client().await; + assert!(matches!(err, NodeError::P2p(P2pError::ShrEx(_)))); - let ns = Namespace::const_v0(rand::random()); - let blob_len = rand::random::() % 4096 + 1; - let blob = Blob::new(ns, random_bytes(blob_len), None, AppVersion::V2).unwrap(); - - let height = blob_submit(&client, &[blob]).await; - let header = node.get_header_by_height(height).await.unwrap(); - let eds = client - .share_get_eds(header.height(), header.app_version()) - .await - .unwrap(); - let square_width = header.square_width(); - - // check existing row namespace data - let rows_with_ns: Vec<_> = header - .dah - .row_roots() - .iter() - .enumerate() - .filter_map(|(n, hash)| { - hash.contains::(*ns) - .then_some(n as u16) - }) - .collect(); - let eds_ns_data = eds.get_namespace_data(ns, &header.dah, height).unwrap(); - - for (n, &row) in rows_with_ns.iter().enumerate() { - let row_ns_data = node - .request_row_namespace_data(ns, row, height, Some(Duration::from_secs(1))) - .await - .unwrap(); - assert_eq!(eds_ns_data[n].1, row_ns_data); - } - - // check nonexisting row row namespace data - let err = node - .request_row_namespace_data(ns, square_width + 1, height, Some(Duration::from_secs(1))) - .await - .unwrap_err(); - assert!(matches!(err, NodeError::P2p(P2pError::RequestTimedOut))); - - // check nonexisting namespace row namespace data - // for namespace that row actually contains - // PFB (0x04) < 0x05 < Primary ns padding (0x255) - let unknown_ns = Namespace::const_v0([0, 0, 0, 0, 0, 0, 0, 0, 0, 5]); - let row = node - .request_row_namespace_data(unknown_ns, 0, height, Some(Duration::from_secs(1))) - .await - .unwrap(); - assert!(row.shares.is_empty()); - - // check nonexisting namespace row namespace data - // for namespace that row doesn't contain - let unknown_ns = Namespace::TAIL_PADDING; - let err = node - .request_row_namespace_data(unknown_ns, 0, height, Some(Duration::from_secs(1))) - .await - .unwrap_err(); - assert!(matches!(err, NodeError::P2p(P2pError::RequestTimedOut))); + node.stop().await; } -#[tokio::test] +#[async_test] async fn shwap_request_all_blobs() { + #[cfg(target_arch = "wasm32")] + use tracing_subscriber::prelude::*; + + #[cfg(target_arch = "wasm32")] + let fmt_layer = tracing_subscriber::fmt::layer() + .with_ansi(false) + .with_timer(tracing_subscriber::fmt::time::UtcTime::rfc_3339()) + .with_writer(tracing_web::MakeConsoleWriter) // write events to the console + .with_filter(tracing_subscriber::filter::LevelFilter::DEBUG); + + #[cfg(target_arch = "wasm32")] + let _ = tracing_subscriber::registry().with(fmt_layer).try_init(); let (node, _) = new_connected_node().await; let client = bridge_client().await; @@ -275,6 +281,7 @@ async fn shwap_request_all_blobs() { .collect(); let height = blob_submit(&client, &blobs).await; + wait_for_height(&node, height).await; // check existing namespace let received = node @@ -292,142 +299,94 @@ async fn shwap_request_all_blobs() { .unwrap(); assert!(received.is_empty()); -} - -#[tokio::test] -async fn shwap_request_sample_should_cleanup_unneeded_samples() { - // submit some blobs to celestia to get bigger square, so that daser - // doesn't sample whole block - let ns = Namespace::const_v0(rand::random()); - let blobs: Vec<_> = (0..5) - .map(|_| { - let blob_len = rand::random::() % 4096 + 1; - Blob::new(ns, random_bytes(blob_len), None, AppVersion::V2).unwrap() - }) - .collect(); - // we submit before creating a node because it's quite slow and events can lag - let client = bridge_client().await; - let submitted_height = blob_submit(&client, &blobs).await; - let header = client.header_get_by_height(submitted_height).await.unwrap(); - - let (removed_sender, mut removed_receiver) = mpsc::unbounded_channel(); - let builder = test_node_builder() - // explicitely set pruning window to something big so that - // pruner doesn't kick in - .pruning_window(Duration::from_secs(60 * 60 * 24)) - .blockstore(TestBlockstore::new(removed_sender)); + node.stop().await; +} - let (node, mut events) = new_connected_node_with_builder(builder).await; +#[async_test] +async fn shwap_request_concurrent() { + #[cfg(target_arch = "wasm32")] + use tracing_subscriber::prelude::*; - // wait for node to sample the height we just submitted - timeout(Duration::from_secs(10), async { - loop { - let ev = events.recv().await.unwrap(); - let NodeEvent::SamplingResult { height, .. } = ev.event else { - continue; - }; + #[cfg(target_arch = "wasm32")] + let fmt_layer = tracing_subscriber::fmt::layer() + .with_ansi(false) + .with_timer(tracing_subscriber::fmt::time::UtcTime::rfc_3339()) + .with_writer(tracing_web::MakeConsoleWriter) // write events to the console + .with_filter(tracing_subscriber::filter::LevelFilter::DEBUG); - if height == submitted_height { - break; - } - } - }) - .await - .unwrap(); + #[cfg(target_arch = "wasm32")] + let _ = tracing_subscriber::registry().with(fmt_layer).try_init(); - // get the cids selected by daser - let cids = node - .get_sampling_metadata(submitted_height) - .await - .unwrap() - .unwrap() - .cids; - - // try to request a sample that wasn't selected by daser - let (row, col, cid) = loop { - let (row, col, cid) = random_sample(&header); - if !cids.contains(&cid) { - break (row, col, cid); - } - }; + let builder = test_node_builder().pruning_window(Duration::from_secs(60)); + let (node, _) = new_connected_node_with_builder(builder).await; + let client = bridge_client().await; - node.request_sample(row, col, submitted_height, None) - .await - .unwrap(); + let blob_ns = Namespace::const_v0(rand::random()); - // it should already be removed from the blockstore - let removed_cid = removed_receiver.try_recv().unwrap(); - assert_eq!(removed_cid, cid); + let mut headers = Vec::new(); - // now try to get a cid that was selected by daser - let id = SampleId::try_from(cids[0]).unwrap(); - assert_eq!(id.block_height(), submitted_height); + for _ in 0..3 { + let blobs: Vec<_> = (0..3) + .map(|_| { + let blob_len = rand::random::() % 512 + 1; + Blob::new(blob_ns, random_bytes(blob_len), None, AppVersion::V2).unwrap() + }) + .collect(); - node.request_sample(id.row_index(), id.column_index(), submitted_height, None) - .await - .unwrap(); + let height = blob_submit(&client, &blobs).await; + headers.push(wait_for_height(&node, height).await); + } - // it shouldn't be removed from blockstore within pruning window - removed_receiver.try_recv().unwrap_err(); -} + let mut futs = FuturesUnordered::new(); -struct TestBlockstore { - blockstore: InMemoryBlockstore, - removed_sender: mpsc::UnboundedSender, -} + for hdr in &headers { + // samples for each header + for _ in 0..4 { + let row = rand::random::() % hdr.square_width(); + let col = rand::random::() % hdr.square_width(); -impl TestBlockstore { - fn new(removed_sender: mpsc::UnboundedSender) -> Self { - Self { - blockstore: InMemoryBlockstore::new(), - removed_sender, + futs.push( + node.request_sample(row, col, hdr.height(), Some(Duration::from_secs(1))) + .map(|res| res.map(|_| ())) + .boxed(), + ); } - } -} -impl Blockstore for TestBlockstore { - async fn get( - &self, - cid: &CidGeneric, - ) -> blockstore::Result>> { - self.blockstore.get(cid).await - } - - async fn put_keyed( - &self, - cid: &CidGeneric, - data: &[u8], - ) -> blockstore::Result<()> { - self.blockstore.put_keyed(cid, data).await - } + // rows + for _ in 0..2 { + let row = rand::random::() % hdr.square_width(); - async fn remove(&self, cid: &CidGeneric) -> blockstore::Result<()> { - self.blockstore.remove(cid).await?; + futs.push( + node.request_row(row, hdr.height(), Some(Duration::from_secs(1))) + .map(|res| res.map(|_| ())) + .boxed(), + ); + } - let cid = convert_cid(cid).unwrap(); - self.removed_sender.send(cid).unwrap(); + // namespace data + for ns in [Namespace::PAY_FOR_BLOB, blob_ns] { + futs.push( + node.request_namespace_data(ns, hdr.height(), Some(Duration::from_secs(2))) + .map(|res| res.map(|_| ())) + .boxed(), + ); + } - Ok(()) + // eds + futs.push( + node.request_extended_data_square(hdr.height(), Some(Duration::from_secs(2))) + .map(|res| res.map(|_| ())) + .boxed(), + ); } - async fn close(self) -> blockstore::Result<()> { - self.blockstore.close().await + while let Some(res) = futs.next().await { + res.expect("Request failed"); } -} -fn random_sample(header: &ExtendedHeader) -> (u16, u16, Cid) { - let square = header.square_width(); - let id = SampleId::new( - rand::random::() % square, - rand::random::() % square, - header.height(), - ) - .unwrap(); - - let cid = convert_cid(&id.into()).unwrap(); - - (id.row_index(), id.column_index(), cid) + drop(futs); + node.stop().await; } fn random_bytes(len: usize) -> Vec { @@ -435,3 +394,28 @@ fn random_bytes(len: usize) -> Vec { rand::thread_rng().fill_bytes(&mut bytes); bytes } + +async fn wait_for_height( + node: &Node, + height: u64, +) -> ExtendedHeader { + if let Ok(hdr) = node.get_header_by_height(height).await { + return hdr; + } + + // we didn't find header, so let's wait for it on subscription + let mut sub = node.header_subscribe().await.unwrap(); + loop { + let hdr = sub.recv().await.unwrap(); + + match hdr.height().cmp(&height) { + Ordering::Less => continue, + Ordering::Equal => return hdr, + Ordering::Greater => break, + } + } + + // check last time with get by height, maybe it was inserted in a moment that + // we didn't get it previously yet but also missed it on subscription + node.get_header_by_height(height).await.unwrap() +} diff --git a/node/tests/utils/mod.rs b/node/tests/utils/mod.rs index cd5c7f1a0..8b264125b 100644 --- a/node/tests/utils/mod.rs +++ b/node/tests/utils/mod.rs @@ -6,6 +6,7 @@ use std::time::Duration; use blockstore::Blockstore; use celestia_rpc::{Client, TxConfig, prelude::*}; use celestia_types::Blob; +use libp2p::identity::Keypair; use libp2p::{Multiaddr, PeerId, multiaddr::Protocol}; use lumina_node::NodeBuilder; use lumina_node::blockstore::InMemoryBlockstore; @@ -13,30 +14,42 @@ use lumina_node::events::EventSubscriber; use lumina_node::node::Node; use lumina_node::store::{InMemoryStore, Store}; use lumina_node::test_utils::test_node_builder; +use lumina_utils::test_utils::env_var; +use lumina_utils::time::sleep; use tokio::sync::Mutex; -use tokio::time::sleep; -const WS_URL: &str = "ws://localhost:26658"; +#[cfg(not(target_arch = "wasm32"))] +const RPC_URL: &str = "ws://localhost:36658"; +#[cfg(target_arch = "wasm32")] +const RPC_URL: &str = "http://localhost:36658"; + +#[cfg(target_arch = "wasm32")] +wasm_bindgen_test::wasm_bindgen_test_configure!(run_in_browser); pub async fn bridge_client() -> Client { - Client::new(WS_URL, None, None, None).await.unwrap() + let token = env_var("CELESTIA_NODE_AUTH_TOKEN_ADMIN") + .expect("Token not found, run ./tools/gen_auth_tokens.sh"); + Client::new(RPC_URL, Some(&token), None, None) + .await + .unwrap() } -pub async fn fetch_bridge_info() -> (PeerId, Multiaddr) { +pub async fn fetch_bridge_info() -> (PeerId, Vec) { let client = bridge_client().await; let bridge_info = client.p2p_info().await.unwrap(); - let mut ma = bridge_info + let addrs = bridge_info .addrs .into_iter() - .find(|ma| ma.protocol_stack().any(|protocol| protocol == "tcp")) - .expect("Bridge doesn't listen on tcp"); - - if !ma.protocol_stack().any(|protocol| protocol == "p2p") { - ma.push(Protocol::P2p(bridge_info.id.into())) - } - - (bridge_info.id.into(), ma) + .map(|mut ma| { + if !ma.protocol_stack().any(|protocol| protocol == "p2p") { + ma.push(Protocol::P2p(bridge_info.id.into())) + } + ma + }) + .collect(); + + (bridge_info.id.into(), addrs) } pub async fn new_connected_node_with_builder( @@ -46,10 +59,17 @@ where B: Blockstore + 'static, S: Store + 'static, { - let (_, bridge_ma) = fetch_bridge_info().await; + let (_, bridge_addrs) = fetch_bridge_info().await; + + // TODO: tried protecting peer id on the go side but doesn't help either + let keypair = Keypair::generate_ed25519(); + let peer_id: PeerId = keypair.public().into(); + let client = bridge_client().await; + client.p2p_protect(&peer_id.into(), "test").await.unwrap(); let (node, events) = builder - .bootnodes([bridge_ma]) + .keypair(keypair) + .bootnodes(bridge_addrs) .start_subscribed() .await .unwrap(); diff --git a/rpc/Cargo.toml b/rpc/Cargo.toml index 27bfdee9e..a165bc430 100644 --- a/rpc/Cargo.toml +++ b/rpc/Cargo.toml @@ -57,7 +57,6 @@ tokio = { workspace = true, features = ["sync"] } tracing.workspace = true [target.'cfg(not(target_arch = "wasm32"))'.dev-dependencies] -dotenvy.workspace = true libp2p = { workspace = true, features = [ "tokio", "macros", @@ -69,7 +68,6 @@ tokio = { workspace = true, features = ["rt", "macros"] } [target.'cfg(target_arch = "wasm32")'.dev-dependencies] getrandom_02.workspace = true -rust-embed = { workspace = true, features = ["debug-embed", "include-exclude", "interpolate-folder-path"] } wasm-bindgen-test.workspace = true [features] diff --git a/rpc/tests/utils/client.rs b/rpc/tests/utils/client.rs index 4c2112c50..b82785205 100644 --- a/rpc/tests/utils/client.rs +++ b/rpc/tests/utils/client.rs @@ -6,6 +6,7 @@ use celestia_rpc::{Client, TxConfig}; use celestia_types::Blob; use jsonrpsee::core::ClientError; use jsonrpsee::core::client::ClientT; +use lumina_utils::test_utils::env_var; use tokio::sync::{Mutex, MutexGuard}; // Use node-2 (light node) as the default RPC URL @@ -27,66 +28,24 @@ pub enum AuthLevel { Admin, } -#[cfg(not(target_arch = "wasm32"))] -fn token_from_env(auth_level: AuthLevel) -> Result> { +fn token_from_env(auth_level: AuthLevel) -> Option { match auth_level { - AuthLevel::Skip => Ok(None), - AuthLevel::Read => Ok(Some(std::env::var("CELESTIA_NODE_AUTH_TOKEN_READ")?)), - AuthLevel::Write => Ok(Some(std::env::var("CELESTIA_NODE_AUTH_TOKEN_WRITE")?)), - AuthLevel::Admin => Ok(Some(std::env::var("CELESTIA_NODE_AUTH_TOKEN_ADMIN")?)), + AuthLevel::Skip => None, + AuthLevel::Read => env_var("CELESTIA_NODE_AUTH_TOKEN_READ"), + AuthLevel::Write => env_var("CELESTIA_NODE_AUTH_TOKEN_WRITE"), + AuthLevel::Admin => env_var("CELESTIA_NODE_AUTH_TOKEN_ADMIN"), } } -#[cfg(target_arch = "wasm32")] -fn token_from_env(auth_level: AuthLevel) -> Result> { - #[derive(rust_embed::Embed)] - #[folder = "$CARGO_MANIFEST_DIR/../"] - #[allow_missing = true] - #[include = ".env*"] - struct Env; - - let token_pattern = match auth_level { - AuthLevel::Skip => return Ok(None), - AuthLevel::Read => "CELESTIA_NODE_AUTH_TOKEN_READ=", - AuthLevel::Write => "CELESTIA_NODE_AUTH_TOKEN_WRITE=", - AuthLevel::Admin => "CELESTIA_NODE_AUTH_TOKEN_ADMIN=", - }; - - let env = Env::get(".env") - .or_else(|| Env::get(".env.sample")) - .ok_or(anyhow::anyhow!("Couldn't find .env file"))?; - - let token = str::from_utf8(env.data.as_ref())? - .lines() - .find_map(|line| line.strip_prefix(token_pattern)); - - if token.is_some_and(|t| !t.is_empty()) { - Ok(token.map(ToOwned::to_owned)) - } else { - anyhow::bail!( - "CELESTIA_NODE_AUTH_TOKEN_ for {auth_level:?} variable not found. \ - Make sure to run 'tools/gen_auth_tokens.sh'" - ) - } -} - -#[cfg(not(target_arch = "wasm32"))] fn env_or(var_name: &str, or_value: &str) -> String { - std::env::var(var_name).unwrap_or_else(|_| or_value.to_owned()) -} - -#[cfg(target_arch = "wasm32")] -fn env_or(_var_name: &str, or_value: &str) -> String { - or_value.to_owned() + env_var(var_name).unwrap_or_else(|| or_value.to_owned()) } pub async fn new_test_client_with_url( auth_level: AuthLevel, celestia_rpc_url: &str, ) -> Result { - #[cfg(not(target_arch = "wasm32"))] - let _ = dotenvy::dotenv(); - let token = token_from_env(auth_level)?; + let token = token_from_env(auth_level); let url = env_or("CELESTIA_RPC_URL", celestia_rpc_url); let client = Client::new(&url, token.as_deref(), None, None).await?; diff --git a/types/src/test_utils.rs b/types/src/test_utils.rs index 97dd99aa6..1cbacb184 100644 --- a/types/src/test_utils.rs +++ b/types/src/test_utils.rs @@ -166,7 +166,7 @@ impl ExtendedHeaderGenerator { let time = self .spoofed_block_time .map(|t| t.0) - .unwrap_or_else(Time::now); + .unwrap_or_else(|| self.now()); generate_next(1, header, time, &self.key, None) } @@ -197,7 +197,7 @@ impl ExtendedHeaderGenerator { let time = self .spoofed_block_time .map(|t| t.0) - .unwrap_or_else(Time::now); + .unwrap_or_else(|| self.now()); generate_next(1, header, time, &self.key, Some(dah)) } @@ -307,11 +307,22 @@ impl ExtendedHeaderGenerator { self.spoofed_block_time = None; } + pub fn now(&self) -> Time { + let now = Time::now(); + if let Some(hdr) = &self.current_header + && hdr.time() >= now + { + (hdr.time() + Duration::from_millis(1)).expect("time overflow") + } else { + now + } + } + // private function which gets and increments generator time, since we cannot have multiple headers on the // exact same timestamp fn get_and_increment_time(&mut self, amount: u64) -> Time { let Some((spoofed_time, block_time)) = self.spoofed_block_time.take() else { - return Time::now(); + return self.now(); }; let block_time_ms: u64 = block_time.as_millis().try_into().expect("u64 overflow"); diff --git a/utils/Cargo.toml b/utils/Cargo.toml index f90d40ec2..5cd36a871 100644 --- a/utils/Cargo.toml +++ b/utils/Cargo.toml @@ -18,6 +18,11 @@ futures = { workspace = true, optional = true } gloo-timers = { workspace = true, optional = true } js-sys = { workspace = true, optional = true } pin-project = { workspace = true, optional = true } +rust-embed = { workspace = true, optional = true, features = [ + "debug-embed", + "include-exclude", + "interpolate-folder-path" +] } send_wrapper = { workspace = true, optional = true } wasm-bindgen = { workspace = true, optional = true } wasm-bindgen-futures = { workspace = true, optional = true } @@ -25,6 +30,7 @@ wasm-bindgen-test = { workspace = true, optional = true } web-time = { version = "1.1", optional = true } [target.'cfg(not(target_arch = "wasm32"))'.dependencies] +dotenvy = { workspace = true, optional = true } # === This is a workaround! === # `lumina-utils` is used as an intermidiate crate and does not really use `tonic`, # see [features] below for explanation. @@ -42,7 +48,7 @@ default = [] executor = ["token", "dep:tokio-util", "dep:tokio", "tokio/rt", "dep:wasm-bindgen", "dep:wasm-bindgen-futures", "dep:send_wrapper"] token = ["dep:tokio-util"] time = ["dep:gloo-timers", "dep:tokio", "tokio/time", "dep:send_wrapper", "dep:futures", "dep:pin-project", "dep:web-time"] -test-utils = ["dep:tokio", "tokio/macros", "dep:wasm-bindgen-test"] +test-utils = ["dep:dotenvy", "dep:tokio", "tokio/macros", "dep:rust-embed", "dep:wasm-bindgen-test"] make-object = ["dep:js-sys"] # Workaround is required, because we want to pass through `tls-*` feature flags to tonic, diff --git a/utils/src/test_utils.rs b/utils/src/test_utils.rs index 95ff4579c..1b44613b6 100644 --- a/utils/src/test_utils.rs +++ b/utils/src/test_utils.rs @@ -2,3 +2,38 @@ pub use tokio::test as async_test; #[cfg(target_arch = "wasm32")] pub use wasm_bindgen_test::wasm_bindgen_test as async_test; + +/// Source the `.env` file and get environment variable. +#[cfg(not(target_arch = "wasm32"))] +pub fn env_var(var_name: &str) -> Option { + use std::sync::Once; + + static DOTENV: Once = Once::new(); + DOTENV.call_once(|| { + let _ = dotenvy::dotenv(); + }); + + std::env::var(var_name).ok() +} + +/// Source the `.env` file and get environment variable. +#[cfg(target_arch = "wasm32")] +pub fn env_var(var_name: &str) -> Option { + #[derive(rust_embed::Embed)] + #[folder = "$CARGO_MANIFEST_DIR/../"] + #[allow_missing = true] + #[include = ".env*"] + struct Env; + + let env = Env::get(".env").expect("Couldn't find .env file"); + + str::from_utf8(env.data.as_ref()) + .unwrap() + .lines() + .find_map(|line| { + line.split_once('=') + .take_if(|_| !line.starts_with('#')) + .take_if(|(name, _)| *name == var_name) + .map(|(_, var)| var.to_owned()) + }) +}