Merge pull request #3062 from autonomys/piece-download-refactor

Refactor piece and segment downloading into subspace-data-retrieval

crates/subspace-archiving/src/reconstructor.rs

@@ -72,15 +72,13 @@ impl Reconstructor {
 
     /// Given a set of pieces of a segment of the archived history (any half of all pieces are
     /// required to be present, the rest will be recovered automatically due to use of erasure
-    /// coding if needed), reconstructs and returns segment header and a list of encoded blocks with
-    /// corresponding block numbers.
+    /// coding if needed), reconstructs and returns the segment itself.
     ///
-    /// It is possible to start with any segment, but when next segment is pushed, it needs to
-    /// follow the previous one or else error will be returned.
-    pub fn add_segment(
-        &mut self,
+    /// Does not modify the internal state of the reconstructor.
+    pub fn reconstruct_segment(
+        &self,
         segment_pieces: &[Option<Piece>],
-    ) -> Result<ReconstructedContents, ReconstructorError> {
+    ) -> Result<Segment, ReconstructorError> {
         let mut segment_data = RecordedHistorySegment::new_boxed();
 
         if !segment_pieces
@@ -151,9 +149,24 @@ impl Reconstructor {
             }
         }
 
-        let Segment::V0 { items } =
-            Segment::decode(&mut AsRef::<[u8]>::as_ref(segment_data.as_ref()))
-                .map_err(ReconstructorError::SegmentDecoding)?;
+        let segment = Segment::decode(&mut AsRef::<[u8]>::as_ref(segment_data.as_ref()))
+            .map_err(ReconstructorError::SegmentDecoding)?;
+
+        Ok(segment)
+    }
+
+    /// Given a set of pieces of a segment of the archived history (any half of all pieces are
+    /// required to be present, the rest will be recovered automatically due to use of erasure
+    /// coding if needed), reconstructs and returns segment header and a list of encoded blocks with
+    /// corresponding block numbers.
+    ///
+    /// It is possible to start with any segment, but when next segment is pushed, it needs to
+    /// follow the previous one or else error will be returned.
+    pub fn add_segment(
+        &mut self,
+        segment_pieces: &[Option<Piece>],
+    ) -> Result<ReconstructedContents, ReconstructorError> {
+        let Segment::V0 { items } = self.reconstruct_segment(segment_pieces)?;
 
         let mut reconstructed_contents = ReconstructedContents::default();
         let mut next_block_number = 0;

crates/subspace-service/src/sync_from_dsn/import_blocks.rs

@@ -234,6 +234,7 @@ where
     Ok(imported_blocks)
 }
 
+/// Downloads and reconstructs blocks from a DSN segment, by concurrently downloading its pieces.
 pub(super) async fn download_and_reconstruct_blocks<PG>(
     segment_index: SegmentIndex,
     piece_getter: &PG,

shared/subspace-data-retrieval/Cargo.toml

@@ -12,12 +12,15 @@ include = [
 ]
 
 [dependencies]
+async-lock = "3.3.0"
 async-trait = "0.1.83"
+futures = "0.3.29"
 parity-scale-codec = { version = "3.6.12", features = ["derive"] }
 subspace-archiving = { version = "0.1.0", path = "../../crates/subspace-archiving" }
 subspace-core-primitives = { version = "0.1.0", path = "../../crates/subspace-core-primitives" }
 subspace-erasure-coding = { version = "0.1.0", path = "../../crates/subspace-erasure-coding" }
 thiserror = "1.0.63"
+tokio = { version = "1.39.2", features = ["sync"] }
 tracing = "0.1.40"
 
 [dev-dependencies]

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

@@ -17,4 +17,5 @@
 
 pub mod object_fetcher;
 pub mod piece_fetcher;
+pub mod piece_getter;
 pub mod segment_fetcher;