Merge pull request #1068 from subspace/dsn-node-provider-storage

node: Add provider record persistence.

Author: shamil-gadelshin

Diff for: Cargo.lock

@@ -236,9 +236,9 @@ impl ParityDbProviderStorage {
         }
 
         if heap.size() > 0 {
-            info!(size = heap.size(), "Record cache loaded.");
+            info!(size = heap.size(), ?path, "Record cache loaded.");
         } else {
-            info!("New record cache initialized.");
+            info!(?path, "New record cache initialized.");
         }
 
         Ok(Self {

Diff for: crates/subspace-networking/src/behavior/provider_storage/providers.rs

@@ -284,6 +284,9 @@ pub enum CreationError {
     /// Transport error when attempting to listen on multiaddr.
     #[error("Transport error when attempting to listen on multiaddr: {0}")]
     TransportError(#[from] TransportError<io::Error>),
+    /// ParityDb storage error
+    #[error("ParityDb storage error: {0}")]
+    ParityDbStorageError(#[from] parity_db::Error),
 }
 
 /// Converts public key from keypair to PeerId.

Diff for: crates/subspace-networking/src/create.rs

@@ -421,6 +421,11 @@ fn main() -> Result<(), Error> {
 
                         DsnConfig {
                             keypair,
+                            base_path: cli.run.base_path()?.map(|base_path| {
+                                base_path
+                                    .config_dir(primary_chain_config.chain_spec.id())
+                                    .join("dsn")
+                            }),
                             listen_on: cli.dsn_listen_on,
                             bootstrap_nodes: dsn_bootstrap_nodes,
                             reserved_peers: cli.dsn_reserved_peers,

Diff for: crates/subspace-node/src/bin/subspace-node.rs

@@ -19,11 +19,13 @@ targets = ["x86_64-unknown-linux-gnu"]
 backoff = { version = "0.4.0", features = ["futures", "tokio"] }
 derive_more = "0.99.17"
 domain-runtime-primitives = { version = "0.1.0", path = "../../domains/primitives/runtime" }
+either = "1.8.0"
 frame-support = { version = "4.0.0-dev", git = "https://github.com/subspace/substrate", rev = "100d6c90d4122578006a47c1dcaf155b9c685f34" }
 futures = "0.3.25"
 jsonrpsee = { version = "0.16.2", features = ["server"] }
 pallet-transaction-payment-rpc = { version = "4.0.0-dev", git = "https://github.com/subspace/substrate", rev = "100d6c90d4122578006a47c1dcaf155b9c685f34" }
 parity-scale-codec = "3.2.1"
+parking_lot = "0.12.1"
 sc-basic-authorship = { version = "0.10.0-dev", git = "https://github.com/subspace/substrate", rev = "100d6c90d4122578006a47c1dcaf155b9c685f34" }
 sc-chain-spec = { version = "4.0.0-dev", git = "https://github.com/subspace/substrate", rev = "100d6c90d4122578006a47c1dcaf155b9c685f34" }
 sc-client-api = { version = "4.0.0-dev", git = "https://github.com/subspace/substrate", rev = "100d6c90d4122578006a47c1dcaf155b9c685f34" }

Diff for: crates/subspace-service/Cargo.toml

@@ -1,6 +1,10 @@
+mod node_provider_storage;
+
+use crate::dsn::node_provider_storage::NodeProviderStorage;
 use crate::piece_cache::PieceCache;
 use backoff::future::retry;
 use backoff::ExponentialBackoff;
+use either::Either;
 use futures::stream::FuturesUnordered;
 use futures::{Stream, StreamExt};
 use sc_client_api::AuxStore;
@@ -9,14 +13,16 @@ use sp_core::traits::SpawnNamed;
 use sp_runtime::traits::Block as BlockT;
 use std::error::Error;
 use std::num::NonZeroUsize;
+use std::path::PathBuf;
 use std::sync::Arc;
 use std::time::Duration;
 use subspace_archiving::archiver::ArchivedSegment;
 use subspace_core_primitives::{PieceIndex, PieceIndexHash, PIECES_IN_SEGMENT};
 use subspace_networking::libp2p::{identity, Multiaddr};
 use subspace_networking::{
     peer_id, BootstrappedNetworkingParameters, CreationError, MemoryProviderStorage, Node,
-    NodeRunner, PieceByHashRequestHandler, PieceByHashResponse, PieceKey, ToMultihash,
+    NodeRunner, ParityDbProviderStorage, PieceByHashRequestHandler, PieceByHashResponse, PieceKey,
+    ToMultihash,
 };
 use tokio::sync::Semaphore;
 use tokio::time::error::Elapsed;
@@ -29,6 +35,8 @@ const PUBLISH_PIECE_TIMEOUT: Duration = Duration::from_secs(120);
 const PUBLISH_PIECE_INITIAL_INTERVAL: Duration = Duration::from_secs(1);
 /// Defines max duration between put_piece calls.
 const PUBLISH_PIECE_MAX_INTERVAL: Duration = Duration::from_secs(30);
+/// Provider records cache size
+const MAX_PROVIDER_RECORDS_LIMIT: usize = 100000; // ~ 10 MB
 
 /// DSN configuration parameters.
 #[derive(Clone, Debug)]
@@ -47,20 +55,38 @@ pub struct DsnConfig {
 
     /// Determines whether we allow keeping non-global (private, shared, loopback..) addresses in Kademlia DHT.
     pub allow_non_global_addresses_in_dht: bool,
+
+    /// System base path.
+    pub base_path: Option<PathBuf>,
 }
 
+type DsnProviderStorage<AS> =
+    NodeProviderStorage<PieceCache<AS>, Either<ParityDbProviderStorage, MemoryProviderStorage>>;
+
 pub(crate) async fn create_dsn_instance<Block, AS>(
     dsn_config: DsnConfig,
     piece_cache: PieceCache<AS>,
-) -> Result<(Node, NodeRunner<MemoryProviderStorage>), CreationError>
+) -> Result<(Node, NodeRunner<DsnProviderStorage<AS>>), CreationError>
 where
     Block: BlockT,
     AS: AuxStore + Sync + Send + 'static,
 {
     trace!("Subspace networking starting.");
 
-    // TODO: This should be a wrapper that handles locally cached pieces
-    let provider_storage = MemoryProviderStorage::new(peer_id(&dsn_config.keypair));
+    let peer_id = peer_id(&dsn_config.keypair);
+
+    let external_provider_storage = if let Some(path) = dsn_config.base_path {
+        let db_path = path.join("storage_providers_db").into_boxed_path();
+
+        let cache_size: NonZeroUsize = NonZeroUsize::new(MAX_PROVIDER_RECORDS_LIMIT)
+            .expect("Manual value should be greater than zero.");
+
+        Either::Left(ParityDbProviderStorage::new(&db_path, cache_size, peer_id)?)
+    } else {
+        Either::Right(MemoryProviderStorage::new(peer_id))
+    };
+
+    let provider_storage = NodeProviderStorage::new(piece_cache.clone(), external_provider_storage);
 
     let networking_config = subspace_networking::Config {
         keypair: dsn_config.keypair.clone(),

Diff for: crates/subspace-service/src/dsn.rs

@@ -0,0 +1,62 @@
+use subspace_networking::libp2p::kad::record::Key;
+use subspace_networking::libp2p::kad::ProviderRecord;
+use subspace_networking::libp2p::PeerId;
+use subspace_networking::ProviderStorage;
+
+pub(crate) struct NodeProviderStorage<ImplicitProviderStorage, PersistentProviderStorage> {
+    /// Provider records from local cache
+    implicit_provider_storage: ImplicitProviderStorage,
+    /// External provider records
+    persistent_provider_storage: PersistentProviderStorage,
+}
+
+impl<ImplicitProviderStorage, ExternalProviderStorage>
+    NodeProviderStorage<ImplicitProviderStorage, ExternalProviderStorage>
+{
+    pub(crate) fn new(
+        implicit_provider_storage: ImplicitProviderStorage,
+        persistent_provider_storage: ExternalProviderStorage,
+    ) -> Self {
+        Self {
+            implicit_provider_storage,
+            persistent_provider_storage,
+        }
+    }
+}
+
+impl<ImplicitProviderStorage, PersistentProviderStorage> ProviderStorage
+    for NodeProviderStorage<ImplicitProviderStorage, PersistentProviderStorage>
+where
+    ImplicitProviderStorage: ProviderStorage,
+    PersistentProviderStorage: ProviderStorage,
+{
+    type ProvidedIter<'a> = ImplicitProviderStorage::ProvidedIter<'a> where Self:'a;
+
+    fn add_provider(
+        &mut self,
+        record: ProviderRecord,
+    ) -> subspace_networking::libp2p::kad::store::Result<()> {
+        // only external provider records
+        self.persistent_provider_storage.add_provider(record)
+    }
+
+    fn providers(&self, key: &Key) -> Vec<ProviderRecord> {
+        let mut local_provider_records = self.implicit_provider_storage.providers(key);
+        let mut external_provider_records = self.persistent_provider_storage.providers(key);
+
+        local_provider_records.append(&mut external_provider_records);
+
+        local_provider_records
+    }
+
+    fn provided(&self) -> Self::ProvidedIter<'_> {
+        // only local cached provider records
+        self.implicit_provider_storage.provided()
+    }
+
+    fn remove_provider(&mut self, key: &Key, peer_id: &PeerId) {
+        // only external provider records
+        self.persistent_provider_storage
+            .remove_provider(key, peer_id)
+    }
+}

Diff for: crates/subspace-service/src/dsn/node_provider_storage.rs

@@ -15,7 +15,7 @@
 // along with this program. If not, see <https://www.gnu.org/licenses/>.
 
 //! Service and ServiceFactory implementation. Specialized wrapper over substrate service.
-#![feature(type_changing_struct_update)]
+#![feature(type_alias_impl_trait, type_changing_struct_update)]
 
 mod dsn;
 pub mod piece_cache;

Diff for: crates/subspace-service/src/lib.rs

@@ -2,6 +2,7 @@
 mod tests;
 
 use parity_scale_codec::{Decode, Encode};
+use parking_lot::Mutex;
 use sc_client_api::backend::AuxStore;
 use std::borrow::Cow;
 use std::collections::BTreeSet;
@@ -25,7 +26,7 @@ pub struct PieceCache<AS> {
     /// Peer ID of the current node.
     local_peer_id: PeerId,
     /// Local provided keys
-    local_provided_keys: BTreeSet<PieceIndex>,
+    local_provided_keys: Arc<Mutex<BTreeSet<PieceIndex>>>,
 }
 
 impl<AS> Clone for PieceCache<AS> {
@@ -65,7 +66,7 @@ where
             aux_store,
             max_pieces_in_cache,
             local_peer_id,
-            local_provided_keys,
+            local_provided_keys: Arc::new(Mutex::new(local_provided_keys)),
         }
     }
 
@@ -80,14 +81,17 @@ where
         }))
     }
 
-    fn write_local_provided_keys(&self) -> Result<(), Box<dyn Error>> {
+    fn write_local_provided_keys(
+        &self,
+        local_provided_keys: BTreeSet<PieceIndex>,
+    ) -> Result<(), Box<dyn Error>> {
         // TODO: Could be a slow process. We need to optimize it ASAP!
         self.aux_store
             .insert_aux(
                 &vec![(
                     LOCAL_PROVIDED_KEYS,
                     ParityDbKeyCollection {
-                        set: self.local_provided_keys.clone(),
+                        set: local_provided_keys,
                     }
                     .encode()
                     .as_slice(),
@@ -150,15 +154,21 @@ where
                 .collect::<Vec<_>>(),
         )?;
 
-        for piece_index in delete_indexes {
-            self.local_provided_keys.remove(&piece_index);
-        }
+        let local_provided_keys = {
+            let mut local_provided_keys = self.local_provided_keys.lock();
 
-        for piece_index in insert_indexes {
-            self.local_provided_keys.insert(piece_index);
-        }
+            for piece_index in delete_indexes {
+                local_provided_keys.remove(&piece_index);
+            }
+
+            for piece_index in insert_indexes {
+                local_provided_keys.insert(piece_index);
+            }
+
+            local_provided_keys.clone()
+        };
 
-        self.write_local_provided_keys()?;
+        self.write_local_provided_keys(local_provided_keys)?;
 
         Ok(())
     }
@@ -249,7 +259,13 @@ where
     }
 
     fn provided(&self) -> Self::ProvidedIter<'_> {
-        let pieces_indexes = self.local_provided_keys.iter();
+        let pieces_indexes = {
+            self.local_provided_keys
+                .lock()
+                .iter()
+                .cloned()
+                .collect::<Vec<_>>()
+        };
 
         AuxStoreProviderRecordIterator::new(pieces_indexes, self.clone())
     }
@@ -264,18 +280,17 @@ where
 }
 
 pub struct AuxStoreProviderRecordIterator<'a, AS> {
-    piece_index_iter: std::collections::btree_set::Iter<'a, PieceIndex>,
+    piece_indexes: Vec<PieceIndex>,
+    piece_indexes_cursor: usize,
     piece_cache: PieceCache<AS>,
     marker: PhantomData<&'a ()>,
 }
 
 impl<'a, AS: AuxStore> AuxStoreProviderRecordIterator<'a, AS> {
-    pub fn new(
-        piece_index_iter: std::collections::btree_set::Iter<'a, PieceIndex>,
-        piece_cache: PieceCache<AS>,
-    ) -> Self {
+    pub fn new(piece_indexes: Vec<PieceIndex>, piece_cache: PieceCache<AS>) -> Self {
         Self {
-            piece_index_iter,
+            piece_indexes,
+            piece_indexes_cursor: 0,
             piece_cache,
             marker: PhantomData,
         }
@@ -286,23 +301,31 @@ impl<'a, AS: AuxStore> Iterator for AuxStoreProviderRecordIterator<'a, AS> {
     type Item = Cow<'a, ProviderRecord>;
 
     fn next(&mut self) -> Option<Self::Item> {
+        if self.piece_indexes.len() == self.piece_indexes_cursor {
+            return None; // iterator finished
+        }
+
         let peer_id = self.piece_cache.local_peer_id;
+        let piece_index = self.piece_indexes[self.piece_indexes_cursor];
+        let piece_index_hash = PieceIndexHash::from_index(piece_index);
+        let key = Key::from(piece_index_hash.to_multihash());
+
+        let result = self
+            .piece_cache
+            .get_piece(piece_index_hash)
+            .ok()
+            .flatten()
+            .map(move |_| ProviderRecord {
+                key: key.clone(),
+                provider: peer_id,
+                expires: None,
+                addresses: vec![], // TODO: add address hints
+            })
+            .map(Cow::Owned);
 
-        self.piece_index_iter.next().and_then(|piece_index| {
-            let piece_index_hash = PieceIndexHash::from_index(*piece_index);
-            let key = Key::from(piece_index_hash.to_multihash());
+        // Move iterator cursor forward
+        self.piece_indexes_cursor += 1;
 
-            self.piece_cache
-                .get_piece(piece_index_hash)
-                .ok()
-                .flatten()
-                .map(move |_| ProviderRecord {
-                    key: key.clone(),
-                    provider: peer_id,
-                    expires: None,
-                    addresses: vec![], // TODO: add address hints
-                })
-                .map(Cow::Owned)
-        })
+        result
     }
 }

Diff for: crates/subspace-service/src/piece_cache.rs

@@ -128,6 +128,7 @@ async fn run_executor(
             force_new_slot_notifications: true,
             subspace_networking: SubspaceNetworking::Create {
                 config: DsnConfig {
+                    base_path: None,
                     listen_on: vec!["/ip4/127.0.0.1/tcp/0"
                         .parse()
                         .expect("Correct multiaddr; qed")],

Diff for: domains/test/service/src/lib.rs

@@ -193,6 +193,7 @@ pub async fn run_validator_node(
             force_new_slot_notifications: true,
             subspace_networking: SubspaceNetworking::Create {
                 config: DsnConfig {
+                    base_path: None,
                     listen_on: vec!["/ip4/127.0.0.1/tcp/0"
                         .parse()
                         .expect("Correct multiaddr; qed")],