Merge branch 'main' into gateway-dsn

crates/pallet-subspace/Cargo.toml

@@ -31,7 +31,7 @@ subspace-verification = { version = "0.1.0", path = "../subspace-verification",
 
 [dev-dependencies]
 env_logger = "0.11.5"
-futures = "0.3.30"
+futures = "0.3.31"
 pallet-balances = { git = "https://github.com/subspace/polkadot-sdk", rev = "5871818e1d736f1843eb9078f886290695165c42" }
 rand = { version = "0.8.5", features = ["min_const_gen"] }
 sp-io = { git = "https://github.com/subspace/polkadot-sdk", rev = "5871818e1d736f1843eb9078f886290695165c42" }

crates/sc-consensus-subspace-rpc/Cargo.toml

@@ -14,7 +14,7 @@ targets = ["x86_64-unknown-linux-gnu"]
 
 [dependencies]
 async-oneshot = "0.5.9"
-futures = "0.3.30"
+futures = "0.3.31"
 futures-timer = "3.0.3"
 jsonrpsee = { version = "0.24.5", features = ["server", "macros"] }
 parking_lot = "0.12.2"

crates/sc-consensus-subspace-rpc/src/lib.rs

@@ -172,7 +172,7 @@ pub trait SubspaceRpcApi {
     #[method(name = "subspace_lastSegmentHeaders")]
     async fn last_segment_headers(&self, limit: u32) -> Result<Vec<Option<SegmentHeader>>, Error>;
 
-    /// Block/transaction object mappings subscription
+    /// DSN object mappings subscription
     #[subscription(
         name = "subspace_subscribeObjectMappings" => "subspace_object_mappings",
         unsubscribe = "subspace_unsubscribeObjectMappings",
@@ -181,7 +181,7 @@ pub trait SubspaceRpcApi {
     )]
     fn subscribe_object_mappings(&self);
 
-    /// Filtered block/transaction object mappings subscription
+    /// Filtered DSN object mappings subscription
     #[subscription(
         name = "subspace_subscribeFilteredObjectMappings" => "subspace_filtered_object_mappings",
         unsubscribe = "subspace_unsubscribeFilteredObjectMappings",

crates/sc-consensus-subspace/Cargo.toml

@@ -15,7 +15,7 @@ targets = ["x86_64-unknown-linux-gnu"]
 [dependencies]
 async-trait = "0.1.83"
 codec = { package = "parity-scale-codec", version = "3.6.12", features = ["derive"] }
-futures = "0.3.30"
+futures = "0.3.31"
 parking_lot = "0.12.2"
 rand = "0.8.5"
 rand_chacha = "0.3.1"

crates/sc-proof-of-time/Cargo.toml

@@ -13,7 +13,7 @@ include = [
 [dependencies]
 core_affinity = "0.8.1"
 derive_more = { version = "1.0.0", features = ["full"] }
-futures = "0.3.30"
+futures = "0.3.31"
 parity-scale-codec = { version = "3.6.12", features = ["derive"] }
 parking_lot = "0.12.2"
 rayon = "1.10.0"

crates/sc-subspace-block-relay/Cargo.toml

@@ -15,7 +15,7 @@ async-channel = "1.9.0"
 async-trait = "0.1.83"
 codec = { package = "parity-scale-codec", version = "3.6.12", default-features = false, features = ["derive"] }
 derive_more = { version = "1.0.0", features = ["full"] }
-futures = "0.3.30"
+futures = "0.3.31"
 parking_lot = "0.12.2"
 sc-client-api = { git = "https://github.com/subspace/polkadot-sdk", rev = "5871818e1d736f1843eb9078f886290695165c42" }
 sc-network = { git = "https://github.com/subspace/polkadot-sdk", rev = "5871818e1d736f1843eb9078f886290695165c42" }

crates/sp-domains-fraud-proof/Cargo.toml

@@ -49,7 +49,7 @@ fp-rpc = { version = "3.0.0-dev", git = "https://github.com/autonomys/frontier",
 fp-self-contained = { version = "1.0.0-dev", git = "https://github.com/autonomys/frontier", rev = "f80f9e2bad338f3bf3854b256b3c4edea23e5968", features = ['default'] }
 frame-support = { default-features = false, git = "https://github.com/subspace/polkadot-sdk", rev = "5871818e1d736f1843eb9078f886290695165c42" }
 frame-system = { default-features = false, git = "https://github.com/subspace/polkadot-sdk", rev = "5871818e1d736f1843eb9078f886290695165c42" }
-futures = "0.3.30"
+futures = "0.3.31"
 libsecp256k1 = { version = "0.7.1", features = ["static-context", "hmac"] }
 pallet-balances = { git = "https://github.com/subspace/polkadot-sdk", rev = "5871818e1d736f1843eb9078f886290695165c42" }
 pallet-ethereum = { git = "https://github.com/autonomys/frontier", rev = "f80f9e2bad338f3bf3854b256b3c4edea23e5968", features = ['default'] }

crates/subspace-archiving/src/piece_reconstructor.rs

@@ -212,6 +212,7 @@ impl PiecesReconstructor {
             return Err(ReconstructorError::IncorrectPiecePosition);
         }
 
+        // TODO: Early exit if position already exists and doesn't need reconstruction
         let (reconstructed_records, polynomial) = self.reconstruct_shards(segment_pieces)?;
 
         let mut piece = Piece::from(&reconstructed_records[piece_position]);

crates/subspace-farmer-components/Cargo.toml

@@ -16,13 +16,14 @@ include = [
 bench = false
 
 [dependencies]
+anyhow = "1.0.89"
 async-lock = "3.4.0"
 async-trait = "0.1.83"
 backoff = { version = "0.4.0", features = ["futures", "tokio"] }
 bitvec = "1.0.1"
 # TODO: Switch to fs4 once https://github.com/al8n/fs4-rs/issues/15 is resolved
 fs2 = "0.4.3"
-futures = "0.3.30"
+futures = "0.3.31"
 hex = "0.4.3"
 libc = "0.2.159"
 parity-scale-codec = "3.6.12"
@@ -47,7 +48,7 @@ winapi = "0.3.9"
 
 [dev-dependencies]
 criterion = "0.5.1"
-futures = "0.3.30"
+futures = "0.3.31"
 subspace-archiving = { version = "0.1.0", path = "../subspace-archiving" }
 subspace-proof-of-space = { version = "0.1.0", path = "../subspace-proof-of-space" }
 

crates/subspace-farmer-components/src/lib.rs

@@ -6,6 +6,7 @@
     array_chunks,
     const_option,
     const_trait_impl,
+    exact_size_is_empty,
     int_roundings,
     iter_collect_into,
     never_type,
@@ -25,10 +26,11 @@ mod segment_reconstruction;
 
 use crate::file_ext::FileExt;
 use async_trait::async_trait;
+use futures::stream::FuturesUnordered;
+use futures::Stream;
 use parity_scale_codec::{Decode, Encode};
 use serde::{Deserialize, Serialize};
 use static_assertions::const_assert;
-use std::error::Error;
 use std::fs::File;
 use std::future::Future;
 use std::io;
@@ -40,35 +42,80 @@ use subspace_core_primitives::segments::{ArchivedHistorySegment, HistorySize};
 #[async_trait]
 pub trait PieceGetter {
     /// Get piece by index
-    async fn get_piece(
-        &self,
-        piece_index: PieceIndex,
-    ) -> Result<Option<Piece>, Box<dyn Error + Send + Sync + 'static>>;
+    async fn get_piece(&self, piece_index: PieceIndex) -> anyhow::Result<Option<Piece>>;
+
+    /// Get pieces with provided indices
+    async fn get_pieces<'a, PieceIndices>(
+        &'a self,
+        piece_indices: PieceIndices,
+    ) -> anyhow::Result<
+        Box<dyn Stream<Item = (PieceIndex, anyhow::Result<Option<Piece>>)> + Send + Unpin + 'a>,
+    >
+    where
+        PieceIndices: IntoIterator<Item = PieceIndex, IntoIter: Send> + Send + 'a,
+    {
+        // TODO: Remove default impl here
+        Ok(Box::new(
+            piece_indices
+                .into_iter()
+                .map(|piece_index| async move {
+                    let result = self.get_piece(piece_index).await;
+                    (piece_index, result)
+                })
+                .collect::<FuturesUnordered<_>>(),
+        ) as Box<_>)
+    }
 }
 
 #[async_trait]
 impl<T> PieceGetter for Arc<T>
 where
     T: PieceGetter + Send + Sync,
 {
-    async fn get_piece(
-        &self,
-        piece_index: PieceIndex,
-    ) -> Result<Option<Piece>, Box<dyn Error + Send + Sync + 'static>> {
+    async fn get_piece(&self, piece_index: PieceIndex) -> anyhow::Result<Option<Piece>> {
         self.as_ref().get_piece(piece_index).await
     }
+
+    async fn get_pieces<'a, PieceIndices>(
+        &'a self,
+        piece_indices: PieceIndices,
+    ) -> anyhow::Result<
+        Box<dyn Stream<Item = (PieceIndex, anyhow::Result<Option<Piece>>)> + Send + Unpin + 'a>,
+    >
+    where
+        PieceIndices: IntoIterator<Item = PieceIndex, IntoIter: Send> + Send + 'a,
+    {
+        self.as_ref().get_pieces(piece_indices).await
+    }
 }
 
 #[async_trait]
 impl PieceGetter for ArchivedHistorySegment {
-    async fn get_piece(
-        &self,
-        piece_index: PieceIndex,
-    ) -> Result<Option<Piece>, Box<dyn Error + Send + Sync + 'static>> {
+    async fn get_piece(&self, piece_index: PieceIndex) -> anyhow::Result<Option<Piece>> {
         let position = usize::try_from(u64::from(piece_index))?;
 
         Ok(self.pieces().nth(position))
     }
+
+    async fn get_pieces<'a, PieceIndices>(
+        &'a self,
+        piece_indices: PieceIndices,
+    ) -> anyhow::Result<
+        Box<dyn Stream<Item = (PieceIndex, anyhow::Result<Option<Piece>>)> + Send + Unpin + 'a>,
+    >
+    where
+        PieceIndices: IntoIterator<Item = PieceIndex, IntoIter: Send> + Send + 'a,
+    {
+        Ok(Box::new(
+            piece_indices
+                .into_iter()
+                .map(|piece_index| async move {
+                    let result = self.get_piece(piece_index).await;
+                    (piece_index, result)
+                })
+                .collect::<FuturesUnordered<_>>(),
+        ) as Box<_>)
+    }
 }
 
 /// Enum to encapsulate the selection between [`ReadAtSync`] and [`ReadAtAsync]` variants