fud, fu: autoseed, `get` rpc subscriber, progress bar, bug fixes

bin/fud/fu/src/main.rs

@@ -17,14 +17,23 @@
  */
 
 use clap::{Parser, Subcommand};
-use log::info;
+use log::error;
 use simplelog::{ColorChoice, TermLogger, TerminalMode};
-use std::{collections::HashMap, sync::Arc};
+use std::{
+    collections::HashMap,
+    io::{stdout, Write},
+    sync::Arc,
+};
 use url::Url;
 
 use darkfi::{
     cli_desc,
-    rpc::{client::RpcClient, jsonrpc::JsonRequest, util::JsonValue},
+    rpc::{
+        client::RpcClient,
+        jsonrpc::{ErrorCode, JsonError, JsonRequest, JsonResult},
+        util::JsonValue,
+    },
+    system::{ExecutorPtr, Publisher, StoppableTask},
     util::cli::{get_log_config, get_log_level},
     Error, Result,
 };
@@ -69,20 +78,132 @@ enum Subcmd {
 }
 
 struct Fu {
-    pub rpc_client: RpcClient,
+    pub rpc_client: Arc<RpcClient>,
 }
 
 impl Fu {
-    async fn close_connection(&self) {
-        self.rpc_client.stop().await;
-    }
+    async fn get(
+        &self,
+        file_hash: String,
+        file_name: Option<String>,
+        ex: ExecutorPtr,
+    ) -> Result<()> {
+        let publisher = Publisher::new();
+        let subscription = Arc::new(publisher.clone().subscribe().await);
+        let subscriber_task = StoppableTask::new();
+        let file_hash_ = file_hash.clone();
+        let publisher_ = publisher.clone();
+        let rpc_client_ = self.rpc_client.clone();
+        subscriber_task.clone().start(
+            async move {
+                let req = JsonRequest::new(
+                    "get",
+                    JsonValue::Array(vec![
+                        JsonValue::String(file_hash_),
+                        JsonValue::String(file_name.unwrap_or_default()),
+                    ]),
+                );
+                rpc_client_.subscribe(req, publisher).await
+            },
+            move |res| async move {
+                match res {
+                    Ok(()) | Err(Error::DetachedTaskStopped) => { /* Do nothing */ }
+                    Err(e) => {
+                        error!("{}", e);
+                        publisher_
+                            .notify(JsonResult::Error(JsonError::new(
+                                ErrorCode::InternalError,
+                                None,
+                                0,
+                            )))
+                            .await;
+                    }
+                }
+            },
+            Error::DetachedTaskStopped,
+            ex,
+        );
 
-    async fn get(&self, file_hash: String, file_name: Option<String>) -> Result<()> {
-        let req = JsonRequest::new("get", JsonValue::Array(vec![JsonValue::String(file_hash), JsonValue::String(file_name.unwrap_or_default())]));
-        let rep = self.rpc_client.request(req).await?;
-        let path: String = rep.try_into().unwrap();
-        println!("{}", path);
-        Ok(())
+        let progress_bar_width = 20;
+        let mut chunks_total = 0;
+        let mut chunks_downloaded = 0;
+
+        let print_progress_bar = |chunks_downloaded: usize, chunks_total: usize| {
+            let completed = (chunks_downloaded as f64 / chunks_total as f64 *
+                progress_bar_width as f64) as usize;
+            let remaining = progress_bar_width - completed;
+            let bar = "=".repeat(completed) + &" ".repeat(remaining);
+            print!("\r[{}] {}/{} chunks", bar, chunks_downloaded, chunks_total);
+            stdout().flush().unwrap();
+        };
+
+        loop {
+            match subscription.receive().await {
+                JsonResult::Notification(n) => {
+                    let params = n.params.get::<HashMap<String, JsonValue>>().unwrap();
+                    match params.get("event").unwrap().get::<String>().unwrap().as_str() {
+                        "file_download_completed" => {
+                            let info = params
+                                .get("info")
+                                .unwrap()
+                                .get::<HashMap<String, JsonValue>>()
+                                .unwrap();
+                            chunks_total =
+                                *info.get("chunk_count").unwrap().get::<f64>().unwrap() as usize;
+                            print_progress_bar(chunks_downloaded, chunks_total);
+                        }
+                        "chunk_download_completed" => {
+                            chunks_downloaded += 1;
+                            print_progress_bar(chunks_downloaded, chunks_total);
+                        }
+                        "download_completed" => {
+                            let info = params
+                                .get("info")
+                                .unwrap()
+                                .get::<HashMap<String, JsonValue>>()
+                                .unwrap();
+                            let file_path = info.get("file_path").unwrap().get::<String>().unwrap();
+                            chunks_downloaded = chunks_total;
+                            print_progress_bar(chunks_downloaded, chunks_total);
+                            println!("\nDownload completed:\n{}", file_path);
+                            return Ok(());
+                        }
+                        "file_not_found" => {
+                            return Err(Error::Custom(format!("Could not find file {}", file_hash)));
+                        }
+                        "chunk_not_found" => {
+                            let info = params
+                                .get("info")
+                                .unwrap()
+                                .get::<HashMap<String, JsonValue>>()
+                                .unwrap();
+                            let chunk_hash =
+                                info.get("chunk_hash").unwrap().get::<String>().unwrap();
+                            println!();
+                            return Err(Error::Custom(format!(
+                                "Could not find chunk {}",
+                                chunk_hash
+                            )));
+                        }
+                        "missing_chunks" => {
+                            println!();
+                            return Err(Error::Custom("Missing chunks".to_string()));
+                        }
+                        _ => {}
+                    }
+                }
+
+                JsonResult::Error(e) => {
+                    return Err(Error::UnexpectedJsonRpc(format!("Got error from JSON-RPC: {e:?}")))
+                }
+
+                x => {
+                    return Err(Error::UnexpectedJsonRpc(format!(
+                        "Got unexpected data from JSON-RPC: {x:?}"
+                    )))
+                }
+            }
+        }
     }
 
     async fn put(&self, file: String) -> Result<()> {
@@ -101,11 +222,13 @@ impl Fu {
         let req = JsonRequest::new("list_buckets", JsonValue::Array(vec![]));
         let rep = self.rpc_client.request(req).await?;
         let buckets: Vec<JsonValue> = rep.try_into().unwrap();
+        let mut empty = true;
         for (bucket_i, bucket) in buckets.into_iter().enumerate() {
             let nodes: Vec<JsonValue> = bucket.try_into().unwrap();
-            if nodes.len() == 0 {
+            if nodes.is_empty() {
                 continue
             }
+            empty = false;
 
             println!("Bucket {}", bucket_i);
             for n in nodes.clone() {
@@ -120,6 +243,10 @@ impl Fu {
             }
         }
 
+        if empty {
+            println!("All buckets are empty");
+        }
+
         Ok(())
     }
 
@@ -129,9 +256,8 @@ impl Fu {
 
         let files: HashMap<String, JsonValue> = rep["seeders"].clone().try_into().unwrap();
 
-        println!("Seeders:");
         if files.is_empty() {
-            println!("No records");
+            println!("No known seeders");
         } else {
             for (file_hash, node_ids) in files {
                 println!("{}", file_hash);
@@ -157,20 +283,16 @@ fn main() -> Result<()> {
     let ex = Arc::new(smol::Executor::new());
     smol::block_on(async {
         ex.run(async {
-            let rpc_client = RpcClient::new(args.endpoint, ex.clone()).await?;
+            let rpc_client = Arc::new(RpcClient::new(args.endpoint.clone(), ex.clone()).await?);
             let fu = Fu { rpc_client };
 
             match args.command {
-                // Subcmd::List => fu.list().await,
-                // Subcmd::Sync => fu.sync().await,
-                Subcmd::Get { file, name } => fu.get(file, name).await,
+                Subcmd::Get { file, name } => fu.get(file, name, ex.clone()).await,
                 Subcmd::Put { file } => fu.put(file).await,
-                Subcmd::ListBuckets { } => fu.list_buckets().await,
-                Subcmd::ListSeeders { } => fu.list_seeders().await,
+                Subcmd::ListBuckets {} => fu.list_buckets().await,
+                Subcmd::ListSeeders {} => fu.list_seeders().await,
             }?;
 
-            fu.close_connection().await;
-
             Ok(())
         })
         .await

bin/fud/fud/fud_config.toml

@@ -29,10 +29,10 @@ hostlist = "~/.local/share/darkfi/fud/p2p_hostlist.tsv"
 # inbound = ["tcp://0.0.0.0:13337"]
 
 ## Outbound connection slots
-# outbound_connections = 8
+# outbound_connections = 16
 
 ## Inbound connection slots
-#inbound_connections = 8
+#inbound_connections = 16
 
 ## White connection percent
 # gold_connect_count = 2

bin/fud/fud/src/dht.rs

@@ -18,6 +18,7 @@
 
 use std::{
     collections::{HashMap, HashSet},
+    hash::{Hash, Hasher},
     sync::Arc,
 };
 
@@ -25,6 +26,7 @@ use async_trait::async_trait;
 use darkfi::{
     net::{connector::Connector, session::Session, ChannelPtr, Message, P2pPtr},
     system::{sleep, ExecutorPtr},
+    util::time::Timestamp,
     Error, Result,
 };
 use darkfi_serial::{SerialDecodable, SerialEncodable};
@@ -34,18 +36,54 @@ use num_bigint::BigUint;
 use smol::lock::RwLock;
 use url::Url;
 
-#[derive(Debug, Clone, SerialEncodable, SerialDecodable, PartialEq, Eq, Hash)]
+#[derive(Debug, Clone, SerialEncodable, SerialDecodable, Eq)]
 pub struct DhtNode {
     pub id: blake3::Hash,
     pub addresses: Vec<Url>,
 }
 
+impl Hash for DhtNode {
+    fn hash<H: Hasher>(&self, state: &mut H) {
+        self.id.hash(state);
+    }
+}
+
+impl PartialEq for DhtNode {
+    fn eq(&self, other: &Self) -> bool {
+        self.id == other.id
+    }
+}
+
 pub struct DhtBucket {
     pub nodes: Vec<DhtNode>,
 }
 
-/// "Router" means: Key -> Set of nodes
-pub type DhtRouter = Arc<RwLock<HashMap<blake3::Hash, HashSet<DhtNode>>>>;
+/// "Router" means: Key -> Set of nodes (+ additional data for each node)
+pub type DhtRouterPtr = Arc<RwLock<HashMap<blake3::Hash, HashSet<DhtRouterItem>>>>;
+
+#[derive(Debug, Clone, SerialEncodable, SerialDecodable, Eq)]
+pub struct DhtRouterItem {
+    pub node: DhtNode,
+    pub timestamp: u64,
+}
+
+impl Hash for DhtRouterItem {
+    fn hash<H: Hasher>(&self, state: &mut H) {
+        self.node.id.hash(state);
+    }
+}
+
+impl PartialEq for DhtRouterItem {
+    fn eq(&self, other: &Self) -> bool {
+        self.node.id == other.node.id
+    }
+}
+
+impl From<DhtNode> for DhtRouterItem {
+    fn from(node: DhtNode) -> Self {
+        DhtRouterItem { node, timestamp: Timestamp::current_time().inner() }
+    }
+}
 
 // TODO: Add a DhtSettings
 pub struct Dht {
@@ -106,7 +144,7 @@ impl Dht {
 
     pub async fn is_bootstrapped(&self) -> bool {
         let bootstrapped = self.bootstrapped.read().await;
-        return *bootstrapped;
+        *bootstrapped
     }
 
     pub async fn set_bootstrapped(&self) {
@@ -129,7 +167,7 @@ impl Dht {
     }
 
     // Sort `nodes`
-    pub fn sort_by_distance(&self, nodes: &mut Vec<DhtNode>, key: &blake3::Hash) {
+    pub fn sort_by_distance(&self, nodes: &mut [DhtNode], key: &blake3::Hash) {
         nodes.sort_by(|a, b| {
             let distance_a = BigUint::from_bytes_be(&self.distance(key, &a.id));
             let distance_b = BigUint::from_bytes_be(&self.distance(key, &b.id));
@@ -179,11 +217,28 @@ impl Dht {
         neighbors
     }
 
+    // Channel ID -> DhtNode
     pub async fn get_node_from_channel(&self, channel_id: u32) -> Option<DhtNode> {
         let node_cache_lock = self.node_cache.clone();
         let node_cache = node_cache_lock.read().await;
         node_cache.get(&channel_id).cloned()
     }
+
+    // Remove nodes in router that are older than expiry_secs
+    pub async fn prune_router(&self, router: DhtRouterPtr, expiry_secs: u32) {
+        let expiry_timestamp = Timestamp::current_time().inner() - (expiry_secs as u64);
+        let mut router_write = router.write().await;
+
+        let keys: Vec<_> = router_write.keys().cloned().collect();
+
+        for key in keys {
+            let items = router_write.get_mut(&key).unwrap();
+            items.retain(|item| item.timestamp > expiry_timestamp);
+            if items.is_empty() {
+                router_write.remove(&key);
+            }
+        }
+    }
 }
 
 #[async_trait]
@@ -204,9 +259,13 @@ pub trait DhtHandler {
         &self,
         key: &blake3::Hash,
         message: &M,
-        router: DhtRouter,
+        router: DhtRouterPtr,
     ) -> Result<()> {
-        self.add_to_router(router.clone(), key, vec![self.dht().node.clone()]).await;
+        if self.dht().node.addresses.is_empty() {
+            return Err(().into()); // TODO
+        }
+
+        self.add_to_router(router.clone(), key, vec![self.dht().node.clone().into()]).await;
         let nodes = self.lookup_nodes(key).await?;
 
         for node in nodes {
@@ -243,9 +302,10 @@ pub trait DhtHandler {
                     node_cache.insert(channel.info.id, n.clone());
                     drop(node_cache);
 
-                    self.add_node(n.clone()).await;
-
-                    let _ = self.on_new_node(&n.clone()).await;
+                    if !n.addresses.is_empty() {
+                        self.add_node(n.clone()).await;
+                        let _ = self.on_new_node(&n.clone()).await;
+                    }
                 }
             }
         }
@@ -269,10 +329,16 @@ pub trait DhtHandler {
 
     // Add a node in the correct bucket
     async fn add_node(&self, node: DhtNode) {
+        // Do not add ourselves to the buckets
         if node.id == self.dht().node.id {
             return;
         }
 
+        // Do not add a node to the buckets if it does not have an address
+        if node.addresses.is_empty() {
+            return;
+        }
+
         let bucket_index = self.dht().get_bucket_index(&node.id).await;
         let buckets_lock = self.dht().buckets.clone();
         let mut buckets = buckets_lock.write().await;
@@ -446,8 +512,16 @@ pub trait DhtHandler {
     }
 
     // Add nodes as a provider for a key
-    async fn add_to_router(&self, router: DhtRouter, key: &blake3::Hash, nodes: Vec<DhtNode>) {
-        debug!(target: "dht::DhtHandler::add_to_router()", "Inserting {} nodes to key {}", nodes.len(), key);
+    async fn add_to_router(
+        &self,
+        router: DhtRouterPtr,
+        key: &blake3::Hash,
+        router_items: Vec<DhtRouterItem>,
+    ) {
+        let mut router_items = router_items.clone();
+        router_items.retain(|item| !item.node.addresses.is_empty());
+
+        debug!(target: "dht::DhtHandler::add_to_router()", "Inserting {} nodes to key {}", router_items.len(), key);
 
         let mut router_write = router.write().await;
         let key_r = router_write.get_mut(key);
@@ -457,22 +531,22 @@ pub trait DhtHandler {
 
         // Add to router
         if let Some(k) = key_r {
-            k.extend(nodes.clone());
+            k.extend(router_items.clone());
         } else {
             let mut hs = HashSet::new();
-            hs.extend(nodes.clone());
+            hs.extend(router_items.clone());
             router_write.insert(*key, hs);
         }
 
         // Add to router_cache
-        for node in nodes {
-            let keys = router_cache.get_mut(&node.id);
+        for router_item in router_items {
+            let keys = router_cache.get_mut(&router_item.node.id);
             if let Some(k) = keys {
                 k.insert(*key);
             } else {
                 let mut keys = HashSet::new();
                 keys.insert(*key);
-                router_cache.insert(node.id, keys);
+                router_cache.insert(router_item.node.id, keys);
             }
         }
     }

bin/fud/fud/src/main.rs

@@ -24,37 +24,35 @@ use std::{
 };
 
 use num_bigint::BigUint;
+use tasks::FetchReply;
 
+use crate::rpc::FudEvent;
 use async_trait::async_trait;
-use dht::{Dht, DhtHandler, DhtNode, DhtRouter};
-use futures::{
-    future::{try_select, Either, FutureExt},
-    pin_mut,
-};
+use dht::{Dht, DhtHandler, DhtNode, DhtRouterItem, DhtRouterPtr};
+use futures::{future::FutureExt, pin_mut, select};
 use log::{debug, error, info, warn};
 use rand::{rngs::OsRng, RngCore};
 use smol::{
     channel,
     fs::{File, OpenOptions},
     io::{AsyncReadExt, AsyncWriteExt},
-    lock::{Mutex, MutexGuard, RwLock},
+    lock::{Mutex, RwLock},
     stream::StreamExt,
     Executor,
 };
 use structopt_toml::{structopt::StructOpt, StructOptToml};
-use tinyjson::JsonValue;
 
 use darkfi::{
     async_daemonize, cli_desc,
     geode::Geode,
     net::{session::SESSION_DEFAULT, settings::SettingsOpt, ChannelPtr, P2p, P2pPtr},
     rpc::{
-        jsonrpc::{ErrorCode, JsonError, JsonRequest, JsonResponse, JsonResult, JsonSubscriber},
+        jsonrpc::JsonSubscriber,
         p2p_method::HandlerP2p,
         server::{listen_and_serve, RequestHandler},
         settings::{RpcSettings, RpcSettingsOpt},
     },
-    system::{StoppableTask, StoppableTaskPtr},
+    system::{Publisher, PublisherPtr, StoppableTask, StoppableTaskPtr},
     util::path::expand_path,
     Error, Result,
 };
@@ -63,11 +61,13 @@ use darkfi::{
 mod proto;
 use proto::{
     FudAnnounce, FudChunkReply, FudFileReply, FudFindNodesReply, FudFindNodesRequest,
-    FudFindRequest, FudFindSeedersReply, FudFindSeedersRequest, FudPingReply, FudPingRequest,
-    ProtocolFud,
+    FudFindRequest, FudFindSeedersReply, FudFindSeedersRequest, FudNotFound, FudPingReply,
+    FudPingRequest, ProtocolFud,
 };
 
 mod dht;
+mod rpc;
+mod tasks;
 
 const CONFIG_FILE: &str = "fud_config.toml";
 const CONFIG_FILE_CONTENTS: &str = include_str!("../fud_config.toml");
@@ -105,7 +105,7 @@ struct Args {
 
 pub struct Fud {
     /// Key -> Seeders
-    seeders_router: DhtRouter,
+    seeders_router: DhtRouterPtr,
 
     /// Pointer to the P2P network instance
     p2p: P2pPtr,
@@ -116,6 +116,8 @@ pub struct Fud {
     /// The DHT instance
     dht: Arc<Dht>,
 
+    get_tx: channel::Sender<(u16, blake3::Hash, Option<String>, Result<()>)>,
+    get_rx: channel::Receiver<(u16, blake3::Hash, Option<String>, Result<()>)>,
     file_fetch_tx: channel::Sender<(blake3::Hash, Result<()>)>,
     file_fetch_rx: channel::Receiver<(blake3::Hash, Result<()>)>,
     file_fetch_end_tx: channel::Sender<(blake3::Hash, Result<()>)>,
@@ -129,283 +131,11 @@ pub struct Fud {
 
     /// dnet JSON-RPC subscriber
     dnet_sub: JsonSubscriber,
-}
-
-#[async_trait]
-impl RequestHandler<()> for Fud {
-    async fn handle_request(&self, req: JsonRequest) -> JsonResult {
-        return match req.method.as_str() {
-            "ping" => self.pong(req.id, req.params).await,
-
-            "put" => self.put(req.id, req.params).await,
-            "get" => self.get(req.id, req.params).await,
-
-            "dnet.switch" => self.dnet_switch(req.id, req.params).await,
-            "dnet.subscribe_events" => self.dnet_subscribe_events(req.id, req.params).await,
-            "p2p.get_info" => self.p2p_get_info(req.id, req.params).await,
-            "list_buckets" => self.list_buckets(req.id, req.params).await,
-            "list_seeders" => self.list_seeders(req.id, req.params).await,
-            _ => JsonError::new(ErrorCode::MethodNotFound, None, req.id).into(),
-        }
-    }
-
-    async fn connections_mut(&self) -> MutexGuard<'_, HashSet<StoppableTaskPtr>> {
-        self.rpc_connections.lock().await
-    }
-}
-
-impl Fud {
-    // RPCAPI:
-    // Put a file onto the network. Takes a local filesystem path as a parameter.
-    // Returns the file hash that serves as a pointer to the uploaded file.
-    //
-    // --> {"jsonrpc": "2.0", "method": "put", "params": ["/foo.txt"], "id": 42}
-    // <-- {"jsonrpc": "2.0", "result: "df4...3db7", "id": 42}
-    async fn put(&self, id: u16, params: JsonValue) -> JsonResult {
-        let params = params.get::<Vec<JsonValue>>().unwrap();
-        if params.len() != 1 || !params[0].is_string() {
-            return JsonError::new(ErrorCode::InvalidParams, None, id).into()
-        }
-
-        let path = params[0].get::<String>().unwrap();
-        let path = match expand_path(path.as_str()) {
-            Ok(v) => v,
-            Err(_) => return JsonError::new(ErrorCode::InvalidParams, None, id).into(),
-        };
-
-        // A valid path was passed. Let's see if we can read it, and if so,
-        // add it to Geode.
-        let fd = match File::open(&path).await {
-            Ok(v) => v,
-            Err(e) => {
-                error!("Failed to open {:?}: {}", path, e);
-                return JsonError::new(ErrorCode::InvalidParams, None, id).into()
-            }
-        };
-
-        let (file_hash, chunk_hashes) = match self.geode.insert(fd).await {
-            Ok(v) => v,
-            Err(e) => {
-                error!("Failed inserting file {:?} to geode: {}", path, e);
-                return JsonError::new(ErrorCode::InternalError, None, id).into()
-            }
-        };
-
-        // Announce file
-        let self_node = self.dht.node.clone();
-        let fud_announce = FudAnnounce { key: file_hash, nodes: vec![self_node.clone()] };
-        let _ = self.announce(&file_hash, &fud_announce, self.seeders_router.clone()).await;
-
-        // Announce chunks
-        for chunk_hash in chunk_hashes {
-            let fud_announce = FudAnnounce { key: chunk_hash, nodes: vec![self_node.clone()] };
-            let _ = self.announce(&chunk_hash, &fud_announce, self.seeders_router.clone()).await;
-        }
-
-        JsonResponse::new(JsonValue::String(file_hash.to_hex().to_string()), id).into()
-    }
-
-    // RPCAPI:
-    // Fetch a file from the network. Takes a file hash as parameter.
-    // Returns the path to the assembled file, if found/fetched.
-    //
-    // --> {"jsonrpc": "2.0", "method": "get", "params": ["1211...abfd"], "id": 42}
-    // <-- {"jsonrpc": "2.0", "result: "~/.local/share/darkfi/fud/downloads/fab1...2314", "id": 42}
-    async fn get(&self, id: u16, params: JsonValue) -> JsonResult {
-        let params = params.get::<Vec<JsonValue>>().unwrap();
-        if params.len() != 2 || !params[0].is_string() || !params[1].is_string() {
-            return JsonError::new(ErrorCode::InvalidParams, None, id).into()
-        }
-
-        let file_name: Option<String> = match params[1].get::<String>() {
-            Some(name) => match name.is_empty() {
-                true => None,
-                false => Some(name.clone()),
-            },
-            None => None,
-        };
-
-        let self_node = self.dht.node.clone();
-
-        let file_hash = match blake3::Hash::from_hex(params[0].get::<String>().unwrap()) {
-            Ok(v) => v,
-            Err(_) => return JsonError::new(ErrorCode::InvalidParams, None, id).into(),
-        };
-
-        let chunked_file = match self.geode.get(&file_hash).await {
-            Ok(v) => v,
-            Err(Error::GeodeNeedsGc) => todo!(),
-            Err(Error::GeodeFileNotFound) => {
-                info!("Requested file {} not found in Geode, triggering fetch", file_hash);
-                self.file_fetch_tx.send((file_hash, Ok(()))).await.unwrap();
-                info!("Waiting for background file fetch task...");
-                let (i_file_hash, status) = self.file_fetch_end_rx.recv().await.unwrap();
-                match status {
-                    Ok(()) => self.geode.get(&i_file_hash).await.unwrap(),
-
-                    Err(Error::GeodeFileRouteNotFound) => {
-                        // TODO: Return FileNotFound error
-                        return JsonError::new(ErrorCode::InternalError, None, id).into()
-                    }
-
-                    Err(e) => panic!("{}", e),
-                }
-            }
-
-            Err(e) => panic!("{}", e),
-        };
-
-        if chunked_file.is_complete() {
-            let fud_announce = FudAnnounce { key: file_hash, nodes: vec![self_node.clone()] };
-            let _ = self.announce(&file_hash, &fud_announce, self.seeders_router.clone()).await;
-
-            return match self.geode.assemble_file(&file_hash, &chunked_file, file_name).await {
-                Ok(file_path) => JsonResponse::new(
-                    JsonValue::String(file_path.to_string_lossy().to_string()),
-                    id,
-                )
-                .into(),
-                Err(_) => JsonError::new(ErrorCode::InternalError, None, id).into(),
-            }
-        }
-
-        // Fetch any missing chunks
-        let mut missing_chunks = vec![];
-        for (chunk, path) in chunked_file.iter() {
-            if path.is_none() {
-                missing_chunks.push(*chunk);
-            }
-        }
-
-        for chunk in missing_chunks {
-            self.chunk_fetch_tx.send((chunk, Ok(()))).await.unwrap();
-            let (i_chunk_hash, status) = self.chunk_fetch_end_rx.recv().await.unwrap();
-
-            match status {
-                Ok(()) => {
-                    let fud_announce =
-                        FudAnnounce { key: i_chunk_hash, nodes: vec![self_node.clone()] };
-                    let _ = self
-                        .announce(&i_chunk_hash, &fud_announce, self.seeders_router.clone())
-                        .await;
-                }
-                Err(Error::GeodeChunkRouteNotFound) => continue,
-
-                Err(e) => panic!("{}", e),
-            };
-        }
-
-        let chunked_file = match self.geode.get(&file_hash).await {
-            Ok(v) => v,
-            Err(e) => panic!("{}", e),
-        };
-
-        if !chunked_file.is_complete() {
-            todo!();
-            // TODO: Return JsonError missing chunks
-        }
-
-        return match self.geode.assemble_file(&file_hash, &chunked_file, file_name).await {
-            Ok(file_path) => {
-                JsonResponse::new(JsonValue::String(file_path.to_string_lossy().to_string()), id)
-                    .into()
-            }
-            Err(_) => JsonError::new(ErrorCode::InternalError, None, id).into(),
-        }
-    }
-
-    // RPCAPI:
-    // Activate or deactivate dnet in the P2P stack.
-    // By sending `true`, dnet will be activated, and by sending `false` dnet
-    // will be deactivated. Returns `true` on success.
-    //
-    // --> {"jsonrpc": "2.0", "method": "dnet_switch", "params": [true], "id": 42}
-    // <-- {"jsonrpc": "2.0", "result": true, "id": 42}
-    async fn dnet_switch(&self, id: u16, params: JsonValue) -> JsonResult {
-        let params = params.get::<Vec<JsonValue>>().unwrap();
-        if params.len() != 1 || !params[0].is_bool() {
-            return JsonError::new(ErrorCode::InvalidParams, None, id).into()
-        }
-
-        let switch = params[0].get::<bool>().unwrap();
-
-        if *switch {
-            self.p2p.dnet_enable();
-        } else {
-            self.p2p.dnet_disable();
-        }
-
-        JsonResponse::new(JsonValue::Boolean(true), id).into()
-    }
-
-    // RPCAPI:
-    // Initializes a subscription to p2p dnet events.
-    // Once a subscription is established, `fud` will send JSON-RPC notifications of
-    // new network events to the subscriber.
-    //
-    // --> {"jsonrpc": "2.0", "method": "dnet.subscribe_events", "params": [], "id": 1}
-    // <-- {"jsonrpc": "2.0", "method": "dnet.subscribe_events", "params": [`event`]}
-    pub async fn dnet_subscribe_events(&self, id: u16, params: JsonValue) -> JsonResult {
-        let params = params.get::<Vec<JsonValue>>().unwrap();
-        if !params.is_empty() {
-            return JsonError::new(ErrorCode::InvalidParams, None, id).into()
-        }
-
-        self.dnet_sub.clone().into()
-    }
 
-    // RPCAPI:
-    // Returns the current buckets
-    //
-    // --> {"jsonrpc": "2.0", "method": "list_buckets", "params": [], "id": 1}
-    // <-- {"jsonrpc": "2.0", "result": [[["abcdef", ["tcp://127.0.0.1:13337"]]]], "id": 1}
-    pub async fn list_buckets(&self, id: u16, params: JsonValue) -> JsonResult {
-        let params = params.get::<Vec<JsonValue>>().unwrap();
-        if !params.is_empty() {
-            return JsonError::new(ErrorCode::InvalidParams, None, id).into()
-        }
-        let mut buckets = vec![];
-        for bucket in self.dht.buckets.read().await.iter() {
-            let mut nodes = vec![];
-            for node in bucket.nodes.clone() {
-                let mut addresses = vec![];
-                for addr in node.addresses {
-                    addresses.push(JsonValue::String(addr.to_string()));
-                }
-                nodes.push(JsonValue::Array(vec![
-                    JsonValue::String(node.id.to_hex().to_string()),
-                    JsonValue::Array(addresses),
-                ]));
-            }
-            buckets.push(JsonValue::Array(nodes));
-        }
+    /// Download JSON-RPC subscriber
+    download_sub: JsonSubscriber,
 
-        JsonResponse::new(JsonValue::Array(buckets), id).into()
-    }
-
-    // RPCAPI:
-    // Returns the content of the seeders router
-    //
-    // --> {"jsonrpc": "2.0", "method": "list_routes", "params": [], "id": 1}
-    // <-- {"jsonrpc": "2.0", "result": {"seeders": {"abcdef": ["ghijkl"]}}, "id": 1}
-    pub async fn list_seeders(&self, id: u16, params: JsonValue) -> JsonResult {
-        let params = params.get::<Vec<JsonValue>>().unwrap();
-        if !params.is_empty() {
-            return JsonError::new(ErrorCode::InvalidParams, None, id).into()
-        }
-        let mut seeders_router: HashMap<String, JsonValue> = HashMap::new();
-        for (hash, nodes) in self.seeders_router.read().await.iter() {
-            let mut node_ids = vec![];
-            for node in nodes {
-                node_ids.push(JsonValue::String(node.id.to_hex().to_string()));
-            }
-            seeders_router.insert(hash.to_hex().to_string(), JsonValue::Array(node_ids));
-        }
-        let mut res: HashMap<String, JsonValue> = HashMap::new();
-        res.insert("seeders".to_string(), JsonValue::Object(seeders_router));
-
-        JsonResponse::new(JsonValue::Object(res), id).into()
-    }
+    download_publisher: PublisherPtr<FudEvent>,
 }
 
 impl HandlerP2p for Fud {
@@ -414,215 +144,6 @@ impl HandlerP2p for Fud {
     }
 }
 
-enum FetchReply {
-    File(FudFileReply),
-    Chunk(FudChunkReply),
-}
-
-/// Fetch a file or chunk from the network
-/// 1. Lookup nodes close to the key
-/// 2. Request seeders for the file/chunk from those nodes
-/// 3. Request the file/chunk from the seeders
-async fn fetch(fud: Arc<Fud>, key: blake3::Hash) -> Option<FetchReply> {
-    let mut queried_seeders: HashSet<blake3::Hash> = HashSet::new();
-    let closest_nodes = fud.lookup_nodes(&key).await; // 1
-    let mut result: Option<FetchReply> = None;
-    if closest_nodes.is_err() {
-        return None
-    }
-
-    for node in closest_nodes.unwrap() {
-        // 2. Request list of seeders
-        let channel = match fud.get_channel(&node).await {
-            Ok(channel) => channel,
-            Err(e) => {
-                warn!(target: "fud::fetch()", "Could not get a channel for node {}: {}", node.id, e);
-                continue;
-            }
-        };
-        let msg_subsystem = channel.message_subsystem();
-        msg_subsystem.add_dispatch::<FudFindSeedersReply>().await;
-
-        let msg_subscriber = match channel.subscribe_msg::<FudFindSeedersReply>().await {
-            Ok(msg_subscriber) => msg_subscriber,
-            Err(e) => {
-                warn!(target: "fud::fetch()", "Error subscribing to msg: {}", e);
-                continue;
-            }
-        };
-
-        let _ = channel.send(&FudFindSeedersRequest { key }).await;
-
-        let reply = match msg_subscriber.receive_with_timeout(fud.dht().timeout).await {
-            Ok(reply) => reply,
-            Err(e) => {
-                warn!(target: "fud::fetch()", "Error waiting for reply: {}", e);
-                continue;
-            }
-        };
-
-        let mut seeders = reply.nodes.clone();
-        info!(target: "fud::fetch()", "Found seeders for {}: {:?}", key, seeders);
-
-        msg_subscriber.unsubscribe().await;
-
-        // 3. Request the file/chunk from the seeders
-        while let Some(seeder) = seeders.pop() {
-            // Only query a seeder once
-            if queried_seeders.iter().any(|s| *s == seeder.id) {
-                continue;
-            }
-            queried_seeders.insert(seeder.id);
-
-            if let Ok(channel) = fud.get_channel(&seeder).await {
-                let msg_subsystem = channel.message_subsystem();
-                msg_subsystem.add_dispatch::<FudChunkReply>().await;
-                msg_subsystem.add_dispatch::<FudFileReply>().await;
-                let msg_subscriber_chunk = channel.subscribe_msg::<FudChunkReply>().await.unwrap();
-                let msg_subscriber_file = channel.subscribe_msg::<FudFileReply>().await.unwrap();
-
-                let _ = channel.send(&FudFindRequest { key }).await;
-
-                let chunk_recv =
-                    msg_subscriber_chunk.receive_with_timeout(fud.dht().timeout).fuse();
-                let file_recv = msg_subscriber_file.receive_with_timeout(fud.dht().timeout).fuse();
-
-                pin_mut!(chunk_recv, file_recv);
-
-                // Wait for a FudChunkReply or a FudFileReply
-                match try_select(chunk_recv, file_recv).await {
-                    Ok(Either::Left((chunk_reply, _))) => {
-                        info!(target: "fud::fetch()", "Received chunk {} from seeder {:?}", key, seeder.id);
-                        msg_subscriber.unsubscribe().await;
-                        result = Some(FetchReply::Chunk((*chunk_reply).clone()));
-                        break;
-                    }
-                    Ok(Either::Right((file_reply, _))) => {
-                        info!(target: "fud::fetch()", "Received file {} from seeder {:?}", key, seeder.id);
-                        msg_subscriber.unsubscribe().await;
-                        result = Some(FetchReply::File((*file_reply).clone()));
-                        break;
-                    }
-                    Err(e) => {
-                        match e {
-                            Either::Left((chunk_err, _)) => {
-                                warn!(target: "fud::fetch()", "Error waiting for chunk reply: {}", chunk_err);
-                            }
-                            Either::Right((file_err, _)) => {
-                                warn!(target: "fud::fetch()", "Error waiting for file reply: {}", file_err);
-                            }
-                        };
-                        msg_subscriber.unsubscribe().await;
-                        continue;
-                    }
-                };
-            }
-        }
-
-        if result.is_some() {
-            break;
-        }
-    }
-
-    result
-}
-
-/// Background task that receives file fetch requests and tries to
-/// fetch objects from the network using the routing table.
-/// TODO: This can be optimised a lot for connection reuse, etc.
-async fn fetch_file_task(fud: Arc<Fud>, _: Arc<Executor<'_>>) -> Result<()> {
-    info!(target: "fud::fetch_file_task()", "Started background file fetch task");
-    loop {
-        let (file_hash, _) = fud.file_fetch_rx.recv().await.unwrap();
-        info!(target: "fud::fetch_file_task()", "Fetching file {}", file_hash);
-
-        let result = fetch(fud.clone(), file_hash).await;
-
-        match result {
-            Some(reply) => {
-                match reply {
-                    FetchReply::File(FudFileReply { chunk_hashes }) => {
-                        if let Err(e) = fud.geode.insert_file(&file_hash, &chunk_hashes).await {
-                            error!("Failed inserting file {} to Geode: {}", file_hash, e);
-                        }
-                        fud.file_fetch_end_tx.send((file_hash, Ok(()))).await.unwrap();
-                    }
-                    // Looked for a file but got a chunk, meaning that file_hash = chunk_hash, the file fits in a single chunk
-                    FetchReply::Chunk(FudChunkReply { chunk }) => {
-                        // TODO: Verify chunk
-                        info!(target: "fud::fetch()", "File fits in a single chunk");
-                        let _ = fud.geode.insert_file(&file_hash, &[file_hash]).await;
-                        match fud.geode.insert_chunk(&chunk).await {
-                            Ok(inserted_hash) => {
-                                if inserted_hash != file_hash {
-                                    warn!("Received chunk does not match requested file");
-                                }
-                            }
-                            Err(e) => {
-                                error!("Failed inserting chunk {} to Geode: {}", file_hash, e);
-                            }
-                        };
-                        fud.file_fetch_end_tx.send((file_hash, Ok(()))).await.unwrap();
-                    }
-                }
-            }
-            None => {
-                fud.file_fetch_end_tx
-                    .send((file_hash, Err(Error::GeodeFileRouteNotFound)))
-                    .await
-                    .unwrap();
-            }
-        };
-    }
-}
-
-/// Background task that receives chunk fetch requests and tries to
-/// fetch objects from the network using the routing table.
-/// TODO: This can be optimised a lot for connection reuse, etc.
-async fn fetch_chunk_task(fud: Arc<Fud>, _: Arc<Executor<'_>>) -> Result<()> {
-    info!(target: "fud::fetch_chunk_task()", "Started background chunk fetch task");
-    loop {
-        let (chunk_hash, _) = fud.chunk_fetch_rx.recv().await.unwrap();
-        info!(target: "fud::fetch_chunk_task()", "Fetching chunk {}", chunk_hash);
-
-        let result = fetch(fud.clone(), chunk_hash).await;
-
-        match result {
-            Some(reply) => {
-                match reply {
-                    FetchReply::Chunk(FudChunkReply { chunk }) => {
-                        // TODO: Verify chunk
-                        match fud.geode.insert_chunk(&chunk).await {
-                            Ok(inserted_hash) => {
-                                if inserted_hash != chunk_hash {
-                                    warn!("Received chunk does not match requested chunk");
-                                }
-                            }
-                            Err(e) => {
-                                error!("Failed inserting chunk {} to Geode: {}", chunk_hash, e);
-                            }
-                        };
-                        fud.chunk_fetch_end_tx.send((chunk_hash, Ok(()))).await.unwrap();
-                    }
-                    _ => {
-                        // Looked for a chunk but got a file instead, not supposed to happen
-                        fud.chunk_fetch_end_tx
-                            .send((chunk_hash, Err(Error::GeodeChunkRouteNotFound)))
-                            .await
-                            .unwrap();
-                    }
-                }
-            }
-            None => {
-                fud.chunk_fetch_end_tx
-                    .send((chunk_hash, Err(Error::GeodeChunkRouteNotFound)))
-                    .await
-                    .unwrap();
-            }
-        };
-    }
-}
-
 #[async_trait]
 impl DhtHandler for Fud {
     fn dht(&self) -> Arc<Dht> {
@@ -630,7 +151,7 @@ impl DhtHandler for Fud {
     }
 
     async fn ping(&self, channel: ChannelPtr) -> Result<dht::DhtNode> {
-        debug!(target: "fud::Fud::DhtHandler::ping()", "Sending ping to channel {}", channel.info.id);
+        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();
@@ -647,7 +168,7 @@ impl DhtHandler for Fud {
 
     // TODO: Optimize this
     async fn on_new_node(&self, node: &DhtNode) -> Result<()> {
-        debug!(target: "fud::Fud::DhtHandler::on_new_node()", "New node {}", node.id);
+        debug!(target: "fud::DhtHandler::on_new_node()", "New node {}", node.id);
 
         // If this is the first node we know about, then bootstrap
         if !self.dht().is_bootstrapped().await {
@@ -655,19 +176,22 @@ impl DhtHandler for Fud {
 
             // Lookup our own node id
             let self_node = self.dht().node.clone();
-            debug!(target: "fud::Fud::DhtHandler::on_new_node()", "DHT bootstrapping {}", self_node.id);
+            debug!(target: "fud::DhtHandler::on_new_node()", "DHT bootstrapping {}", self_node.id);
             let _ = self.lookup_nodes(&self_node.id).await;
         }
 
         // Send keys that are closer to this node than we are
         let self_id = self.dht().node.id;
         let channel = self.get_channel(node).await?;
-        for (key, nodes) in self.seeders_router.read().await.iter() {
+        for (key, seeders) in self.seeders_router.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, nodes: nodes.iter().cloned().collect() })
+                    .send(&FudAnnounce {
+                        key: *key,
+                        seeders: seeders.clone().into_iter().collect(),
+                    })
                     .await;
             }
         }
@@ -676,7 +200,7 @@ impl DhtHandler for Fud {
     }
 
     async fn fetch_nodes(&self, node: &DhtNode, key: &blake3::Hash) -> Result<Vec<DhtNode>> {
-        debug!(target: "fud::Fud::DhtHandler::fetch_value()", "Fetching nodes close to {} from node {}", key, node.id);
+        debug!(target: "fud::DhtHandler::fetch_value()", "Fetching nodes close to {} from node {}", key, node.id);
 
         let channel = self.get_channel(node).await?;
         let msg_subsystem = channel.message_subsystem();
@@ -694,6 +218,165 @@ impl DhtHandler for Fud {
     }
 }
 
+impl Fud {
+    /// Add ourselves to `seeders_router` for the files and chunks we already have.
+    /// Skipped if we have no external address.
+    async fn init(&self) -> Result<()> {
+        if self.dht().node.clone().addresses.is_empty() {
+            return Ok(());
+        }
+        let self_router_items: Vec<DhtRouterItem> = vec![self.dht().node.clone().into()];
+        let mut hashes = self.geode.list_chunks().await?;
+        hashes.extend(self.geode.list_files().await?);
+
+        for hash in hashes {
+            self.add_to_router(self.seeders_router.clone(), &hash, self_router_items.clone()).await;
+        }
+
+        Ok(())
+    }
+
+    /// Fetch a file or chunk from the network
+    /// 1. Lookup nodes close to the key
+    /// 2. Request seeders for the file/chunk from those nodes
+    /// 3. Request the file/chunk from the seeders
+    async fn fetch(&self, key: blake3::Hash) -> Option<FetchReply> {
+        let mut queried_seeders: HashSet<blake3::Hash> = HashSet::new();
+        let closest_nodes = self.lookup_nodes(&key).await; // 1
+        let mut result: Option<FetchReply> = None;
+        if closest_nodes.is_err() {
+            return None
+        }
+
+        for node in closest_nodes.unwrap() {
+            // 2. Request list of seeders
+            let channel = match self.get_channel(&node).await {
+                Ok(channel) => channel,
+                Err(e) => {
+                    warn!(target: "fud::fetch()", "Could not get a channel for node {}: {}", node.id, e);
+                    continue;
+                }
+            };
+            let msg_subsystem = channel.message_subsystem();
+            msg_subsystem.add_dispatch::<FudFindSeedersReply>().await;
+
+            let msg_subscriber = match channel.subscribe_msg::<FudFindSeedersReply>().await {
+                Ok(msg_subscriber) => msg_subscriber,
+                Err(e) => {
+                    warn!(target: "fud::fetch()", "Error subscribing to msg: {}", e);
+                    continue;
+                }
+            };
+
+            let send_res = channel.send(&FudFindSeedersRequest { key }).await;
+            if let Err(e) = send_res {
+                warn!(target: "fud::fetch()", "Error while sending FudFindSeedersRequest: {}", e);
+                msg_subscriber.unsubscribe().await;
+                continue;
+            }
+
+            let reply = match msg_subscriber.receive_with_timeout(self.dht().timeout).await {
+                Ok(reply) => reply,
+                Err(e) => {
+                    warn!(target: "fud::fetch()", "Error waiting for reply: {}", e);
+                    continue;
+                }
+            };
+
+            let mut seeders = reply.seeders.clone();
+            info!(target: "fud::fetch()", "Found {} seeders for {}", seeders.len(), key);
+
+            msg_subscriber.unsubscribe().await;
+
+            // 3. Request the file/chunk from the seeders
+            while let Some(seeder) = seeders.pop() {
+                // Only query a seeder once
+                if queried_seeders.iter().any(|s| *s == seeder.node.id) {
+                    continue;
+                }
+                queried_seeders.insert(seeder.node.id);
+
+                if let Ok(channel) = self.get_channel(&seeder.node).await {
+                    let msg_subsystem = channel.message_subsystem();
+                    msg_subsystem.add_dispatch::<FudChunkReply>().await;
+                    msg_subsystem.add_dispatch::<FudFileReply>().await;
+                    msg_subsystem.add_dispatch::<FudNotFound>().await;
+                    let msg_subscriber_chunk =
+                        channel.subscribe_msg::<FudChunkReply>().await.unwrap();
+                    let msg_subscriber_file =
+                        channel.subscribe_msg::<FudFileReply>().await.unwrap();
+                    let msg_subscriber_notfound =
+                        channel.subscribe_msg::<FudNotFound>().await.unwrap();
+
+                    let send_res = channel.send(&FudFindRequest { key }).await;
+                    if let Err(e) = send_res {
+                        warn!(target: "fud::fetch()", "Error while sending FudFindRequest: {}", e);
+                        msg_subscriber_chunk.unsubscribe().await;
+                        msg_subscriber_file.unsubscribe().await;
+                        msg_subscriber_notfound.unsubscribe().await;
+                        continue;
+                    }
+
+                    let chunk_recv =
+                        msg_subscriber_chunk.receive_with_timeout(self.dht().timeout).fuse();
+                    let file_recv =
+                        msg_subscriber_file.receive_with_timeout(self.dht().timeout).fuse();
+                    let notfound_recv =
+                        msg_subscriber_notfound.receive_with_timeout(self.dht().timeout).fuse();
+
+                    pin_mut!(chunk_recv, file_recv, notfound_recv);
+
+                    // Wait for a FudChunkReply, FudFileReply, or FudNotFound
+                    select! {
+                        chunk_reply = chunk_recv => {
+                            if let Err(e) = chunk_reply {
+                                warn!(target: "fud::fetch()", "Error waiting for chunk reply: {}", e);
+                                continue;
+                            }
+                            let reply = chunk_reply.unwrap();
+                            info!(target: "fud::fetch()", "Received chunk {} from seeder {}", key, seeder.node.id.to_hex().to_string());
+                            msg_subscriber_chunk.unsubscribe().await;
+                            msg_subscriber_file.unsubscribe().await;
+                            msg_subscriber_notfound.unsubscribe().await;
+                            result = Some(FetchReply::Chunk((*reply).clone()));
+                            break;
+                        }
+                        file_reply = file_recv => {
+                            if let Err(e) = file_reply {
+                                warn!(target: "fud::fetch()", "Error waiting for file reply: {}", e);
+                                continue;
+                            }
+                            let reply = file_reply.unwrap();
+                            info!(target: "fud::fetch()", "Received file {} from seeder {}", key, seeder.node.id.to_hex().to_string());
+                            msg_subscriber_chunk.unsubscribe().await;
+                            msg_subscriber_file.unsubscribe().await;
+                            msg_subscriber_notfound.unsubscribe().await;
+                            result = Some(FetchReply::File((*reply).clone()));
+                            break;
+                        }
+                        notfound_reply = notfound_recv => {
+                            if let Err(e) = notfound_reply {
+                                warn!(target: "fud::fetch()", "Error waiting for NOTFOUND reply: {}", e);
+                                continue;
+                            }
+                            info!(target: "fud::fetch()", "Received NOTFOUND {} from seeder {}", key, seeder.node.id.to_hex().to_string());
+                            msg_subscriber_chunk.unsubscribe().await;
+                            msg_subscriber_file.unsubscribe().await;
+                            msg_subscriber_notfound.unsubscribe().await;
+                        }
+                    };
+                }
+            }
+
+            if result.is_some() {
+                break;
+            }
+        }
+
+        result
+    }
+}
+
 // TODO: This is not Sybil-resistant
 fn generate_node_id() -> Result<blake3::Hash> {
     let mut rng = OsRng;
@@ -720,11 +403,7 @@ async fn realmain(args: Args, ex: Arc<Executor<'static>>) -> Result<()> {
     let external_addrs = p2p.hosts().external_addrs().await;
 
     if external_addrs.is_empty() {
-        error!(
-            target: "fud::realmain",
-            "External addrs not configured. Stopping",
-        );
-        return Ok(())
+        warn!(target: "fud::realmain", "No external addresses, you won't be able to seed")
     }
 
     info!("Starting dnet subs task");
@@ -780,17 +459,21 @@ async fn realmain(args: Args, ex: Arc<Executor<'static>>) -> Result<()> {
     info!(target: "fud", "Your node ID: {}", node_id_);
 
     // Daemon instantiation
+    let download_sub = JsonSubscriber::new("get");
+    let (get_tx, get_rx) = smol::channel::unbounded();
     let (file_fetch_tx, file_fetch_rx) = smol::channel::unbounded();
     let (file_fetch_end_tx, file_fetch_end_rx) = smol::channel::unbounded();
     let (chunk_fetch_tx, chunk_fetch_rx) = smol::channel::unbounded();
     let (chunk_fetch_end_tx, chunk_fetch_end_rx) = smol::channel::unbounded();
     // TODO: Add DHT settings in the config file
-    let dht = Arc::new(Dht::new(&node_id_, 4, 16, 15, p2p.clone(), ex.clone()).await);
+    let dht = Arc::new(Dht::new(&node_id_, 4, 16, 60, p2p.clone(), ex.clone()).await);
     let fud = Arc::new(Fud {
         seeders_router,
         p2p: p2p.clone(),
         geode,
         dht: dht.clone(),
+        get_tx,
+        get_rx,
         file_fetch_tx,
         file_fetch_rx,
         file_fetch_end_tx,
@@ -801,12 +484,38 @@ async fn realmain(args: Args, ex: Arc<Executor<'static>>) -> Result<()> {
         chunk_fetch_end_rx,
         rpc_connections: Mutex::new(HashSet::new()),
         dnet_sub,
+        download_sub: download_sub.clone(),
+        download_publisher: Publisher::new(),
     });
+    fud.init().await?;
+
+    info!("Starting download subs task");
+    let download_sub_ = download_sub.clone();
+    let fud_ = fud.clone();
+    let download_task = StoppableTask::new();
+    download_task.clone().start(
+        async move {
+            let download_sub = fud_.download_publisher.clone().subscribe().await;
+            loop {
+                let event = download_sub.receive().await;
+                debug!("Got download event: {:?}", event);
+                download_sub_.notify(event.into()).await;
+            }
+        },
+        |res| async {
+            match res {
+                Ok(()) | Err(Error::DetachedTaskStopped) => { /* Do nothing */ }
+                Err(e) => panic!("{}", e),
+            }
+        },
+        Error::DetachedTaskStopped,
+        ex.clone(),
+    );
 
     info!(target: "fud", "Starting fetch file task");
     let file_task = StoppableTask::new();
     file_task.clone().start(
-        fetch_file_task(fud.clone(), ex.clone()),
+        tasks::fetch_file_task(fud.clone()),
         |res| async {
             match res {
                 Ok(()) | Err(Error::DetachedTaskStopped) => { /* Do nothing */ }
@@ -820,7 +529,7 @@ async fn realmain(args: Args, ex: Arc<Executor<'static>>) -> Result<()> {
     info!(target: "fud", "Starting fetch chunk task");
     let chunk_task = StoppableTask::new();
     chunk_task.clone().start(
-        fetch_chunk_task(fud.clone(), ex.clone()),
+        tasks::fetch_chunk_task(fud.clone()),
         |res| async {
             match res {
                 Ok(()) | Err(Error::DetachedTaskStopped) => { /* Do nothing */ }
@@ -831,6 +540,20 @@ async fn realmain(args: Args, ex: Arc<Executor<'static>>) -> Result<()> {
         ex.clone(),
     );
 
+    info!(target: "fud", "Starting get task");
+    let get_task_ = StoppableTask::new();
+    get_task_.clone().start(
+        tasks::get_task(fud.clone()),
+        |res| async {
+            match res {
+                Ok(()) | Err(Error::DetachedTaskStopped) => { /* Do nothing */ }
+                Err(e) => error!(target: "fud", "Failed starting get task: {}", e),
+            }
+        },
+        Error::DetachedTaskStopped,
+        ex.clone(),
+    );
+
     let rpc_settings: RpcSettings = args.rpc.into();
     info!(target: "fud", "Starting JSON-RPC server on {}", rpc_settings.listen);
     let rpc_task = StoppableTask::new();
@@ -885,6 +608,32 @@ async fn realmain(args: Args, ex: Arc<Executor<'static>>) -> Result<()> {
         Error::DetachedTaskStopped,
         ex.clone(),
     );
+    let prune_task = StoppableTask::new();
+    let fud_ = fud.clone();
+    prune_task.clone().start(
+        async move { tasks::prune_seeders_task(fud_.clone()).await },
+        |res| async {
+            match res {
+                Ok(()) | Err(Error::DetachedTaskStopped) => { /* Do nothing */ }
+                Err(e) => error!(target: "fud", "Failed starting prune seeders task: {}", e),
+            }
+        },
+        Error::DetachedTaskStopped,
+        ex.clone(),
+    );
+    let announce_task = StoppableTask::new();
+    let fud_ = fud.clone();
+    announce_task.clone().start(
+        async move { tasks::announce_seed_task(fud_.clone()).await },
+        |res| async {
+            match res {
+                Ok(()) | Err(Error::DetachedTaskStopped) => { /* Do nothing */ }
+                Err(e) => error!(target: "fud", "Failed starting announce task: {}", e),
+            }
+        },
+        Error::DetachedTaskStopped,
+        ex.clone(),
+    );
 
     // Signal handling for graceful termination.
     let (signals_handler, signals_task) = SignalHandler::new(ex)?;
@@ -897,6 +646,9 @@ async fn realmain(args: Args, ex: Arc<Executor<'static>>) -> Result<()> {
     info!(target: "fud", "Stopping fetch chunk task...");
     chunk_task.stop().await;
 
+    info!(target: "fud", "Stopping get task...");
+    get_task_.stop().await;
+
     info!(target: "fud", "Stopping JSON-RPC server...");
     rpc_task.stop().await;
 
@@ -906,6 +658,8 @@ async fn realmain(args: Args, ex: Arc<Executor<'static>>) -> Result<()> {
     info!(target: "fud", "Stopping DHT tasks");
     dht_channel_task.stop().await;
     dht_disconnect_task.stop().await;
+    prune_task.stop().await;
+    announce_task.stop().await;
 
     info!("Bye!");
     Ok(())

bin/fud/fud/src/proto.rs

@@ -23,7 +23,7 @@ use darkfi::{
     geode::{read_until_filled, MAX_CHUNK_SIZE},
     impl_p2p_message,
     net::{
-        metering::{DEFAULT_METERING_CONFIGURATION, MeteringConfiguration},
+        metering::{MeteringConfiguration, DEFAULT_METERING_CONFIGURATION},
         ChannelPtr, Message, MessageSubscription, P2pPtr, ProtocolBase, ProtocolBasePtr,
         ProtocolJobsManager, ProtocolJobsManagerPtr,
     },
@@ -34,7 +34,7 @@ use log::{debug, error, info};
 use smol::{fs::File, Executor};
 
 use super::Fud;
-use crate::dht::{DhtHandler, DhtNode};
+use crate::dht::{DhtHandler, DhtNode, DhtRouterItem};
 
 /// Message representing a file reply from the network
 #[derive(Debug, Clone, SerialEncodable, SerialDecodable)]
@@ -47,7 +47,7 @@ impl_p2p_message!(FudFileReply, "FudFileReply", 0, 0, DEFAULT_METERING_CONFIGURA
 #[derive(Debug, Clone, SerialEncodable, SerialDecodable)]
 pub struct FudAnnounce {
     pub key: blake3::Hash,
-    pub nodes: Vec<DhtNode>,
+    pub seeders: Vec<DhtRouterItem>,
 }
 impl_p2p_message!(FudAnnounce, "FudAnnounce", 0, 0, DEFAULT_METERING_CONFIGURATION);
 
@@ -61,22 +61,12 @@ impl_p2p_message!(FudChunkReply, "FudChunkReply", 0, 0, DEFAULT_METERING_CONFIGU
 
 /// Message representing a chunk reply when a file is not found
 #[derive(Debug, Clone, SerialEncodable, SerialDecodable)]
-pub struct FudFileNotFound;
-impl_p2p_message!(FudFileNotFound, "FudFileNotFound", 0, 0, DEFAULT_METERING_CONFIGURATION);
-
-/// Message representing a chunk reply when a chunk is not found
-#[derive(Debug, Clone, SerialEncodable, SerialDecodable)]
-pub struct FudChunkNotFound;
-impl_p2p_message!(FudChunkNotFound, "FudChunkNotFound", 0, 0, DEFAULT_METERING_CONFIGURATION);
-
-/// Message representing a seeders reply when seeders are not found
-#[derive(Debug, Clone, SerialEncodable, SerialDecodable)]
-pub struct FudSeedersNotFound;
-impl_p2p_message!(FudSeedersNotFound, "FudSeedersNotFound", 0, 0, DEFAULT_METERING_CONFIGURATION);
+pub struct FudNotFound;
+impl_p2p_message!(FudNotFound, "FudNotFound", 0, 0, DEFAULT_METERING_CONFIGURATION);
 
 /// Message representing a ping request on the network
 #[derive(Debug, Clone, SerialEncodable, SerialDecodable)]
-pub struct FudPingRequest {}
+pub struct FudPingRequest;
 impl_p2p_message!(FudPingRequest, "FudPingRequest", 0, 0, DEFAULT_METERING_CONFIGURATION);
 
 /// Message representing a ping reply on the network
@@ -112,12 +102,18 @@ impl_p2p_message!(FudFindNodesReply, "FudFindNodesReply", 0, 0, DEFAULT_METERING
 pub struct FudFindSeedersRequest {
     pub key: blake3::Hash,
 }
-impl_p2p_message!(FudFindSeedersRequest, "FudFindSeedersRequest", 0, 0, DEFAULT_METERING_CONFIGURATION);
+impl_p2p_message!(
+    FudFindSeedersRequest,
+    "FudFindSeedersRequest",
+    0,
+    0,
+    DEFAULT_METERING_CONFIGURATION
+);
 
 /// Message representing a find seeders reply on the network
 #[derive(Debug, Clone, SerialEncodable, SerialDecodable)]
 pub struct FudFindSeedersReply {
-    pub nodes: Vec<DhtNode>,
+    pub seeders: Vec<DhtRouterItem>,
 }
 impl_p2p_message!(FudFindSeedersReply, "FudFindSeedersReply", 0, 0, DEFAULT_METERING_CONFIGURATION);
 
@@ -218,6 +214,7 @@ impl ProtocolFud {
                     let bytes_read = read_until_filled(&mut chunk_fd, &mut buf).await.unwrap();
                     let chunk_slice = &buf[..bytes_read];
                     let reply = FudChunkReply { chunk: chunk_slice.to_vec() };
+                    info!(target: "fud::ProtocolFud::handle_fud_find_request()", "Sending chunk");
                     let _ = self.channel.send(&reply).await;
                     continue;
                 }
@@ -233,27 +230,14 @@ impl ProtocolFud {
                     let reply = FudFileReply {
                         chunk_hashes: chunked_file.iter().map(|(chunk, _)| *chunk).collect(),
                     };
+                    info!(target: "fud::ProtocolFud::handle_fud_find_request()", "Sending file");
                     let _ = self.channel.send(&reply).await;
                     continue;
                 }
             }
 
-            // Peers
-            {
-                let router = self.fud.seeders_router.read().await;
-                let peers = router.get(&request.key);
-
-                if let Some(nodes) = peers {
-                    let reply = FudFindNodesReply { nodes: nodes.clone().into_iter().collect() };
-                    let _ = self.channel.send(&reply).await;
-                    continue;
-                }
-            }
-
-            // Nodes
-            let reply = FudFindNodesReply {
-                nodes: self.fud.dht().find_neighbors(&request.key, self.fud.dht().k).await,
-            };
+            let reply = FudNotFound {};
+            info!(target: "fud::ProtocolFud::handle_fud_find_request()", "We do not have {}", request.key.to_hex().to_string());
             let _ = self.channel.send(&reply).await;
         }
     }
@@ -310,14 +294,14 @@ impl ProtocolFud {
             let peers = router.get(&request.key);
 
             match peers {
-                Some(nodes) => {
+                Some(seeders) => {
                     let _ = self
                         .channel
-                        .send(&FudFindSeedersReply { nodes: nodes.iter().cloned().collect() })
+                        .send(&FudFindSeedersReply { seeders: seeders.iter().cloned().collect() })
                         .await;
                 }
                 None => {
-                    let _ = self.channel.send(&FudSeedersNotFound {}).await;
+                    let _ = self.channel.send(&FudFindSeedersReply { seeders: vec![] }).await;
                 }
             };
         }
@@ -342,9 +326,17 @@ impl ProtocolFud {
                 self.fud.update_node(&node).await;
             }
 
-            self.fud
-                .add_to_router(self.fud.seeders_router.clone(), &request.key, request.nodes.clone())
-                .await;
+            let mut seeders = vec![];
+
+            for seeder in request.seeders.clone() {
+                if seeder.node.addresses.is_empty() {
+                    continue
+                }
+                // TODO: Verify each address
+                seeders.push(seeder);
+            }
+
+            self.fud.add_to_router(self.fud.seeders_router.clone(), &request.key, seeders).await;
         }
     }
 }

bin/fud/fud/src/rpc.rs

@@ -0,0 +1,471 @@
+/* This file is part of DarkFi (https://dark.fi)
+ *
+ * Copyright (C) 2020-2025 Dyne.org foundation
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU Affero General Public License as
+ * published by the Free Software Foundation, either version 3 of the
+ * License, or (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU Affero General Public License for more details.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with this program.  If not, see <https://www.gnu.org/licenses/>.
+ */
+
+use std::{
+    collections::{HashMap, HashSet},
+    path::PathBuf,
+};
+
+use crate::{dht::DhtHandler, proto::FudAnnounce, Fud};
+use async_trait::async_trait;
+use darkfi::{
+    rpc::{
+        jsonrpc::{ErrorCode, JsonError, JsonRequest, JsonResponse, JsonResult},
+        p2p_method::HandlerP2p,
+        server::RequestHandler,
+        util::{json_map, json_str},
+    },
+    system::StoppableTaskPtr,
+    util::path::expand_path,
+    Error,
+};
+use log::{error, info};
+use smol::{fs::File, lock::MutexGuard};
+use tinyjson::JsonValue;
+
+#[async_trait]
+impl RequestHandler<()> for Fud {
+    async fn handle_request(&self, req: JsonRequest) -> JsonResult {
+        return match req.method.as_str() {
+            "ping" => self.pong(req.id, req.params).await,
+
+            "put" => self.put(req.id, req.params).await,
+            "get" => self.get(req.id, req.params).await,
+
+            "dnet.switch" => self.dnet_switch(req.id, req.params).await,
+            "dnet.subscribe_events" => self.dnet_subscribe_events(req.id, req.params).await,
+            "p2p.get_info" => self.p2p_get_info(req.id, req.params).await,
+            "list_buckets" => self.list_buckets(req.id, req.params).await,
+            "list_seeders" => self.list_seeders(req.id, req.params).await,
+            _ => JsonError::new(ErrorCode::MethodNotFound, None, req.id).into(),
+        }
+    }
+
+    async fn connections_mut(&self) -> MutexGuard<'_, HashSet<StoppableTaskPtr>> {
+        self.rpc_connections.lock().await
+    }
+}
+
+/// Fud RPC methods
+impl Fud {
+    // RPCAPI:
+    // Put a file onto the network. Takes a local filesystem path as a parameter.
+    // Returns the file hash that serves as a pointer to the uploaded file.
+    //
+    // --> {"jsonrpc": "2.0", "method": "put", "params": ["/foo.txt"], "id": 42}
+    // <-- {"jsonrpc": "2.0", "result: "df4...3db7", "id": 42}
+    async fn put(&self, id: u16, params: JsonValue) -> JsonResult {
+        if self.dht().node.addresses.is_empty() {
+            error!(target: "fud::put()", "Cannot put file, you don't have any external address");
+            return JsonError::new(ErrorCode::InternalError, None, id).into()
+        }
+
+        let params = params.get::<Vec<JsonValue>>().unwrap();
+        if params.len() != 1 || !params[0].is_string() {
+            return JsonError::new(ErrorCode::InvalidParams, None, id).into()
+        }
+
+        let path = params[0].get::<String>().unwrap();
+        let path = match expand_path(path.as_str()) {
+            Ok(v) => v,
+            Err(_) => return JsonError::new(ErrorCode::InvalidParams, None, id).into(),
+        };
+
+        // A valid path was passed. Let's see if we can read it, and if so,
+        // add it to Geode.
+        let fd = match File::open(&path).await {
+            Ok(v) => v,
+            Err(e) => {
+                error!(target: "fud::put()", "Failed to open {:?}: {}", path, e);
+                return JsonError::new(ErrorCode::InvalidParams, None, id).into()
+            }
+        };
+
+        let (file_hash, chunk_hashes) = match self.geode.insert(fd).await {
+            Ok(v) => v,
+            Err(e) => {
+                error!(target: "fud::put()", "Failed inserting file {:?} to geode: {}", path, e);
+                return JsonError::new(ErrorCode::InternalError, None, id).into()
+            }
+        };
+
+        // Announce file
+        let self_node = self.dht.node.clone();
+        let fud_announce = FudAnnounce { key: file_hash, seeders: vec![self_node.clone().into()] };
+        let _ = self.announce(&file_hash, &fud_announce, self.seeders_router.clone()).await;
+
+        // Announce chunks
+        for chunk_hash in chunk_hashes {
+            let fud_announce =
+                FudAnnounce { key: chunk_hash, seeders: vec![self_node.clone().into()] };
+            let _ = self.announce(&chunk_hash, &fud_announce, self.seeders_router.clone()).await;
+        }
+
+        JsonResponse::new(JsonValue::String(file_hash.to_hex().to_string()), id).into()
+    }
+
+    // RPCAPI:
+    // Fetch a file from the network, and subscribe to download events. Takes a file hash as parameter.
+    //
+    // --> {"jsonrpc": "2.0", "method": "get", "params": ["1211...abfd"], "id": 42}
+    // <-- {"jsonrpc": "2.0", "method": "get", "params": `event`}
+    async fn get(&self, id: u16, params: JsonValue) -> JsonResult {
+        let params = params.get::<Vec<JsonValue>>().unwrap();
+        if params.len() != 2 || !params[0].is_string() || !params[1].is_string() {
+            return JsonError::new(ErrorCode::InvalidParams, None, id).into()
+        }
+
+        let file_name: Option<String> = match params[1].get::<String>() {
+            Some(name) => match name.is_empty() {
+                true => None,
+                false => Some(name.clone()),
+            },
+            None => None,
+        };
+
+        let file_hash = match blake3::Hash::from_hex(params[0].get::<String>().unwrap()) {
+            Ok(v) => v,
+            Err(_) => return JsonError::new(ErrorCode::InvalidParams, None, id).into(),
+        };
+
+        let _ = self.get_tx.send((id, file_hash, file_name, Ok(()))).await;
+
+        self.download_sub.clone().into()
+    }
+
+    // RPCAPI:
+    // Activate or deactivate dnet in the P2P stack.
+    // By sending `true`, dnet will be activated, and by sending `false` dnet
+    // will be deactivated. Returns `true` on success.
+    //
+    // --> {"jsonrpc": "2.0", "method": "dnet_switch", "params": [true], "id": 42}
+    // <-- {"jsonrpc": "2.0", "result": true, "id": 42}
+    async fn dnet_switch(&self, id: u16, params: JsonValue) -> JsonResult {
+        let params = params.get::<Vec<JsonValue>>().unwrap();
+        if params.len() != 1 || !params[0].is_bool() {
+            return JsonError::new(ErrorCode::InvalidParams, None, id).into()
+        }
+
+        let switch = params[0].get::<bool>().unwrap();
+
+        if *switch {
+            self.p2p.dnet_enable();
+        } else {
+            self.p2p.dnet_disable();
+        }
+
+        JsonResponse::new(JsonValue::Boolean(true), id).into()
+    }
+
+    // RPCAPI:
+    // Initializes a subscription to p2p dnet events.
+    // Once a subscription is established, `fud` will send JSON-RPC notifications of
+    // new network events to the subscriber.
+    //
+    // --> {"jsonrpc": "2.0", "method": "dnet.subscribe_events", "params": [], "id": 1}
+    // <-- {"jsonrpc": "2.0", "method": "dnet.subscribe_events", "params": [`event`]}
+    pub async fn dnet_subscribe_events(&self, id: u16, params: JsonValue) -> JsonResult {
+        let params = params.get::<Vec<JsonValue>>().unwrap();
+        if !params.is_empty() {
+            return JsonError::new(ErrorCode::InvalidParams, None, id).into()
+        }
+
+        self.dnet_sub.clone().into()
+    }
+
+    // RPCAPI:
+    // Returns the current buckets
+    //
+    // --> {"jsonrpc": "2.0", "method": "list_buckets", "params": [], "id": 1}
+    // <-- {"jsonrpc": "2.0", "result": [[["abcdef", ["tcp://127.0.0.1:13337"]]]], "id": 1}
+    pub async fn list_buckets(&self, id: u16, params: JsonValue) -> JsonResult {
+        let params = params.get::<Vec<JsonValue>>().unwrap();
+        if !params.is_empty() {
+            return JsonError::new(ErrorCode::InvalidParams, None, id).into()
+        }
+        let mut buckets = vec![];
+        for bucket in self.dht.buckets.read().await.iter() {
+            let mut nodes = vec![];
+            for node in bucket.nodes.clone() {
+                let mut addresses = vec![];
+                for addr in node.addresses {
+                    addresses.push(JsonValue::String(addr.to_string()));
+                }
+                nodes.push(JsonValue::Array(vec![
+                    JsonValue::String(node.id.to_hex().to_string()),
+                    JsonValue::Array(addresses),
+                ]));
+            }
+            buckets.push(JsonValue::Array(nodes));
+        }
+
+        JsonResponse::new(JsonValue::Array(buckets), id).into()
+    }
+
+    // RPCAPI:
+    // Returns the content of the seeders router
+    //
+    // --> {"jsonrpc": "2.0", "method": "list_routes", "params": [], "id": 1}
+    // <-- {"jsonrpc": "2.0", "result": {"seeders": {"abcdef": ["ghijkl"]}}, "id": 1}
+    pub async fn list_seeders(&self, id: u16, params: JsonValue) -> JsonResult {
+        let params = params.get::<Vec<JsonValue>>().unwrap();
+        if !params.is_empty() {
+            return JsonError::new(ErrorCode::InvalidParams, None, id).into()
+        }
+        let mut seeders_router: HashMap<String, JsonValue> = HashMap::new();
+        for (hash, items) in self.seeders_router.read().await.iter() {
+            let mut node_ids = vec![];
+            for item in items {
+                node_ids.push(JsonValue::String(item.node.id.to_hex().to_string()));
+            }
+            seeders_router.insert(hash.to_hex().to_string(), JsonValue::Array(node_ids));
+        }
+        let mut res: HashMap<String, JsonValue> = HashMap::new();
+        res.insert("seeders".to_string(), JsonValue::Object(seeders_router));
+
+        JsonResponse::new(JsonValue::Object(res), id).into()
+    }
+}
+
+#[derive(Clone, Debug)]
+pub struct ChunkDownloadCompleted {
+    pub file_hash: blake3::Hash,
+    pub chunk_hash: blake3::Hash,
+}
+#[derive(Clone, Debug)]
+pub struct FileDownloadCompleted {
+    pub file_hash: blake3::Hash,
+    pub chunk_count: usize,
+}
+#[derive(Clone, Debug)]
+pub struct DownloadCompleted {
+    pub file_hash: blake3::Hash,
+    pub file_path: PathBuf,
+}
+#[derive(Clone, Debug)]
+pub struct ChunkNotFound {
+    pub file_hash: blake3::Hash,
+    pub chunk_hash: blake3::Hash,
+}
+#[derive(Clone, Debug)]
+pub struct FileNotFound {
+    pub file_hash: blake3::Hash,
+}
+#[derive(Clone, Debug)]
+pub struct MissingChunks {}
+
+#[derive(Clone, Debug)]
+pub enum FudEvent {
+    ChunkDownloadCompleted(ChunkDownloadCompleted),
+    FileDownloadCompleted(FileDownloadCompleted),
+    DownloadCompleted(DownloadCompleted),
+    ChunkNotFound(ChunkNotFound),
+    FileNotFound(FileNotFound),
+    MissingChunks(MissingChunks),
+}
+
+impl From<ChunkDownloadCompleted> for JsonValue {
+    fn from(info: ChunkDownloadCompleted) -> JsonValue {
+        json_map([
+            ("file_hash", JsonValue::String(info.file_hash.to_hex().to_string())),
+            ("chunk_hash", JsonValue::String(info.chunk_hash.to_hex().to_string())),
+        ])
+    }
+}
+impl From<FileDownloadCompleted> for JsonValue {
+    fn from(info: FileDownloadCompleted) -> JsonValue {
+        json_map([
+            ("file_hash", JsonValue::String(info.file_hash.to_hex().to_string())),
+            ("chunk_count", JsonValue::Number(info.chunk_count as f64)),
+        ])
+    }
+}
+impl From<DownloadCompleted> for JsonValue {
+    fn from(info: DownloadCompleted) -> JsonValue {
+        json_map([
+            ("file_hash", JsonValue::String(info.file_hash.to_hex().to_string())),
+            ("file_path", JsonValue::String(info.file_path.to_string_lossy().to_string())),
+        ])
+    }
+}
+impl From<ChunkNotFound> for JsonValue {
+    fn from(info: ChunkNotFound) -> JsonValue {
+        json_map([
+            ("file_hash", JsonValue::String(info.file_hash.to_hex().to_string())),
+            ("chunk_hash", JsonValue::String(info.chunk_hash.to_hex().to_string())),
+        ])
+    }
+}
+impl From<FileNotFound> for JsonValue {
+    fn from(info: FileNotFound) -> JsonValue {
+        json_map([("file_hash", JsonValue::String(info.file_hash.to_hex().to_string()))])
+    }
+}
+impl From<FudEvent> for JsonValue {
+    fn from(event: FudEvent) -> JsonValue {
+        match event {
+            FudEvent::ChunkDownloadCompleted(info) => {
+                json_map([("event", json_str("chunk_download_completed")), ("info", info.into())])
+            }
+            FudEvent::FileDownloadCompleted(info) => {
+                json_map([("event", json_str("file_download_completed")), ("info", info.into())])
+            }
+            FudEvent::DownloadCompleted(info) => {
+                json_map([("event", json_str("download_completed")), ("info", info.into())])
+            }
+            FudEvent::ChunkNotFound(info) => {
+                json_map([("event", json_str("chunk_not_found")), ("info", info.into())])
+            }
+            FudEvent::FileNotFound(info) => {
+                json_map([("event", json_str("file_not_found")), ("info", info.into())])
+            }
+            FudEvent::MissingChunks(_) => json_map([("event", json_str("missing_chunks"))]),
+        }
+    }
+}
+
+impl Fud {
+    /// Handle `get` RPC request
+    pub async fn handle_get(&self, file_hash: blake3::Hash, file_name: Option<String>) {
+        let self_node = self.dht().node.clone();
+
+        let chunked_file = match self.geode.get(&file_hash).await {
+            Ok(v) => v,
+            Err(Error::GeodeNeedsGc) => todo!(),
+            Err(Error::GeodeFileNotFound) => {
+                info!(target: "self::get()", "Requested file {} not found in Geode, triggering fetch", file_hash);
+                self.file_fetch_tx.send((file_hash, Ok(()))).await.unwrap();
+                info!(target: "self::get()", "Waiting for background file fetch task...");
+                let (i_file_hash, status) = self.file_fetch_end_rx.recv().await.unwrap();
+                match status {
+                    Ok(()) => self.geode.get(&i_file_hash).await.unwrap(),
+
+                    Err(Error::GeodeFileRouteNotFound) => {
+                        self.download_publisher
+                            .notify(FudEvent::FileNotFound(FileNotFound { file_hash }))
+                            .await;
+                        return;
+                    }
+
+                    Err(e) => panic!("{}", e),
+                }
+            }
+
+            Err(e) => panic!("{}", e),
+        };
+
+        self.download_publisher
+            .notify(FudEvent::FileDownloadCompleted(FileDownloadCompleted {
+                file_hash,
+                chunk_count: chunked_file.len(),
+            }))
+            .await;
+
+        if chunked_file.is_complete() {
+            let self_announce =
+                FudAnnounce { key: file_hash, seeders: vec![self_node.clone().into()] };
+            let _ = self.announce(&file_hash, &self_announce, self.seeders_router.clone()).await;
+
+            return match self.geode.assemble_file(&file_hash, &chunked_file, file_name).await {
+                Ok(file_path) => {
+                    self.download_publisher
+                        .notify(FudEvent::DownloadCompleted(DownloadCompleted {
+                            file_hash,
+                            file_path: file_path.clone(),
+                        }))
+                        .await;
+                }
+                Err(e) => {
+                    error!(target: "fud::handle_get()", "{}", e);
+                }
+            };
+        }
+
+        // Fetch any missing chunks
+        let mut missing_chunks = vec![];
+        for (chunk, path) in chunked_file.iter() {
+            if path.is_none() {
+                missing_chunks.push(*chunk);
+            } else {
+                self.download_publisher
+                    .notify(FudEvent::ChunkDownloadCompleted(ChunkDownloadCompleted {
+                        file_hash,
+                        chunk_hash: *chunk,
+                    }))
+                    .await;
+            }
+        }
+
+        for chunk in missing_chunks {
+            self.chunk_fetch_tx.send((chunk, Ok(()))).await.unwrap();
+            let (i_chunk_hash, status) = self.chunk_fetch_end_rx.recv().await.unwrap();
+
+            match status {
+                Ok(()) => {
+                    self.download_publisher
+                        .notify(FudEvent::ChunkDownloadCompleted(ChunkDownloadCompleted {
+                            file_hash,
+                            chunk_hash: i_chunk_hash,
+                        }))
+                        .await;
+                    let self_announce =
+                        FudAnnounce { key: i_chunk_hash, seeders: vec![self_node.clone().into()] };
+                    let _ = self
+                        .announce(&i_chunk_hash, &self_announce, self.seeders_router.clone())
+                        .await;
+                }
+                Err(Error::GeodeChunkRouteNotFound) => {
+                    self.download_publisher
+                        .notify(FudEvent::ChunkNotFound(ChunkNotFound {
+                            file_hash,
+                            chunk_hash: i_chunk_hash,
+                        }))
+                        .await;
+                    return;
+                }
+
+                Err(e) => panic!("{}", e),
+            };
+        }
+
+        let chunked_file = match self.geode.get(&file_hash).await {
+            Ok(v) => v,
+            Err(e) => panic!("{}", e),
+        };
+
+        // We fetched all chunks, but the file is not complete?
+        if !chunked_file.is_complete() {
+            self.download_publisher.notify(FudEvent::MissingChunks(MissingChunks {})).await;
+            return;
+        }
+
+        match self.geode.assemble_file(&file_hash, &chunked_file, file_name).await {
+            Ok(file_path) => {
+                self.download_publisher
+                    .notify(FudEvent::DownloadCompleted(DownloadCompleted {
+                        file_hash,
+                        file_path: file_path.clone(),
+                    }))
+                    .await;
+            }
+            Err(e) => {
+                error!(target: "fud::handle_get()", "{}", e);
+            }
+        };
+    }
+}

bin/fud/fud/src/tasks.rs

@@ -0,0 +1,186 @@
+/* This file is part of DarkFi (https://dark.fi)
+ *
+ * Copyright (C) 2020-2025 Dyne.org foundation
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU Affero General Public License as
+ * published by the Free Software Foundation, either version 3 of the
+ * License, or (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU Affero General Public License for more details.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with this program.  If not, see <https://www.gnu.org/licenses/>.
+ */
+
+use std::sync::Arc;
+
+use darkfi::{system::sleep, Error, Result};
+
+use crate::{
+    dht::DhtHandler,
+    proto::{FudAnnounce, FudChunkReply, FudFileReply},
+    Fud,
+};
+use log::{error, info, warn};
+
+/// Triggered when calling the `get` RPC method
+pub async fn get_task(fud: Arc<Fud>) -> Result<()> {
+    loop {
+        let (_, file_hash, file_name, _) = fud.get_rx.recv().await.unwrap();
+
+        let _ = fud.handle_get(file_hash, file_name).await;
+    }
+}
+
+pub enum FetchReply {
+    File(FudFileReply),
+    Chunk(FudChunkReply),
+}
+
+/// Background task that receives file fetch requests and tries to
+/// fetch objects from the network using the routing table.
+/// TODO: This can be optimised a lot for connection reuse, etc.
+pub async fn fetch_file_task(fud: Arc<Fud>) -> Result<()> {
+    info!(target: "fud::fetch_file_task()", "Started background file fetch task");
+    loop {
+        let (file_hash, _) = fud.file_fetch_rx.recv().await.unwrap();
+        info!(target: "fud::fetch_file_task()", "Fetching file {}", file_hash);
+
+        let result = fud.fetch(file_hash).await;
+
+        match result {
+            Some(reply) => {
+                match reply {
+                    FetchReply::File(FudFileReply { chunk_hashes }) => {
+                        if let Err(e) = fud.geode.insert_file(&file_hash, &chunk_hashes).await {
+                            error!("Failed inserting file {} to Geode: {}", file_hash, e);
+                        }
+                        fud.file_fetch_end_tx.send((file_hash, Ok(()))).await.unwrap();
+                    }
+                    // Looked for a file but got a chunk, meaning that file_hash = chunk_hash, the file fits in a single chunk
+                    FetchReply::Chunk(FudChunkReply { chunk }) => {
+                        // TODO: Verify chunk
+                        info!(target: "fud::fetch()", "File fits in a single chunk");
+                        let _ = fud.geode.insert_file(&file_hash, &[file_hash]).await;
+                        match fud.geode.insert_chunk(&chunk).await {
+                            Ok(inserted_hash) => {
+                                if inserted_hash != file_hash {
+                                    warn!("Received chunk does not match requested file");
+                                }
+                            }
+                            Err(e) => {
+                                error!("Failed inserting chunk {} to Geode: {}", file_hash, e);
+                            }
+                        };
+                        fud.file_fetch_end_tx.send((file_hash, Ok(()))).await.unwrap();
+                    }
+                }
+            }
+            None => {
+                fud.file_fetch_end_tx
+                    .send((file_hash, Err(Error::GeodeFileRouteNotFound)))
+                    .await
+                    .unwrap();
+            }
+        };
+    }
+}
+
+/// Background task that receives chunk fetch requests and tries to
+/// fetch objects from the network using the routing table.
+/// TODO: This can be optimised a lot for connection reuse, etc.
+pub async fn fetch_chunk_task(fud: Arc<Fud>) -> Result<()> {
+    info!(target: "fud::fetch_chunk_task()", "Started background chunk fetch task");
+    loop {
+        let (chunk_hash, _) = fud.chunk_fetch_rx.recv().await.unwrap();
+        info!(target: "fud::fetch_chunk_task()", "Fetching chunk {}", chunk_hash);
+
+        let result = fud.fetch(chunk_hash).await;
+
+        match result {
+            Some(reply) => {
+                match reply {
+                    FetchReply::Chunk(FudChunkReply { chunk }) => {
+                        // TODO: Verify chunk
+                        match fud.geode.insert_chunk(&chunk).await {
+                            Ok(inserted_hash) => {
+                                if inserted_hash != chunk_hash {
+                                    warn!("Received chunk does not match requested chunk");
+                                }
+                            }
+                            Err(e) => {
+                                error!("Failed inserting chunk {} to Geode: {}", chunk_hash, e);
+                            }
+                        };
+                        fud.chunk_fetch_end_tx.send((chunk_hash, Ok(()))).await.unwrap();
+                    }
+                    _ => {
+                        // Looked for a chunk but got a file instead, not supposed to happen
+                        fud.chunk_fetch_end_tx
+                            .send((chunk_hash, Err(Error::GeodeChunkRouteNotFound)))
+                            .await
+                            .unwrap();
+                    }
+                }
+            }
+            None => {
+                fud.chunk_fetch_end_tx
+                    .send((chunk_hash, Err(Error::GeodeChunkRouteNotFound)))
+                    .await
+                    .unwrap();
+            }
+        };
+    }
+}
+
+/// Background task that removes seeders that did not announce a file/chunk
+/// for more than an hour.
+pub async fn prune_seeders_task(fud: Arc<Fud>) -> Result<()> {
+    loop {
+        sleep(1800).await; // TODO: Make a setting
+
+        info!(target: "fud::prune_seeders_task()", "Pruning seeders...");
+        fud.dht().prune_router(fud.seeders_router.clone(), 3600).await;
+    }
+}
+
+/// Background task that announces our files and chunks once every hour.
+pub async fn announce_seed_task(fud: Arc<Fud>) -> Result<()> {
+    loop {
+        sleep(3600).await; // TODO: Make a setting
+
+        let seeders = vec![fud.dht().node.clone().into()];
+
+        info!(target: "fud::announce_task()", "Announcing chunks...");
+        let chunk_hashes = fud.geode.list_chunks().await;
+        if let Ok(chunks) = chunk_hashes {
+            for chunk in chunks {
+                let _ = fud
+                    .announce(
+                        &chunk,
+                        &FudAnnounce { key: chunk, seeders: seeders.clone() },
+                        fud.seeders_router.clone(),
+                    )
+                    .await;
+            }
+        }
+
+        info!(target: "fud::announce_task()", "Announcing files...");
+        let file_hashes = fud.geode.list_files().await;
+        if let Ok(files) = file_hashes {
+            for file in files {
+                let _ = fud
+                    .announce(
+                        &file,
+                        &FudAnnounce { key: file, seeders: seeders.clone() },
+                        fud.seeders_router.clone(),
+                    )
+                    .await;
+            }
+        }
+    }
+}

src/geode/mod.rs

@@ -105,6 +105,16 @@ impl ChunkedFile {
     pub fn iter(&self) -> core::slice::Iter<'_, (blake3::Hash, Option<PathBuf>)> {
         self.0.iter()
     }
+
+    /// Return the number of chunks.
+    pub fn len(&self) -> usize {
+        self.0.len()
+    }
+
+    /// Return `true` if the chunked file contains no chunk.
+    pub fn is_empty(&self) -> bool {
+        self.0.is_empty()
+    }
 }
 
 /// Chunk-based file storage interface.
@@ -505,8 +515,47 @@ impl Geode {
             let bytes_read = chunk_fd.read_to_end(&mut buf).await?;
             let chunk_slice = &buf[..bytes_read];
             file_fd.write(chunk_slice).await?;
+            file_fd.flush().await?;
         }
 
         Ok(file_path)
     }
+
+    /// List file hashes.
+    pub async fn list_files(&self) -> Result<Vec<blake3::Hash>> {
+        info!(target: "geode::list_files()", "[Geode] Listing files");
+
+        let mut dir = fs::read_dir(&self.files_path).await?;
+
+        let mut file_hashes = vec![];
+
+        while let Some(file) = dir.try_next().await? {
+            let os_file_name = file.file_name();
+            let file_name = os_file_name.to_string_lossy();
+            if let Ok(file_hash) = blake3::Hash::from_hex(file_name.to_string()) {
+                file_hashes.push(file_hash);
+            }
+        }
+
+        Ok(file_hashes)
+    }
+
+    /// List chunk hashes.
+    pub async fn list_chunks(&self) -> Result<Vec<blake3::Hash>> {
+        info!(target: "geode::list_chunks()", "[Geode] Listing chunks");
+
+        let mut dir = fs::read_dir(&self.chunks_path).await?;
+
+        let mut chunk_hashes = vec![];
+
+        while let Some(chunk) = dir.try_next().await? {
+            let os_file_name = chunk.file_name();
+            let file_name = os_file_name.to_string_lossy();
+            if let Ok(chunk_hash) = blake3::Hash::from_hex(file_name.to_string()) {
+                chunk_hashes.push(chunk_hash);
+            }
+        }
+
+        Ok(chunk_hashes)
+    }
 }