net: Create datastore in P2p::new()

bin/darkfid/src/main.rs

@@ -288,7 +288,8 @@ async fn realmain(args: Args, ex: Arc<smol::Executor<'static>>) -> Result<()> {
     subscribers.insert("proposals", JsonSubscriber::new("blockchain.subscribe_proposals"));
 
     // Initialize P2P network
-    let p2p = spawn_p2p(&blockchain_config.net.into(), &validator, &subscribers, ex.clone()).await;
+    let p2p =
+        spawn_p2p(&blockchain_config.net.into(), &validator, &subscribers, ex.clone()).await?;
 
     // Initialize JSON-RPC client to perform requests to minerd
     let rpc_client = if blockchain_config.miner {

bin/darkfid/src/tests/harness.rs

@@ -241,7 +241,7 @@ pub async fn generate_node(
     // We initialize a dnet subscriber but do not activate it.
     let dnet_sub = JsonSubscriber::new("dnet.subscribe_events");
 
-    let p2p = spawn_p2p(settings, &validator, &subscribers, ex.clone()).await;
+    let p2p = spawn_p2p(settings, &validator, &subscribers, ex.clone()).await?;
     let node = Darkfid::new(p2p.clone(), validator, miner, 50, subscribers, None, dnet_sub).await;
 
     p2p.start().await?;

bin/darkfid/src/utils.rs

@@ -41,9 +41,9 @@ pub async fn spawn_p2p(
     validator: &ValidatorPtr,
     subscribers: &HashMap<&'static str, JsonSubscriber>,
     executor: Arc<Executor<'static>>,
-) -> P2pPtr {
+) -> Result<P2pPtr> {
     info!(target: "darkfid", "Registering sync network P2P protocols...");
-    let p2p = P2p::new(settings.clone(), executor.clone()).await;
+    let p2p = P2p::new(settings.clone(), executor.clone()).await?;
     let registry = p2p.protocol_registry();
 
     let _validator = validator.clone();
@@ -78,7 +78,7 @@ pub async fn spawn_p2p(
         })
         .await;
 
-    p2p
+    Ok(p2p)
 }
 
 /// Auxiliary function to parse darkfid configuration file and extract requested

bin/darkirc/src/main.rs

@@ -248,7 +248,7 @@ async fn realmain(args: Args, ex: Arc<Executor<'static>>) -> Result<()> {
     let sled_db = sled::open(datastore)?;
     let mut p2p_settings: darkfi::net::Settings = args.net.into();
     p2p_settings.app_version = semver::Version::parse(env!("CARGO_PKG_VERSION")).unwrap();
-    let p2p = P2p::new(p2p_settings, ex.clone()).await;
+    let p2p = P2p::new(p2p_settings, ex.clone()).await?;
     let event_graph = EventGraph::new(
         p2p.clone(),
         sled_db.clone(),

bin/genev/genevd/src/main.rs

@@ -116,7 +116,7 @@ async fn realmain(settings: Args, executor: Arc<smol::Executor<'static>>) -> Res
     let replay_mode = settings.replay_mode;
 
     let sled_db = sled::open(datastore_path.clone())?;
-    let p2p = P2p::new(settings.net.into(), executor.clone()).await;
+    let p2p = P2p::new(settings.net.into(), executor.clone()).await?;
     let event_graph = EventGraph::new(
         p2p.clone(),
         sled_db.clone(),

bin/lilith/src/main.rs

@@ -145,6 +145,8 @@ struct NetInfo {
     pub version: Version,
     /// Enable localnet hosts
     pub localnet: bool,
+    /// Path to P2P datastore
+    pub datastore: String,
     /// Path to hostlist
     pub hostlist: String,
 }
@@ -322,9 +324,12 @@ fn parse_configured_networks(data: &str) -> Result<HashMap<String, NetInfo>> {
                     semver::Version::parse(option_env!("CARGO_PKG_VERSION").unwrap_or("0.0.0"))?
                 };
 
+                let datastore: String = table["datastore"].as_str().unwrap().to_string();
+
                 let hostlist: String = table["hostlist"].as_str().unwrap().to_string();
 
-                let net_info = NetInfo { accept_addrs, seeds, peers, version, localnet, hostlist };
+                let net_info =
+                    NetInfo { accept_addrs, seeds, peers, version, localnet, datastore, hostlist };
                 ret.insert(name, net_info);
             }
         }
@@ -351,6 +356,7 @@ async fn spawn_net(name: String, info: &NetInfo, ex: Arc<Executor<'static>>) ->
         inbound_connections: 512,
         app_version: info.version.clone(),
         localnet: info.localnet,
+        datastore: Some(info.datastore.clone()),
         hostlist: Some(info.hostlist.clone()),
         allowed_transports: vec![
             "tcp".to_string(),
@@ -364,7 +370,7 @@ async fn spawn_net(name: String, info: &NetInfo, ex: Arc<Executor<'static>>) ->
     };
 
     // Create P2P instance
-    let p2p = P2p::new(settings, ex.clone()).await;
+    let p2p = P2p::new(settings, ex.clone()).await?;
 
     let addrs_str: Vec<&str> = listen_urls.iter().map(|x| x.as_str()).collect();
     info!(target: "lilith", "Starting seed network node for \"{}\" on {:?}", name, addrs_str);

bin/tau/taud/src/main.rs

@@ -357,7 +357,7 @@ async fn realmain(settings: Args, executor: Arc<smol::Executor<'static>>) -> Res
 
     info!("Instantiating event DAG");
     let sled_db = sled::open(datastore)?;
-    let p2p = P2p::new(settings.net.into(), executor.clone()).await;
+    let p2p = P2p::new(settings.net.into(), executor.clone()).await?;
     let event_graph = EventGraph::new(
         p2p.clone(),
         sled_db.clone(),

example/dchat/dchatd/src/main.rs

@@ -99,7 +99,7 @@ impl Dchat {
 // ANCHOR: main
 async_daemonize!(realmain);
 async fn realmain(args: Args, ex: Arc<smol::Executor<'static>>) -> Result<()> {
-    let p2p = net::P2p::new(args.net.into(), ex.clone()).await;
+    let p2p = net::P2p::new(args.net.into(), ex.clone()).await?;
 
     // ANCHOR: dnet
     info!("Starting dnet subs task");

src/event_graph/event.rs

@@ -214,7 +214,7 @@ mod tests {
 
     async fn make_event_graph() -> Result<EventGraphPtr> {
         let ex = Arc::new(Executor::new());
-        let p2p = P2p::new(Settings::default(), ex.clone()).await;
+        let p2p = P2p::new(Settings::default(), ex.clone()).await?;
         let sled_db = sled::Config::new().temporary(true).open().unwrap();
         EventGraph::new(p2p, sled_db, "/tmp".into(), false, "dag", 1, ex).await
     }

src/event_graph/tests.rs

@@ -89,7 +89,7 @@ async fn spawn_node(
         ..Default::default()
     };
 
-    let p2p = P2p::new(settings, ex.clone()).await;
+    let p2p = P2p::new(settings, ex.clone()).await.unwrap();
     let sled_db = sled::Config::new().temporary(true).open().unwrap();
     let event_graph =
         EventGraph::new(p2p.clone(), sled_db, "/tmp".into(), false, "dag", 1, ex.clone())

src/net/p2p.rs

@@ -21,7 +21,11 @@ use std::sync::Arc;
 use futures::{stream::FuturesUnordered, TryFutureExt};
 use futures_rustls::rustls::crypto::{ring, CryptoProvider};
 use log::{debug, error, info, warn};
-use smol::{lock::Mutex, stream::StreamExt};
+use smol::{
+    fs::{self, unix::PermissionsExt},
+    lock::Mutex,
+    stream::StreamExt,
+};
 use url::Url;
 
 use super::{
@@ -38,6 +42,7 @@ use super::{
 };
 use crate::{
     system::{ExecutorPtr, Publisher, PublisherPtr, Subscription},
+    util::path::expand_path,
     Result,
 };
 
@@ -79,9 +84,16 @@ impl P2p {
     ///
     /// Creates a weak pointer to self that is used by all sessions to access the
     /// p2p parent class.
-    pub async fn new(settings: Settings, executor: ExecutorPtr) -> P2pPtr {
+    pub async fn new(settings: Settings, executor: ExecutorPtr) -> Result<P2pPtr> {
         let settings = Arc::new(settings);
 
+        // Create the datastore
+        if let Some(ref datastore) = settings.datastore {
+            let datastore = expand_path(datastore)?;
+            fs::create_dir_all(&datastore).await?;
+            fs::set_permissions(&datastore, PermissionsExt::from_mode(0o700)).await?;
+        }
+
         // Register a CryptoProvider for rustls
         let _ = CryptoProvider::install_default(ring::default_provider());
 
@@ -108,7 +120,7 @@ impl P2p {
 
         register_default_protocols(self_.clone()).await;
 
-        self_
+        Ok(self_)
     }
 
     /// Starts inbound, outbound, and manual sessions.

src/net/tests.rs

@@ -113,7 +113,7 @@ async fn spawn_seed_session(seed_addr: Url, ex: Arc<Executor<'static>>) -> Vec<A
             ..Default::default()
         };
 
-        let p2p = P2p::new(settings, ex.clone()).await;
+        let p2p = P2p::new(settings, ex.clone()).await.unwrap();
         info!("========================================================");
         info!("Starting node={}", p2p.settings().external_addrs[0]);
         info!("========================================================");
@@ -162,7 +162,7 @@ async fn spawn_manual_session(ex: Arc<Executor<'static>>) -> Vec<Arc<P2p>> {
             ..Default::default()
         };
 
-        let p2p = P2p::new(settings, ex.clone()).await;
+        let p2p = P2p::new(settings, ex.clone()).await.unwrap();
         info!("========================================================");
         info!("Starting node={}", p2p.settings().external_addrs[0]);
         info!("========================================================");
@@ -308,7 +308,7 @@ async fn p2p_test_real(ex: Arc<Executor<'static>>) {
         ..Default::default()
     };
 
-    let seed = P2p::new(settings, ex.clone()).await;
+    let seed = P2p::new(settings, ex.clone()).await.unwrap();
     info!("========================================================");
     info!("Starting seed node on {}", seed_addr);
     info!("========================================================");

src/net/transport/tor.rs

@@ -169,7 +169,7 @@ impl TorListener {
         // the Tor dialer
         let client = match TOR_CLIENT
             .get_or_try_init(|| async {
-                debug!(target: "net::tor::do_dial", "Bootstrapping...");
+                debug!(target: "net::tor::do_listen", "Bootstrapping...");
                 if let Some(datadir) = &self.datastore {
                     let datadir = expand_path(datadir).unwrap();