Merge pull request #3472 from autonomys/object-piece-reuse

Re-use the last piece for the next object in a batch

Author: teor2345

Cargo.lock

@@ -4,15 +4,12 @@
 //! * for objects within a block
 //! * for global objects in the global history of the blockchain (inside a piece)
 
-#[cfg(not(feature = "std"))]
-extern crate alloc;
-
 use crate::hashes::Blake3Hash;
 use crate::pieces::PieceIndex;
-#[cfg(not(feature = "std"))]
-use alloc::vec::Vec;
 use core::default::Default;
 use parity_scale_codec::{Decode, Encode};
+use scale_info::prelude::vec;
+use scale_info::prelude::vec::Vec;
 use scale_info::TypeInfo;
 #[cfg(feature = "serde")]
 use serde::{Deserialize, Serialize};
@@ -143,6 +140,14 @@ impl GlobalObjectMapping {
         }
     }
 
+    /// Returns a newly created GlobalObjectMapping from a single object mapping
+    #[inline]
+    pub fn from_object(object: GlobalObject) -> Self {
+        Self::V0 {
+            objects: vec![object],
+        }
+    }
+
     /// Returns the object mappings
     pub fn objects(&self) -> &[GlobalObject] {
         match self {

crates/subspace-core-primitives/src/objects.rs

@@ -4,6 +4,8 @@ RPC API for Subspace Gateway.
 
 ### Using the gateway RPCs
 
+#### Getting Object Mappings
+
 The gateway RPCs can fetch data using object mappings supplied by a node.
 
 Launch a node with `--create-object-mappings blockNumber --sync full`, and wait for mappings from
@@ -41,7 +43,9 @@ subspace_subscribeObjectMappings
 }
 ```
 
-Then use those mappings to get object data from the gateway RPCs:
+#### Using Object Mappings to get Objects
+
+Use those mappings to get object data from the gateway RPCs:
 ```sh
 $ websocat --jsonrpc ws://127.0.0.1:9955
 subspace_fetchObject {"mappings": {"v0": {"objects": [["0000000000000000000000000000000000000000000000000000000000000000", 0, 0]]}}}
@@ -54,6 +58,16 @@ subspace_fetchObject {"mappings": {"v0": {"objects": [["000000000000000000000000
 }
 ```
 
+For efficiency, objects in a batch should be sorted by increasing piece index. And objects with
+the same piece index should be sorted by increasing offset. This allows the last piece to be
+re-used for the next object in the batch.
+
+Batches should be split if the gap between object piece indexes is 6 or more. Those objects
+can't share any pieces, because a maximum-sized object only uses 6 pieces. (Batches should also
+be split so that the response stays within the RPC response size limit.)
+
+### Advanced Usage
+
 #### Missed Mappings
 
 The node doesn't make sure the client has processed the previous mapping before generating the next

crates/subspace-gateway-rpc/README.md

@@ -83,8 +83,15 @@ impl DerefMut for HexData {
 /// Provides rpc methods for interacting with a Subspace DSN Gateway.
 #[rpc(client, server)]
 pub trait SubspaceGatewayRpcApi {
-    /// Get object data from DSN object mappings.
+    /// Get object data from a DSN object mapping batch.
     /// Returns an error if any object fetch was unsuccessful.
+    ///
+    /// For efficiency, objects in a batch should be sorted by increasing piece index. Objects with
+    /// the same piece index should be sorted by increasing offset. This allows the last piece to
+    /// be re-used for the next object in the batch.
+    ///
+    /// Batches should be split if the gap between object piece indexes is 6 or more. Those objects
+    /// can't share any pieces, because a maximum-sized object only uses 6 pieces.
     #[method(name = "subspace_fetchObject")]
     async fn fetch_object(&self, mappings: GlobalObjectMapping) -> Result<Vec<HexData>, Error>;
 }

crates/subspace-gateway-rpc/src/lib.rs

@@ -109,7 +109,11 @@ where
         }
     };
 
-    // TODO: return a multi-part response, with one part per object
+    // TODO:
+    // - return a multi-part response, with one part per object.
+    // - add the object hash to each part, so we can sort mappings by piece index and offset,
+    //   for more efficient piece re-use. See the `ObjectFetcher::fetch_objects` performance docs
+    //   for more details.
     HttpResponse::Ok()
         .content_type("application/octet-stream")
         .body(objects.concat())

crates/subspace-gateway/src/commands/http/server.rs

@@ -102,6 +102,9 @@ sp-session.workspace = true
 frame-system-rpc-runtime-api.workspace = true
 pallet-transaction-payment-rpc-runtime-api.workspace = true
 
+[dev-dependencies]
+static_assertions.workspace = true
+
 [features]
 runtime-benchmarks = [
     "dep:frame-benchmarking",

crates/subspace-service/Cargo.toml

@@ -1392,3 +1392,20 @@ fn extract_confirmation_depth(chain_spec: &dyn ChainSpec) -> Option<u32> {
     .ok()?;
     u32::decode(&mut encoded_confirmation_depth.as_slice()).ok()
 }
+
+#[cfg(test)]
+mod test {
+    use static_assertions::const_assert_eq;
+    use subspace_data_retrieval::object_fetcher::MAX_BLOCK_LENGTH as ARCHIVER_MAX_BLOCK_LENGTH;
+    use subspace_runtime_primitives::MAX_BLOCK_LENGTH as CONSENSUS_RUNTIME_MAX_BLOCK_LENGTH;
+
+    /// Runtime and archiver code must agree on the consensus block length.
+    /// (This avoids importing all the runtime primitives code into the farmer and gateway.)
+    #[test]
+    fn max_block_length_consistent() {
+        const_assert_eq!(
+            CONSENSUS_RUNTIME_MAX_BLOCK_LENGTH,
+            ARCHIVER_MAX_BLOCK_LENGTH,
+        );
+    }
+}

crates/subspace-service/src/lib.rs

@@ -20,13 +20,13 @@ parity-scale-codec = { workspace = true, features = ["derive"] }
 subspace-archiving.workspace = true
 subspace-core-primitives = { workspace = true, features = ["std"] }
 subspace-erasure-coding.workspace = true
-subspace-runtime-primitives = { workspace = true, features = ["std"] }
+# This crate can't depend on any runtime code, because it needs to be independent of Substrate.
 thiserror.workspace = true
 tokio = { workspace = true, features = ["sync", "rt"] }
 tracing = { workspace = true, features = ["std"] }
 
 [dev-dependencies]
-rand.workspace = true
+rand = { workspace = true, features = ["std", "std_rng"] }
 subspace-logging.workspace = true
 tokio = { workspace = true, features = ["rt-multi-thread", "macros"] }
 

shared/subspace-data-retrieval/Cargo.toml

@@ -20,6 +20,8 @@ mod segment_header;
 #[cfg(test)]
 mod tests;
 
+pub use segment_header::MAX_BLOCK_LENGTH;
+
 /// The maximum object length the implementation in this module can reliably handle.
 ///
 /// Currently objects are limited by the largest block size in the consensus chain, which is 5 MB.
@@ -52,6 +54,9 @@ pub fn max_supported_object_length() -> usize {
 /// The length of the compact encoding of `max_supported_object_length()`.
 const MAX_ENCODED_LENGTH_SIZE: usize = 4;
 
+/// Used to store the last piece downloaded in an object fetcher batch.
+pub type LastPieceCache = (PieceIndex, Piece);
+
 /// Object fetching errors.
 #[derive(Debug, PartialEq, Eq, thiserror::Error)]
 pub enum Error {
@@ -238,11 +243,19 @@ where
     /// putting the objects' bytes together.
     ///
     /// Checks the objects' hashes to make sure the correct bytes are returned.
+    ///
+    /// For efficiency, objects in a batch should be sorted by increasing piece index. Objects with
+    /// the same piece index should be sorted by increasing offset. This allows the last piece to
+    /// be re-used for the next object in the batch.
+    ///
+    /// Batches should be split if the gap between object piece indexes is 6 or more. Those objects
+    /// can't share any pieces, because a maximum-sized object only uses 6 pieces.
     pub async fn fetch_objects(
         &self,
         mappings: GlobalObjectMapping,
     ) -> Result<Vec<Vec<u8>>, Error> {
         let mut objects = Vec::with_capacity(mappings.objects().len());
+        let mut piece_cache = None;
 
         // TODO:
         // - keep the last downloaded piece until it's no longer needed
@@ -292,7 +305,7 @@ where
 
             // All objects can be assembled from individual pieces, we handle segments by checking
             // all possible padding, and parsing and discarding segment headers.
-            let data = self.fetch_object(mapping).await?;
+            let data = self.fetch_object(mapping, &mut piece_cache).await?;
 
             objects.push(data);
         }
@@ -312,7 +325,11 @@ where
     /// or fetch more data.
     //
     // TODO: return last downloaded piece from fetch_object() and pass them to the next fetch_object()
-    async fn fetch_object(&self, mapping: GlobalObject) -> Result<Vec<u8>, Error> {
+    async fn fetch_object(
+        &self,
+        mapping: GlobalObject,
+        piece_cache: &mut Option<LastPieceCache>,
+    ) -> Result<Vec<u8>, Error> {
         let GlobalObject {
             piece_index,
             offset,
@@ -327,7 +344,9 @@ where
 
         // Get pieces until we have enough data to calculate the object's length(s).
         // Objects with their length bytes at the end of a piece are a rare edge case.
-        let piece = self.read_piece(next_source_piece_index, mapping).await?;
+        let piece = self
+            .read_piece(next_source_piece_index, mapping, piece_cache)
+            .await?;
 
         // Discard piece data before the offset.
         // If this is the first piece in a segment, this automatically skips the segment header.
@@ -367,7 +386,9 @@ where
             );
 
             // Get the second piece for the object
-            let piece = self.read_piece(next_source_piece_index, mapping).await?;
+            let piece = self
+                .read_piece(next_source_piece_index, mapping, piece_cache)
+                .await?;
             // We want all the piece data
             let piece_data = piece
                 .record()
@@ -427,7 +448,7 @@ where
             // TODO: turn this into a concurrent stream, which cancels piece downloads if they aren't
             // needed
             let pieces = self
-                .read_pieces(remaining_piece_indexes.clone(), mapping)
+                .read_pieces(remaining_piece_indexes.clone(), mapping, piece_cache)
                 .await?
                 .into_iter()
                 .zip(remaining_piece_indexes.iter().copied())
@@ -476,22 +497,31 @@ where
         &self,
         piece_indexes: Arc<[PieceIndex]>,
         mapping: GlobalObject,
+        piece_cache: &mut Option<LastPieceCache>,
     ) -> Result<Vec<Piece>, Error> {
-        download_pieces(piece_indexes.clone(), &self.piece_getter)
-            .await
-            .map_err(|source| {
-                debug!(
-                    ?piece_indexes,
-                    error = ?source,
-                    ?mapping,
-                    "Error fetching pieces during object assembling"
-                );
+        download_pieces(
+            piece_indexes.clone(),
+            &piece_cache.clone().with_fallback(self.piece_getter.clone()),
+        )
+        .await
+        .inspect(|pieces| {
+            if let (Some(piece_index), Some(piece)) = (piece_indexes.last(), pieces.last()) {
+                *piece_cache = Some((*piece_index, piece.clone()))
+            }
+        })
+        .map_err(|source| {
+            debug!(
+                ?piece_indexes,
+                error = ?source,
+                ?mapping,
+                "Error fetching pieces during object assembling"
+            );
 
-                Error::PieceGetterError {
-                    error: format!("{source:?}"),
-                    mapping,
-                }
-            })
+            Error::PieceGetterError {
+                error: format!("{source:?}"),
+                mapping,
+            }
+        })
     }
 
     /// Read and return a single piece.
@@ -501,28 +531,34 @@ where
         &self,
         piece_index: PieceIndex,
         mapping: GlobalObject,
+        piece_cache: &mut Option<LastPieceCache>,
     ) -> Result<Piece, Error> {
-        download_pieces(vec![piece_index].into(), &self.piece_getter)
-            .await
-            .map(|pieces| {
-                pieces
-                    .first()
-                    .expect("download_pieces always returns exact pieces or error")
-                    .clone()
-            })
-            .map_err(|source| {
-                debug!(
-                    %piece_index,
-                    error = ?source,
-                    ?mapping,
-                    "Error fetching piece during object assembling"
-                );
+        let piece_indexes = Arc::<[PieceIndex]>::from(vec![piece_index]);
+        download_pieces(
+            piece_indexes.clone(),
+            &piece_cache.clone().with_fallback(self.piece_getter.clone()),
+        )
+        .await
+        .inspect(|pieces| *piece_cache = Some((piece_index, pieces[0].clone())))
+        .map(|pieces| {
+            pieces
+                .first()
+                .expect("download_pieces always returns exact pieces or error")
+                .clone()
+        })
+        .map_err(|source| {
+            debug!(
+                %piece_index,
+                error = ?source,
+                ?mapping,
+                "Error fetching piece during object assembling"
+            );
 
-                Error::PieceGetterError {
-                    error: format!("{source:?}"),
-                    mapping,
-                }
-            })
+            Error::PieceGetterError {
+                error: format!("{source:?}"),
+                mapping,
+            }
+        })
     }
 }
 

shared/subspace-data-retrieval/src/object_fetcher.rs

@@ -12,7 +12,6 @@ use subspace_core_primitives::objects::GlobalObject;
 use subspace_core_primitives::segments::{
     ArchivedBlockProgress, LastArchivedBlock, SegmentCommitment, SegmentHeader, SegmentIndex,
 };
-use subspace_runtime_primitives::MAX_BLOCK_LENGTH;
 
 /// The maximum amount of segment padding.
 ///
@@ -22,6 +21,10 @@ use subspace_runtime_primitives::MAX_BLOCK_LENGTH;
 /// <https://docs.substrate.io/reference/scale-codec/#fn-1>
 pub const MAX_SEGMENT_PADDING: usize = 3;
 
+/// Maximum block length for non-`Normal` extrinsic is 5 MiB.
+/// This is a copy of the constant in `subspace_runtime_primitives`.
+pub const MAX_BLOCK_LENGTH: u32 = 5 * 1024 * 1024;
+
 /// The segment version this code knows how to parse.
 const SEGMENT_VERSION_VARIANT: u8 = 0;
 
@@ -156,7 +159,6 @@ mod test {
     use parity_scale_codec::{Compact, CompactLen};
     use subspace_archiving::archiver::Segment;
     use subspace_core_primitives::objects::BlockObjectMapping;
-    use subspace_runtime_primitives::MAX_BLOCK_LENGTH;
 
     #[test]
     fn max_segment_padding_constant() {