feat: sector indexing and data retrieval

mater/lib/src/cid.rs

@@ -3,8 +3,10 @@ use tokio::io::{AsyncRead, AsyncReadExt};
 
 use crate::{async_varint::read_varint, IDENTITY_CODE};
 
+/// Extension trait for [`Cid`](ipld_core::cid::Cid)
 pub trait CidExt {
-    async fn read_bytes_async<R>(r: R) -> Result<(Self, usize), Error>
+    /// Reads the bytes from a byte stream.
+    fn read_bytes_async<R>(r: R) -> impl std::future::Future<Output = Result<(Self, usize), Error>>
     where
         Self: Sized,
         R: AsyncRead + Unpin;
@@ -13,8 +15,10 @@ pub trait CidExt {
     fn get_identity_data(&self) -> Option<&[u8]>;
 }
 
+/// Extension trait for [`Multihash`](ipld_core::cid::multihash::Multihash)
 pub trait MultihashExt {
-    async fn read_async<R>(r: R) -> Result<(Self, usize), Error>
+    /// Reads the bytes from a byte stream.
+    fn read_async<R>(r: R) -> impl std::future::Future<Output = Result<(Self, usize), Error>>
     where
         Self: Sized,
         R: AsyncRead + Unpin;
@@ -62,7 +66,6 @@ impl<const S: usize> MultihashExt for Multihash<S> {
     /// https://github.com/multiformats/rust-multihash/blob/90a6c19ec71ced09469eec164a3586aafeddfbbd/src/multihash.rs#L271
     async fn read_async<R>(mut r: R) -> Result<(Self, usize), Error>
     where
-        Self: Sized,
         R: AsyncRead + Unpin,
     {
         let (code, code_bytes_read): (u64, usize) = read_varint(&mut r).await?;

mater/lib/src/lib.rs

@@ -21,11 +21,12 @@ mod v1;
 mod v2;
 
 // We need to re-expose this because `read_block` returns `(Cid, Vec<u8>)`.
+pub use cid::{CidExt, MultihashExt};
 pub use file_reader::CarExtractor;
 pub use ipld_core::cid::Cid;
 pub use multicodec::{DAG_PB_CODE, IDENTITY_CODE, RAW_CODE};
 pub use stores::{create_filestore, Blockstore, Config, FileBlockstore};
-pub use v1::{Header as CarV1Header, Reader as CarV1Reader, Writer as CarV1Writer};
+pub use v1::{BlockMetadata, Header as CarV1Header, Reader as CarV1Reader, Writer as CarV1Writer};
 pub use v2::{
     verify_cid, Characteristics, Header as CarV2Header, Index, IndexEntry, IndexSorted,
     MultihashIndexSorted, Reader as CarV2Reader, SingleWidthIndex, Writer as CarV2Writer,

storage-provider/common/src/sector.rs

@@ -70,6 +70,11 @@ pub struct UnsealedSector {
     /// Indexes match with corresponding deals in [`Sector::deals`].
     pub piece_infos: Vec<PieceInfo>,
 
+    /// Tracks locations of the actual pieces added to the unsealed sector. This
+    /// vector does not contain padding pieces. It contains the actual pieces
+    /// corresponding with the deals from the users.
+    pub pieces_locations: Vec<(Commitment<CommP>, PathBuf)>,
+
     /// Tracks all of the deals that have been added to the sector.
     pub deals: Vec<(DealId, DealProposal)>,
 
@@ -97,6 +102,7 @@ impl UnsealedSector {
             occupied_sector_space: 0,
             piece_infos: vec![],
             deals: vec![],
+            pieces_locations: vec![],
             unsealed_path,
         })
     }
@@ -111,24 +117,29 @@ impl UnsealedSector {
     ) -> Result<(), SectorError> {
         self.deals.push((deal_id, deal));
 
-        // would love to use something like scoped spawn blocking
-        let pieces = self.piece_infos.clone();
-        let unsealed_path = self.unsealed_path.clone();
         let handle: JoinHandle<Result<(PieceInfo, u64), SectorError>> =
-            tokio::task::spawn_blocking(move || {
-                let unsealed_sector = std::fs::File::options().append(true).open(unsealed_path)?;
-
-                tracing::info!("Preparing piece...");
-                let (padded_reader, piece_info) = prepare_piece(piece_path, commitment)?;
-                tracing::info!("Adding piece...");
-                let occupied_piece_space =
-                    add_piece(padded_reader, piece_info, &pieces, unsealed_sector)?;
-
-                Ok((piece_info, occupied_piece_space))
+            tokio::task::spawn_blocking({
+                let pieces = self.piece_infos.clone();
+                let unsealed_path = self.unsealed_path.clone();
+                let piece_path = piece_path.clone();
+
+                move || {
+                    let unsealed_sector =
+                        std::fs::File::options().append(true).open(unsealed_path)?;
+
+                    tracing::info!("Preparing piece...");
+                    let (padded_reader, piece_info) = prepare_piece(piece_path, commitment)?;
+                    tracing::info!("Adding piece...");
+                    let occupied_piece_space =
+                        add_piece(padded_reader, piece_info, &pieces, unsealed_sector)?;
+
+                    Ok((piece_info, occupied_piece_space))
+                }
             });
 
         let (piece_info, occupied_piece_space) = handle.await??;
         self.piece_infos.push(piece_info);
+        self.pieces_locations.push((commitment, piece_path));
         self.occupied_sector_space += occupied_piece_space;
 
         Ok(())
@@ -289,6 +300,11 @@ pub struct PreCommittedSector {
     /// Indexes match with corresponding deals in [`Sector::deals`].
     pub piece_infos: Vec<PieceInfo>,
 
+    /// Tracks locations of the actual pieces added to the unsealed sector. This
+    /// vector does not contain padding pieces. It contains the actual pieces
+    /// corresponding with the deals from the users.
+    pub pieces_locations: Vec<(Commitment<CommP>, PathBuf)>,
+
     /// Tracks all of the deals that have been added to the sector.
     pub deals: Vec<(DealId, DealProposal)>,
 
@@ -349,6 +365,7 @@ impl PreCommittedSector {
             seal_proof: unsealed.seal_proof,
             sector_number: unsealed.sector_number,
             piece_infos: unsealed.piece_infos,
+            pieces_locations: unsealed.pieces_locations,
             deals: unsealed.deals,
             cache_path,
             sealed_path,
@@ -503,6 +520,11 @@ pub struct ProvenSector {
     /// Indexes match with corresponding deals in [`Sector::deals`].
     pub piece_infos: Vec<PieceInfo>,
 
+    /// Tracks locations of the actual pieces added to the unsealed sector. This
+    /// vector does not contain padding pieces. It contains the actual pieces
+    /// corresponding with the deals from the users.
+    pub pieces_locations: Vec<(Commitment<CommP>, PathBuf)>,
+
     /// Tracks all of the deals that have been added to the sector.
     pub deals: Vec<(DealId, DealProposal)>,
 
@@ -526,6 +548,7 @@ impl ProvenSector {
         Self {
             sector_number: sector.sector_number,
             piece_infos: sector.piece_infos,
+            pieces_locations: sector.pieces_locations,
             deals: sector.deals,
             cache_path: sector.cache_path,
             sealed_path: sector.sealed_path,

storage-provider/server/Cargo.toml

@@ -18,12 +18,14 @@ opencl = ["polka-storage-proofs/opencl", "polka-storage-proofs/std", "polka-stor
 mater = { workspace = true }
 polka-storage-proofs = { workspace = true, default-features = false }
 polka-storage-provider-common = { workspace = true, features = ["clap"] }
+polka-storage-retrieval = { workspace = true }
 primitives = { workspace = true, features = ["clap", "serde", "std"] }
 storagext = { workspace = true, features = ["clap"] }
 
 async-trait = { workspace = true }
 axum = { workspace = true, features = ["macros", "multipart"] }
 base64 = { workspace = true }
+blockstore = { workspace = true }
 chrono = { workspace = true, features = ["serde"] }
 ciborium = { workspace = true }
 cid = { workspace = true, features = ["serde", "std"] }

storage-provider/server/src/config.rs

@@ -44,6 +44,12 @@ fn default_node_address() -> Url {
     Url::parse(DEFAULT_NODE_ADDRESS).expect("DEFAULT_NODE_ADDRESS must be a valid Url")
 }
 
+fn default_retrieval_address() -> Multiaddr {
+    "/ip4/127.0.0.1/tcp/8002"
+        .parse()
+        .expect("multiaddres is correct")
+}
+
 #[derive(Debug, Clone, Deserialize, Args)]
 #[group(multiple = true, conflicts_with = "config")]
 #[serde(deny_unknown_fields)]
@@ -63,6 +69,11 @@ pub struct ConfigurationArgs {
     #[arg(long, default_value_t = default_node_address())]
     pub(crate) node_url: Url,
 
+    /// Storage provider retrieval service listen address.
+    #[serde(default = "default_retrieval_address")]
+    #[arg(long, default_value_t = default_retrieval_address())]
+    pub(crate) retrieval_listen_address: Multiaddr,
+
     /// RocksDB storage directory.
     /// Defaults to a temporary random directory, like `/tmp/<random>/deals_database`.
     #[arg(long)]

storage-provider/server/src/local_index_directory/mod.rs → storage-provider/server/src/indexer/local_index_directory/mod.rs

@@ -8,11 +8,11 @@ use cid::{
     multihash::{self, Multihash},
     Cid,
 };
+use primitives::{sector::SectorNumber, DealId};
 use serde::{Deserialize, Serialize};
 use uuid::Uuid;
 
 pub mod rdb;
-pub mod rdb_ext;
 
 /// Convert a [`Multihash`] into a key (converts [`Multihash::digest`] to base-64).
 ///
@@ -182,16 +182,6 @@ impl Default for PieceInfo {
     }
 }
 
-/// Identifier for a retrieval deal (unique to a client)
-#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Serialize, Deserialize)]
-pub struct DealId(u64);
-
-impl From<u64> for DealId {
-    fn from(value: u64) -> Self {
-        Self(value)
-    }
-}
-
 // TODO(@jmg-duarte,14/06/2024): validate miner address
 
 /// The storage provider address.
@@ -217,19 +207,6 @@ impl From<String> for StorageProviderAddress {
     }
 }
 
-/// Numeric identifier for a sector. It is usually relative to a storage provider.
-///
-/// For more information on sectors, see:
-/// <https://spec.filecoin.io/#section-systems.filecoin_mining.sector>
-#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Serialize, Deserialize)]
-pub struct SectorNumber(u64);
-
-impl From<u64> for SectorNumber {
-    fn from(value: u64) -> Self {
-        Self(value)
-    }
-}
-
 /// Information about a single *storage* deal for a given piece.
 ///
 /// Source: