net: cleanup connect loop code reuse by implement connect_slot() method. also prevent infinite loop by doing peer discovery when the hostlist is empty.

also fix tests.

src/net/hosts/store.rs

@@ -692,6 +692,24 @@ impl Hosts {
         self.whitelist.read().await.is_empty()
     }
 
+    /// Check if the anchorlist is empty.
+    pub async fn is_empty_anchorlist(&self) -> bool {
+        self.anchorlist.read().await.is_empty()
+    }
+
+    /// Check if the hostlist is empty.
+    pub async fn is_empty_hostlist(&self) -> bool {
+        if self.is_empty_greylist().await {
+            return true
+        } else if self.is_empty_whitelist().await {
+            return true
+        } else if self.is_empty_anchorlist().await {
+            return true
+        } else {
+            return false
+        }
+    }
+
     // Check whether this peer is in any of the hostlists.
     async fn hostlist_contains(&self, addr: &Url) -> bool {
         if self.greylist_contains(addr).await {

src/net/session/outbound_session.rs

@@ -199,6 +199,8 @@ impl Slot {
     //
     // TODO: read white/anchor/greylist from disk instead of getting from a seed node each time.
     // TODO: clean up below code reuse and test! Currently untested.
+    //
+    // Fetch address, try to connect to it, call setup_channel.
     async fn run(self: Arc<Self>) {
         let hosts = self.p2p().hosts();
         let slot_count = self.p2p().settings().outbound_connections;
@@ -218,152 +220,40 @@ impl Slot {
             // Get the active connection count.
             let connect_count = self.get_active_connect_count().await;
 
+            // Do peer discovery if we don't have a hostlist (first time connecting
+            // to the network).
+            if hosts.is_empty_hostlist().await {
+                dnetev!(self, OutboundSlotSleeping, {
+                    slot: self.slot,
+                });
+
+                self.wakeup_self.reset();
+                // Peer discovery
+                self.session().wakeup_peer_discovery();
+                // Wait to be woken up by peer discovery
+                self.wakeup_self.wait().await;
+                continue
+            }
+
             // For the first 2 loops, connect to a host from the anchorlist.
             if connect_count < DEFAULT_ANCHOR_CONN_COUNT {
                 if let Some(host) =
                     hosts.anchorlist_fetch_address_with_lock(self.p2p(), transports).await
                 {
-                    info!(
-                        target: "net::outbound_session::try_connect()",
-                        "[P2P] Connecting outbound slot #{} [{}]",
-                        self.slot, host.0,
-                    );
-
-                    dnetev!(self, OutboundSlotConnecting, {
-                        slot: self.slot,
-                        addr: host.0.clone(),
-                    });
-
-                    let (addr_final, channel) = match self.try_connect(host.0.clone()).await {
-                        Ok(connect_info) => connect_info,
-                        Err(err) => {
-                            error!(
-                                target: "net::outbound_session",
-                                "[P2P] Outbound slot #{} connection failed: {}",
-                                self.slot, err,
-                            );
-
-                            dnetev!(self, OutboundSlotDisconnected, {
-                                slot: self.slot,
-                                err: err.to_string()
-                            });
-
-                            self.channel_id.store(0, Ordering::Relaxed);
-                            continue
-                        }
-                    };
-
-                    info!(
-                        target: "net::outbound_session::try_connect()",
-                        "[P2P] Outbound slot #{} connected [{}]",
-                        self.slot, addr_final
-                    );
-
-                    dnetev!(self, OutboundSlotConnected, {
-                        slot: self.slot,
-                        addr: addr_final.clone(),
-                        channel_id: channel.info.id
-                    });
-
-                    let stop_sub =
-                        channel.subscribe_stop().await.expect("Channel should not be stopped");
-                    // Setup new channel
-                    if let Err(err) = self.setup_channel(host.0, channel.clone()).await {
-                        info!(
-                            target: "net::outbound_session",
-                            "[P2P] Outbound slot #{} disconnected: {}",
-                            self.slot, err
-                        );
-
-                        dnetev!(self, OutboundSlotDisconnected, {
-                            slot: self.slot,
-                            err: err.to_string()
-                        });
-
-                        self.channel_id.store(0, Ordering::Relaxed);
-                        continue
-                    }
-
-                    self.channel_id.store(channel.info.id, Ordering::Relaxed);
-
-                    // Wait for channel to close
-                    stop_sub.receive().await;
-                    self.channel_id.store(0, Ordering::Relaxed);
+                    // TODO: Error handling.
+                    self.connect_slot(&host.0, self.slot).await.unwrap();
                 } else {
                     continue
                 }
             }
 
-            // For the next N loops, connect to a host from the whitelist.
+            //// For the next N loops, connect to a host from the whitelist.
             if connect_count < white_count {
                 if let Some(host) =
                     hosts.whitelist_fetch_address_with_lock(self.p2p(), transports).await
                 {
-                    info!(
-                        target: "net::outbound_session::try_connect()",
-                        "[P2P] Connecting outbound slot #{} [{}]",
-                        self.slot, host.0,
-                    );
-
-                    dnetev!(self, OutboundSlotConnecting, {
-                        slot: self.slot,
-                        addr: host.0.clone(),
-                    });
-
-                    let (addr_final, channel) = match self.try_connect(host.0.clone()).await {
-                        Ok(connect_info) => connect_info,
-                        Err(err) => {
-                            error!(
-                                target: "net::outbound_session",
-                                "[P2P] Outbound slot #{} connection failed: {}",
-                                self.slot, err,
-                            );
-
-                            dnetev!(self, OutboundSlotDisconnected, {
-                                slot: self.slot,
-                                err: err.to_string()
-                            });
-
-                            self.channel_id.store(0, Ordering::Relaxed);
-                            continue
-                        }
-                    };
-                    info!(
-                        target: "net::outbound_session::try_connect()",
-                        "[P2P] Outbound slot #{} connected [{}]",
-                        self.slot, addr_final
-                    );
-
-                    dnetev!(self, OutboundSlotConnected, {
-                        slot: self.slot,
-                        addr: addr_final.clone(),
-                        channel_id: channel.info.id
-                    });
-
-                    let stop_sub =
-                        channel.subscribe_stop().await.expect("Channel should not be stopped");
-                    // Setup new channel
-                    if let Err(err) = self.setup_channel(host.0, channel.clone()).await {
-                        info!(
-                            target: "net::outbound_session",
-                            "[P2P] Outbound slot #{} disconnected: {}",
-                            self.slot, err
-                        );
-
-                        dnetev!(self, OutboundSlotDisconnected, {
-                            slot: self.slot,
-                            err: err.to_string()
-                        });
-
-                        self.channel_id.store(0, Ordering::Relaxed);
-                        continue
-                    }
-
-                    self.channel_id.store(channel.info.id, Ordering::Relaxed);
-
-                    // Wait for channel to close
-                    stop_sub.receive().await;
-                    self.channel_id.store(0, Ordering::Relaxed);
+                    // TODO: Error handling.
+                    self.connect_slot(&host.0, self.slot).await.unwrap();
                 } else {
                     continue
                 }
@@ -374,70 +264,8 @@ impl Slot {
                 if let Some(host) =
                     hosts.greylist_fetch_address_with_lock(self.p2p(), transports).await
                 {
-                    info!(
-                        target: "net::outbound_session::try_connect()",
-                        "[P2P] Connecting outbound slot #{} [{}]",
-                        self.slot, host.0,
-                    );
-
-                    dnetev!(self, OutboundSlotConnecting, {
-                        slot: self.slot,
-                        addr: host.0.clone(),
-                    });
-                    let (addr_final, channel) = match self.try_connect(host.0.clone()).await {
-                        Ok(connect_info) => connect_info,
-                        Err(err) => {
-                            error!(
-                                target: "net::outbound_session",
-                                "[P2P] Outbound slot #{} connection failed: {}",
-                                self.slot, err,
-                            );
-
-                            dnetev!(self, OutboundSlotDisconnected, {
-                                slot: self.slot,
-                                err: err.to_string()
-                            });
-
-                            self.channel_id.store(0, Ordering::Relaxed);
-                            continue
-                        }
-                    };
-                    info!(
-                        target: "net::outbound_session::try_connect()",
-                        "[P2P] Outbound slot #{} connected [{}]",
-                        self.slot, addr_final
-                    );
-
-                    dnetev!(self, OutboundSlotConnected, {
-                        slot: self.slot,
-                        addr: addr_final.clone(),
-                        channel_id: channel.info.id
-                    });
-
-                    let stop_sub =
-                        channel.subscribe_stop().await.expect("Channel should not be stopped");
-                    // Setup new channel
-                    if let Err(err) = self.setup_channel(host.0, channel.clone()).await {
-                        info!(
-                            target: "net::outbound_session",
-                            "[P2P] Outbound slot #{} disconnected: {}",
-                            self.slot, err
-                        );
-
-                        dnetev!(self, OutboundSlotDisconnected, {
-                            slot: self.slot,
-                            err: err.to_string()
-                        });
-
-                        self.channel_id.store(0, Ordering::Relaxed);
-                        continue
-                    }
-
-                    self.channel_id.store(channel.info.id, Ordering::Relaxed);
-
-                    // Wait for channel to close
-                    stop_sub.receive().await;
-                    self.channel_id.store(0, Ordering::Relaxed);
+                    // TODO: Error handling.
+                    self.connect_slot(&host.0, self.slot).await.unwrap();
                 } else {
                     // We haven't been able to connect to any known peers. Activate peer discovery.
                     dnetev!(self, OutboundSlotSleeping, {
@@ -455,6 +283,76 @@ impl Slot {
         }
     }
 
+    async fn connect_slot(&self, host: &Url, slot: u32) -> Result<()> {
+        info!(
+            target: "net::outbound_session::try_connect()",
+            "[P2P] Connecting outbound slot #{} [{}]",
+            slot, host,
+        );
+
+        dnetev!(self, OutboundSlotConnecting, {
+            slot: slot,
+            addr: host.clone(),
+        });
+
+        let (addr, channel) = match self.try_connect(host.clone()).await {
+            Ok(connect_info) => connect_info,
+            Err(err) => {
+                error!(
+                    target: "net::outbound_session",
+                    "[P2P] Outbound slot #{} connection failed: {}",
+                    slot, err,
+                );
+
+                dnetev!(self, OutboundSlotDisconnected, {
+                    slot,
+                    err: err.to_string()
+                });
+
+                self.channel_id.store(0, Ordering::Relaxed);
+                return Err(err.into())
+            }
+        };
+
+        info!(
+            target: "net::outbound_session::try_connect()",
+            "[P2P] Outbound slot #{} connected [{}]",
+            slot, addr
+        );
+
+        dnetev!(self, OutboundSlotConnected, {
+            slot: self.slot,
+            addr: addr.clone(),
+            channel_id: channel.info.id
+        });
+
+        let stop_sub = channel.subscribe_stop().await.expect("Channel should not be stopped");
+        // Setup new channel
+        if let Err(err) = self.setup_channel(host.clone(), channel.clone()).await {
+            info!(
+                target: "net::outbound_session",
+                "[P2P] Outbound slot #{} disconnected: {}",
+                slot, err
+            );
+
+            dnetev!(self, OutboundSlotDisconnected, {
+                slot: self.slot,
+                err: err.to_string()
+            });
+
+            self.channel_id.store(0, Ordering::Relaxed);
+            return Err(err.into())
+        }
+
+        self.channel_id.store(channel.info.id, Ordering::Relaxed);
+
+        // Wait for channel to close
+        stop_sub.receive().await;
+        self.channel_id.store(0, Ordering::Relaxed);
+
+        Ok(())
+    }
+
     //async fn run(self: Arc<Self>) {
     //    // This is the main outbound connection loop where we try to establish
     //    // a connection in the slot. The `try_connect` function will block in

src/net/tests.rs

@@ -16,7 +16,7 @@
  * along with this program.  If not, see <https://www.gnu.org/licenses/>.
  */
 
-// cargo +nightly test --release --all-features --lib p2p -- --include-ignored
+// cargo +nightly test --release --features=net --lib p2p -- --include-ignored
 
 use std::sync::Arc;
 
@@ -34,30 +34,9 @@ use crate::{
 const N_NODES: usize = 5;
 const N_CONNS: usize = 2;
 
-// TODO: test whitelist propagation between peers
-// TODO: test whitelist propagation from lilith to peers
-// TODO: test greylist storage and sorting
-// TODO: test greylist/ whitelist refining and refreshing
 #[test]
 fn p2p_test() {
     let mut cfg = simplelog::ConfigBuilder::new();
-    //cfg.add_filter_ignore("sled".to_string());
-    //cfg.add_filter_ignore("net::channel::subscribe_stop()".to_string());
-    //cfg.add_filter_ignore("net::hosts".to_string());
-    //cfg.add_filter_ignore("net::session".to_string());
-    //cfg.add_filter_ignore("net::message_subscriber".to_string());
-    //cfg.add_filter_ignore("net::protocol_ping".to_string());
-    //cfg.add_filter_ignore("net::protocol_version".to_string());
-    //cfg.add_filter_ignore("net::protocol_jobs_manager".to_string());
-    //cfg.add_filter_ignore("net::protocol_registry".to_string());
-    //cfg.add_filter_ignore("net::channel::send()".to_string());
-    //cfg.add_filter_ignore("net::channel::start()".to_string());
-    //cfg.add_filter_ignore("net::channel::stop()".to_string());
-    //cfg.add_filter_ignore("net::channel::handle_stop()".to_string());
-    //cfg.add_filter_ignore("net::channel::subscribe_msg()".to_string());
-    //cfg.add_filter_ignore("net::channel::main_receive_loop()".to_string());
-    //cfg.add_filter_ignore("net::greylist_refinery::run()".to_string());
-    //cfg.add_filter_ignore("net::outbound_session::try_connect()".to_string());
 
     simplelog::TermLogger::init(
         //simplelog::LevelFilter::Info,
@@ -126,16 +105,18 @@ async fn hostlist_propagation(ex: Arc<Executor<'static>>) {
     }
 
     info!("Waiting until all peers connect");
-    sleep(30).await;
+    sleep(60).await;
 
     info!("Inspecting hostlists...");
     for p2p in p2p_instances.iter() {
         let hosts = p2p.hosts();
-        assert!(!hosts.is_empty_greylist().await);
+        //assert!(!hosts.is_empty_greylist().await);
         //assert!(!hosts.is_empty_whitelist().await);
+        //assert!(!hosts.is_empty_anchorlist().await);
 
         let greylist = hosts.greylist.read().await;
         let whitelist = hosts.whitelist.read().await;
+        let anchorlist = hosts.anchorlist.read().await;
 
         info!("Node {}", p2p.settings().node_id);
         for (i, (url, last_seen)) in greylist.iter().enumerate() {
@@ -145,6 +126,10 @@ async fn hostlist_propagation(ex: Arc<Executor<'static>>) {
         for (i, (url, last_seen)) in whitelist.iter().enumerate() {
             info!("Whitelist entry {}: {}, {}", i, url, last_seen);
         }
+
+        for (i, (url, last_seen)) in anchorlist.iter().enumerate() {
+            info!("Anchorlist entry {}: {}, {}", i, url, last_seen);
+        }
     }
 
     // Stop the P2P network