timeout for querying seeds

src/bin/dfi.rs

@@ -10,6 +10,7 @@ use smol::Async;
 use std::net::SocketAddr;
 use std::net::TcpListener;
 use std::sync::Arc;
+use log::*;
 
 use sapvi::{net, Result};
 
@@ -152,6 +153,7 @@ async fn start(executor: Arc<Executor<'_>>, options: ProgramOptions) -> Result<(
     *rpc.started.lock().await = true;
 
     p2p.clone().start(executor.clone()).await?;
+
     p2p.run(executor).await?;
 
     rpc.wait_for_quit().await?;
@@ -269,7 +271,6 @@ impl ProgramOptions {
             (@arg CONNECTS: -c --connect ... "Manual connections")
             (@arg CONNECT_SLOTS: --slots +takes_value "Connection slots")
             (@arg LOG_PATH: --log +takes_value "Logfile path")
-            (@arg DISABLE_SEED: -D --disable_seed "Disable seed process")
             (@arg RPC_PORT: -r --rpc +takes_value "RPC port")
         )
         .get_matches();
@@ -309,12 +310,6 @@ impl ProgramOptions {
             .to_path_buf(),
         );
 
-        let skip_seed_sync = if app.is_present("DISABLE_SEED") {
-            true
-        } else {
-            false
-        };
-
         let rpc_port = if let Some(rpc_port) = app.value_of("RPC_PORT") {
             rpc_port.parse()?
         } else {
@@ -325,13 +320,13 @@ impl ProgramOptions {
             network_settings: net::Settings {
                 inbound: accept_addr,
                 outbound_connections: connection_slots,
+                seed_query_timeout_seconds: 8,
                 connect_timeout_seconds: 10,
                 channel_handshake_seconds: 4,
                 channel_heartbeat_seconds: 10,
                 external_addr: accept_addr,
                 peers: manual_connects,
                 seeds: seed_addrs,
-                skip_seed_sync,
             },
             log_path,
             rpc_port,

src/net/hosts.rs

@@ -35,4 +35,8 @@ impl Hosts {
     pub async fn load_all(&self) -> Vec<SocketAddr> {
         self.addrs.lock().await.clone()
     }
+
+    pub async fn is_empty(&self) -> bool {
+        self.addrs.lock().await.is_empty()
+    }
 }

src/net/sessions/seed_session.rs

@@ -1,4 +1,5 @@
 use async_executor::Executor;
+use futures::FutureExt;
 use log::*;
 use std::net::SocketAddr;
 use std::sync::{Arc, Weak};
@@ -7,6 +8,7 @@ use crate::net::error::{NetError, NetResult};
 use crate::net::protocols::{ProtocolPing, ProtocolSeed};
 use crate::net::sessions::Session;
 use crate::net::{ChannelPtr, Connector, HostsPtr, P2p, SettingsPtr};
+use crate::net::utility::sleep;
 
 pub struct SeedSession {
     p2p: Weak<P2p>,
@@ -19,40 +21,43 @@ impl SeedSession {
 
     pub async fn start(self: Arc<Self>, executor: Arc<Executor<'_>>) -> NetResult<()> {
         debug!(target: "net", "SeedSession::start() [START]");
-        let settings = {
-            let p2p = self.p2p.upgrade().unwrap();
-            p2p.settings()
-        };
+        let settings = self.p2p().settings();
 
-        if settings.skip_seed_sync {
-            info!("Configured to skip seed synchronization process.");
+        if settings.seeds.is_empty() {
+            warn!("Skipping seed sync process since no seeds are configured.");
             return Ok(());
         }
 
         // if cached addresses then quit
 
-        // if seeds empty then seeding required but empty
-        if settings.seeds.is_empty() {
-            error!("Seeding is required but no seeds are configured.");
-            return Err(NetError::OperationFailed);
-        }
-
         let mut tasks = Vec::new();
 
         for (i, seed) in settings.seeds.iter().enumerate() {
             tasks.push(executor.spawn(self.clone().start_seed(i, seed.clone(), executor.clone())));
         }
 
-        for (i, task) in tasks.into_iter().enumerate() {
-            // Ignore errors
-            match task.await {
-                Ok(()) => info!("Successfully queried seed #{}", i),
-                Err(err) => warn!("Seed query #{} failed for reason: {}", i, err),
+        futures::select! {
+            _ = async move {
+                for (i, task) in tasks.into_iter().enumerate() {
+                    // Ignore errors
+                    match task.await {
+                        Ok(()) => info!("Successfully queried seed #{}", i),
+                        Err(err) => warn!("Seed query #{} failed for reason: {}", i, err),
+                    }
+                }
+            }.fuse() => {
+            }
+            _ = sleep(settings.seed_query_timeout_seconds).fuse() => {
+                error!("Querying seeds timed out");
+                return Err(NetError::OperationFailed);
             }
         }
 
         // Seed process complete
-        // TODO: check increase count of address
+        if self.p2p().hosts().is_empty().await {
+            error!("Hosts pool still empty after seeding");
+            return Err(NetError::OperationFailed);
+        }
 
         debug!(target: "net", "SeedSession::start() [END]");
         Ok(())

src/net/settings.rs

@@ -8,6 +8,7 @@ pub struct Settings {
     pub inbound: Option<SocketAddr>,
     pub outbound_connections: u32,
 
+    pub seed_query_timeout_seconds: u32,
     pub connect_timeout_seconds: u32,
     pub channel_handshake_seconds: u32,
     pub channel_heartbeat_seconds: u32,
@@ -15,5 +16,4 @@ pub struct Settings {
     pub external_addr: Option<SocketAddr>,
     pub peers: Vec<SocketAddr>,
     pub seeds: Vec<SocketAddr>,
-    pub skip_seed_sync: bool,
 }