dht, fud: integrate direct session, add inbound timeout, use DHT events

bin/fud/fud/fud_config.toml

@@ -51,6 +51,9 @@ btc_electrum_nodes = [
 ## Timeout in seconds
 #dht_timeout = 5
 
+## Timeout in seconds for inbound connections
+#dht_inbound_timeout = 30
+
 # JSON-RPC settings
 [rpc]
 # JSON-RPC listen URL

bin/fud/fud/src/dht.rs

@@ -19,8 +19,8 @@
 use std::sync::Arc;
 
 use async_trait::async_trait;
-use num_bigint::BigUint;
-use rand::{rngs::OsRng, Rng};
+use smol::lock::Mutex;
+use tinyjson::JsonValue;
 use tracing::debug;
 use url::Url;
 
@@ -28,18 +28,16 @@ use darkfi::{
     dht::{impl_dht_node_defaults, Dht, DhtHandler, DhtLookupReply, DhtNode},
     geode::hash_to_string,
     net::ChannelPtr,
+    rpc::util::json_map,
     util::time::Timestamp,
-    Error, Result,
+    Result,
 };
-use darkfi_sdk::crypto::schnorr::SchnorrPublic;
 use darkfi_serial::{SerialDecodable, SerialEncodable};
 
 use crate::{
     pow::VerifiableNodeData,
-    proto::{
-        FudAnnounce, FudFindNodesReply, FudFindNodesRequest, FudFindSeedersReply,
-        FudFindSeedersRequest, FudPingReply, FudPingRequest,
-    },
+    proto::{FudAnnounce, FudNodesReply, FudNodesRequest, FudSeedersReply, FudSeedersRequest},
+    util::receive_resource_msg,
     Fud,
 };
 
@@ -59,6 +57,20 @@ impl DhtNode for FudNode {
     }
 }
 
+impl From<FudNode> for JsonValue {
+    fn from(node: FudNode) -> JsonValue {
+        json_map([
+            ("id", JsonValue::String(hash_to_string(&node.id()))),
+            (
+                "addresses",
+                JsonValue::Array(
+                    node.addresses.iter().map(|addr| JsonValue::String(addr.to_string())).collect(),
+                ),
+            ),
+        ])
+    }
+}
+
 /// The values of the DHT are `Vec<FudSeeder>`, mapping resource hashes to lists of [`FudSeeder`]s
 #[derive(Debug, Clone, SerialEncodable, SerialDecodable, Eq)]
 pub struct FudSeeder {
@@ -78,6 +90,15 @@ impl PartialEq for FudSeeder {
     }
 }
 
+impl From<FudSeeder> for JsonValue {
+    fn from(seeder: FudSeeder) -> JsonValue {
+        json_map([
+            ("key", JsonValue::String(hash_to_string(&seeder.key))),
+            ("node", seeder.node.into()),
+        ])
+    }
+}
+
 /// [`DhtHandler`] implementation for fud
 #[async_trait]
 impl DhtHandler for Fud {
@@ -105,98 +126,74 @@ impl DhtHandler for Fud {
     }
 
     async fn ping(&self, channel: ChannelPtr) -> Result<FudNode> {
-        debug!(target: "fud::DhtHandler::ping()", "Sending ping to channel {}", channel.info.id);
-        let msg_subsystem = channel.message_subsystem();
-        msg_subsystem.add_dispatch::<FudPingReply>().await;
-        let msg_subscriber = channel.subscribe_msg::<FudPingReply>().await.unwrap();
-
-        // Send `FudPingRequest`
-        let mut rng = OsRng;
-        let request = FudPingRequest { random: rng.gen() };
-        channel.send(&request).await?;
+        let lock_map = self.ping_locks.clone();
+        let mut locks = lock_map.lock().await;
 
-        // Wait for `FudPingReply`
-        let reply = msg_subscriber.receive_with_timeout(self.dht().settings.timeout).await;
-        msg_subscriber.unsubscribe().await;
-        let reply = reply?;
+        // Get or create the lock
+        let lock = if let Some(lock) = locks.get(&channel.info.id) {
+            lock.clone()
+        } else {
+            let lock = Arc::new(Mutex::new(None));
+            locks.insert(channel.info.id, lock.clone());
+            lock
+        };
+        drop(locks);
 
-        // Verify the signature
-        if !reply.node.data.public_key.verify(&request.random.to_be_bytes(), &reply.sig) {
-            channel.ban().await;
-            return Err(Error::InvalidSignature)
-        }
+        // Acquire the lock
+        let mut result = lock.lock().await;
 
-        // Verify PoW
-        if let Err(e) = self.pow.write().await.verify_node(&reply.node.data).await {
-            channel.ban().await;
-            return Err(e)
+        if let Some(res) = result.clone() {
+            return res
         }
 
-        Ok(reply.node.clone())
+        // Do the actual pinging process
+        let ping_result = self.do_ping(channel.clone()).await;
+        *result = Some(ping_result.clone());
+        ping_result
     }
 
-    // TODO: Optimize this
-    async fn on_new_node(&self, node: &FudNode) -> Result<()> {
-        debug!(target: "fud::DhtHandler::on_new_node()", "New node {}", hash_to_string(&node.id()));
-
-        // If this is the first node we know about, then bootstrap and announce our files
-        if !self.dht.is_bootstrapped().await {
-            let _ = self.init().await;
-        }
-
-        // Send keys that are closer to this node than we are
-        let self_id = self.node_data.read().await.id();
-        let channel = self.dht.get_channel(node, None).await?;
-        for (key, seeders) in self.dht.hash_table.read().await.iter() {
-            let node_distance = BigUint::from_bytes_be(&self.dht().distance(key, &node.id()));
-            let self_distance = BigUint::from_bytes_be(&self.dht().distance(key, &self_id));
-            if node_distance <= self_distance {
-                let _ = channel.send(&FudAnnounce { key: *key, seeders: seeders.clone() }).await;
-            }
-        }
-        self.dht.cleanup_channel(channel).await;
+    async fn store(
+        &self,
+        channel: ChannelPtr,
+        key: &blake3::Hash,
+        value: &Vec<FudSeeder>,
+    ) -> Result<()> {
+        debug!(target: "fud::DhtHandler::store()", "Announcing {} to {}", hash_to_string(key), channel.display_address());
 
-        Ok(())
+        channel.send(&FudAnnounce { key: *key, seeders: value.clone() }).await
     }
 
-    async fn find_nodes(&self, node: &FudNode, key: &blake3::Hash) -> Result<Vec<FudNode>> {
-        debug!(target: "fud::DhtHandler::find_nodes()", "Fetching nodes close to {} from node {}", hash_to_string(key), hash_to_string(&node.id()));
+    async fn find_nodes(&self, channel: ChannelPtr, key: &blake3::Hash) -> Result<Vec<FudNode>> {
+        debug!(target: "fud::DhtHandler::find_nodes()", "Fetching nodes close to {} from node {}", hash_to_string(key), channel.display_address());
 
-        let channel = self.dht.get_channel(node, None).await?;
-        let msg_subsystem = channel.message_subsystem();
-        msg_subsystem.add_dispatch::<FudFindNodesReply>().await;
-        let msg_subscriber_nodes = channel.subscribe_msg::<FudFindNodesReply>().await.unwrap();
+        let msg_subscriber_nodes = channel.subscribe_msg::<FudNodesReply>().await.unwrap();
 
-        let request = FudFindNodesRequest { key: *key };
+        let request = FudNodesRequest { key: *key };
         channel.send(&request).await?;
 
-        let reply = msg_subscriber_nodes.receive_with_timeout(self.dht().settings.timeout).await;
+        let reply =
+            receive_resource_msg(&msg_subscriber_nodes, *key, self.dht().settings.timeout).await;
 
         msg_subscriber_nodes.unsubscribe().await;
-        self.dht.cleanup_channel(channel).await;
 
         Ok(reply?.nodes.clone())
     }
 
     async fn find_value(
         &self,
-        node: &FudNode,
+        channel: ChannelPtr,
         key: &blake3::Hash,
     ) -> Result<DhtLookupReply<FudNode, Vec<FudSeeder>>> {
-        debug!(target: "fud::DhtHandler::find_value()", "Fetching value {} from node {}", hash_to_string(key), hash_to_string(&node.id()));
+        debug!(target: "fud::DhtHandler::find_value()", "Fetching value {} (or close nodes) from {}", hash_to_string(key), channel.display_address());
 
-        let channel = self.dht.get_channel(node, None).await?;
-        let msg_subsystem = channel.message_subsystem();
-        msg_subsystem.add_dispatch::<FudFindSeedersReply>().await;
-        let msg_subscriber = channel.subscribe_msg::<FudFindSeedersReply>().await.unwrap();
+        let msg_subscriber = channel.subscribe_msg::<FudSeedersReply>().await.unwrap();
 
-        let request = FudFindSeedersRequest { key: *key };
+        let request = FudSeedersRequest { key: *key };
         channel.send(&request).await?;
 
-        let recv = msg_subscriber.receive_with_timeout(self.dht().settings.timeout).await;
+        let recv = receive_resource_msg(&msg_subscriber, *key, self.dht().settings.timeout).await;
 
         msg_subscriber.unsubscribe().await;
-        self.dht.cleanup_channel(channel).await;
 
         let rep = recv?;
         Ok(DhtLookupReply::NodesAndValue(rep.nodes.clone(), rep.seeders.clone()))

bin/fud/fud/src/download.rs

@@ -30,29 +30,31 @@ use rand::{
 use tracing::{error, info, warn};
 
 use darkfi::{
-    dht::DhtNode,
+    dht::{event::DhtEvent, DhtNode},
     geode::{hash_to_string, ChunkedStorage},
     net::ChannelPtr,
-    system::Subscription,
     Error, Result,
 };
 use darkfi_serial::serialize_async;
 
 use crate::{
     event::{self, notify_event, FudEvent},
-    proto::{FudChunkReply, FudDirectoryReply, FudFileReply, FudFindRequest, FudNotFound},
-    util::create_all_files,
+    proto::{
+        FudChunkNotFound, FudChunkReply, FudChunkRequest, FudDirectoryReply, FudFileReply,
+        FudMetadataNotFound, FudMetadataRequest,
+    },
+    util::{create_all_files, receive_resource_msg},
     Fud, FudSeeder, ResourceStatus, ResourceType, Scrap,
 };
 
-/// Receive seeders from a subscription, and execute an async expression for
-/// each deduplicated seeder once (seeder order is random).
+/// Receive seeders from a DHT events subscription, and execute an async
+/// expression for each deduplicated seeder once (seeder order is random).
 /// It will keep going until the expression returns `Ok(())`, or there are
 /// no more seeders.
 /// It has an optional `favored_seeder` argument that will be tried first if
 /// specified.
 macro_rules! seeders_loop {
-    ($seeders_sub:expr, $favored_seeder:expr, $code:expr) => {
+    ($key:expr, $fud:expr, $favored_seeder:expr, $code:expr) => {
         let mut queried_seeders: HashSet<blake3::Hash> = HashSet::new();
         let mut is_done = false;
 
@@ -65,13 +67,20 @@ macro_rules! seeders_loop {
             }
         }
 
-        // Try other seeders using the subscription
+        // Try other seeders using the DHT subscription
+        let dht_sub = $fud.dht.subscribe().await;
         while !is_done {
-            let rep = $seeders_sub.receive().await;
-            if rep.is_none() {
-                break; // None means the lookup is done
+            let event = dht_sub.receive().await;
+            if event.key() != Some($key) {
+                continue // Ignore this event if it's not about the right key
             }
-            let seeders = rep.unwrap().clone();
+            if let DhtEvent::ValueLookupCompleted { .. } = event {
+                break // Lookup is done
+            }
+            if !matches!(event, DhtEvent::ValueFound { .. }) {
+                continue // Ignore this event as it's not a ValueFound
+            }
+            let seeders = event.into_value().unwrap();
             let mut shuffled_seeders = {
                 let mut vec: Vec<_> = seeders.iter().cloned().collect();
                 vec.shuffle(&mut OsRng);
@@ -81,21 +90,22 @@ macro_rules! seeders_loop {
             while let Some(seeder) = shuffled_seeders.pop() {
                 // Only use a seeder once
                 if queried_seeders.iter().any(|s| *s == seeder.node.id()) {
-                    continue;
+                    continue
                 }
                 queried_seeders.insert(seeder.node.id());
 
                 if $code(seeder).await.is_err() {
-                    continue;
+                    continue
                 }
 
                 is_done = true;
-                break;
+                break
             }
         }
+        dht_sub.unsubscribe().await;
     };
-    ($seeders_sub:expr, $code:expr) => {
-        seeders_loop!($seeders_sub, None, $code)
+    ($key:expr, $fud:expr, $code:expr) => {
+        seeders_loop!($key, $fud, None, $code)
     };
 }
 
@@ -117,14 +127,13 @@ pub async fn fetch_chunks(
     fud: &Fud,
     hash: &blake3::Hash,
     chunked: &mut ChunkedStorage,
-    seeders_sub: &Subscription<Option<Vec<FudSeeder>>>,
     favored_seeder: Option<FudSeeder>,
     chunks: &mut HashSet<blake3::Hash>,
 ) -> Result<()> {
     let mut ctx = ChunkFetchContext { fud, hash, chunked, chunks };
 
-    seeders_loop!(seeders_sub, favored_seeder, async |seeder: FudSeeder| -> Result<()> {
-        let channel = match fud.dht.get_channel(&seeder.node, Some(*hash)).await {
+    seeders_loop!(hash, fud, favored_seeder, async |seeder: FudSeeder| -> Result<()> {
+        let (channel, _) = match fud.dht.get_channel(&seeder.node).await {
             Ok(channel) => channel,
             Err(e) => {
                 warn!(target: "fud::download::fetch_chunks()", "Could not get a channel for node {}: {e}", hash_to_string(&seeder.node.id()));
@@ -181,20 +190,19 @@ async fn fetch_chunk(
     chunks_to_query.remove(&chunk_hash);
 
     let start_time = Instant::now();
-    let msg_subsystem = channel.message_subsystem();
-    msg_subsystem.add_dispatch::<FudChunkReply>().await;
-    msg_subsystem.add_dispatch::<FudNotFound>().await;
     let msg_subscriber_chunk = channel.subscribe_msg::<FudChunkReply>().await.unwrap();
-    let msg_subscriber_notfound = channel.subscribe_msg::<FudNotFound>().await.unwrap();
+    let msg_subscriber_notfound = channel.subscribe_msg::<FudChunkNotFound>().await.unwrap();
 
-    let send_res = channel.send(&FudFindRequest { info: Some(*ctx.hash), key: chunk_hash }).await;
+    let send_res = channel.send(&FudChunkRequest { resource: *ctx.hash, chunk: chunk_hash }).await;
     if let Err(e) = send_res {
-        warn!(target: "fud::download::fetch_chunk()", "Error while sending FudFindRequest: {e}");
+        warn!(target: "fud::download::fetch_chunk()", "Error while sending FudChunkRequest: {e}");
         return ChunkFetchControl::NextSeeder;
     }
 
-    let chunk_recv = msg_subscriber_chunk.receive_with_timeout(ctx.fud.chunk_timeout).fuse();
-    let notfound_recv = msg_subscriber_notfound.receive_with_timeout(ctx.fud.chunk_timeout).fuse();
+    let chunk_recv =
+        receive_resource_msg(&msg_subscriber_chunk, *ctx.hash, ctx.fud.chunk_timeout).fuse();
+    let notfound_recv =
+        receive_resource_msg(&msg_subscriber_notfound, *ctx.hash, ctx.fud.chunk_timeout).fuse();
 
     pin_mut!(chunk_recv, notfound_recv);
 
@@ -315,29 +323,19 @@ enum MetadataFetchReply {
 /// 1. Wait for seeders from the subscription
 /// 2. Request the metadata from the seeders
 /// 3. Insert metadata to geode using the reply
-pub async fn fetch_metadata(
-    fud: &Fud,
-    hash: &blake3::Hash,
-    seeders_sub: &Subscription<Option<Vec<FudSeeder>>>,
-    path: &Path,
-) -> Result<FudSeeder> {
+pub async fn fetch_metadata(fud: &Fud, hash: &blake3::Hash, path: &Path) -> Result<FudSeeder> {
     let mut result: Option<(FudSeeder, MetadataFetchReply)> = None;
 
-    seeders_loop!(seeders_sub, async |seeder: FudSeeder| -> Result<()> {
-        let channel = fud.dht.get_channel(&seeder.node, Some(*hash)).await?;
-        let msg_subsystem = channel.message_subsystem();
-        msg_subsystem.add_dispatch::<FudChunkReply>().await;
-        msg_subsystem.add_dispatch::<FudFileReply>().await;
-        msg_subsystem.add_dispatch::<FudDirectoryReply>().await;
-        msg_subsystem.add_dispatch::<FudNotFound>().await;
+    seeders_loop!(hash, fud, async |seeder: FudSeeder| -> Result<()> {
+        let (channel, _) = fud.dht.get_channel(&seeder.node).await?;
         let msg_subscriber_chunk = channel.subscribe_msg::<FudChunkReply>().await.unwrap();
         let msg_subscriber_file = channel.subscribe_msg::<FudFileReply>().await.unwrap();
         let msg_subscriber_dir = channel.subscribe_msg::<FudDirectoryReply>().await.unwrap();
-        let msg_subscriber_notfound = channel.subscribe_msg::<FudNotFound>().await.unwrap();
+        let msg_subscriber_notfound = channel.subscribe_msg::<FudMetadataNotFound>().await.unwrap();
 
-        let send_res = channel.send(&FudFindRequest { info: None, key: *hash }).await;
+        let send_res = channel.send(&FudMetadataRequest { resource: *hash }).await;
         if let Err(e) = send_res {
-            warn!(target: "fud::download::fetch_metadata()", "Error while sending FudFindRequest: {e}");
+            warn!(target: "fud::download::fetch_metadata()", "Error while sending FudMetadataRequest: {e}");
             msg_subscriber_chunk.unsubscribe().await;
             msg_subscriber_file.unsubscribe().await;
             msg_subscriber_dir.unsubscribe().await;
@@ -346,10 +344,12 @@ pub async fn fetch_metadata(
             return Err(e)
         }
 
-        let chunk_recv = msg_subscriber_chunk.receive_with_timeout(fud.chunk_timeout).fuse();
-        let file_recv = msg_subscriber_file.receive_with_timeout(fud.chunk_timeout).fuse();
-        let dir_recv = msg_subscriber_dir.receive_with_timeout(fud.chunk_timeout).fuse();
-        let notfound_recv = msg_subscriber_notfound.receive_with_timeout(fud.chunk_timeout).fuse();
+        let chunk_recv =
+            receive_resource_msg(&msg_subscriber_chunk, *hash, fud.chunk_timeout).fuse();
+        let file_recv = receive_resource_msg(&msg_subscriber_file, *hash, fud.chunk_timeout).fuse();
+        let dir_recv = receive_resource_msg(&msg_subscriber_dir, *hash, fud.chunk_timeout).fuse();
+        let notfound_recv =
+            receive_resource_msg(&msg_subscriber_notfound, *hash, fud.chunk_timeout).fuse();
 
         pin_mut!(chunk_recv, file_recv, dir_recv, notfound_recv);
 
@@ -439,7 +439,7 @@ pub async fn fetch_metadata(
     // At this point the reply content is already verified
     let (seeder, reply) = result.unwrap();
     match reply {
-        MetadataFetchReply::Directory(FudDirectoryReply { files, chunk_hashes }) => {
+        MetadataFetchReply::Directory(FudDirectoryReply { files, chunk_hashes, .. }) => {
             // Convert all file paths from String to PathBuf
             let mut files: Vec<_> =
                 files.into_iter().map(|(path_str, size)| (PathBuf::from(path_str), size)).collect();
@@ -450,14 +450,14 @@ pub async fn fetch_metadata(
                 return Err(e)
             }
         }
-        MetadataFetchReply::File(FudFileReply { chunk_hashes }) => {
+        MetadataFetchReply::File(FudFileReply { chunk_hashes, .. }) => {
             if let Err(e) = fud.geode.insert_metadata(hash, &chunk_hashes, &[]).await {
                 error!(target: "fud::download::fetch_metadata()", "Failed inserting file {} to Geode: {e}", hash_to_string(hash));
                 return Err(e)
             }
         }
         // Looked for a file but got a chunk: the entire file fits in a single chunk
-        MetadataFetchReply::Chunk(FudChunkReply { chunk }) => {
+        MetadataFetchReply::Chunk(FudChunkReply { chunk, .. }) => {
             info!(target: "fud::download::fetch_metadata()", "File fits in a single chunk");
             let chunk_hash = blake3::hash(&chunk);
             if let Err(e) = fud.geode.insert_metadata(hash, &[chunk_hash], &[]).await {

bin/fud/fud/src/lib.rs

@@ -21,25 +21,32 @@ use std::{
     io::ErrorKind,
     path::{Path, PathBuf},
     sync::Arc,
+    time::Duration,
 };
 
+use rand::{rngs::OsRng, Rng};
 use sled_overlay::sled;
 use smol::{
     channel,
     fs::{self, OpenOptions},
-    lock::RwLock,
+    lock::{Mutex, RwLock},
 };
-use tracing::{error, info, warn};
+use tracing::{debug, error, info, warn};
 
 use darkfi::{
-    dht::{tasks as dht_tasks, Dht, DhtHandler, DhtSettings},
+    dht::{
+        event::DhtEvent, tasks as dht_tasks, Dht, DhtHandler, DhtNode, DhtSettings, HostCacheItem,
+    },
     geode::{hash_to_string, ChunkedStorage, FileSequence, Geode, MAX_CHUNK_SIZE},
-    net::P2pPtr,
-    system::{ExecutorPtr, Publisher, PublisherPtr, StoppableTask},
+    net::{
+        session::{SESSION_DIRECT, SESSION_INBOUND, SESSION_MANUAL, SESSION_OUTBOUND},
+        ChannelPtr, P2pPtr,
+    },
+    system::{timeout::timeout, ExecutorPtr, PublisherPtr, StoppableTask},
     util::{path::expand_path, time::Timestamp},
     Error, Result,
 };
-use darkfi_sdk::crypto::SecretKey;
+use darkfi_sdk::crypto::{schnorr::SchnorrPublic, SecretKey};
 use darkfi_serial::{deserialize_async, serialize_async};
 
 /// P2P protocols
@@ -91,10 +98,18 @@ use download::{fetch_chunks, fetch_metadata};
 pub mod dht;
 use dht::FudSeeder;
 
+use crate::{
+    dht::FudNode,
+    pow::PowSettings,
+    proto::{FudPingReply, FudPingRequest},
+};
+
 const SLED_PATH_TREE: &[u8] = b"_fud_paths";
 const SLED_FILE_SELECTION_TREE: &[u8] = b"_fud_file_selections";
 const SLED_SCRAP_TREE: &[u8] = b"_fud_scraps";
 
+type PingLock = Arc<Mutex<Option<Result<FudNode>>>>;
+
 pub struct Fud {
     /// Our own [`VerifiableNodeData`]
     pub node_data: Arc<RwLock<VerifiableNodeData>>,
@@ -125,6 +140,8 @@ pub struct Fud {
     /// is not saved to the filesystem in the downloaded files.
     /// "chunk/scrap hash -> chunk content"
     scrap_tree: sled::Tree,
+    /// Locks that prevent pinging the same channel multiple times at once.
+    ping_locks: Arc<Mutex<HashMap<u32, PingLock>>>,
     /// Get requests sender
     get_tx: channel::Sender<(blake3::Hash, PathBuf, FileSelection)>,
     /// Get requests receiver
@@ -134,9 +151,9 @@ pub struct Fud {
     /// Put requests receiver
     put_rx: channel::Receiver<PathBuf>,
     /// Lookup requests sender
-    lookup_tx: channel::Sender<(blake3::Hash, PublisherPtr<Option<Vec<FudSeeder>>>)>,
+    lookup_tx: channel::Sender<blake3::Hash>,
     /// Lookup requests receiver
-    lookup_rx: channel::Receiver<(blake3::Hash, PublisherPtr<Option<Vec<FudSeeder>>>)>,
+    lookup_rx: channel::Receiver<blake3::Hash>,
     /// Currently active downloading tasks (running the `fud.fetch_resource()` method)
     fetch_tasks: Arc<RwLock<HashMap<blake3::Hash, Arc<StoppableTask>>>>,
     /// Currently active put tasks (running the `fud.insert_resource()` method)
@@ -161,6 +178,16 @@ impl Fud {
         event_publisher: PublisherPtr<FudEvent>,
         executor: ExecutorPtr,
     ) -> Result<Arc<Self>> {
+        let dht_settings: DhtSettings = settings.dht.into();
+        let net_settings_lock = p2p.settings();
+        let mut net_settings = net_settings_lock.write().await;
+        // We do not need any outbound slot
+        net_settings.outbound_connections = 0;
+        // Default GetAddrsMessage's `max` is dht's `k`
+        net_settings.getaddrs_max =
+            Some(net_settings.getaddrs_max.unwrap_or(dht_settings.k.min(u32::MAX as usize) as u32));
+        drop(net_settings);
+
         let basedir = expand_path(&settings.base_dir)?;
         let downloads_path = match settings.downloads_path {
             Some(downloads_path) => expand_path(&downloads_path)?,
@@ -168,8 +195,11 @@ impl Fud {
         };
 
         // Run the PoW and generate a `VerifiableNodeData`
-        let mut pow = FudPow::new(settings.pow.into(), executor.clone());
-        pow.bitcoin_hash_cache.update().await?; // Fetch BTC block hashes
+        let pow_settings: PowSettings = settings.pow.into();
+        let mut pow = FudPow::new(pow_settings.clone(), executor.clone());
+        if pow_settings.btc_enabled {
+            pow.bitcoin_hash_cache.update().await?; // Fetch BTC block hashes
+        }
         let (node_data, secret_key) = pow.generate_node().await?;
         info!(target: "fud::new()", "Your node ID: {}", hash_to_string(&node_data.id()));
 
@@ -178,7 +208,6 @@ impl Fud {
         let geode = Geode::new(&basedir).await?;
 
         // DHT
-        let dht_settings: DhtSettings = settings.dht.into();
         let dht: Arc<Dht<Fud>> =
             Arc::new(Dht::<Fud>::new(&dht_settings, p2p.clone(), executor.clone()).await);
 
@@ -197,6 +226,7 @@ impl Fud {
             file_selection_tree: sled_db.open_tree(SLED_FILE_SELECTION_TREE)?,
             scrap_tree: sled_db.open_tree(SLED_SCRAP_TREE)?,
             resources: Arc::new(RwLock::new(HashMap::new())),
+            ping_locks: Arc::new(Mutex::new(HashMap::new())),
             get_tx,
             get_rx,
             put_tx,
@@ -220,19 +250,26 @@ impl Fud {
         let mut tasks = self.tasks.write().await;
         start_task!(self, "get", tasks::get_task, tasks);
         start_task!(self, "put", tasks::put_task, tasks);
+        start_task!(self, "events", tasks::handle_dht_events, tasks);
+        start_task!(self, "DHT events", dht_tasks::events_task::<Fud>, tasks);
         start_task!(self, "DHT channel", dht_tasks::channel_task::<Fud>, tasks);
+        start_task!(self, "DHT cleanup channels", dht_tasks::cleanup_channels_task::<Fud>, tasks);
+        start_task!(self, "DHT add node", dht_tasks::add_node_task::<Fud>, tasks);
+        start_task!(self, "DHT refinery", dht_tasks::dht_refinery_task::<Fud>, tasks);
+        start_task!(
+            self,
+            "DHT disconnect inbounds",
+            dht_tasks::disconnect_inbounds_task::<Fud>,
+            tasks
+        );
         start_task!(self, "lookup", tasks::lookup_task, tasks);
         start_task!(self, "announce", tasks::announce_seed_task, tasks);
         start_task!(self, "node ID", tasks::node_id_task, tasks);
     }
 
-    /// Bootstrap the DHT, verify our resources, add ourselves to
-    /// the seeders (`dht.hash_table`) for the resources we already have,
-    /// announce our files.
+    /// Verify our resources, add ourselves to the seeders (`dht.hash_table`)
+    /// for the resources we already have, announce our resources.
     async fn init(&self) -> Result<()> {
-        info!(target: "fud::init()", "Bootstrapping the DHT...");
-        self.dht.bootstrap().await;
-
         info!(target: "fud::init()", "Finding resources...");
         let mut resources_write = self.resources.write().await;
         for result in self.path_tree.iter() {
@@ -364,6 +401,110 @@ impl Fud {
         }
     }
 
+    async fn do_ping(&self, channel: ChannelPtr) -> Result<FudNode> {
+        debug!(target: "fud::DhtHandler::do_ping()", "Sending ping to {}", channel.display_address());
+
+        let dht = self.dht();
+
+        // Setup `FudPingReply` subscriber
+        let msg_subscriber = channel.subscribe_msg::<FudPingReply>().await.unwrap();
+
+        // Send `FudPingRequest`
+        let mut rng = OsRng;
+        let request = FudPingRequest { random: rng.gen() };
+        if channel.is_stopped() {
+            return Err(Error::ChannelStopped)
+        }
+        channel.send(&request).await?;
+
+        // Wait for `FudPingReply`
+        let reply = msg_subscriber.receive_with_timeout(dht.settings.timeout).await;
+        msg_subscriber.unsubscribe().await;
+        let reply = reply?;
+        let node = &reply.node;
+
+        // Verify the signature
+        if !node.data.public_key.verify(&request.random.to_be_bytes(), &reply.sig) {
+            warn!(target: "fud::do_ping()", "Received an invalid signature while pinging {}", channel.display_address());
+            self.dht
+                .event_publisher
+                .notify(DhtEvent::PingReceived {
+                    from: channel.clone(),
+                    result: Err(Error::InvalidSignature),
+                })
+                .await;
+            self.dht.cleanup_channel(channel.clone()).await;
+            channel.ban().await;
+            return Err(Error::InvalidSignature)
+        }
+
+        // Verify PoW
+        if let Err(e) = self.pow.write().await.verify_node(&node.data).await {
+            warn!(target: "fud::do_ping()", "Received an invalid PoW while pinging {}: {e}", channel.display_address());
+            self.dht
+                .event_publisher
+                .notify(DhtEvent::PingReceived { from: channel.clone(), result: Err(e.clone()) })
+                .await;
+            self.dht.cleanup_channel(channel.clone()).await;
+            channel.ban().await;
+            return Err(e)
+        }
+        self.dht
+            .event_publisher
+            .notify(DhtEvent::PingReceived { from: channel.clone(), result: Ok(node.id()) })
+            .await;
+
+        if channel.session_type_id() & (SESSION_OUTBOUND | SESSION_DIRECT | SESSION_MANUAL) != 0 {
+            // Wait for the other node to ping us
+            let ping_timeout = Duration::from_secs(10);
+
+            if let Err(e) = timeout(ping_timeout, dht.wait_fully_pinged(channel.info.id)).await {
+                dht.cleanup_channel(channel).await;
+                return Err(e.into())
+            }
+
+            let mut host_cache = dht.host_cache.write().await;
+
+            // If we had another node id for this host in our cache, remove
+            // the old one from the buckets and seeders
+            if let Some(cached) = host_cache.get(channel.address()) {
+                if cached.node_id != node.id() {
+                    dht.remove_node(&cached.node_id).await;
+
+                    for (_, seeders) in dht.hash_table.write().await.iter_mut() {
+                        seeders.retain(|seeder| seeder.node.id() != cached.node_id);
+                    }
+                }
+            }
+
+            // Update host cache
+            host_cache.insert(
+                channel.address().clone(),
+                HostCacheItem { last_ping: Timestamp::current_time(), node_id: node.id() },
+            );
+
+            drop(host_cache);
+
+            // Update our buckets
+            if !node.addresses().is_empty() {
+                dht.update_node(&node.clone(), channel.clone()).await;
+            }
+        } else if channel.session_type_id() & SESSION_INBOUND != 0 {
+            // If it's an inbound connection, verify that we can connect to at
+            // least one of the provided external addresses.
+            // This may try to create a new outbound channel and it will update
+            // our buckets if successful.
+            if let Ok((channel, _)) = dht.create_channel_to_node(node).await {
+                dht.cleanup_channel(channel).await;
+            }
+        }
+
+        // Update the channel cache
+        dht.add_channel_to_cache(channel.info.id, node).await;
+
+        Ok(node.clone())
+    }
+
     /// Verify if resources are complete and uncorrupted.
     /// If a resource is incomplete or corrupted, its status is changed to Incomplete.
     /// If a resource is complete, its status is changed to Seeding.
@@ -527,7 +668,6 @@ impl Fud {
         &self,
         hash: &blake3::Hash,
         path: &Path,
-        seeders_pub: PublisherPtr<Option<Vec<FudSeeder>>>,
     ) -> Result<(ChunkedStorage, Option<FudSeeder>)> {
         match self.geode.get(hash, path).await {
             // We already know the metadata
@@ -538,12 +678,10 @@ impl Fud {
             Err(Error::GeodeFileNotFound) => {
                 // Find nodes close to the file hash
                 info!(target: "fud::get_metadata()", "Requested metadata {} not found in Geode, triggering fetch", hash_to_string(hash));
-                let metadata_sub = seeders_pub.clone().subscribe().await;
-                self.lookup_tx.send((*hash, seeders_pub.clone())).await?;
+                self.lookup_tx.send(*hash).await?;
 
                 // Fetch resource metadata
-                let fetch_res = fetch_metadata(self, hash, &metadata_sub, path).await;
-                metadata_sub.unsubscribe().await;
+                let fetch_res = fetch_metadata(self, hash, path).await;
                 let seeder = fetch_res?;
                 Ok((self.geode.get(hash, path).await?, Some(seeder)))
             }
@@ -619,11 +757,8 @@ impl Fud {
         // Send a DownloadStarted event
         notify_event!(self, DownloadStarted, resource);
 
-        let seeders_pub = Publisher::new();
-        let seeders_sub = seeders_pub.clone().subscribe().await;
-
         // Try to get the chunked file or directory from geode
-        let metadata_result = self.get_metadata(hash, path, seeders_pub.clone()).await;
+        let metadata_result = self.get_metadata(hash, path).await;
 
         if let Err(e) = metadata_result {
             // Set resource status to `Incomplete` and send a `MetadataNotFound` event
@@ -747,19 +882,11 @@ impl Fud {
 
         // Start looking up seeders if we did not need to do it for the metadata
         if metadata_seeder.is_none() {
-            self.lookup_tx.send((*hash, seeders_pub)).await?;
+            self.lookup_tx.send(*hash).await?;
         }
 
         // Fetch missing chunks from seeders
-        let _ = fetch_chunks(
-            self,
-            hash,
-            &mut chunked,
-            &seeders_sub,
-            metadata_seeder,
-            &mut missing_chunks,
-        )
-        .await;
+        let _ = fetch_chunks(self, hash, &mut chunked, metadata_seeder, &mut missing_chunks).await;
 
         // Get chunked file from geode
         let mut chunked = self.geode.get(hash, path).await?;

bin/fud/fud/src/main.rs

@@ -24,7 +24,10 @@ use tracing::{debug, error, info, warn};
 
 use darkfi::{
     async_daemonize,
-    net::{session::SESSION_DEFAULT, P2p, Settings as NetSettings},
+    net::{
+        session::{SESSION_DIRECT, SESSION_INBOUND, SESSION_MANUAL},
+        P2p, Settings as NetSettings,
+    },
     rpc::{
         jsonrpc::JsonSubscriber,
         server::{listen_and_serve, RequestHandler},
@@ -54,6 +57,9 @@ async fn realmain(args: Args, ex: Arc<Executor<'static>>) -> Result<()> {
     let sled_db = sled::open(basedir.join("db"))?;
 
     info!(target: "fud", "Instantiating P2P network");
+    // We will use the peers defined in the settings as direct connections (instead of manual)
+    // let direct_peers = net_settings.peers.clone();
+    // net_settings.peers = vec![];
     let net_settings: NetSettings = args.net.into();
     let p2p = P2p::new(net_settings.clone(), ex.clone()).await?;
 
@@ -133,13 +139,15 @@ async fn realmain(args: Args, ex: Arc<Executor<'static>>) -> Result<()> {
     let registry = p2p.protocol_registry();
     let fud_ = fud.clone();
     registry
-        .register(SESSION_DEFAULT, move |channel, p2p| {
+        .register(SESSION_DIRECT | SESSION_INBOUND | SESSION_MANUAL, move |channel, p2p| {
             let fud_ = fud_.clone();
             async move { ProtocolFud::init(fud_, channel, p2p).await.unwrap() }
         })
         .await;
     p2p.clone().start().await?;
 
+    p2p.session_direct().start_peer_discovery();
+
     let p2p_settings_lock = p2p.settings();
     let p2p_settings = p2p_settings_lock.read().await;
     if p2p_settings.external_addrs.is_empty() {

bin/fud/fud/src/proto.rs

@@ -22,11 +22,12 @@ use std::{path::StripPrefixError, sync::Arc};
 use tracing::{debug, error, info};
 
 use darkfi::{
-    dht::DhtHandler,
+    dht::{event::DhtEvent, DhtHandler},
     geode::hash_to_string,
     impl_p2p_message,
     net::{
         metering::{MeteringConfiguration, DEFAULT_METERING_CONFIGURATION},
+        session::SESSION_INBOUND,
         ChannelPtr, Message, MessageSubscription, P2pPtr, ProtocolBase, ProtocolBasePtr,
         ProtocolJobsManager, ProtocolJobsManagerPtr,
     },
@@ -40,20 +41,42 @@ use crate::{
     Fud,
 };
 
+/// Trait for resource-specific messages.
+/// Adds a method to get the resource's hash from the message.
+pub trait ResourceMessage {
+    fn resource_hash(&self) -> blake3::Hash;
+}
+macro_rules! impl_resource_msg {
+    ($msg:ty, $field:ident) => {
+        impl ResourceMessage for $msg {
+            fn resource_hash(&self) -> blake3::Hash {
+                self.$field
+            }
+        }
+    };
+    ($msg:ty) => {
+        impl_resource_msg!($msg, resource);
+    };
+}
+
 /// Message representing a file reply from the network
 #[derive(Debug, Clone, SerialEncodable, SerialDecodable)]
 pub struct FudFileReply {
+    pub resource: blake3::Hash,
     pub chunk_hashes: Vec<blake3::Hash>,
 }
 impl_p2p_message!(FudFileReply, "FudFileReply", 0, 0, DEFAULT_METERING_CONFIGURATION);
+impl_resource_msg!(FudFileReply);
 
 /// Message representing a directory reply from the network
 #[derive(Debug, Clone, SerialEncodable, SerialDecodable)]
 pub struct FudDirectoryReply {
+    pub resource: blake3::Hash,
     pub chunk_hashes: Vec<blake3::Hash>,
     pub files: Vec<(String, u64)>, // Vec of (file path, file size)
 }
 impl_p2p_message!(FudDirectoryReply, "FudDirectoryReply", 0, 0, DEFAULT_METERING_CONFIGURATION);
+impl_resource_msg!(FudDirectoryReply);
 
 /// Message representing a node announcing a key on the network
 #[derive(Debug, Clone, SerialEncodable, SerialDecodable)]
@@ -66,15 +89,29 @@ impl_p2p_message!(FudAnnounce, "FudAnnounce", 0, 0, DEFAULT_METERING_CONFIGURATI
 /// Message representing a chunk reply from the network
 #[derive(Debug, Clone, SerialEncodable, SerialDecodable)]
 pub struct FudChunkReply {
+    pub resource: blake3::Hash,
     // TODO: This should be a chunk-sized array, but then we need padding?
     pub chunk: Vec<u8>,
 }
 impl_p2p_message!(FudChunkReply, "FudChunkReply", 0, 0, DEFAULT_METERING_CONFIGURATION);
+impl_resource_msg!(FudChunkReply);
 
-/// Message representing a chunk reply when a file is not found
+/// Message representing a reply when a metadata is not found
 #[derive(Debug, Clone, SerialEncodable, SerialDecodable)]
-pub struct FudNotFound;
-impl_p2p_message!(FudNotFound, "FudNotFound", 0, 0, DEFAULT_METERING_CONFIGURATION);
+pub struct FudMetadataNotFound {
+    pub resource: blake3::Hash,
+}
+impl_p2p_message!(FudMetadataNotFound, "FudMetadataNotFound", 0, 0, DEFAULT_METERING_CONFIGURATION);
+impl_resource_msg!(FudMetadataNotFound);
+
+/// Message representing a reply when a chunk is not found
+#[derive(Debug, Clone, SerialEncodable, SerialDecodable)]
+pub struct FudChunkNotFound {
+    pub resource: blake3::Hash,
+    pub chunk: blake3::Hash,
+}
+impl_p2p_message!(FudChunkNotFound, "FudChunkNotFound", 0, 0, DEFAULT_METERING_CONFIGURATION);
+impl_resource_msg!(FudChunkNotFound);
 
 /// Message representing a ping request on the network
 #[derive(Debug, Clone, SerialEncodable, SerialDecodable)]
@@ -87,61 +124,70 @@ impl_p2p_message!(FudPingRequest, "FudPingRequest", 0, 0, DEFAULT_METERING_CONFI
 #[derive(Debug, Clone, SerialEncodable, SerialDecodable)]
 pub struct FudPingReply {
     pub node: FudNode,
+    pub random: u64,
     /// Signature of the random u64 from the ping request
     pub sig: Signature,
 }
 impl_p2p_message!(FudPingReply, "FudPingReply", 0, 0, DEFAULT_METERING_CONFIGURATION);
 
-/// Message representing a find file/chunk request from the network
+/// Message representing a find file/directory request from the network
 #[derive(Debug, Clone, SerialEncodable, SerialDecodable)]
-pub struct FudFindRequest {
-    pub info: Option<blake3::Hash>,
-    pub key: blake3::Hash,
+pub struct FudMetadataRequest {
+    pub resource: blake3::Hash,
 }
-impl_p2p_message!(FudFindRequest, "FudFindRequest", 0, 0, DEFAULT_METERING_CONFIGURATION);
+impl_p2p_message!(FudMetadataRequest, "FudMetadataRequest", 0, 0, DEFAULT_METERING_CONFIGURATION);
+impl_resource_msg!(FudMetadataRequest);
+
+/// Message representing a find chunk request from the network
+#[derive(Debug, Clone, SerialEncodable, SerialDecodable)]
+pub struct FudChunkRequest {
+    pub resource: blake3::Hash,
+    pub chunk: blake3::Hash,
+}
+impl_p2p_message!(FudChunkRequest, "FudChunkRequest", 0, 0, DEFAULT_METERING_CONFIGURATION);
+impl_resource_msg!(FudChunkRequest);
 
 /// Message representing a find nodes request on the network
 #[derive(Debug, Clone, SerialEncodable, SerialDecodable)]
-pub struct FudFindNodesRequest {
+pub struct FudNodesRequest {
     pub key: blake3::Hash,
 }
-impl_p2p_message!(FudFindNodesRequest, "FudFindNodesRequest", 0, 0, DEFAULT_METERING_CONFIGURATION);
+impl_p2p_message!(FudNodesRequest, "FudNodesRequest", 0, 0, DEFAULT_METERING_CONFIGURATION);
 
 /// Message representing a find nodes reply on the network
 #[derive(Debug, Clone, SerialEncodable, SerialDecodable)]
-pub struct FudFindNodesReply {
+pub struct FudNodesReply {
+    pub key: blake3::Hash,
     pub nodes: Vec<FudNode>,
 }
-impl_p2p_message!(FudFindNodesReply, "FudFindNodesReply", 0, 0, DEFAULT_METERING_CONFIGURATION);
+impl_p2p_message!(FudNodesReply, "FudNodesReply", 0, 0, DEFAULT_METERING_CONFIGURATION);
+impl_resource_msg!(FudNodesReply, key);
 
 /// Message representing a find seeders request on the network
 #[derive(Debug, Clone, SerialEncodable, SerialDecodable)]
-pub struct FudFindSeedersRequest {
+pub struct FudSeedersRequest {
     pub key: blake3::Hash,
 }
-impl_p2p_message!(
-    FudFindSeedersRequest,
-    "FudFindSeedersRequest",
-    0,
-    0,
-    DEFAULT_METERING_CONFIGURATION
-);
+impl_p2p_message!(FudSeedersRequest, "FudSeedersRequest", 0, 0, DEFAULT_METERING_CONFIGURATION);
 
 /// Message representing a find seeders reply on the network
 #[derive(Debug, Clone, SerialEncodable, SerialDecodable)]
-pub struct FudFindSeedersReply {
+pub struct FudSeedersReply {
+    pub key: blake3::Hash,
     pub seeders: Vec<FudSeeder>,
     pub nodes: Vec<FudNode>,
 }
-impl_p2p_message!(FudFindSeedersReply, "FudFindSeedersReply", 0, 0, DEFAULT_METERING_CONFIGURATION);
+impl_p2p_message!(FudSeedersReply, "FudSeedersReply", 0, 0, DEFAULT_METERING_CONFIGURATION);
+impl_resource_msg!(FudSeedersReply, key);
 
 /// P2P protocol implementation for fud.
 pub struct ProtocolFud {
     channel: ChannelPtr,
     ping_request_sub: MessageSubscription<FudPingRequest>,
-    find_request_sub: MessageSubscription<FudFindRequest>,
-    find_nodes_request_sub: MessageSubscription<FudFindNodesRequest>,
-    find_seeders_request_sub: MessageSubscription<FudFindSeedersRequest>,
+    find_metadata_request_sub: MessageSubscription<FudMetadataRequest>,
+    find_chunk_request_sub: MessageSubscription<FudChunkRequest>,
+    find_nodes_request_sub: MessageSubscription<FudNodesRequest>,
+    find_seeders_request_sub: MessageSubscription<FudSeedersRequest>,
     announce_sub: MessageSubscription<FudAnnounce>,
     fud: Arc<Fud>,
     jobsman: ProtocolJobsManagerPtr,
@@ -156,21 +202,32 @@ impl ProtocolFud {
 
         let msg_subsystem = channel.message_subsystem();
         msg_subsystem.add_dispatch::<FudPingRequest>().await;
-        msg_subsystem.add_dispatch::<FudFindRequest>().await;
-        msg_subsystem.add_dispatch::<FudFindNodesRequest>().await;
-        msg_subsystem.add_dispatch::<FudFindSeedersRequest>().await;
+        msg_subsystem.add_dispatch::<FudPingReply>().await;
+        msg_subsystem.add_dispatch::<FudMetadataRequest>().await;
+        msg_subsystem.add_dispatch::<FudChunkRequest>().await;
+        msg_subsystem.add_dispatch::<FudChunkReply>().await;
+        msg_subsystem.add_dispatch::<FudChunkNotFound>().await;
+        msg_subsystem.add_dispatch::<FudFileReply>().await;
+        msg_subsystem.add_dispatch::<FudDirectoryReply>().await;
+        msg_subsystem.add_dispatch::<FudMetadataNotFound>().await;
+        msg_subsystem.add_dispatch::<FudNodesRequest>().await;
+        msg_subsystem.add_dispatch::<FudNodesReply>().await;
+        msg_subsystem.add_dispatch::<FudSeedersRequest>().await;
+        msg_subsystem.add_dispatch::<FudSeedersReply>().await;
         msg_subsystem.add_dispatch::<FudAnnounce>().await;
 
         let ping_request_sub = channel.subscribe_msg::<FudPingRequest>().await?;
-        let find_request_sub = channel.subscribe_msg::<FudFindRequest>().await?;
-        let find_nodes_request_sub = channel.subscribe_msg::<FudFindNodesRequest>().await?;
-        let find_seeders_request_sub = channel.subscribe_msg::<FudFindSeedersRequest>().await?;
+        let find_metadata_request_sub = channel.subscribe_msg::<FudMetadataRequest>().await?;
+        let find_chunk_request_sub = channel.subscribe_msg::<FudChunkRequest>().await?;
+        let find_nodes_request_sub = channel.subscribe_msg::<FudNodesRequest>().await?;
+        let find_seeders_request_sub = channel.subscribe_msg::<FudSeedersRequest>().await?;
         let announce_sub = channel.subscribe_msg::<FudAnnounce>().await?;
 
         Ok(Arc::new(Self {
             channel: channel.clone(),
             ping_request_sub,
-            find_request_sub,
+            find_metadata_request_sub,
+            find_chunk_request_sub,
             find_nodes_request_sub,
             find_seeders_request_sub,
             announce_sub,
@@ -186,187 +243,193 @@ impl ProtocolFud {
             let ping_req = match self.ping_request_sub.receive().await {
                 Ok(v) => v,
                 Err(Error::ChannelStopped) => continue,
-                Err(e) => {
-                    error!("{e}");
-                    continue
-                }
+                Err(_) => continue,
             };
-            info!(target: "fud::ProtocolFud::handle_fud_ping_request()", "Received PING REQUEST");
+            info!(target: "fud::ProtocolFud::handle_fud_ping_request()", "Received PING REQUEST from {}", self.channel.address());
+            self.fud.dht.update_channel(self.channel.info.id).await;
 
             let reply = FudPingReply {
                 node: self.fud.node().await,
+                random: ping_req.random,
                 sig: self.fud.secret_key.read().await.sign(&ping_req.random.to_be_bytes()),
             };
-            match self.channel.send(&reply).await {
-                Ok(()) => continue,
-                Err(_e) => continue,
+            if let Err(e) = self.channel.send(&reply).await {
+                self.fud
+                    .dht
+                    .event_publisher
+                    .notify(DhtEvent::PingSent { to: self.channel.clone(), result: Err(e) })
+                    .await;
+                continue;
+            }
+            self.fud
+                .dht
+                .event_publisher
+                .notify(DhtEvent::PingSent { to: self.channel.clone(), result: Ok(()) })
+                .await;
+
+            // Ping the peer if this is an inbound connection
+            if self.channel.session_type_id() & SESSION_INBOUND != 0 {
+                let _ = self.fud.ping(self.channel.clone()).await;
             }
         }
     }
 
-    async fn handle_fud_find_request(self: Arc<Self>) -> Result<()> {
-        debug!(target: "fud::ProtocolFud::handle_fud_find_request()", "START");
+    async fn handle_fud_metadata_request(self: Arc<Self>) -> Result<()> {
+        debug!(target: "fud::ProtocolFud::handle_fud_metadata_request()", "START");
 
         loop {
-            let request = match self.find_request_sub.receive().await {
+            let request = match self.find_metadata_request_sub.receive().await {
                 Ok(v) => v,
                 Err(Error::ChannelStopped) => continue,
-                Err(e) => {
-                    error!("{e}");
-                    continue
-                }
+                Err(_) => continue,
             };
-            info!(target: "fud::ProtocolFud::handle_fud_find_request()", "Received FIND for {}", hash_to_string(&request.key));
+            info!(target: "fud::ProtocolFud::handle_fud_request()", "Received METADATA REQUEST for {}", hash_to_string(&request.resource));
+            self.fud.dht.update_channel(self.channel.info.id).await;
 
-            let node = self.fud.dht().get_node_from_channel(self.channel.info.id).await;
-            if let Some(node) = node {
-                self.fud.dht.update_node(&node).await;
-            }
+            let notfound = async || {
+                let reply = FudMetadataNotFound { resource: request.resource };
+                info!(target: "fud::ProtocolFud::handle_fud_metadata_request()", "We do not have the metadata of {}", hash_to_string(&request.resource));
+                let _ = self.channel.send(&reply).await;
+            };
 
-            if self.handle_fud_chunk_request(&request).await {
-                continue;
+            let path = self.fud.hash_to_path(&request.resource).ok().flatten();
+            if path.is_none() {
+                notfound().await;
+                continue
             }
+            let path = path.unwrap();
 
-            if self.handle_fud_metadata_request(&request).await {
-                continue;
+            let chunked_file = self.fud.geode.get(&request.resource, &path).await.ok();
+            if chunked_file.is_none() {
+                notfound().await;
+                continue
+            }
+            let mut chunked_file = chunked_file.unwrap();
+
+            // If it's a file with a single chunk, just reply with the chunk
+            if chunked_file.len() == 1 && !chunked_file.is_dir() {
+                let chunk_hash = chunked_file.get_chunks()[0].0;
+                let chunk = self.fud.geode.get_chunk(&mut chunked_file, &chunk_hash).await;
+                if let Ok(chunk) = chunk {
+                    if blake3::hash(blake3::hash(&chunk).as_bytes()) != request.resource {
+                        // TODO: Run geode GC
+                        notfound().await;
+                        continue
+                    }
+                    let reply = FudChunkReply { resource: request.resource, chunk };
+                    info!(target: "fud::ProtocolFud::handle_fud_metadata_request()", "Sending chunk (file has a single chunk) {}", hash_to_string(&chunk_hash));
+                    let _ = self.channel.send(&reply).await;
+                    continue
+                }
+                // We don't have the chunk, but we can still reply with the metadata
             }
 
-            // Request did not match anything we have
-            let reply = FudNotFound {};
-            info!(target: "fud::ProtocolFud::handle_fud_find_request()", "We do not have {}", hash_to_string(&request.key));
-            let _ = self.channel.send(&reply).await;
+            // Reply with the metadata
+            match chunked_file.is_dir() {
+                false => {
+                    let reply = FudFileReply {
+                        resource: request.resource,
+                        chunk_hashes: chunked_file
+                            .get_chunks()
+                            .iter()
+                            .map(|(chunk, _)| *chunk)
+                            .collect(),
+                    };
+                    info!(target: "fud::ProtocolFud::handle_fud_metadata_request()", "Sending file metadata {}", hash_to_string(&request.resource));
+                    let _ = self.channel.send(&reply).await;
+                }
+                true => {
+                    let files = chunked_file
+                        .get_files()
+                        .iter()
+                        .map(|(file_path, size)| match file_path.strip_prefix(path.clone()) {
+                            Ok(rel_path) => Ok((rel_path.to_string_lossy().to_string(), *size)),
+                            Err(e) => Err(e),
+                        })
+                        .collect::<std::result::Result<Vec<_>, StripPrefixError>>();
+                    if let Err(e) = files {
+                        error!(target: "fud::ProtocolFud::handle_fud_metadata_request()", "Error parsing file paths before sending directory metadata: {e}");
+                        notfound().await;
+                        continue
+                    }
+                    let reply = FudDirectoryReply {
+                        resource: request.resource,
+                        chunk_hashes: chunked_file
+                            .get_chunks()
+                            .iter()
+                            .map(|(chunk, _)| *chunk)
+                            .collect(),
+                        files: files.unwrap(),
+                    };
+                    info!(target: "fud::ProtocolFud::handle_fud_metadata_request()", "Sending directory metadata {}", hash_to_string(&request.resource));
+                    let _ = self.channel.send(&reply).await;
+                }
+            };
         }
     }
 
-    /// If the FudFindRequest matches a chunk we have, handle it.
-    /// Returns true if the chunk was found.
-    async fn handle_fud_chunk_request(&self, request: &FudFindRequest) -> bool {
-        let hash = request.info;
-        if hash.is_none() {
-            return false;
-        }
-        let hash = hash.unwrap();
+    async fn handle_fud_chunk_request(self: Arc<Self>) -> Result<()> {
+        debug!(target: "fud::ProtocolFud::handle_fud_chunk_request()", "START");
 
-        let path = self.fud.hash_to_path(&hash).ok().flatten();
-        if path.is_none() {
-            return false;
-        }
-        let path = path.unwrap();
+        loop {
+            let request = match self.find_chunk_request_sub.receive().await {
+                Ok(v) => v,
+                Err(Error::ChannelStopped) => continue,
+                Err(_) => continue,
+            };
+            info!(target: "fud::ProtocolFud::handle_fud_chunk_request()", "Received CHUNK REQUEST for {}", hash_to_string(&request.resource));
+            self.fud.dht.update_channel(self.channel.info.id).await;
 
-        let chunked = self.fud.geode.get(&hash, &path).await;
-        if chunked.is_err() {
-            return false;
-        }
+            let notfound = async || {
+                let reply = FudChunkNotFound { resource: request.resource, chunk: request.chunk };
+                info!(target: "fud::ProtocolFud::handle_fud_chunk_request()", "We do not have chunk {} of resource {}", hash_to_string(&request.resource), hash_to_string(&request.chunk));
+                let _ = self.channel.send(&reply).await;
+            };
 
-        let chunk = self.fud.geode.get_chunk(&mut chunked.unwrap(), &request.key).await;
-        if let Ok(chunk) = chunk {
-            if !self.fud.geode.verify_chunk(&request.key, &chunk) {
-                // TODO: Run geode GC
-                return false;
+            let path = self.fud.hash_to_path(&request.resource).ok().flatten();
+            if path.is_none() {
+                notfound().await;
+                continue
             }
-            let reply = FudChunkReply { chunk };
-            info!(target: "fud::ProtocolFud::handle_fud_chunk_request()", "Sending chunk {}", hash_to_string(&request.key));
-            let _ = self.channel.send(&reply).await;
-            return true;
-        }
-
-        false
-    }
-
-    /// If the FudFindRequest matches a file we have, handle it
-    /// Returns true if the file was found.
-    async fn handle_fud_metadata_request(&self, request: &FudFindRequest) -> bool {
-        let path = self.fud.hash_to_path(&request.key).ok().flatten();
-        if path.is_none() {
-            return false;
-        }
-        let path = path.unwrap();
+            let path = path.unwrap();
 
-        let chunked_file = self.fud.geode.get(&request.key, &path).await.ok();
-        if chunked_file.is_none() {
-            return false;
-        }
-        let mut chunked_file = chunked_file.unwrap();
+            let chunked = self.fud.geode.get(&request.resource, &path).await;
+            if chunked.is_err() {
+                notfound().await;
+                continue
+            }
 
-        // If it's a file with a single chunk, just reply with the chunk
-        if chunked_file.len() == 1 && !chunked_file.is_dir() {
-            let chunk_hash = chunked_file.get_chunks()[0].0;
-            let chunk = self.fud.geode.get_chunk(&mut chunked_file, &chunk_hash).await;
+            let chunk = self.fud.geode.get_chunk(&mut chunked.unwrap(), &request.chunk).await;
             if let Ok(chunk) = chunk {
-                if blake3::hash(blake3::hash(&chunk).as_bytes()) != request.key {
+                if !self.fud.geode.verify_chunk(&request.chunk, &chunk) {
                     // TODO: Run geode GC
-                    return false;
-                }
-                let reply = FudChunkReply { chunk };
-                info!(target: "fud::ProtocolFud::handle_fud_metadata_request()", "Sending chunk (file has a single chunk) {}", hash_to_string(&chunk_hash));
-                let _ = self.channel.send(&reply).await;
-                return true;
-            }
-            return false;
-        }
-
-        // Otherwise reply with the metadata
-        match chunked_file.is_dir() {
-            false => {
-                let reply = FudFileReply {
-                    chunk_hashes: chunked_file
-                        .get_chunks()
-                        .iter()
-                        .map(|(chunk, _)| *chunk)
-                        .collect(),
-                };
-                info!(target: "fud::ProtocolFud::handle_fud_metadata_request()", "Sending file metadata {}", hash_to_string(&request.key));
-                let _ = self.channel.send(&reply).await;
-            }
-            true => {
-                let files = chunked_file
-                    .get_files()
-                    .iter()
-                    .map(|(file_path, size)| match file_path.strip_prefix(path.clone()) {
-                        Ok(rel_path) => Ok((rel_path.to_string_lossy().to_string(), *size)),
-                        Err(e) => Err(e),
-                    })
-                    .collect::<std::result::Result<Vec<_>, StripPrefixError>>();
-                if let Err(e) = files {
-                    error!(target: "fud::ProtocolFud::handle_fud_metadata_request()", "Error parsing file paths before sending directory metadata: {e}");
-                    return false;
+                    notfound().await;
+                    continue
                 }
-                let reply = FudDirectoryReply {
-                    chunk_hashes: chunked_file
-                        .get_chunks()
-                        .iter()
-                        .map(|(chunk, _)| *chunk)
-                        .collect(),
-                    files: files.unwrap(),
-                };
-                info!(target: "fud::ProtocolFud::handle_fud_metadata_request()", "Sending directory metadata {}", hash_to_string(&request.key));
+                let reply = FudChunkReply { resource: request.resource, chunk };
+                info!(target: "fud::ProtocolFud::handle_fud_chunk_request()", "Sending chunk {}", hash_to_string(&request.chunk));
                 let _ = self.channel.send(&reply).await;
+                continue
             }
-        };
 
-        true
+            notfound().await;
+        }
     }
 
-    async fn handle_fud_find_nodes_request(self: Arc<Self>) -> Result<()> {
-        debug!(target: "fud::ProtocolFud::handle_fud_find_nodes_request()", "START");
+    async fn handle_fud_nodes_request(self: Arc<Self>) -> Result<()> {
+        debug!(target: "fud::ProtocolFud::handle_fud_nodes_request()", "START");
 
         loop {
             let request = match self.find_nodes_request_sub.receive().await {
                 Ok(v) => v,
                 Err(Error::ChannelStopped) => continue,
-                Err(e) => {
-                    error!("{e}");
-                    continue
-                }
+                Err(_) => continue,
             };
-            info!(target: "fud::ProtocolFud::handle_fud_find_nodes_request()", "Received FIND NODES for {}", hash_to_string(&request.key));
-
-            let node = self.fud.dht().get_node_from_channel(self.channel.info.id).await;
-            if let Some(node) = node {
-                self.fud.dht.update_node(&node).await;
-            }
+            info!(target: "fud::ProtocolFud::handle_fud_nodes_request()", "Received FIND NODES for {}", hash_to_string(&request.key));
+            self.fud.dht.update_channel(self.channel.info.id).await;
 
-            let reply = FudFindNodesReply {
+            let reply = FudNodesReply {
+                key: request.key,
                 nodes: self.fud.dht().find_neighbors(&request.key, self.fud.dht().settings.k).await,
             };
             match self.channel.send(&reply).await {
@@ -376,24 +439,17 @@ impl ProtocolFud {
         }
     }
 
-    async fn handle_fud_find_seeders_request(self: Arc<Self>) -> Result<()> {
-        debug!(target: "fud::ProtocolFud::handle_fud_find_seeders_request()", "START");
+    async fn handle_fud_seeders_request(self: Arc<Self>) -> Result<()> {
+        debug!(target: "fud::ProtocolFud::handle_fud_seeders_request()", "START");
 
         loop {
             let request = match self.find_seeders_request_sub.receive().await {
                 Ok(v) => v,
                 Err(Error::ChannelStopped) => continue,
-                Err(e) => {
-                    error!("{e}");
-                    continue
-                }
+                Err(_) => continue,
             };
-            info!(target: "fud::ProtocolFud::handle_fud_find_seeders_request()", "Received FIND SEEDERS for {}", hash_to_string(&request.key));
-
-            let node = self.fud.dht().get_node_from_channel(self.channel.info.id).await;
-            if let Some(node) = node {
-                self.fud.dht.update_node(&node).await;
-            }
+            info!(target: "fud::ProtocolFud::handle_fud_seeders_request()", "Received FIND SEEDERS for {} from {:?}", hash_to_string(&request.key), self.channel);
+            self.fud.dht.update_channel(self.channel.info.id).await;
 
             let router = self.fud.dht.hash_table.read().await;
             let peers = router.get(&request.key);
@@ -402,7 +458,8 @@ impl ProtocolFud {
                 Some(seeders) => {
                     let _ = self
                         .channel
-                        .send(&FudFindSeedersReply {
+                        .send(&FudSeedersReply {
+                            key: request.key,
                             seeders: seeders.to_vec(),
                             nodes: self
                                 .fud
@@ -415,7 +472,8 @@ impl ProtocolFud {
                 None => {
                     let _ = self
                         .channel
-                        .send(&FudFindSeedersReply {
+                        .send(&FudSeedersReply {
+                            key: request.key,
                             seeders: vec![],
                             nodes: self
                                 .fud
@@ -436,17 +494,10 @@ impl ProtocolFud {
             let request = match self.announce_sub.receive().await {
                 Ok(v) => v,
                 Err(Error::ChannelStopped) => continue,
-                Err(e) => {
-                    error!("{e}");
-                    continue
-                }
+                Err(_) => continue,
             };
             info!(target: "fud::ProtocolFud::handle_fud_announce()", "Received ANNOUNCE for {}", hash_to_string(&request.key));
-
-            let node = self.fud.dht().get_node_from_channel(self.channel.info.id).await;
-            if let Some(node) = node {
-                self.fud.dht.update_node(&node).await;
-            }
+            self.fud.dht.update_channel(self.channel.info.id).await;
 
             let mut seeders = vec![];
 
@@ -454,6 +505,8 @@ impl ProtocolFud {
                 if seeder.node.addresses.is_empty() {
                     continue
                 }
+
+                // TODO: Limit the number of addresses
                 // TODO: Verify each address
                 seeders.push(seeder);
             }
@@ -469,14 +522,15 @@ impl ProtocolBase for ProtocolFud {
         debug!(target: "fud::ProtocolFud::start()", "START");
         self.jobsman.clone().start(executor.clone());
         self.jobsman.clone().spawn(self.clone().handle_fud_ping_request(), executor.clone()).await;
-        self.jobsman.clone().spawn(self.clone().handle_fud_find_request(), executor.clone()).await;
         self.jobsman
             .clone()
-            .spawn(self.clone().handle_fud_find_nodes_request(), executor.clone())
+            .spawn(self.clone().handle_fud_metadata_request(), executor.clone())
             .await;
+        self.jobsman.clone().spawn(self.clone().handle_fud_chunk_request(), executor.clone()).await;
+        self.jobsman.clone().spawn(self.clone().handle_fud_nodes_request(), executor.clone()).await;
         self.jobsman
             .clone()
-            .spawn(self.clone().handle_fud_find_seeders_request(), executor.clone())
+            .spawn(self.clone().handle_fud_seeders_request(), executor.clone())
             .await;
         self.jobsman.clone().spawn(self.clone().handle_fud_announce(), executor.clone()).await;
         debug!(target: "fud::ProtocolFud::start()", "END");

bin/fud/fud/src/tasks.rs

@@ -21,7 +21,7 @@ use std::sync::Arc;
 use tracing::{error, info, warn};
 
 use darkfi::{
-    dht::{DhtHandler, DhtNode},
+    dht::{event::DhtEvent, DhtHandler, DhtNode},
     geode::hash_to_string,
     system::{sleep, StoppableTask},
     Error, Result,
@@ -29,6 +29,29 @@ use darkfi::{
 
 use crate::{event, event::notify_event, proto::FudAnnounce, Fud, FudEvent};
 
+/// Handle DHT events in fud.
+pub async fn handle_dht_events(fud: Arc<Fud>) -> Result<()> {
+    let sub = fud.dht().subscribe().await;
+    loop {
+        let event = sub.receive().await;
+
+        match event {
+            DhtEvent::ValueLookupCompleted { key, values, .. } => {
+                let mut seeders: Vec<_> = values.into_iter().flatten().collect();
+                seeders.dedup_by_key(|seeder| seeder.node.id());
+                notify_event!(fud, SeedersFound, {
+                    hash: key,
+                    seeders
+                });
+            }
+            DhtEvent::BootstrapCompleted => {
+                let _ = fud.init().await;
+            }
+            _ => {}
+        }
+    }
+}
+
 /// Triggered when calling the `fud.get()` method.
 /// It creates a new StoppableTask (running `fud.fetch_resource()`) and inserts
 /// it into the `fud.fetch_tasks` hashmap. When the task is stopped it's
@@ -123,7 +146,7 @@ pub async fn put_task(fud: Arc<Fud>) -> Result<()> {
 /// Triggered when you need to lookup seeders for a resource.
 pub async fn lookup_task(fud: Arc<Fud>) -> Result<()> {
     loop {
-        let (key, seeders_pub) = fud.lookup_rx.recv().await.unwrap();
+        let key = fud.lookup_rx.recv().await.unwrap();
 
         let mut lookup_tasks = fud.lookup_tasks.write().await;
         let task = StoppableTask::new();
@@ -134,7 +157,7 @@ pub async fn lookup_task(fud: Arc<Fud>) -> Result<()> {
         let fud_2 = fud.clone();
         task.start(
             async move {
-                fud_1.dht.lookup_value(&key, seeders_pub).await?;
+                fud_1.dht.lookup_value(&key).await;
                 Ok(())
             },
             move |res| async move {

bin/fud/fud/src/util.rs

@@ -23,10 +23,17 @@ use smol::{
 use std::{
     collections::HashSet,
     path::{Path, PathBuf},
+    sync::Arc,
+    time::Instant,
 };
 
 pub use darkfi::geode::hash_to_string;
-use darkfi::Result;
+use darkfi::{
+    net::{Message, MessageSubscription},
+    Error, Result,
+};
+
+use crate::proto::ResourceMessage;
 
 pub async fn get_all_files(dir: &Path) -> Result<Vec<(PathBuf, u64)>> {
     let mut files = Vec::new();
@@ -76,3 +83,26 @@ impl FromIterator<PathBuf> for FileSelection {
         FileSelection::Set(paths)
     }
 }
+
+/// Wait for a [`crate::proto::ResourceMessage`] on `msg_subscriber` with a timeout.
+/// If we receive a message with the wrong resource hash, it's skipped.
+pub async fn receive_resource_msg<M: Message + ResourceMessage + std::fmt::Debug>(
+    msg_subscriber: &MessageSubscription<M>,
+    resource_hash: blake3::Hash,
+    timeout_seconds: u64,
+) -> Result<Arc<M>> {
+    let start = Instant::now();
+    loop {
+        let elapsed = start.elapsed().as_secs();
+        if elapsed >= timeout_seconds {
+            return Err(Error::ConnectTimeout);
+        }
+        let remaining_timeout = timeout_seconds - elapsed;
+
+        let reply = msg_subscriber.receive_with_timeout(remaining_timeout).await?;
+        // Done if it's the right resource hash
+        if reply.resource_hash() == resource_hash {
+            return Ok(reply)
+        }
+    }
+}

src/dht/handler.rs

@@ -17,6 +17,7 @@
  */
 
 use std::{
+    fmt::Debug,
     marker::{Send, Sync},
     sync::Arc,
 };
@@ -29,7 +30,7 @@ use crate::{net::ChannelPtr, Result};
 /// Trait for application-specific behaviors over a [`Dht`]
 #[async_trait]
 pub trait DhtHandler: Send + Sync + Sized {
-    type Value: Clone;
+    type Value: Clone + Debug;
     type Node: DhtNode;
 
     /// The [`Dht`] instance
@@ -38,20 +39,25 @@ pub trait DhtHandler: Send + Sync + Sized {
     /// Get our own node
     async fn node(&self) -> Self::Node;
 
-    /// Send a DHT ping request, which is used to know the node data of a peer
+    /// Send PING request, which is used to know the node data of a peer
     /// (and most importantly, its ID/key in the DHT keyspace)
     async fn ping(&self, channel: ChannelPtr) -> Result<Self::Node>;
 
-    /// Triggered when we find a new node
-    async fn on_new_node(&self, node: &Self::Node) -> Result<()>;
+    /// Send STORE request to instruct a peer to store a key-value pair
+    async fn store(
+        &self,
+        channel: ChannelPtr,
+        key: &blake3::Hash,
+        value: &Self::Value,
+    ) -> Result<()>;
 
     /// Send FIND NODES request to a peer to get nodes close to `key`
-    async fn find_nodes(&self, node: &Self::Node, key: &blake3::Hash) -> Result<Vec<Self::Node>>;
+    async fn find_nodes(&self, channel: ChannelPtr, key: &blake3::Hash) -> Result<Vec<Self::Node>>;
 
     /// Send FIND VALUE request to a peer to get a value and/or nodes close to `key`
     async fn find_value(
         &self,
-        node: &Self::Node,
+        channel: ChannelPtr,
         key: &blake3::Hash,
     ) -> Result<DhtLookupReply<Self::Node, Self::Value>>;
 

src/dht/mod.rs

@@ -23,26 +23,27 @@ use std::{
     hash::Hash,
     marker::{Send, Sync},
     sync::{Arc, Weak},
-    time::Duration,
 };
 
 use futures::stream::FuturesUnordered;
 use num_bigint::BigUint;
 use smol::{
-    lock::{RwLock, Semaphore},
+    channel,
+    lock::{Mutex, RwLock, Semaphore},
     stream::StreamExt,
 };
-use tracing::{debug, info, warn};
+use tracing::{info, warn};
 use url::Url;
 
 use crate::{
     dht::event::DhtEvent,
     net::{
         connector::Connector,
-        session::{Session, SESSION_REFINE, SESSION_SEED},
+        session::{SESSION_DIRECT, SESSION_MANUAL},
         ChannelPtr, Message, P2pPtr,
     },
     system::{msleep, ExecutorPtr, Publisher, PublisherPtr, Subscription},
+    util::time::Timestamp,
     Error, Result,
 };
 
@@ -80,9 +81,9 @@ macro_rules! impl_dht_node_defaults {
 }
 pub use impl_dht_node_defaults;
 
-enum DhtLookupType<V> {
-    Nodes(blake3::Hash),
-    Value(blake3::Hash, PublisherPtr<Option<V>>),
+enum DhtLookupType {
+    Nodes,
+    Value,
 }
 
 pub enum DhtLookupReply<N: DhtNode, V> {
@@ -98,22 +99,25 @@ pub struct DhtBucket<N: DhtNode> {
 /// Our local hash table, storing DHT keys and values
 pub type DhtHashTable<V> = Arc<RwLock<HashMap<blake3::Hash, V>>>;
 
-#[derive(Clone)]
+#[derive(Clone, Debug)]
 pub struct ChannelCacheItem<N: DhtNode> {
     /// The DHT node the channel is connected to.
-    pub node: N,
-
-    /// Topic is a hash that you set to remember what the channel is about,
-    /// it's not shared with the peer. If you ask for a channel (with
-    /// `dht.get_channel()`) for a specific topic, it will give you a
-    /// channel that has no topic, has the same topic, or a new
-    /// channel.
-    topic: Option<blake3::Hash>,
-
-    /// Usage count increments when you call `handler.get_channel()` and
-    /// decrements when you call `handler.cleanup_channel()`. A channel's
-    /// topic is cleared on cleanup if its usage count is zero.
-    usage_count: u32,
+    pub node: Option<N>,
+    /// The last time this channel was used by the [`DhtHandler`]. It's used
+    /// to stop inbound connections in [`crate::dht::tasks::disconnect_inbounds_task()`].
+    pub last_used: Timestamp,
+    /// Have we already received a DHT ping from this channel?
+    pub ping_received: bool,
+    /// Have we already sent a DHT ping to this channel?
+    pub ping_sent: bool,
+}
+
+#[derive(Clone, Debug)]
+pub struct HostCacheItem {
+    /// The last time we tried to send a DHT ping to this host.
+    pub last_ping: Timestamp,
+    /// The last known node id for this host.
+    pub node_id: blake3::Hash,
 }
 
 pub struct Dht<H: DhtHandler> {
@@ -129,12 +133,20 @@ pub struct Dht<H: DhtHandler> {
     pub n_buckets: usize,
     /// Channel ID -> ChannelCacheItem
     pub channel_cache: Arc<RwLock<HashMap<u32, ChannelCacheItem<H::Node>>>>,
+    /// Host address -> ChannelCacheItem
+    pub host_cache: Arc<RwLock<HashMap<Url, HostCacheItem>>>,
+    /// Add node sender
+    pub add_node_tx: channel::Sender<(H::Node, ChannelPtr)>,
+    /// Add node receiver
+    pub add_node_rx: channel::Receiver<(H::Node, ChannelPtr)>,
     /// DHT settings
     pub settings: DhtSettings,
     /// DHT event publisher
     pub event_publisher: PublisherPtr<DhtEvent<H::Node, H::Value>>,
     /// P2P network pointer
     pub p2p: P2pPtr,
+    /// Connector to create manual connections
+    pub connector: Connector,
     /// Global multithreaded executor reference
     pub executor: ExecutorPtr,
 }
@@ -147,6 +159,11 @@ impl<H: DhtHandler> Dht<H> {
             buckets.push(DhtBucket { nodes: vec![] })
         }
 
+        let (add_node_tx, add_node_rx) = smol::channel::unbounded();
+
+        let session_weak = Arc::downgrade(&p2p.session_manual());
+        let connector = Connector::new(p2p.settings(), session_weak);
+
         Self {
             handler: RwLock::new(Weak::new()),
             buckets: Arc::new(RwLock::new(buckets)),
@@ -154,12 +171,16 @@ impl<H: DhtHandler> Dht<H> {
             n_buckets: 256,
             bootstrapped: Arc::new(RwLock::new(false)),
             channel_cache: Arc::new(RwLock::new(HashMap::new())),
+            host_cache: Arc::new(RwLock::new(HashMap::new())),
+            add_node_tx,
+            add_node_rx,
 
             event_publisher: Publisher::new(),
 
             settings: settings.clone(),
 
             p2p: p2p.clone(),
+            connector,
             executor: ex,
         }
     }
@@ -208,7 +229,7 @@ impl<H: DhtHandler> Dht<H> {
     /// `key` -> bucket index
     pub async fn get_bucket_index(&self, self_node_id: &blake3::Hash, key: &blake3::Hash) -> usize {
         if key == self_node_id {
-            return 0
+            return 0;
         }
         let distance = self.distance(self_node_id, key);
         let mut leading_zeros = 0;
@@ -252,7 +273,7 @@ impl<H: DhtHandler> Dht<H> {
         let channel_cache_lock = self.channel_cache.clone();
         let channel_cache = channel_cache_lock.read().await;
         if let Some(cached) = channel_cache.get(&channel_id) {
-            return Some(cached.node.clone())
+            return cached.node.clone();
         }
 
         None
@@ -315,135 +336,152 @@ impl<H: DhtHandler> Dht<H> {
         }
     }
 
-    /// Add a node in the correct bucket
-    pub async fn add_node(&self, node: H::Node) {
-        let self_node = self.handler().await.node().await;
-
-        // Do not add ourselves to the buckets
-        if node.id() == self_node.id() {
-            return;
-        }
-
-        // Don't add this node if it has any external address that is the same as one of ours
-        let node_addresses = node.addresses();
-        if self_node.addresses().iter().any(|addr| node_addresses.contains(addr)) {
-            return;
-        }
+    // TODO: Optimize this
+    async fn on_new_node(&self, node: &H::Node, channel: ChannelPtr) {
+        info!(target: "dht::on_new_node()", "[DHT] Found new node {}", H::key_to_string(&node.id()));
 
-        // Do not add a node to the buckets if it does not have an address
-        if node.addresses().is_empty() {
-            return;
+        // If this is the first node we know about then bootstrap
+        if !self.is_bootstrapped().await {
+            self.bootstrap().await;
         }
 
-        let bucket_index =
-            self.get_bucket_index(&self.handler().await.node().await.id(), &node.id()).await;
-        let buckets_lock = self.buckets.clone();
-        let mut buckets = buckets_lock.write().await;
-        let bucket = &mut buckets[bucket_index];
-
-        // Node is already in the bucket
-        if bucket.nodes.iter().any(|n| n.id() == node.id()) {
-            return;
-        }
-
-        // Bucket is full
-        if bucket.nodes.len() >= self.settings.k {
-            // Ping the least recently seen node
-            if let Ok(channel) = self.get_channel(&bucket.nodes[0], None).await {
-                let ping_res = self.handler().await.ping(channel.clone()).await;
-                self.cleanup_channel(channel).await;
-                if ping_res.is_ok() {
-                    // Ping was successful, move the least recently seen node to the tail
-                    let n = bucket.nodes.remove(0);
-                    bucket.nodes.push(n);
-                    return;
-                }
+        // Send keys that are closer to this node than we are
+        let self_id = self.handler().await.node().await.id();
+        for (key, value) in self.hash_table.read().await.iter() {
+            let node_distance = BigUint::from_bytes_be(&self.distance(key, &node.id()));
+            let self_distance = BigUint::from_bytes_be(&self.distance(key, &self_id));
+            if node_distance <= self_distance {
+                let _ = self.handler().await.store(channel.clone(), key, value).await;
             }
-
-            // Ping was not successful, remove the least recently seen node and add the new node
-            bucket.nodes.remove(0);
-            bucket.nodes.push(node);
-            return;
         }
-
-        // Bucket is not full
-        bucket.nodes.push(node);
     }
 
     /// Move a node to the tail in its bucket,
     /// to show that it is the most recently seen in the bucket.
-    /// If the node is not in a bucket it will be added using `add_node`
-    pub async fn update_node(&self, node: &H::Node) {
-        let bucket_index =
-            self.get_bucket_index(&self.handler().await.node().await.id(), &node.id()).await;
-        let buckets_lock = self.buckets.clone();
-        let mut buckets = buckets_lock.write().await;
-        let bucket = &mut buckets[bucket_index];
-
-        let node_index = bucket.nodes.iter().position(|n| n.id() == node.id());
-        if node_index.is_none() {
-            drop(buckets);
-            self.add_node(node.clone()).await;
-            return;
+    /// If the node is not in a bucket it will be added using `add_node`.
+    pub async fn update_node(&self, node: &H::Node, channel: ChannelPtr) {
+        if let Err(e) = self.add_node_tx.send((node.clone(), channel.clone())).await {
+            warn!(target: "dht::update_node()", "[DHT] Cannot add node {}: {e}", H::key_to_string(&node.id()))
         }
+    }
 
-        let n = bucket.nodes.remove(node_index.unwrap());
-        bucket.nodes.push(n);
+    /// Remove a node from the buckets.
+    pub async fn remove_node(&self, node_id: &blake3::Hash) {
+        let handler = self.handler().await;
+        let self_node = handler.node().await;
+        let bucket_index = handler.dht().get_bucket_index(&self_node.id(), node_id).await;
+        let buckets_lock = handler.dht().buckets.clone();
+        let mut buckets = buckets_lock.write().await;
+        let bucket = &mut buckets[bucket_index];
+        bucket.nodes.retain(|node| node.id() != *node_id);
     }
 
-    /// Lookup algorithm for both nodes lookup and value lookup
-    async fn lookup(&self, lookup_type: DhtLookupType<H::Value>) -> Result<Vec<H::Node>> {
-        let (key, value_pub) = match lookup_type {
-            DhtLookupType::Nodes(key) => (key, None),
-            DhtLookupType::Value(key, ref pub_ptr) => (key, Some(pub_ptr)),
-        };
+    /// Lookup algorithm for both nodes lookup and value lookup.
+    async fn lookup(
+        &self,
+        key: blake3::Hash,
+        lookup_type: DhtLookupType,
+    ) -> (Vec<H::Node>, Vec<H::Value>) {
+        let net_settings = self.p2p.settings().read_arc().await;
+        let allowed_transports = net_settings.allowed_transports.clone();
+        drop(net_settings);
 
         let (k, a) = (self.settings.k, self.settings.alpha);
         let semaphore = Arc::new(Semaphore::new(self.settings.concurrency));
-
-        let mut unique_nodes = HashSet::new();
+        let queried_addrs = Arc::new(Mutex::new(HashSet::new()));
+        let mut seen_nodes = HashSet::new();
         let mut nodes_to_visit = self.find_neighbors(&key, k).await;
         let mut result = Vec::new();
         let mut futures = FuturesUnordered::new();
+        let mut consecutive_stalls = 0;
+
+        let mut values = Vec::new();
 
         let distance_check = |(furthest, next): (&H::Node, &H::Node)| {
             BigUint::from_bytes_be(&self.distance(&key, &furthest.id())) <
                 BigUint::from_bytes_be(&self.distance(&key, &next.id()))
         };
 
+        // Create a channel if necessary and send a FIND NODES or FIND VALUE
+        // request to `addr`
         let lookup = async |node: H::Node, key| {
             let _permit = semaphore.acquire().await;
-            let n = node.clone();
-            let handler = self.handler().await;
-            match &lookup_type {
-                DhtLookupType::Nodes(_) => {
-                    (n, handler.find_nodes(&node, key).await.map(DhtLookupReply::Nodes))
+
+            // Filter and try all valid addresses for the node
+            let valid_addrs: Vec<Url> = node
+                .addresses()
+                .iter()
+                .filter(|addr| allowed_transports.contains(&addr.scheme().to_string()))
+                .cloned()
+                .collect();
+
+            let mut last_err = None;
+            for addr in valid_addrs {
+                let mut queried_addrs_set = queried_addrs.lock().await;
+                // Skip if this address has already been queried
+                if queried_addrs_set.contains(&addr) {
+                    continue;
+                }
+                queried_addrs_set.insert(addr.clone());
+                drop(queried_addrs_set);
+
+                // Try to create or find an existing channel
+                let channel = self.create_channel(&addr).await.map(|(ch, _)| ch);
+
+                if let Err(e) = channel {
+                    last_err = Some(e);
+                    continue
                 }
-                DhtLookupType::Value(_, _) => (n, handler.find_value(&node, key).await),
+                let channel = channel.unwrap();
+
+                let handler = self.handler().await;
+                let res = match &lookup_type {
+                    DhtLookupType::Nodes => {
+                        info!(target: "dht::lookup()", "[DHT] [LOOKUP] Querying node {} for nodes lookup of key {}", H::key_to_string(&node.id()), H::key_to_string(key));
+                        handler.find_nodes(channel.clone(), key).await.map(DhtLookupReply::Nodes)
+                    }
+                    DhtLookupType::Value => {
+                        info!(target: "dht::lookup()", "[DHT] [LOOKUP] Querying node {} for value lookup of key {}", H::key_to_string(&node.id()), H::key_to_string(key));
+                        handler.find_value(channel.clone(), key).await
+                    }
+                };
+
+                self.cleanup_channel(channel).await;
+                if res.is_ok() {
+                    return (node, res)
+                }
+                last_err = res.err();
+            }
+            if let Some(e) = last_err {
+                return (node, Err(e))
             }
+
+            (node, Err(Error::Custom("All node's addresses failed".to_string())))
         };
 
+        // Spawn up to `alpha` futures for lookup()
         let spawn_futures = async |nodes_to_visit: &mut Vec<H::Node>,
-                                   unique_nodes: &mut HashSet<_>,
                                    futures: &mut FuturesUnordered<_>| {
             for _ in 0..a {
-                if let Some(node) = nodes_to_visit.pop() {
-                    unique_nodes.insert(node.id());
+                if !nodes_to_visit.is_empty() {
+                    let node = nodes_to_visit.remove(0);
                     futures.push(Box::pin(lookup(node, &key)));
                 }
             }
         };
 
-        spawn_futures(&mut nodes_to_visit, &mut unique_nodes, &mut futures).await; // Initial alpha tasks
+        // Initial futures
+        spawn_futures(&mut nodes_to_visit, &mut futures).await;
 
+        // Process lookup responses
         while let Some((queried_node, res)) = futures.next().await {
             if let Err(e) = res {
-                warn!(target: "dht::lookup()", "Error in DHT lookup: {e}");
+                warn!(target: "dht::lookup()", "[DHT] [LOOKUP] Error in lookup: {e}");
 
                 // Spawn next `alpha` futures if there are no more futures but
                 // we still have nodes to visit
                 if futures.is_empty() {
-                    spawn_futures(&mut nodes_to_visit, &mut unique_nodes, &mut futures).await;
+                    spawn_futures(&mut nodes_to_visit, &mut futures).await;
                 }
 
                 continue;
@@ -455,178 +493,213 @@ impl<H: DhtHandler> Dht<H> {
                 DhtLookupReply::NodesAndValue(nodes, value) => (Some(nodes), Some(value)),
             };
 
+            // Send the value we found to the publisher
             if let Some(value) = value {
-                if let Some(publisher) = value_pub {
-                    publisher.notify(Some(value)).await;
-                }
+                info!(target: "dht::lookup()", "[DHT] [LOOKUP] Found value for {} from {}", H::key_to_string(&key), H::key_to_string(&queried_node.id()));
+                values.push(value.clone());
+                self.event_publisher.notify(DhtEvent::ValueFound { key, value }).await;
             }
 
+            // Update nodes_to_visit
             if let Some(mut nodes) = nodes {
-                let self_id = self.handler().await.node().await.id();
-                nodes.retain(|node| node.id() != self_id && unique_nodes.insert(node.id()));
-
-                nodes_to_visit.extend(nodes.clone());
-                self.sort_by_distance(&mut nodes_to_visit, &key);
+                if !nodes.is_empty() {
+                    info!(target: "dht::lookup()", "[DHT] [LOOKUP] Found {} nodes from {}", nodes.len(), H::key_to_string(&queried_node.id()));
+
+                    self.event_publisher
+                        .notify(DhtEvent::NodesFound { key, nodes: nodes.clone() })
+                        .await;
+
+                    // Remove our own node and duplicates
+                    let self_id = self.handler().await.node().await.id();
+                    nodes.retain(|node: &H::Node| {
+                        node.id() != self_id && seen_nodes.insert(node.id())
+                    });
+
+                    // Add new nodes to the list of nodes to visit
+                    nodes_to_visit.extend(nodes.clone());
+                    self.sort_by_distance(&mut nodes_to_visit, &key);
+                }
             }
 
             result.push(queried_node);
             self.sort_by_distance(&mut result, &key);
 
-            // Early termination logic
+            // Early termination logic:
+            // The closest node to visit must be further than the furthest
+            // queried node, 3 consecutive times
             if result.len() >= k &&
                 result.last().zip(nodes_to_visit.first()).is_some_and(distance_check)
             {
-                break;
+                consecutive_stalls += 1;
+                if consecutive_stalls >= 3 {
+                    break;
+                }
+            } else {
+                consecutive_stalls = 0;
             }
 
             // Spawn next `alpha` futures
-            spawn_futures(&mut nodes_to_visit, &mut unique_nodes, &mut futures).await;
+            spawn_futures(&mut nodes_to_visit, &mut futures).await;
         }
 
-        if let Some(publisher) = value_pub {
-            publisher.notify(None).await;
-        }
+        info!(target: "dht::lookup()", "[DHT] [LOOKUP] Lookup for {} completed", H::key_to_string(&key));
 
-        Ok(result.into_iter().take(k).collect())
+        let nodes: Vec<_> = result.into_iter().take(k).collect();
+        (nodes, values)
     }
 
     /// Find `k` nodes closest to a key
-    pub async fn lookup_nodes(&self, key: &blake3::Hash) -> Result<Vec<H::Node>> {
-        info!(target: "dht::lookup_nodes()", "Starting node lookup for key {}", H::key_to_string(key));
-        self.lookup(DhtLookupType::Nodes(*key)).await
+    pub async fn lookup_nodes(&self, key: &blake3::Hash) -> Vec<H::Node> {
+        info!(target: "dht::lookup_nodes()", "[DHT] [LOOKUP] Starting node lookup for key {}", H::key_to_string(key));
+
+        self.event_publisher.notify(DhtEvent::NodesLookupStarted { key: *key }).await;
+
+        let (nodes, _) = self.lookup(*key, DhtLookupType::Nodes).await;
+
+        self.event_publisher
+            .notify(DhtEvent::NodesLookupCompleted { key: *key, nodes: nodes.clone() })
+            .await;
+
+        nodes
     }
 
     /// Find value for `key`
-    pub async fn lookup_value(
-        &self,
-        key: &blake3::Hash,
-        value_pub: PublisherPtr<Option<H::Value>>,
-    ) -> Result<Vec<H::Node>> {
-        info!(target: "dht::lookup_value()", "Starting value lookup for key {}", H::key_to_string(key));
-        self.lookup(DhtLookupType::Value(*key, value_pub)).await
+    pub async fn lookup_value(&self, key: &blake3::Hash) -> (Vec<H::Node>, Vec<H::Value>) {
+        info!(target: "dht::lookup_value()", "[DHT] [LOOKUP] Starting value lookup for key {}", H::key_to_string(key));
+
+        self.event_publisher.notify(DhtEvent::ValueLookupStarted { key: *key }).await;
+
+        let (nodes, values) = self.lookup(*key, DhtLookupType::Value).await;
+
+        self.event_publisher
+            .notify(DhtEvent::ValueLookupCompleted {
+                key: *key,
+                nodes: nodes.clone(),
+                values: values.clone(),
+            })
+            .await;
+
+        (nodes, values)
     }
 
-    /// Get a channel (existing or create a new one) to `node` about `topic`.
-    /// Don't forget to call `cleanup_channel()` once you are done with it.
-    pub async fn get_channel(
-        &self,
-        node: &H::Node,
-        topic: Option<blake3::Hash>,
-    ) -> Result<ChannelPtr> {
+    /// Update a channel's `last_used` field in the channel cache.
+    pub async fn update_channel(&self, channel_id: u32) {
         let channel_cache_lock = self.channel_cache.clone();
         let mut channel_cache = channel_cache_lock.write().await;
 
-        // Get existing channels for this node, regardless of topic
-        let channels: HashMap<u32, ChannelCacheItem<H::Node>> = channel_cache
+        if let Some(cached) = channel_cache.get_mut(&channel_id) {
+            cached.last_used = Timestamp::current_time();
+        }
+    }
+
+    /// Get a channel (existing or create a new one) to `node`.
+    /// Don't forget to call `cleanup_channel()` once you are done with it.
+    pub async fn get_channel(&self, node: &H::Node) -> Result<(ChannelPtr, H::Node)> {
+        let node_id = node.id();
+
+        // Look in the channel cache for a channel connected to this node.
+        // We skip direct session channels, for those we will call
+        // `create_channel()` which increments the sessions's usage counter.
+        let channel_cache = self.channel_cache.read().await.clone();
+        if let Some((channel_id, cached)) = channel_cache
+            .clone()
             .iter()
-            .filter(|&(_, item)| item.node == *node)
-            .map(|(&key, item)| (key, item.clone()))
-            .collect();
-
-        let (channel_id, topic, usage_count) =
-            // If we already have a channel for this node and topic, use it
-            if let Some((cid, cached)) = channels.iter().find(|&(_, c)| c.topic == topic) {
-                (Some(*cid), cached.topic, cached.usage_count)
-            }
-            // If we have a topicless channel for this node, use it
-            else if let Some((cid, cached)) = channels.iter().find(|&(_, c)| c.topic.is_none()) {
-                (Some(*cid), topic, cached.usage_count)
-            }
-            // If we don't need any specific topic, use the first channel we have
-            else if topic.is_none() {
-                match channels.iter().next() {
-                    Some((cid, cached)) => (Some(*cid), cached.topic, cached.usage_count),
-                    _ => (None, topic, 0),
+            .find(|(_, cached)| cached.node.clone().is_some_and(|n| n.id() == node_id))
+        {
+            if let Some(channel) = self.p2p.get_channel(*channel_id) {
+                if channel.session_type_id() & SESSION_DIRECT == 0 {
+                    if channel.is_stopped() {
+                        self.cleanup_channel(channel).await;
+                    } else {
+                        return Ok((channel, cached.node.clone().unwrap()))
+                    }
                 }
             }
-            // There is no existing channel we can use, we will create one
-            else {
-                (None, topic, 0)
-            };
-
-        // If we found an existing channel we can use, try to use it
-        if let Some(channel_id) = channel_id {
-            if let Some(channel) = self.p2p.get_channel(channel_id) {
-                if channel.session_type_id() & (SESSION_SEED | SESSION_REFINE) != 0 {
-                    return Err(Error::Custom(
-                        "Could not get a channel (for DHT) as this is a seed or refine session"
-                            .to_string(),
-                    ));
-                }
+        }
 
-                if channel.is_stopped() {
-                    channel.clone().start(self.executor.clone());
-                }
+        self.create_channel_to_node(node).await
+    }
 
-                channel_cache.insert(
-                    channel_id,
-                    ChannelCacheItem { node: node.clone(), topic, usage_count: usage_count + 1 },
-                );
-                return Ok(channel);
+    /// Create a channel in the direct session, ping the peer, add the
+    /// DHT node to our buckets and the channel to our channel cache.
+    pub async fn create_channel(&self, addr: &Url) -> Result<(ChannelPtr, H::Node)> {
+        let channel = self.p2p.session_direct().get_channel(addr).await?;
+        let channel_cache = self.channel_cache.read().await;
+        if let Some(cached) = channel_cache.get(&channel.info.id) {
+            if let Some(node) = &cached.node {
+                return Ok((channel, node.clone()))
             }
         }
-
         drop(channel_cache);
 
-        // Create a channel
-        for addr in node.addresses().clone() {
-            let session_out = self.p2p.session_outbound();
-            let session_weak = Arc::downgrade(&self.p2p.session_outbound());
-
-            let connector = Connector::new(self.p2p.settings(), session_weak);
-            let dur = Duration::from_secs(self.settings.timeout);
-            let Ok(connect_res) = timeout(dur, connector.connect(&addr)).await else {
-                warn!(target: "dht::get_channel()", "Timeout trying to connect to {addr}");
-                return Err(Error::ConnectTimeout);
-            };
-            if connect_res.is_err() {
-                warn!(target: "dht::get_channel()", "Error while connecting: {}", connect_res.unwrap_err());
-                continue;
-            }
-            let (_, channel) = connect_res.unwrap();
+        let node = self.handler().await.ping(channel.clone()).await;
+        // If ping failed, cleanup the channel and abort
+        if let Err(e) = node {
+            self.cleanup_channel(channel).await;
+            return Err(e);
+        }
+        let node = node.unwrap();
+        self.add_channel_to_cache(channel.info.id, &node).await;
+        Ok((channel, node))
+    }
 
-            if channel.session_type_id() & (SESSION_SEED | SESSION_REFINE) != 0 {
-                return Err(Error::Custom(
-                    "Could not create a channel (for DHT) as this is a seed or refine session"
-                        .to_string(),
-                ));
-            }
+    pub async fn create_channel_to_node(&self, node: &H::Node) -> Result<(ChannelPtr, H::Node)> {
+        let net_settings = self.p2p.settings().read_arc().await;
+        let allowed_transports = net_settings.allowed_transports.clone();
+        drop(net_settings);
+
+        // Create a channel
+        let mut addrs = node.addresses().clone();
+        addrs.retain(|addr| allowed_transports.contains(&addr.scheme().to_string()));
+        for addr in addrs {
+            let res = self.create_channel(&addr).await;
 
-            let register_res =
-                session_out.register_channel(channel.clone(), self.executor.clone()).await;
-            if register_res.is_err() {
-                channel.clone().stop().await;
-                warn!(target: "dht::get_channel()", "Error while registering channel {}: {}", channel.info.id, register_res.unwrap_err());
+            if res.is_err() {
                 continue;
             }
 
-            let mut channel_cache = channel_cache_lock.write().await;
-            channel_cache.insert(
-                channel.info.id,
-                ChannelCacheItem { node: node.clone(), topic, usage_count: 1 },
-            );
-
-            return Ok(channel)
+            let (channel, node) = res.unwrap();
+            return Ok((channel, node));
         }
 
         Err(Error::Custom("Could not create channel".to_string()))
     }
 
-    /// Decrement the channel usage count, if it becomes 0 then set the topic
-    /// to None, so that this channel is available for another task
-    pub async fn cleanup_channel(&self, channel: ChannelPtr) {
-        let channel_cache_lock = self.channel_cache.clone();
-        let mut channel_cache = channel_cache_lock.write().await;
+    /// Insert a channel to the DHT's channel cache. If the channel is already
+    /// in the cache, `last_used` is updated.
+    pub async fn add_channel_to_cache(&self, channel_id: u32, node: &H::Node) {
+        let mut channel_cache = self.channel_cache.write().await;
+        channel_cache
+            .entry(channel_id)
+            .and_modify(|c| c.last_used = Timestamp::current_time())
+            .or_insert(ChannelCacheItem {
+                node: Some(node.clone()),
+                last_used: Timestamp::current_time(),
+                ping_received: false,
+                ping_sent: false,
+            });
+    }
 
-        if let Some(cached) = channel_cache.get_mut(&channel.info.id) {
-            if cached.usage_count > 0 {
-                cached.usage_count -= 1;
+    pub async fn wait_fully_pinged(&self, channel_id: u32) -> Result<()> {
+        loop {
+            let channel_cache = self.channel_cache.read().await;
+            let cached = channel_cache
+                .get(&channel_id)
+                .ok_or(Error::Custom("Missing channel".to_string()))?;
+            if cached.ping_received && cached.ping_sent {
+                return Ok(())
             }
+            drop(channel_cache);
+            msleep(100).await; // Wait for completion
+        }
+    }
 
-            // If the channel is not used by anything, remove the topic
-            if cached.usage_count == 0 {
-                cached.topic = None;
-            }
+    /// Call [`crate::net::session::DirectSession::cleanup_channel()`] and cleanup the DHT caches.
+    pub async fn cleanup_channel(&self, channel: ChannelPtr) {
+        let channel_cache_lock = self.channel_cache.clone();
+        let mut channel_cache = channel_cache_lock.write().await;
+        if self.p2p.session_direct().cleanup_channel(channel.clone()).await {
+            channel_cache.remove(&channel.info.id);
         }
     }
 }

src/dht/settings.rs

@@ -28,11 +28,13 @@ pub struct DhtSettings {
     pub concurrency: usize,
     /// Timeout in seconds
     pub timeout: u64,
+    /// Timeout in seconds for inbound connections
+    pub inbound_timeout: u64,
 }
 
 impl Default for DhtSettings {
     fn default() -> Self {
-        Self { k: 16, alpha: 4, concurrency: 10, timeout: 5 }
+        Self { k: 16, alpha: 4, concurrency: 10, timeout: 5, inbound_timeout: 30 }
     }
 }
 
@@ -55,6 +57,10 @@ pub struct DhtSettingsOpt {
     /// Timeout in seconds
     #[structopt(long)]
     pub dht_timeout: Option<u64>,
+
+    /// Timeout in seconds for inbound connections
+    #[structopt(long)]
+    pub dht_inbound_timeout: Option<u64>,
 }
 
 impl From<DhtSettingsOpt> for DhtSettings {
@@ -66,6 +72,7 @@ impl From<DhtSettingsOpt> for DhtSettings {
             alpha: opt.dht_alpha.unwrap_or(def.alpha),
             concurrency: opt.dht_concurrency.unwrap_or(def.concurrency),
             timeout: opt.dht_timeout.unwrap_or(def.timeout),
+            inbound_timeout: opt.dht_inbound_timeout.unwrap_or(def.inbound_timeout),
         }
     }
 }

src/dht/tasks.rs

@@ -16,58 +16,282 @@
  * along with this program.  If not, see <https://www.gnu.org/licenses/>.
  */
 
-use std::sync::Arc;
-use tracing::warn;
+use std::{sync::Arc, time::UNIX_EPOCH};
+use tracing::{error, info, warn};
 
 use crate::{
-    dht::{ChannelCacheItem, DhtHandler, DhtNode},
-    net::session::{SESSION_REFINE, SESSION_SEED},
+    dht::{event::DhtEvent, ChannelCacheItem, DhtHandler, DhtNode, SESSION_MANUAL},
+    net::{
+        hosts::HostColor,
+        session::{SESSION_INBOUND, SESSION_OUTBOUND},
+    },
+    system::sleep,
+    util::time::Timestamp,
     Result,
 };
 
+/// Handle DHT events.
+pub async fn events_task<H: DhtHandler>(handler: Arc<H>) -> Result<()> {
+    let dht = handler.dht();
+    let sub = dht.event_publisher.clone().subscribe().await;
+    loop {
+        let event = sub.receive().await;
+
+        match event {
+            // On [`DhtEvent::PingReceived`] set channel_cache.ping_received = true
+            DhtEvent::PingReceived { from, .. } => {
+                let channel_cache_lock = dht.channel_cache.clone();
+                let mut channel_cache = channel_cache_lock.write().await;
+                if let Some(cached) = channel_cache.get_mut(&from.info.id) {
+                    cached.ping_received = true;
+                }
+            }
+            // On [`DhtEvent::PingSent`] set channel_cache.ping_sent = true
+            DhtEvent::PingSent { to, .. } => {
+                let channel_cache_lock = dht.channel_cache.clone();
+                let mut channel_cache = channel_cache_lock.write().await;
+                if let Some(cached) = channel_cache.get_mut(&to.info.id) {
+                    cached.ping_sent = true;
+                }
+            }
+            _ => {}
+        }
+    }
+}
+
 /// Send a DHT ping request when there is a new channel, to know the node id of the new peer,
 /// Then fill the channel cache and the buckets
 pub async fn channel_task<H: DhtHandler>(handler: Arc<H>) -> Result<()> {
+    let channel_sub = handler.dht().p2p.hosts().subscribe_channel().await;
     loop {
-        let channel_sub = handler.dht().p2p.hosts().subscribe_channel().await;
         let res = channel_sub.receive().await;
-        channel_sub.unsubscribe().await;
         if res.is_err() {
             continue;
         }
         let channel = res.unwrap();
+
         let channel_cache_lock = handler.dht().channel_cache.clone();
         let mut channel_cache = channel_cache_lock.write().await;
 
-        // Skip this channel if it's stopped or not new.
-        if channel.is_stopped() || channel_cache.keys().any(|&k| k == channel.info.id) {
+        // Skip this channel if it's not new
+        if channel_cache.keys().any(|&k| k == channel.info.id) {
             continue;
         }
-        // Skip this channel if it's a seed or refine session.
-        if channel.session_type_id() & (SESSION_SEED | SESSION_REFINE) != 0 {
+
+        channel_cache.insert(
+            channel.info.id,
+            ChannelCacheItem {
+                node: None,
+                last_used: Timestamp::current_time(),
+                ping_received: false,
+                ping_sent: false,
+            },
+        );
+        drop(channel_cache);
+
+        // It's a manual connection
+        if channel.session_type_id() & SESSION_MANUAL != 0 {
+            let ping_res = handler.ping(channel.clone()).await;
+
+            if let Err(e) = ping_res {
+                warn!(target: "dht::channel_task()", "Error while pinging manual connection (requesting node id) {}: {e}", channel.address());
+                continue;
+            }
+        }
+
+        // It's an outbound connection
+        if channel.session_type_id() & SESSION_OUTBOUND != 0 {
+            let node = handler.ping(channel.clone()).await;
+            if node.is_err() {
+                continue;
+            }
+
             continue;
         }
+    }
+}
+
+/// Periodically send a DHT ping to known hosts. If the ping is successful, we
+/// move the host to the whitelist (updating the last seen field).
+///
+/// This is necessary to prevent unresponsive nodes staying on the whitelist,
+/// as the DHT does not require any outbound slot.
+pub async fn dht_refinery_task<H: DhtHandler>(handler: Arc<H>) -> Result<()> {
+    let interval = 60; // TODO: Make a setting
+    let min_ping_interval = 10 * 60; // TODO: Make a setting
+    let dht = handler.dht();
+    let hosts = dht.p2p.hosts();
+
+    loop {
+        let mut hostlist = hosts.container.fetch_all(HostColor::Gold);
+        hostlist.extend(hosts.container.fetch_all(HostColor::White));
+
+        // Include the greylist only if the DHT is not bootstrapped yet
+        if !handler.dht().is_bootstrapped().await {
+            hostlist.extend(hosts.container.fetch_all(HostColor::Grey));
+        }
+
+        for entry in &hostlist {
+            let url = &entry.0;
+            let host_cache = dht.host_cache.read().await;
+            let last_ping = host_cache.get(url).map(|h| h.last_ping.inner());
+            if last_ping.is_some() &&
+                last_ping.unwrap() > Timestamp::current_time().inner() - min_ping_interval
+            {
+                continue
+            }
+            drop(host_cache);
+
+            let res = dht.create_channel(url).await;
+            if res.is_err() {
+                continue
+            }
+            let (channel, _) = res.unwrap();
+            dht.cleanup_channel(channel).await;
+
+            let last_seen = UNIX_EPOCH.elapsed().unwrap().as_secs();
+            if let Err(e) = hosts.whitelist_host(url, last_seen).await {
+                error!(target: "dht::tasks::whitelist_refinery_task()", "Could not send {url} to the whitelist: {e}");
+            }
+            break
+        }
+
+        match hostlist.is_empty() {
+            true => sleep(5).await,
+            false => sleep(interval).await,
+        }
+    }
+}
+
+/// Add a node to the DHT buckets.
+/// If the bucket is already full, we ping the least recently seen node in the
+/// bucket: if successful it becomes the most recently seen node, if the ping
+/// fails we remove it and add the new node.
+pub async fn add_node_task<H: DhtHandler>(handler: Arc<H>) -> Result<()> {
+    let dht = handler.dht();
+    loop {
+        let (node, channel) = dht.add_node_rx.recv().await.unwrap();
+
+        let self_node = handler.node().await;
 
-        let ping_res = handler.ping(channel.clone()).await;
+        let bucket_index = dht.get_bucket_index(&self_node.id(), &node.id()).await;
+        let buckets_lock = dht.buckets.clone();
+        let mut buckets = buckets_lock.write().await;
+        let bucket = &mut buckets[bucket_index];
 
-        if let Err(e) = ping_res {
-            warn!(target: "dht::channel_task()", "Error while pinging (requesting node id) {}: {e}", channel.display_address());
-            // channel.stop().await;
+        // Do not add ourselves to the buckets
+        if node.id() == self_node.id() {
             continue;
         }
 
-        let node = ping_res.unwrap();
+        // Don't add this node if it has any external address that is the same as one of ours
+        let node_addresses = node.addresses();
+        if self_node.addresses().iter().any(|addr| node_addresses.contains(addr)) {
+            continue;
+        }
 
-        channel_cache.entry(channel.info.id).or_insert_with(|| ChannelCacheItem {
-            node: node.clone(),
-            topic: None,
-            usage_count: 0,
-        });
-        drop(channel_cache);
+        // Do not add a node to the buckets if it does not have an address
+        if node.addresses().is_empty() {
+            continue;
+        }
+
+        // We already have this node, move it to the tail of the bucket
+        if let Some(node_index) = bucket.nodes.iter().position(|n| n.id() == node.id()) {
+            bucket.nodes.remove(node_index);
+            bucket.nodes.push(node);
+            continue;
+        }
+
+        // Bucket is full
+        if bucket.nodes.len() >= handler.dht().settings.k {
+            // Ping the least recently seen node
+            if let Ok((channel, node)) = handler.dht().get_channel(&bucket.nodes[0]).await {
+                // Ping was successful, move the least recently seen node to the tail
+                let n = bucket.nodes.remove(0);
+                bucket.nodes.push(n);
+                drop(buckets);
+                dht.on_new_node(&node.clone(), channel.clone()).await;
+                handler.dht().cleanup_channel(channel).await;
+                continue;
+            }
+
+            // Ping was not successful, remove the least recently seen node and add the new node
+            bucket.nodes.remove(0);
+            bucket.nodes.push(node.clone());
+            drop(buckets);
+            dht.on_new_node(&node.clone(), channel.clone()).await;
+            continue;
+        }
+
+        // Bucket is not full, just add the node
+        bucket.nodes.push(node.clone());
+        drop(buckets);
+        dht.on_new_node(&node.clone(), channel.clone()).await;
+    }
+}
+
+/// Close inbound connections that are unused for too long.
+pub async fn disconnect_inbounds_task<H: DhtHandler>(handler: Arc<H>) -> Result<()> {
+    let interval = 10; // TODO: Make a setting
+    let dht = handler.dht();
+
+    loop {
+        sleep(interval).await;
+
+        let min_last_used = Timestamp::current_time().inner() - dht.settings.inbound_timeout;
+
+        let channel_cache_lock = dht.channel_cache.clone();
+        let mut channel_cache = channel_cache_lock.write().await;
+
+        for (channel_id, cached) in channel_cache.clone() {
+            // Check that:
+            // The channel timed out,
+            if cached.last_used.inner() >= min_last_used {
+                continue;
+            }
+            // The channel exists,
+            let channel = dht.p2p.get_channel(channel_id);
+            if channel.is_none() {
+                channel_cache.remove(&channel_id);
+                continue;
+            }
+            let channel = channel.unwrap();
+            // And the channel is inbound.
+            if channel.session_type_id() & SESSION_INBOUND == 0 {
+                continue;
+            }
+
+            // Now we can stop it and remove it from the channel cache
+            info!(target: "dht::disconnect_inbounds_task()", "Closing expired inbound channel [{}]", channel.address());
+            channel.stop().await;
+            channel_cache.remove(&channel.info.id);
+        }
+    }
+}
+
+/// Removes entries from [`crate::dht::Dht::channel_cache`] when a channel is
+/// stopped.
+pub async fn cleanup_channels_task<H: DhtHandler>(handler: Arc<H>) -> Result<()> {
+    let interval = 60; // TODO: Make a setting
+    let dht = handler.dht();
+
+    loop {
+        sleep(interval).await;
+
+        let channel_cache_lock = dht.channel_cache.clone();
+        let mut channel_cache = channel_cache_lock.write().await;
 
-        if !node.addresses().is_empty() {
-            handler.dht().add_node(node.clone()).await;
-            let _ = handler.on_new_node(&node.clone()).await;
+        for (channel_id, _) in channel_cache.clone() {
+            match dht.p2p.get_channel(channel_id) {
+                Some(channel) => {
+                    if channel.is_stopped() {
+                        channel_cache.remove(&channel_id);
+                    }
+                }
+                None => {
+                    channel_cache.remove(&channel_id);
+                }
+            }
         }
     }
 }