feat: refactor p2p node & merge retrieval functionality

Cargo.toml

@@ -20,11 +20,10 @@ members = [
   "pallets/storage-provider/benchmarks",
   "primitives",
   "runtime",
+  "storage-fetch",
   "storage-provider/client",
   "storage-provider/common",
   "storage-provider/server",
-  "storage-retrieval/cli",
-  "storage-retrieval/lib",
   "storagext/cli",
   "storagext/lib",
 ]
@@ -149,7 +148,6 @@ pallet-storage-provider = { path = "pallets/storage-provider", default-features
 pallet-storage-provider-benchmarks = { path = "pallets/storage-provider/benchmarks", default-features = false }
 polka-storage-proofs = { path = "lib/polka-storage-proofs", default-features = false }
 polka-storage-provider-common = { path = "storage-provider/common" }
-polka-storage-retrieval = { path = "storage-retrieval/lib" }
 polka-storage-runtime = { path = "runtime" }
 primitives = { path = "primitives", default-features = false }
 storagext = { path = "storagext/lib" }

storage-retrieval/lib/Cargo.toml → storage-fetch/Cargo.toml

@@ -3,31 +3,29 @@ authors.workspace = true
 edition.workspace = true
 homepage.workspace = true
 license-file.workspace = true
-name = "polka-storage-retrieval"
+name = "polka-fetch"
 repository.workspace = true
 version = "0.1.0"
 
-[lints]
-workspace = true
 
 [dependencies]
 anyhow = { workspace = true }
 beetswap = { workspace = true }
 blockstore = { workspace = true }
 cid = { workspace = true }
+clap = { workspace = true, features = ["derive"] }
 futures = { workspace = true }
 ipld-core = { workspace = true, features = ["serde"] }
 ipld-dagpb.workspace = true
 libp2p = { workspace = true, features = ["macros", "noise", "tcp", "tokio", "yamux"] }
 libp2p-core = { workspace = true }
 libp2p-swarm = { workspace = true }
 mater = { workspace = true, features = ["blockstore"] }
+multihash-codetable = { workspace = true, features = ["sha2"] }
 thiserror = { workspace = true }
 tokio = { workspace = true, features = ["macros", "rt", "rt-multi-thread", "sync", "time"] }
 tracing = { workspace = true }
-
-[dev-dependencies]
-multihash-codetable = { workspace = true, features = ["sha2"] }
-rand = { workspace = true, default-features = true }
-tracing-appender = { workspace = true }
 tracing-subscriber = { workspace = true, features = ["env-filter"] }
+
+[lints]
+workspace = true

storage-retrieval/lib/src/client.rs → storage-fetch/src/client.rs

@@ -21,17 +21,14 @@ use tokio::{
 };
 use tracing::{debug, error, info, instrument, trace};
 
-use crate::{new_swarm, Behaviour, BehaviourEvent, InitSwarmError};
+use crate::p2p::{new_swarm, Behaviour, BehaviourEvent, InitSwarmError};
 
 /// Errors that can occur while retrieving some content.
 #[derive(Debug, Error)]
 pub enum ClientError {
     /// Error occurred while initialing swarm
     #[error("Swarm initialization error: {0}")]
     InitSwarm(#[from] InitSwarmError),
-    /// This error indicates that the download was timed out.
-    #[error("Download timeout")]
-    DownloadTimeout,
     /// Error occurred when trying to establish or upgrade an outbound connection.
     #[error("Dial error: {0}")]
     Dial(#[from] DialError),

storage-retrieval/cli/src/main.rs → storage-fetch/src/main.rs

@@ -2,14 +2,17 @@ use std::{path::PathBuf, time::Duration};
 
 use cid::Cid;
 use clap::{command, Parser};
+use client::{Client, ClientSettings};
 use libp2p::Multiaddr;
-use polka_storage_retrieval::client::{Client, ClientSettings};
 use tokio::time::timeout;
 use tracing::{error, info, level_filters::LevelFilter};
 use tracing_subscriber::{
     filter::FromEnvError, fmt, layer::SubscriberExt, util::SubscriberInitExt, EnvFilter, Layer,
 };
 
+mod client;
+mod p2p;
+
 #[derive(Parser, Debug)]
 #[command()]
 struct Cli {

storage-retrieval/lib/src/lib.rs → storage-fetch/src/p2p.rs

@@ -1,23 +1,18 @@
-pub mod client;
-pub mod server;
-
 use std::{sync::Arc, time::Duration};
 
 use ::blockstore::Blockstore;
-pub use client::Client;
 use libp2p::{noise, swarm::NetworkBehaviour, tcp, yamux, Swarm, SwarmBuilder};
-pub use server::Server;
 use thiserror::Error;
 
 const MAX_MULTIHASH_LENGTH: usize = 64;
 
 /// Custom Behaviour used by the server and client.
 #[derive(NetworkBehaviour)]
-struct Behaviour<B>
+pub struct Behaviour<B>
 where
     B: Blockstore + 'static,
 {
-    bitswap: beetswap::Behaviour<MAX_MULTIHASH_LENGTH, B>,
+    pub bitswap: beetswap::Behaviour<MAX_MULTIHASH_LENGTH, B>,
 }
 
 /// Error that can occur while initializing a swarm
@@ -29,7 +24,7 @@ pub enum InitSwarmError {
 }
 
 /// Initialize a new swarm with our custom Behaviour.
-fn new_swarm<B>(blockstore: Arc<B>) -> Result<Swarm<Behaviour<B>>, InitSwarmError>
+pub fn new_swarm<B>(blockstore: Arc<B>) -> Result<Swarm<Behaviour<B>>, InitSwarmError>
 where
     B: Blockstore + 'static,
 {

storage-provider/server/Cargo.toml

@@ -18,13 +18,13 @@ 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 }
+beetswap = { workspace = true }
 blockstore = { workspace = true }
 chrono = { workspace = true, features = ["serde"] }
 ciborium = { workspace = true }

storage-provider/server/src/config.rs

@@ -2,22 +2,20 @@ use std::{
     net::{IpAddr, Ipv4Addr, SocketAddr},
     num::NonZero,
     path::PathBuf,
+    str::FromStr,
 };
 
 use clap::Args;
 use libp2p::{identity::Keypair, Multiaddr, PeerId};
 use polka_storage_provider_common::config::sealing::SealingConfiguration;
 use primitives::{
-    p2p::{keypair_value_parser, DEFAULT_REGISTRATION_TTL},
+    p2p::keypair_value_parser,
     proofs::{RegisteredPoStProof, RegisteredSealProof},
 };
-use serde::Deserialize;
+use serde::{de::Error, Deserialize, Deserializer};
 use url::Url;
 
-use crate::{
-    p2p::{deser_keypair, deserialize_string_to_peer_id},
-    DEFAULT_NODE_ADDRESS,
-};
+use crate::DEFAULT_NODE_ADDRESS;
 
 /// Default address to bind the RPC server to.
 const fn default_rpc_listen_address() -> SocketAddr {
@@ -35,16 +33,11 @@ const fn default_parallel_prove_commits() -> NonZero<usize> {
     unsafe { NonZero::new_unchecked(2) }
 }
 
-/// Default registration TTL, how long the node is registered.
-const fn default_registration_ttl() -> u64 {
-    DEFAULT_REGISTRATION_TTL
-}
-
 fn default_node_address() -> Url {
     Url::parse(DEFAULT_NODE_ADDRESS).expect("DEFAULT_NODE_ADDRESS must be a valid Url")
 }
 
-fn default_retrieval_address() -> Multiaddr {
+fn default_p2p_listen_address() -> Multiaddr {
     "/ip4/127.0.0.1/tcp/8002"
         .parse()
         .expect("multiaddres is correct")
@@ -69,11 +62,6 @@ 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)]
@@ -128,22 +116,35 @@ pub struct ConfigurationArgs {
     #[arg(long, value_parser = keypair_value_parser, required = false)]
     pub(crate) p2p_key: Keypair,
 
-    /// Rendezvous point address that the registration node connects to
-    /// or the bootstrap node binds to.
+    /// P2P listen address.
+    #[serde(default = "default_p2p_listen_address")]
+    #[arg(long, default_value_t = default_p2p_listen_address())]
+    pub(crate) p2p_listen_address: Multiaddr,
+
+    /// Rendezvous multiaddr that the node registers to.
     #[arg(long, required = false)]
     pub(crate) rendezvous_point_address: Multiaddr,
 
-    /// PeerID of the bootstrap node used by the registration node.
+    /// PeerID of the rendezvous node used.
     #[serde(deserialize_with = "deserialize_string_to_peer_id")]
     #[arg(long, required = false)]
     pub(crate) rendezvous_point: PeerId,
 
-    /// TTL of the p2p registration in seconds
-    #[serde(default = "default_registration_ttl")]
-    #[arg(long, default_value_t = DEFAULT_REGISTRATION_TTL, required = false)]
-    pub(crate) registration_ttl: u64,
-
     #[clap(flatten)]
     #[serde(default)]
     pub(crate) sealing_configuration: SealingConfiguration,
 }
+
+/// Deserializes a ED25519 private key into a Keypair.
+/// Can either be the private key as a string or the path of a PEM file with an @ prefixed
+/// Calls `keypair_value_parser` after deserializing the source string
+fn deser_keypair<'de, D: Deserializer<'de>>(d: D) -> Result<Keypair, D::Error> {
+    let src: String = Deserialize::deserialize(d)?;
+    keypair_value_parser(&src).map_err(Error::custom)
+}
+
+/// Parses a string to an Peer ID.
+fn deserialize_string_to_peer_id<'de, D: Deserializer<'de>>(d: D) -> Result<PeerId, D::Error> {
+    let s: String = Deserialize::deserialize(d)?;
+    PeerId::from_str(&s).map_err(Error::custom)
+}