fud: add lib target, cleanups

bin/fud/fud/Cargo.toml

@@ -8,6 +8,14 @@ license = "AGPL-3.0-only"
 homepage = "https://dark.fi"
 repository = "https://codeberg.org/darkrenaissance/darkfi"
 
+[lib]
+name = "fud"
+path = "src/lib.rs"
+
+[[bin]]
+name = "fud"
+path = "src/main.rs"
+
 [dependencies]
 darkfi = {path = "../../../", features = ["async-daemonize", "geode", "rpc", "dht", "sled-overlay"]}
 darkfi-serial = {version = "0.4.2", features = ["hash"]}

bin/fud/fud/Makefile

@@ -17,7 +17,7 @@ SRC = \
 	$(shell find src -type f -name '*.rs') \
 	$(shell find ../../../src -type f -name '*.rs') \
 
-BIN = $(shell grep '^name = ' Cargo.toml | cut -d' ' -f3 | tr -d '"')
+BIN = $(shell grep '^name = ' Cargo.toml | sed 1q | cut -d' ' -f3 | tr -d '"')
 
 all: $(BIN)
 

bin/fud/fud/src/lib.rs

@@ -0,0 +1,968 @@
+/* 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 async_trait::async_trait;
+use futures::{future::FutureExt, pin_mut, select};
+use log::{debug, error, info, warn};
+use num_bigint::BigUint;
+use rand::{prelude::IteratorRandom, rngs::OsRng, seq::SliceRandom, RngCore};
+use sled_overlay::sled;
+use smol::{
+    channel,
+    fs::{File, OpenOptions},
+    io::{AsyncReadExt, AsyncWriteExt},
+    lock::RwLock,
+};
+use std::{
+    collections::{HashMap, HashSet},
+    io::ErrorKind,
+    path::{Path, PathBuf},
+    sync::Arc,
+};
+
+use darkfi::{
+    dht::{Dht, DhtHandler, DhtNode, DhtRouterItem, DhtRouterPtr},
+    geode::{hash_to_string, ChunkedFile, Geode},
+    net::{ChannelPtr, P2pPtr},
+    system::PublisherPtr,
+    util::path::expand_path,
+    Error, Result,
+};
+
+/// P2P protocols
+pub mod proto;
+use proto::{
+    FudAnnounce, FudChunkReply, FudFileReply, FudFindNodesReply, FudFindNodesRequest,
+    FudFindRequest, FudFindSeedersReply, FudFindSeedersRequest, FudNotFound, FudPingReply,
+    FudPingRequest,
+};
+
+/// FudEvent
+pub mod event;
+use event::{ChunkDownloadCompleted, ChunkNotFound, FudEvent, ResourceUpdated};
+
+/// Resource definition
+pub mod resource;
+use resource::{Resource, ResourceStatus};
+
+/// JSON-RPC related methods
+pub mod rpc;
+
+/// Background tasks
+pub mod tasks;
+use tasks::FetchReply;
+
+// TODO: This is not Sybil-resistant
+fn generate_node_id() -> Result<blake3::Hash> {
+    let mut rng = OsRng;
+    let mut random_data = [0u8; 32];
+    rng.fill_bytes(&mut random_data);
+    let node_id = blake3::Hash::from_bytes(random_data);
+    Ok(node_id)
+}
+
+/// Get or generate the node id.
+/// Fetches and saves the node id from/to a file.
+pub async fn get_node_id(node_id_path: &Path) -> Result<blake3::Hash> {
+    match File::open(node_id_path).await {
+        Ok(mut file) => {
+            let mut buffer = Vec::new();
+            file.read_to_end(&mut buffer).await?;
+            let mut out_buf = [0u8; 32];
+            bs58::decode(buffer).onto(&mut out_buf)?;
+            let node_id = blake3::Hash::from_bytes(out_buf);
+            Ok(node_id)
+        }
+        Err(e) if e.kind() == ErrorKind::NotFound => {
+            let node_id = generate_node_id()?;
+            let mut file = OpenOptions::new().write(true).create(true).open(node_id_path).await?;
+            file.write_all(&bs58::encode(node_id.as_bytes()).into_vec()).await?;
+            file.flush().await?;
+            Ok(node_id)
+        }
+        Err(e) => Err(e.into()),
+    }
+}
+
+pub struct Fud {
+    /// Key -> Seeders
+    seeders_router: DhtRouterPtr,
+
+    /// Pointer to the P2P network instance
+    p2p: P2pPtr,
+
+    /// The Geode instance
+    geode: Geode,
+
+    /// Default download directory
+    downloads_path: PathBuf,
+
+    /// Chunk transfer timeout in seconds
+    chunk_timeout: u64,
+
+    /// The DHT instance
+    dht: Arc<Dht>,
+
+    /// Resources (current status of all downloads/seeds)
+    resources: Arc<RwLock<HashMap<blake3::Hash, Resource>>>,
+
+    /// Sled tree containing "resource hash -> path on the filesystem"
+    path_tree: sled::Tree,
+
+    get_tx: channel::Sender<(u16, blake3::Hash, PathBuf, Result<()>)>,
+    get_rx: channel::Receiver<(u16, blake3::Hash, PathBuf, Result<()>)>,
+    file_fetch_tx: channel::Sender<(Vec<DhtNode>, blake3::Hash, PathBuf, Result<()>)>,
+    file_fetch_rx: channel::Receiver<(Vec<DhtNode>, blake3::Hash, PathBuf, Result<()>)>,
+    file_fetch_end_tx: channel::Sender<(blake3::Hash, Result<()>)>,
+    file_fetch_end_rx: channel::Receiver<(blake3::Hash, Result<()>)>,
+
+    event_publisher: PublisherPtr<FudEvent>,
+}
+
+#[async_trait]
+impl DhtHandler for Fud {
+    fn dht(&self) -> Arc<Dht> {
+        self.dht.clone()
+    }
+
+    async fn ping(&self, channel: ChannelPtr) -> Result<DhtNode> {
+        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();
+        let request = FudPingRequest {};
+
+        channel.send(&request).await?;
+
+        let reply = msg_subscriber.receive_with_timeout(self.dht().settings.timeout).await?;
+
+        msg_subscriber.unsubscribe().await;
+
+        Ok(reply.node.clone())
+    }
+
+    // TODO: Optimize this
+    async fn on_new_node(&self, node: &DhtNode) -> 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
+        if !self.dht().is_bootstrapped().await {
+            self.dht().set_bootstrapped().await;
+
+            // Lookup our own node id
+            debug!(target: "fud::DhtHandler::on_new_node()", "DHT bootstrapping {}", hash_to_string(&self.dht().node_id));
+            let _ = self.lookup_nodes(&self.dht().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, 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,
+                        seeders: seeders.clone().into_iter().collect(),
+                    })
+                    .await;
+            }
+        }
+
+        Ok(())
+    }
+
+    async fn fetch_nodes(&self, node: &DhtNode, key: &blake3::Hash) -> Result<Vec<DhtNode>> {
+        debug!(target: "fud::DhtHandler::fetch_value()", "Fetching nodes close to {} from node {}", hash_to_string(key), hash_to_string(&node.id));
+
+        let channel = self.get_channel(node).await?;
+        let msg_subsystem = channel.message_subsystem();
+        msg_subsystem.add_dispatch::<FudFindNodesReply>().await;
+        let msg_subscriber_nodes = channel.subscribe_msg::<FudFindNodesReply>().await.unwrap();
+
+        let request = FudFindNodesRequest { key: *key };
+        channel.send(&request).await?;
+
+        let reply = msg_subscriber_nodes.receive_with_timeout(self.dht().settings.timeout).await?;
+
+        msg_subscriber_nodes.unsubscribe().await;
+
+        Ok(reply.nodes.clone())
+    }
+}
+
+impl Fud {
+    pub async fn new(
+        p2p: P2pPtr,
+        basedir: PathBuf,
+        downloads_path: PathBuf,
+        chunk_timeout: u64,
+        dht: Arc<Dht>,
+        path_tree: sled::Tree,
+        event_publisher: PublisherPtr<FudEvent>,
+    ) -> Result<Self> {
+        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();
+
+        // Hashmap used for routing
+        let seeders_router = Arc::new(RwLock::new(HashMap::new()));
+
+        info!("Instantiating Geode instance");
+        let geode = Geode::new(&basedir).await?;
+
+        info!("Instantiating DHT instance");
+
+        let fud = Self {
+            seeders_router,
+            p2p,
+            geode,
+            downloads_path,
+            chunk_timeout,
+            dht,
+            path_tree,
+            resources: Arc::new(RwLock::new(HashMap::new())),
+            get_tx,
+            get_rx,
+            file_fetch_tx,
+            file_fetch_rx,
+            file_fetch_end_tx,
+            file_fetch_end_rx,
+            event_publisher,
+        };
+
+        fud.init().await?;
+
+        Ok(fud)
+    }
+
+    /// Add ourselves to `seeders_router` for the files we already have.
+    /// Skipped if we have no external address.
+    async fn init(&self) -> Result<()> {
+        info!(target: "fud::init()", "Finding resources...");
+        let mut resources_write = self.resources.write().await;
+        for result in self.path_tree.iter() {
+            if result.is_err() {
+                continue;
+            }
+
+            // Parse hash
+            let (hash, path) = result.unwrap();
+            let hash_bytes: [u8; 32] = match hash.to_vec().try_into() {
+                Ok(v) => v,
+                Err(_) => continue,
+            };
+            let hash = blake3::Hash::from_bytes(hash_bytes);
+
+            // Parse path
+            let path_bytes = path.to_vec();
+            let path_str = match std::str::from_utf8(&path_bytes) {
+                Ok(v) => v,
+                Err(_) => continue,
+            };
+            let path: PathBuf = match expand_path(path_str) {
+                Ok(v) => v,
+                Err(_) => continue,
+            };
+
+            // Add resource
+            resources_write.insert(
+                hash,
+                Resource {
+                    hash,
+                    path,
+                    status: ResourceStatus::Incomplete,
+                    chunks_total: 0,
+                    chunks_downloaded: 0,
+                },
+            );
+        }
+        drop(resources_write);
+
+        info!(target: "fud::init()", "Verifying resources...");
+        let resources = self.verify_resources(None).await?;
+
+        let self_node = self.dht().node().await;
+
+        if self_node.addresses.is_empty() {
+            return Ok(());
+        }
+
+        info!(target: "fud::init()", "Start seeding...");
+        let self_router_items: Vec<DhtRouterItem> = vec![self_node.into()];
+
+        for resource in resources {
+            self.add_to_router(
+                self.seeders_router.clone(),
+                &resource.hash,
+                self_router_items.clone(),
+            )
+            .await;
+        }
+
+        Ok(())
+    }
+
+    /// Get resource path from hash using the sled db
+    pub fn hash_to_path(&self, hash: &blake3::Hash) -> Result<Option<PathBuf>> {
+        if let Some(value) = self.path_tree.get(hash.as_bytes())? {
+            let path: PathBuf = expand_path(std::str::from_utf8(&value)?)?;
+            return Ok(Some(path));
+        }
+
+        Ok(None)
+    }
+
+    /// 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.
+    /// Takes an optional list of hashes.
+    /// If no hash is given (None), it verifies all resources.
+    /// Returns the list of verified and uncorrupted/complete seeding resources.
+    pub async fn verify_resources(
+        &self,
+        hashes: Option<Vec<blake3::Hash>>,
+    ) -> Result<Vec<Resource>> {
+        let mut resources_write = self.resources.write().await;
+
+        let update_resource =
+            async |resource: &mut Resource,
+                   status: ResourceStatus,
+                   chunked_file: Option<&ChunkedFile>| {
+                resource.status = status;
+                resource.chunks_total = match chunked_file {
+                    Some(chunked_file) => chunked_file.len() as u64,
+                    None => 0,
+                };
+                resource.chunks_downloaded = match chunked_file {
+                    Some(chunked_file) => chunked_file.local_chunks() as u64,
+                    None => 0,
+                };
+
+                self.event_publisher
+                    .notify(FudEvent::ResourceUpdated(ResourceUpdated {
+                        hash: resource.hash,
+                        resource: resource.clone(),
+                    }))
+                    .await;
+            };
+
+        let mut seeding_resources: Vec<Resource> = vec![];
+        for (_, mut resource) in resources_write.iter_mut() {
+            if let Some(ref hashes_list) = hashes {
+                if !hashes_list.contains(&resource.hash) {
+                    continue;
+                }
+            }
+
+            match resource.status {
+                ResourceStatus::Seeding => {}
+                ResourceStatus::Incomplete => {}
+                _ => continue,
+            };
+
+            // Make sure the resource is not corrupted or incomplete
+            let resource_path = match self.hash_to_path(&resource.hash) {
+                Ok(Some(v)) => v,
+                Ok(None) | Err(_) => {
+                    update_resource(&mut resource, ResourceStatus::Incomplete, None).await;
+                    continue;
+                }
+            };
+            let chunked_file = match self.geode.get(&resource.hash, &resource_path).await {
+                Ok(v) => v,
+                Err(_) => {
+                    update_resource(&mut resource, ResourceStatus::Incomplete, None).await;
+                    continue;
+                }
+            };
+            if !chunked_file.is_complete() {
+                update_resource(&mut resource, ResourceStatus::Incomplete, Some(&chunked_file))
+                    .await;
+                continue;
+            }
+
+            update_resource(&mut resource, ResourceStatus::Seeding, Some(&chunked_file)).await;
+            seeding_resources.push(resource.clone());
+        }
+
+        Ok(seeding_resources)
+    }
+
+    /// Query `nodes` to find the seeders for `key`
+    async fn fetch_seeders(
+        &self,
+        nodes: &Vec<DhtNode>,
+        key: &blake3::Hash,
+    ) -> HashSet<DhtRouterItem> {
+        let mut seeders: HashSet<DhtRouterItem> = HashSet::new();
+
+        for node in nodes {
+            let channel = match self.get_channel(node).await {
+                Ok(channel) => channel,
+                Err(e) => {
+                    warn!(target: "fud::fetch_seeders()", "Could not get a channel for node {}: {}", hash_to_string(&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_seeders()", "Error subscribing to msg: {}", e);
+                    continue;
+                }
+            };
+
+            let send_res = channel.send(&FudFindSeedersRequest { key: *key }).await;
+            if let Err(e) = send_res {
+                warn!(target: "fud::fetch_seeders()", "Error while sending FudFindSeedersRequest: {}", e);
+                msg_subscriber.unsubscribe().await;
+                continue;
+            }
+
+            let reply = match msg_subscriber.receive_with_timeout(self.dht().settings.timeout).await
+            {
+                Ok(reply) => reply,
+                Err(e) => {
+                    warn!(target: "fud::fetch_seeders()", "Error waiting for reply: {}", e);
+                    msg_subscriber.unsubscribe().await;
+                    continue;
+                }
+            };
+
+            msg_subscriber.unsubscribe().await;
+
+            seeders.extend(reply.seeders.clone());
+        }
+
+        info!(target: "fud::fetch_seeders()", "Found {} seeders for {}", seeders.len(), hash_to_string(key));
+        seeders
+    }
+
+    /// Fetch chunks for a file from `seeders`
+    async fn fetch_chunks(
+        &self,
+        file_path: &PathBuf,
+        file_hash: &blake3::Hash,
+        chunk_hashes: &HashSet<blake3::Hash>,
+        seeders: &HashSet<DhtRouterItem>,
+    ) -> Result<()> {
+        let mut remaining_chunks = chunk_hashes.clone();
+        let mut shuffled_seeders = {
+            let mut vec: Vec<_> = seeders.iter().cloned().collect();
+            vec.shuffle(&mut OsRng);
+            vec
+        };
+
+        while let Some(seeder) = shuffled_seeders.pop() {
+            let channel = match self.get_channel(&seeder.node).await {
+                Ok(channel) => channel,
+                Err(e) => {
+                    warn!(target: "fud::fetch_chunks()", "Could not get a channel for node {}: {}", hash_to_string(&seeder.node.id), e);
+                    continue;
+                }
+            };
+            let mut chunks_to_query = remaining_chunks.clone();
+            info!("Requesting chunks from seeder {}", hash_to_string(&seeder.node.id));
+            loop {
+                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();
+
+                // Select a chunk to request
+                let mut chunk_hash: Option<blake3::Hash> = None;
+                if let Some(&random_chunk) = chunks_to_query.iter().choose(&mut OsRng) {
+                    chunk_hash = Some(random_chunk);
+                }
+
+                if chunk_hash.is_none() {
+                    // No more chunks to request from this seeder
+                    break; // Switch to another seeder
+                }
+                let chunk_hash = chunk_hash.unwrap();
+                chunks_to_query.remove(&chunk_hash);
+
+                let send_res =
+                    channel.send(&FudFindRequest { info: Some(*file_hash), key: chunk_hash }).await;
+                if let Err(e) = send_res {
+                    warn!(target: "fud::fetch_chunks()", "Error while sending FudFindRequest: {}", e);
+                    break; // Switch to another seeder
+                }
+
+                let chunk_recv =
+                    msg_subscriber_chunk.receive_with_timeout(self.chunk_timeout).fuse();
+                let notfound_recv =
+                    msg_subscriber_notfound.receive_with_timeout(self.chunk_timeout).fuse();
+
+                pin_mut!(chunk_recv, notfound_recv);
+
+                // Wait for a FudChunkReply or FudNotFound
+                select! {
+                    chunk_reply = chunk_recv => {
+                        if let Err(e) = chunk_reply {
+                            warn!(target: "fud::fetch_chunks()", "Error waiting for chunk reply: {}", e);
+                            break; // Switch to another seeder
+                        }
+                        let reply = chunk_reply.unwrap();
+
+                        match self.geode.write_chunk(file_hash, file_path, &reply.chunk).await {
+                            Ok(inserted_hash) => {
+                                if inserted_hash != chunk_hash {
+                                    warn!("Received chunk does not match requested chunk");
+                                    msg_subscriber_chunk.unsubscribe().await;
+                                    msg_subscriber_notfound.unsubscribe().await;
+                                    continue; // Skip to next chunk, will retry this chunk later
+                                }
+
+                                // Update resource `chunks_downloaded`
+                                let mut resources_write = self.resources.write().await;
+                                let resource = match resources_write.get_mut(file_hash) {
+                                    Some(resource) => {
+                                        resource.status = ResourceStatus::Downloading;
+                                        resource.chunks_downloaded += 1;
+                                        resource.clone()
+                                    }
+                                    None => return Ok(()) // Resource was removed, abort
+                                };
+                                drop(resources_write);
+
+                                info!(target: "fud::fetch_chunks()", "Received chunk {} from seeder {}", hash_to_string(&chunk_hash), hash_to_string(&seeder.node.id));
+                                self.event_publisher
+                                    .notify(FudEvent::ChunkDownloadCompleted(ChunkDownloadCompleted {
+                                        hash: *file_hash,
+                                        chunk_hash,
+                                        resource,
+                                    }))
+                                    .await;
+                                remaining_chunks.remove(&chunk_hash);
+                            }
+                            Err(e) => {
+                                error!("Failed inserting chunk {} to Geode: {}", hash_to_string(&chunk_hash), e);
+                            }
+                        };
+                    }
+                    notfound_reply = notfound_recv => {
+                        if let Err(e) = notfound_reply {
+                            warn!(target: "fud::fetch_chunks()", "Error waiting for NOTFOUND reply: {}", e);
+                            msg_subscriber_chunk.unsubscribe().await;
+                            msg_subscriber_notfound.unsubscribe().await;
+                            break; // Switch to another seeder
+                        }
+                        info!(target: "fud::fetch_chunks()", "Received NOTFOUND {} from seeder {}", hash_to_string(&chunk_hash), hash_to_string(&seeder.node.id));
+                        self.event_publisher
+                            .notify(FudEvent::ChunkNotFound(ChunkNotFound {
+                                hash: *file_hash,
+                                chunk_hash,
+                            }))
+                        .await;
+                    }
+                };
+
+                msg_subscriber_chunk.unsubscribe().await;
+                msg_subscriber_notfound.unsubscribe().await;
+            }
+
+            // Stop when there are no missing chunks
+            if remaining_chunks.is_empty() {
+                break;
+            }
+        }
+
+        Ok(())
+    }
+
+    /// Fetch a single file metadata from `nodes`.
+    /// If the file is smaller than a single chunk then the chunk is returned.
+    /// 1. Request seeders for the file from those nodes
+    /// 2. Request the file from the seeders
+    pub async fn fetch_file_metadata(
+        &self,
+        nodes: Vec<DhtNode>,
+        file_hash: blake3::Hash,
+    ) -> Option<FetchReply> {
+        let mut queried_seeders: HashSet<blake3::Hash> = HashSet::new();
+        let mut result: Option<FetchReply> = None;
+
+        for node in nodes {
+            // 1. Request list of seeders
+            let channel = match self.get_channel(&node).await {
+                Ok(channel) => channel,
+                Err(e) => {
+                    warn!(target: "fud::fetch_file_metadata()", "Could not get a channel for node {}: {}", hash_to_string(&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_file_metadata()", "Error subscribing to msg: {}", e);
+                    continue;
+                }
+            };
+
+            let send_res = channel.send(&FudFindSeedersRequest { key: file_hash }).await;
+            if let Err(e) = send_res {
+                warn!(target: "fud::fetch_file_metadata()", "Error while sending FudFindSeedersRequest: {}", e);
+                msg_subscriber.unsubscribe().await;
+                continue;
+            }
+
+            let reply = match msg_subscriber.receive_with_timeout(self.dht().settings.timeout).await
+            {
+                Ok(reply) => reply,
+                Err(e) => {
+                    warn!(target: "fud::fetch_file_metadata()", "Error waiting for reply: {}", e);
+                    msg_subscriber.unsubscribe().await;
+                    continue;
+                }
+            };
+
+            let mut seeders = reply.seeders.clone();
+            info!(target: "fud::fetch_file_metadata()", "Found {} seeders for {} (from {})", seeders.len(), hash_to_string(&file_hash), hash_to_string(&node.id));
+
+            msg_subscriber.unsubscribe().await;
+
+            // 2. 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 { info: None, key: file_hash }).await;
+                    if let Err(e) = send_res {
+                        warn!(target: "fud::fetch_file_metadata()", "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.chunk_timeout).fuse();
+                    let file_recv =
+                        msg_subscriber_file.receive_with_timeout(self.chunk_timeout).fuse();
+                    let notfound_recv =
+                        msg_subscriber_notfound.receive_with_timeout(self.chunk_timeout).fuse();
+
+                    pin_mut!(chunk_recv, file_recv, notfound_recv);
+
+                    let cleanup = async || {
+                        msg_subscriber_chunk.unsubscribe().await;
+                        msg_subscriber_file.unsubscribe().await;
+                        msg_subscriber_notfound.unsubscribe().await;
+                    };
+
+                    // Wait for a FudChunkReply, FudFileReply, or FudNotFound
+                    select! {
+                        // Received a chunk while requesting a file, this is allowed to
+                        // optimize fetching files smaller than a single chunk
+                        chunk_reply = chunk_recv => {
+                            cleanup().await;
+                            if let Err(e) = chunk_reply {
+                                warn!(target: "fud::fetch_file_metadata()", "Error waiting for chunk reply: {}", e);
+                                continue;
+                            }
+                            let reply = chunk_reply.unwrap();
+                            let chunk_hash = blake3::hash(&reply.chunk);
+                            // Check that this is the only chunk in the file
+                            if !self.geode.verify_file(&file_hash, &[chunk_hash]) {
+                                warn!(target: "fud::fetch_file_metadata()", "Received a chunk while fetching a file, the chunk did not match the file hash");
+                                continue;
+                            }
+                            info!(target: "fud::fetch_file_metadata()", "Received chunk {} (for file {}) from seeder {}", hash_to_string(&chunk_hash), hash_to_string(&file_hash), hash_to_string(&seeder.node.id));
+                            result = Some(FetchReply::Chunk((*reply).clone()));
+                            break;
+                        }
+                        file_reply = file_recv => {
+                            cleanup().await;
+                            if let Err(e) = file_reply {
+                                warn!(target: "fud::fetch_file_metadata()", "Error waiting for file reply: {}", e);
+                                continue;
+                            }
+                            let reply = file_reply.unwrap();
+                            if !self.geode.verify_file(&file_hash, &reply.chunk_hashes) {
+                                warn!(target: "fud::fetch_file_metadata()", "Received invalid file metadata");
+                                continue;
+                            }
+                            info!(target: "fud::fetch_file_metadata()", "Received file {} from seeder {}", hash_to_string(&file_hash), hash_to_string(&seeder.node.id));
+                            result = Some(FetchReply::File((*reply).clone()));
+                            break;
+                        }
+                        notfound_reply = notfound_recv => {
+                            cleanup().await;
+                            if let Err(e) = notfound_reply {
+                                warn!(target: "fud::fetch_file_metadata()", "Error waiting for NOTFOUND reply: {}", e);
+                                continue;
+                            }
+                            info!(target: "fud::fetch_file_metadata()", "Received NOTFOUND {} from seeder {}", hash_to_string(&file_hash), hash_to_string(&seeder.node.id));
+                        }
+                    };
+                }
+            }
+
+            if result.is_some() {
+                break;
+            }
+        }
+
+        result
+    }
+
+    /// Download a file from the network to `file_path`.
+    pub async fn get(&self, file_hash: &blake3::Hash, file_path: &PathBuf) -> Result<()> {
+        let self_node = self.dht().node().await;
+        let mut closest_nodes = vec![];
+
+        // Add path to the sled db
+        self.path_tree
+            .insert(file_hash.as_bytes(), file_path.to_string_lossy().to_string().as_bytes())?;
+
+        // Add resource to `self.resources`
+        let resource = Resource {
+            hash: *file_hash,
+            path: file_path.clone(),
+            status: ResourceStatus::Discovering,
+            chunks_total: 0,
+            chunks_downloaded: 0,
+        };
+        let mut resources_write = self.resources.write().await;
+        resources_write.insert(*file_hash, resource.clone());
+        drop(resources_write);
+
+        // Send a DownloadStarted event
+        self.event_publisher
+            .notify(FudEvent::DownloadStarted(event::DownloadStarted {
+                hash: *file_hash,
+                resource,
+            }))
+            .await;
+
+        // Try to get the chunked file from geode
+        let chunked_file = match self.geode.get(file_hash, file_path).await {
+            // We already know the list of chunk hashes for this file
+            Ok(v) => v,
+            // The metadata in geode is invalid or corrupted
+            Err(Error::GeodeNeedsGc) => todo!(),
+            // If we could not find the file in geode, get the file metadata from the network
+            Err(Error::GeodeFileNotFound) => {
+                // Find nodes close to the file hash
+                info!(target: "self::get()", "Requested file {} not found in Geode, triggering fetch", hash_to_string(file_hash));
+                closest_nodes = self.lookup_nodes(file_hash).await.unwrap_or_default();
+
+                // Fetch file metadata (list of chunk hashes)
+                self.file_fetch_tx
+                    .send((closest_nodes.clone(), *file_hash, file_path.clone(), 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 {
+                    // The file metadata was found and inserted into geode
+                    Ok(()) => self.geode.get(&i_file_hash, file_path).await.unwrap(),
+                    // We could not find the file metadata
+                    Err(Error::GeodeFileRouteNotFound) => {
+                        // Set resource status to `Incomplete` and send FudEvent::FileNotFound
+                        let mut resources_write = self.resources.write().await;
+                        if let Some(resource) = resources_write.get_mut(file_hash) {
+                            resource.status = ResourceStatus::Incomplete;
+
+                            self.event_publisher
+                                .notify(FudEvent::FileNotFound(event::FileNotFound {
+                                    hash: *file_hash,
+                                    resource: resource.clone(),
+                                }))
+                                .await;
+                        }
+                        drop(resources_write);
+                        return Err(Error::GeodeFileRouteNotFound);
+                    }
+
+                    Err(e) => {
+                        error!(target: "fud::handle_get()", "{}", e);
+                        return Err(e);
+                    }
+                }
+            }
+
+            Err(e) => {
+                error!(target: "fud::handle_get()", "{}", e);
+                return Err(e);
+            }
+        };
+
+        // Set resource status to `Downloading`
+        let mut resources_write = self.resources.write().await;
+        let resource = match resources_write.get_mut(file_hash) {
+            Some(resource) => {
+                resource.status = ResourceStatus::Downloading;
+                resource.chunks_downloaded = chunked_file.local_chunks() as u64;
+                resource.chunks_total = chunked_file.len() as u64;
+                resource.clone()
+            }
+            None => return Ok(()), // Resource was removed, abort
+        };
+        drop(resources_write);
+
+        // Send a FileDownloadCompleted event
+        self.event_publisher
+            .notify(FudEvent::FileDownloadCompleted(event::FileDownloadCompleted {
+                hash: *file_hash,
+                resource: resource.clone(),
+            }))
+            .await;
+
+        // If the file is already complete, we don't need to download any chunk
+        if chunked_file.is_complete() {
+            // Announce the file
+            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;
+
+            // Set resource status to `Seeding`
+            let mut resources_write = self.resources.write().await;
+            let resource = match resources_write.get_mut(file_hash) {
+                Some(resource) => {
+                    resource.status = ResourceStatus::Seeding;
+                    resource.chunks_downloaded = chunked_file.len() as u64;
+                    resource.clone()
+                }
+                None => return Ok(()), // Resource was removed, abort
+            };
+            drop(resources_write);
+
+            // Send a DownloadCompleted event
+            self.event_publisher
+                .notify(FudEvent::DownloadCompleted(event::DownloadCompleted {
+                    hash: *file_hash,
+                    resource,
+                }))
+                .await;
+
+            return Ok(());
+        }
+
+        // Find nodes close to the file hash if we didn't previously fetched them
+        if closest_nodes.is_empty() {
+            closest_nodes = self.lookup_nodes(file_hash).await.unwrap_or_default();
+        }
+
+        // Find seeders and remove ourselves from the result
+        let seeders = self
+            .fetch_seeders(&closest_nodes, file_hash)
+            .await
+            .iter()
+            .filter(|seeder| seeder.node.id != self_node.id)
+            .cloned()
+            .collect();
+
+        // List missing chunks
+        let mut missing_chunks = HashSet::new();
+        for (chunk, path) in chunked_file.iter() {
+            if path.is_none() {
+                missing_chunks.insert(*chunk);
+            }
+        }
+
+        // Fetch missing chunks from seeders
+        self.fetch_chunks(file_path, file_hash, &missing_chunks, &seeders).await?;
+
+        // Get chunked file from geode
+        let chunked_file = match self.geode.get(file_hash, file_path).await {
+            Ok(v) => v,
+            Err(e) => {
+                error!(target: "fud::handle_get()", "{}", e);
+                return Err(e);
+            }
+        };
+
+        // We fetched all chunks, but the file is not complete
+        // (some chunks were missing from all seeders)
+        if !chunked_file.is_complete() {
+            // Set resource status to `Incomplete`
+            let mut resources_write = self.resources.write().await;
+            let resource = match resources_write.get_mut(file_hash) {
+                Some(resource) => {
+                    resource.status = ResourceStatus::Incomplete;
+                    resource.clone()
+                }
+                None => return Ok(()), // Resource was removed, abort
+            };
+            drop(resources_write);
+
+            // Send a MissingChunks event
+            self.event_publisher
+                .notify(FudEvent::MissingChunks(event::MissingChunks {
+                    hash: *file_hash,
+                    resource,
+                }))
+                .await;
+            return Ok(());
+        }
+
+        // Announce the file
+        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;
+
+        // Set resource status to `Seeding`
+        let mut resources_write = self.resources.write().await;
+        let resource = match resources_write.get_mut(file_hash) {
+            Some(resource) => {
+                resource.status = ResourceStatus::Seeding;
+                resource.chunks_downloaded = chunked_file.len() as u64;
+                resource.clone()
+            }
+            None => return Ok(()), // Resource was removed, abort
+        };
+        drop(resources_write);
+
+        // Send a DownloadCompleted event
+        self.event_publisher
+            .notify(FudEvent::DownloadCompleted(event::DownloadCompleted {
+                hash: *file_hash,
+                resource,
+            }))
+            .await;
+
+        Ok(())
+    }
+}

bin/fud/fud/src/main.rs

@@ -16,67 +16,37 @@
  * along with this program.  If not, see <https://www.gnu.org/licenses/>.
  */
 
-use async_trait::async_trait;
-use futures::{future::FutureExt, pin_mut, select};
 use log::{debug, error, info, warn};
-use num_bigint::BigUint;
-use rand::{prelude::IteratorRandom, rngs::OsRng, seq::SliceRandom, RngCore};
 use sled_overlay::sled;
-use smol::{
-    channel,
-    fs::{File, OpenOptions},
-    io::{AsyncReadExt, AsyncWriteExt},
-    lock::{Mutex, RwLock},
-    stream::StreamExt,
-    Executor,
-};
-use std::{
-    collections::{HashMap, HashSet},
-    io::ErrorKind,
-    path::PathBuf,
-    sync::Arc,
-};
+use smol::{stream::StreamExt, Executor};
+use std::sync::Arc;
 use structopt_toml::{structopt::StructOpt, StructOptToml};
 
 use darkfi::{
     async_daemonize, cli_desc,
-    dht::{Dht, DhtHandler, DhtNode, DhtRouterItem, DhtRouterPtr, DhtSettings, DhtSettingsOpt},
-    geode::{hash_to_string, ChunkedFile, Geode},
-    net::{
-        session::SESSION_DEFAULT, settings::SettingsOpt, ChannelPtr, P2p, P2pPtr,
-        Settings as NetSettings,
-    },
+    dht::{Dht, DhtHandler, DhtSettings, DhtSettingsOpt},
+    geode::hash_to_string,
+    net::{session::SESSION_DEFAULT, settings::SettingsOpt, P2p, Settings as NetSettings},
     rpc::{
         jsonrpc::JsonSubscriber,
-        p2p_method::HandlerP2p,
         server::{listen_and_serve, RequestHandler},
         settings::{RpcSettings, RpcSettingsOpt},
     },
-    system::{Publisher, PublisherPtr, StoppableTask, StoppableTaskPtr},
+    system::{Publisher, StoppableTask},
     util::path::expand_path,
     Error, Result,
 };
 
-use event::{ChunkDownloadCompleted, ChunkNotFound, FudEvent, ResourceUpdated};
-use resource::{Resource, ResourceStatus};
-use tasks::FetchReply;
-
-/// P2P protocols
-mod proto;
-use proto::{
-    FudAnnounce, FudChunkReply, FudFileReply, FudFindNodesReply, FudFindNodesRequest,
-    FudFindRequest, FudFindSeedersReply, FudFindSeedersRequest, FudNotFound, FudPingReply,
-    FudPingRequest, ProtocolFud,
+use fud::{
+    get_node_id,
+    proto::{FudFindNodesReply, ProtocolFud},
+    rpc::JsonRpcInterface,
+    tasks::{announce_seed_task, fetch_file_task, get_task},
+    Fud,
 };
 
-mod event;
-mod resource;
-mod rpc;
-mod tasks;
-
 const CONFIG_FILE: &str = "fud_config.toml";
 const CONFIG_FILE_CONTENTS: &str = include_str!("../fud_config.toml");
-
 const NODE_ID_PATH: &str = "node_id";
 
 #[derive(Clone, Debug, serde::Deserialize, StructOpt, StructOptToml)]
@@ -120,631 +90,6 @@ struct Args {
     dht: DhtSettingsOpt,
 }
 
-pub struct Fud {
-    /// Key -> Seeders
-    seeders_router: DhtRouterPtr,
-
-    /// Pointer to the P2P network instance
-    p2p: P2pPtr,
-
-    /// The Geode instance
-    geode: Geode,
-
-    /// Default download directory
-    downloads_path: PathBuf,
-
-    /// Chunk transfer timeout in seconds
-    chunk_timeout: u64,
-
-    /// The DHT instance
-    dht: Arc<Dht>,
-
-    /// Resources (current status of all downloads/seeds)
-    resources: Arc<RwLock<HashMap<blake3::Hash, Resource>>>,
-
-    /// Sled tree containing "resource hash -> path on the filesystem"
-    path_tree: sled::Tree,
-
-    get_tx: channel::Sender<(u16, blake3::Hash, PathBuf, Result<()>)>,
-    get_rx: channel::Receiver<(u16, blake3::Hash, PathBuf, Result<()>)>,
-    file_fetch_tx: channel::Sender<(Vec<DhtNode>, blake3::Hash, PathBuf, Result<()>)>,
-    file_fetch_rx: channel::Receiver<(Vec<DhtNode>, blake3::Hash, PathBuf, Result<()>)>,
-    file_fetch_end_tx: channel::Sender<(blake3::Hash, Result<()>)>,
-    file_fetch_end_rx: channel::Receiver<(blake3::Hash, Result<()>)>,
-
-    rpc_connections: Mutex<HashSet<StoppableTaskPtr>>,
-
-    /// dnet JSON-RPC subscriber
-    dnet_sub: JsonSubscriber,
-
-    /// Download JSON-RPC subscriber
-    event_sub: JsonSubscriber,
-
-    event_publisher: PublisherPtr<FudEvent>,
-}
-
-impl HandlerP2p for Fud {
-    fn p2p(&self) -> P2pPtr {
-        self.p2p.clone()
-    }
-}
-
-#[async_trait]
-impl DhtHandler for Fud {
-    fn dht(&self) -> Arc<Dht> {
-        self.dht.clone()
-    }
-
-    async fn ping(&self, channel: ChannelPtr) -> Result<DhtNode> {
-        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();
-        let request = FudPingRequest {};
-
-        channel.send(&request).await?;
-
-        let reply = msg_subscriber.receive_with_timeout(self.dht().settings.timeout).await?;
-
-        msg_subscriber.unsubscribe().await;
-
-        Ok(reply.node.clone())
-    }
-
-    // TODO: Optimize this
-    async fn on_new_node(&self, node: &DhtNode) -> 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
-        if !self.dht().is_bootstrapped().await {
-            self.dht().set_bootstrapped().await;
-
-            // Lookup our own node id
-            debug!(target: "fud::DhtHandler::on_new_node()", "DHT bootstrapping {}", hash_to_string(&self.dht().node_id));
-            let _ = self.lookup_nodes(&self.dht().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, 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,
-                        seeders: seeders.clone().into_iter().collect(),
-                    })
-                    .await;
-            }
-        }
-
-        Ok(())
-    }
-
-    async fn fetch_nodes(&self, node: &DhtNode, key: &blake3::Hash) -> Result<Vec<DhtNode>> {
-        debug!(target: "fud::DhtHandler::fetch_value()", "Fetching nodes close to {} from node {}", hash_to_string(key), hash_to_string(&node.id));
-
-        let channel = self.get_channel(node).await?;
-        let msg_subsystem = channel.message_subsystem();
-        msg_subsystem.add_dispatch::<FudFindNodesReply>().await;
-        let msg_subscriber_nodes = channel.subscribe_msg::<FudFindNodesReply>().await.unwrap();
-
-        let request = FudFindNodesRequest { key: *key };
-        channel.send(&request).await?;
-
-        let reply = msg_subscriber_nodes.receive_with_timeout(self.dht().settings.timeout).await?;
-
-        msg_subscriber_nodes.unsubscribe().await;
-
-        Ok(reply.nodes.clone())
-    }
-}
-
-impl Fud {
-    /// Add ourselves to `seeders_router` for the files we already have.
-    /// Skipped if we have no external address.
-    async fn init(&self) -> Result<()> {
-        info!(target: "fud::init()", "Finding resources...");
-        let mut resources_write = self.resources.write().await;
-        for result in self.path_tree.iter() {
-            if result.is_err() {
-                continue;
-            }
-
-            // Parse hash
-            let (hash, path) = result.unwrap();
-            let hash_bytes: [u8; 32] = match hash.to_vec().try_into() {
-                Ok(v) => v,
-                Err(_) => continue,
-            };
-            let hash = blake3::Hash::from_bytes(hash_bytes);
-
-            // Parse path
-            let path_bytes = path.to_vec();
-            let path_str = match std::str::from_utf8(&path_bytes) {
-                Ok(v) => v,
-                Err(_) => continue,
-            };
-            let path: PathBuf = match expand_path(path_str) {
-                Ok(v) => v,
-                Err(_) => continue,
-            };
-
-            // Add resource
-            resources_write.insert(
-                hash,
-                Resource {
-                    hash,
-                    path,
-                    status: ResourceStatus::Incomplete,
-                    chunks_total: 0,
-                    chunks_downloaded: 0,
-                },
-            );
-        }
-        drop(resources_write);
-
-        info!(target: "fud::init()", "Verifying resources...");
-        let resources = self.verify_resources(None).await?;
-
-        let self_node = self.dht().node().await;
-
-        if self_node.addresses.is_empty() {
-            return Ok(());
-        }
-
-        info!(target: "fud::init()", "Start seeding...");
-        let self_router_items: Vec<DhtRouterItem> = vec![self_node.into()];
-
-        for resource in resources {
-            self.add_to_router(
-                self.seeders_router.clone(),
-                &resource.hash,
-                self_router_items.clone(),
-            )
-            .await;
-        }
-
-        Ok(())
-    }
-
-    /// Get resource path from hash using the sled db
-    fn hash_to_path(&self, hash: &blake3::Hash) -> Result<Option<PathBuf>> {
-        if let Some(value) = self.path_tree.get(hash.as_bytes())? {
-            let path: PathBuf = expand_path(std::str::from_utf8(&value)?)?;
-            return Ok(Some(path));
-        }
-
-        Ok(None)
-    }
-
-    /// 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.
-    /// Takes an optional list of hashes.
-    /// If no hash is given (None), it verifies all resources.
-    /// Returns the list of verified and uncorrupted/complete seeding resources.
-    async fn verify_resources(&self, hashes: Option<Vec<blake3::Hash>>) -> Result<Vec<Resource>> {
-        let mut resources_write = self.resources.write().await;
-
-        let update_resource =
-            async |resource: &mut Resource,
-                   status: ResourceStatus,
-                   chunked_file: Option<&ChunkedFile>| {
-                resource.status = status;
-                resource.chunks_total = match chunked_file {
-                    Some(chunked_file) => chunked_file.len() as u64,
-                    None => 0,
-                };
-                resource.chunks_downloaded = match chunked_file {
-                    Some(chunked_file) => chunked_file.local_chunks() as u64,
-                    None => 0,
-                };
-
-                self.event_publisher
-                    .notify(FudEvent::ResourceUpdated(ResourceUpdated {
-                        hash: resource.hash,
-                        resource: resource.clone(),
-                    }))
-                    .await;
-            };
-
-        let mut seeding_resources: Vec<Resource> = vec![];
-        for (_, mut resource) in resources_write.iter_mut() {
-            if let Some(ref hashes_list) = hashes {
-                if !hashes_list.contains(&resource.hash) {
-                    continue;
-                }
-            }
-
-            match resource.status {
-                ResourceStatus::Seeding => {}
-                ResourceStatus::Incomplete => {}
-                _ => continue,
-            };
-
-            // Make sure the resource is not corrupted or incomplete
-            let resource_path = match self.hash_to_path(&resource.hash) {
-                Ok(Some(v)) => v,
-                Ok(None) | Err(_) => {
-                    update_resource(&mut resource, ResourceStatus::Incomplete, None).await;
-                    continue;
-                }
-            };
-            let chunked_file = match self.geode.get(&resource.hash, &resource_path).await {
-                Ok(v) => v,
-                Err(_) => {
-                    update_resource(&mut resource, ResourceStatus::Incomplete, None).await;
-                    continue;
-                }
-            };
-            if !chunked_file.is_complete() {
-                update_resource(&mut resource, ResourceStatus::Incomplete, Some(&chunked_file))
-                    .await;
-                continue;
-            }
-
-            update_resource(&mut resource, ResourceStatus::Seeding, Some(&chunked_file)).await;
-            seeding_resources.push(resource.clone());
-        }
-
-        Ok(seeding_resources)
-    }
-
-    /// Query `nodes` to find the seeders for `key`
-    async fn fetch_seeders(
-        &self,
-        nodes: &Vec<DhtNode>,
-        key: &blake3::Hash,
-    ) -> HashSet<DhtRouterItem> {
-        let mut seeders: HashSet<DhtRouterItem> = HashSet::new();
-
-        for node in nodes {
-            let channel = match self.get_channel(node).await {
-                Ok(channel) => channel,
-                Err(e) => {
-                    warn!(target: "fud::fetch_seeders()", "Could not get a channel for node {}: {}", hash_to_string(&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_seeders()", "Error subscribing to msg: {}", e);
-                    continue;
-                }
-            };
-
-            let send_res = channel.send(&FudFindSeedersRequest { key: *key }).await;
-            if let Err(e) = send_res {
-                warn!(target: "fud::fetch_seeders()", "Error while sending FudFindSeedersRequest: {}", e);
-                msg_subscriber.unsubscribe().await;
-                continue;
-            }
-
-            let reply = match msg_subscriber.receive_with_timeout(self.dht().settings.timeout).await
-            {
-                Ok(reply) => reply,
-                Err(e) => {
-                    warn!(target: "fud::fetch_seeders()", "Error waiting for reply: {}", e);
-                    msg_subscriber.unsubscribe().await;
-                    continue;
-                }
-            };
-
-            msg_subscriber.unsubscribe().await;
-
-            seeders.extend(reply.seeders.clone());
-        }
-
-        info!(target: "fud::fetch_seeders()", "Found {} seeders for {}", seeders.len(), hash_to_string(key));
-        seeders
-    }
-
-    /// Fetch chunks for a file from `seeders`
-    async fn fetch_chunks(
-        &self,
-        file_path: &PathBuf,
-        file_hash: &blake3::Hash,
-        chunk_hashes: &HashSet<blake3::Hash>,
-        seeders: &HashSet<DhtRouterItem>,
-    ) -> Result<()> {
-        let mut remaining_chunks = chunk_hashes.clone();
-        let mut shuffled_seeders = {
-            let mut vec: Vec<_> = seeders.iter().cloned().collect();
-            vec.shuffle(&mut OsRng);
-            vec
-        };
-
-        while let Some(seeder) = shuffled_seeders.pop() {
-            let channel = match self.get_channel(&seeder.node).await {
-                Ok(channel) => channel,
-                Err(e) => {
-                    warn!(target: "fud::fetch_chunks()", "Could not get a channel for node {}: {}", hash_to_string(&seeder.node.id), e);
-                    continue;
-                }
-            };
-            let mut chunks_to_query = remaining_chunks.clone();
-            info!("Requesting chunks from seeder {}", hash_to_string(&seeder.node.id));
-            loop {
-                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();
-
-                // Select a chunk to request
-                let mut chunk_hash: Option<blake3::Hash> = None;
-                if let Some(&random_chunk) = chunks_to_query.iter().choose(&mut OsRng) {
-                    chunk_hash = Some(random_chunk);
-                }
-
-                if chunk_hash.is_none() {
-                    // No more chunks to request from this seeder
-                    break; // Switch to another seeder
-                }
-                let chunk_hash = chunk_hash.unwrap();
-                chunks_to_query.remove(&chunk_hash);
-
-                let send_res =
-                    channel.send(&FudFindRequest { info: Some(*file_hash), key: chunk_hash }).await;
-                if let Err(e) = send_res {
-                    warn!(target: "fud::fetch_chunks()", "Error while sending FudFindRequest: {}", e);
-                    break; // Switch to another seeder
-                }
-
-                let chunk_recv =
-                    msg_subscriber_chunk.receive_with_timeout(self.chunk_timeout).fuse();
-                let notfound_recv =
-                    msg_subscriber_notfound.receive_with_timeout(self.chunk_timeout).fuse();
-
-                pin_mut!(chunk_recv, notfound_recv);
-
-                // Wait for a FudChunkReply or FudNotFound
-                select! {
-                    chunk_reply = chunk_recv => {
-                        if let Err(e) = chunk_reply {
-                            warn!(target: "fud::fetch_chunks()", "Error waiting for chunk reply: {}", e);
-                            break; // Switch to another seeder
-                        }
-                        let reply = chunk_reply.unwrap();
-
-                        match self.geode.write_chunk(file_hash, file_path, &reply.chunk).await {
-                            Ok(inserted_hash) => {
-                                if inserted_hash != chunk_hash {
-                                    warn!("Received chunk does not match requested chunk");
-                                    msg_subscriber_chunk.unsubscribe().await;
-                                    msg_subscriber_notfound.unsubscribe().await;
-                                    continue; // Skip to next chunk, will retry this chunk later
-                                }
-
-                                // Update resource `chunks_downloaded`
-                                let mut resources_write = self.resources.write().await;
-                                let resource = match resources_write.get_mut(file_hash) {
-                                    Some(resource) => {
-                                        resource.status = ResourceStatus::Downloading;
-                                        resource.chunks_downloaded += 1;
-                                        resource.clone()
-                                    }
-                                    None => return Ok(()) // Resource was removed, abort
-                                };
-                                drop(resources_write);
-
-                                info!(target: "fud::fetch_chunks()", "Received chunk {} from seeder {}", hash_to_string(&chunk_hash), hash_to_string(&seeder.node.id));
-                                self.event_publisher
-                                    .notify(FudEvent::ChunkDownloadCompleted(ChunkDownloadCompleted {
-                                        hash: *file_hash,
-                                        chunk_hash,
-                                        resource,
-                                    }))
-                                    .await;
-                                remaining_chunks.remove(&chunk_hash);
-                            }
-                            Err(e) => {
-                                error!("Failed inserting chunk {} to Geode: {}", hash_to_string(&chunk_hash), e);
-                            }
-                        };
-                    }
-                    notfound_reply = notfound_recv => {
-                        if let Err(e) = notfound_reply {
-                            warn!(target: "fud::fetch_chunks()", "Error waiting for NOTFOUND reply: {}", e);
-                            msg_subscriber_chunk.unsubscribe().await;
-                            msg_subscriber_notfound.unsubscribe().await;
-                            break; // Switch to another seeder
-                        }
-                        info!(target: "fud::fetch_chunks()", "Received NOTFOUND {} from seeder {}", hash_to_string(&chunk_hash), hash_to_string(&seeder.node.id));
-                        self.event_publisher
-                            .notify(FudEvent::ChunkNotFound(ChunkNotFound {
-                                hash: *file_hash,
-                                chunk_hash,
-                            }))
-                        .await;
-                    }
-                };
-
-                msg_subscriber_chunk.unsubscribe().await;
-                msg_subscriber_notfound.unsubscribe().await;
-            }
-
-            // Stop when there are no missing chunks
-            if remaining_chunks.is_empty() {
-                break;
-            }
-        }
-
-        Ok(())
-    }
-
-    /// Fetch a single file metadata from `nodes`.
-    /// If the file is smaller than a single chunk then the chunk is returned.
-    /// 1. Request seeders for the file from those nodes
-    /// 2. Request the file from the seeders
-    async fn fetch_file_metadata(
-        &self,
-        nodes: Vec<DhtNode>,
-        file_hash: blake3::Hash,
-    ) -> Option<FetchReply> {
-        let mut queried_seeders: HashSet<blake3::Hash> = HashSet::new();
-        let mut result: Option<FetchReply> = None;
-
-        for node in nodes {
-            // 1. Request list of seeders
-            let channel = match self.get_channel(&node).await {
-                Ok(channel) => channel,
-                Err(e) => {
-                    warn!(target: "fud::fetch_file_metadata()", "Could not get a channel for node {}: {}", hash_to_string(&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_file_metadata()", "Error subscribing to msg: {}", e);
-                    continue;
-                }
-            };
-
-            let send_res = channel.send(&FudFindSeedersRequest { key: file_hash }).await;
-            if let Err(e) = send_res {
-                warn!(target: "fud::fetch_file_metadata()", "Error while sending FudFindSeedersRequest: {}", e);
-                msg_subscriber.unsubscribe().await;
-                continue;
-            }
-
-            let reply = match msg_subscriber.receive_with_timeout(self.dht().settings.timeout).await
-            {
-                Ok(reply) => reply,
-                Err(e) => {
-                    warn!(target: "fud::fetch_file_metadata()", "Error waiting for reply: {}", e);
-                    msg_subscriber.unsubscribe().await;
-                    continue;
-                }
-            };
-
-            let mut seeders = reply.seeders.clone();
-            info!(target: "fud::fetch_file_metadata()", "Found {} seeders for {} (from {})", seeders.len(), hash_to_string(&file_hash), hash_to_string(&node.id));
-
-            msg_subscriber.unsubscribe().await;
-
-            // 2. 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 { info: None, key: file_hash }).await;
-                    if let Err(e) = send_res {
-                        warn!(target: "fud::fetch_file_metadata()", "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.chunk_timeout).fuse();
-                    let file_recv =
-                        msg_subscriber_file.receive_with_timeout(self.chunk_timeout).fuse();
-                    let notfound_recv =
-                        msg_subscriber_notfound.receive_with_timeout(self.chunk_timeout).fuse();
-
-                    pin_mut!(chunk_recv, file_recv, notfound_recv);
-
-                    let cleanup = async || {
-                        msg_subscriber_chunk.unsubscribe().await;
-                        msg_subscriber_file.unsubscribe().await;
-                        msg_subscriber_notfound.unsubscribe().await;
-                    };
-
-                    // Wait for a FudChunkReply, FudFileReply, or FudNotFound
-                    select! {
-                        // Received a chunk while requesting a file, this is allowed to
-                        // optimize fetching files smaller than a single chunk
-                        chunk_reply = chunk_recv => {
-                            cleanup().await;
-                            if let Err(e) = chunk_reply {
-                                warn!(target: "fud::fetch_file_metadata()", "Error waiting for chunk reply: {}", e);
-                                continue;
-                            }
-                            let reply = chunk_reply.unwrap();
-                            let chunk_hash = blake3::hash(&reply.chunk);
-                            // Check that this is the only chunk in the file
-                            if !self.geode.verify_file(&file_hash, &[chunk_hash]) {
-                                warn!(target: "fud::fetch_file_metadata()", "Received a chunk while fetching a file, the chunk did not match the file hash");
-                                continue;
-                            }
-                            info!(target: "fud::fetch_file_metadata()", "Received chunk {} (for file {}) from seeder {}", hash_to_string(&chunk_hash), hash_to_string(&file_hash), hash_to_string(&seeder.node.id));
-                            result = Some(FetchReply::Chunk((*reply).clone()));
-                            break;
-                        }
-                        file_reply = file_recv => {
-                            cleanup().await;
-                            if let Err(e) = file_reply {
-                                warn!(target: "fud::fetch_file_metadata()", "Error waiting for file reply: {}", e);
-                                continue;
-                            }
-                            let reply = file_reply.unwrap();
-                            if !self.geode.verify_file(&file_hash, &reply.chunk_hashes) {
-                                warn!(target: "fud::fetch_file_metadata()", "Received invalid file metadata");
-                                continue;
-                            }
-                            info!(target: "fud::fetch_file_metadata()", "Received file {} from seeder {}", hash_to_string(&file_hash), hash_to_string(&seeder.node.id));
-                            result = Some(FetchReply::File((*reply).clone()));
-                            break;
-                        }
-                        notfound_reply = notfound_recv => {
-                            cleanup().await;
-                            if let Err(e) = notfound_reply {
-                                warn!(target: "fud::fetch_file_metadata()", "Error waiting for NOTFOUND reply: {}", e);
-                                continue;
-                            }
-                            info!(target: "fud::fetch_file_metadata()", "Received NOTFOUND {} from seeder {}", hash_to_string(&file_hash), hash_to_string(&seeder.node.id));
-                        }
-                    };
-                }
-            }
-
-            if result.is_some() {
-                break;
-            }
-        }
-
-        result
-    }
-}
-
-// TODO: This is not Sybil-resistant
-fn generate_node_id() -> Result<blake3::Hash> {
-    let mut rng = OsRng;
-    let mut random_data = [0u8; 32];
-    rng.fill_bytes(&mut random_data);
-    let node_id = blake3::Hash::from_bytes(random_data);
-    Ok(node_id)
-}
-
 async_daemonize!(realmain);
 async fn realmain(args: Args, ex: Arc<Executor<'static>>) -> Result<()> {
     // The working directory for this daemon and geode.
@@ -756,16 +101,10 @@ async fn realmain(args: Args, ex: Arc<Executor<'static>>) -> Result<()> {
         None => basedir.join("downloads"),
     };
 
-    // Hashmap used for routing
-    let seeders_router = Arc::new(RwLock::new(HashMap::new()));
-
     // Sled database init
     info!("Instantiating database");
     let sled_db = sled::open(basedir.join("db"))?;
 
-    info!("Instantiating Geode instance");
-    let geode = Geode::new(&basedir /*, sled_db, "geode"*/).await?;
-
     info!("Instantiating P2P network");
     let net_settings: NetSettings = args.net.into();
     let p2p = P2p::new(net_settings.clone(), ex.clone()).await?;
@@ -794,71 +133,36 @@ async fn realmain(args: Args, ex: Arc<Executor<'static>>) -> Result<()> {
         ex.clone(),
     );
 
-    // Get or generate the node id
-    let node_id: Result<blake3::Hash> = {
-        let mut node_id_path: PathBuf = basedir.clone();
-        node_id_path.push(NODE_ID_PATH);
-        match File::open(node_id_path.clone()).await {
-            Ok(mut file) => {
-                let mut buffer = Vec::new();
-                file.read_to_end(&mut buffer).await?;
-                let mut out_buf = [0u8; 32];
-                bs58::decode(buffer).onto(&mut out_buf)?;
-                let node_id = blake3::Hash::from_bytes(out_buf);
-                Ok(node_id)
-            }
-            Err(e) if e.kind() == ErrorKind::NotFound => {
-                let node_id = generate_node_id()?;
-                let mut file =
-                    OpenOptions::new().write(true).create(true).open(node_id_path).await?;
-                file.write_all(&bs58::encode(node_id.as_bytes()).into_vec()).await?;
-                file.flush().await?;
-                Ok(node_id)
-            }
-            Err(e) => Err(e.into()),
-        }
-    };
-
-    let node_id_ = node_id?;
+    let mut node_id_path = basedir.to_path_buf();
+    node_id_path.push(NODE_ID_PATH);
+    let node_id = get_node_id(&node_id_path).await?;
 
-    info!(target: "fud", "Your node ID: {}", hash_to_string(&node_id_));
+    info!(target: "fud", "Your node ID: {}", hash_to_string(&node_id));
 
     // Daemon instantiation
-    let event_sub = JsonSubscriber::new("event");
-    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 event_pub = Publisher::new();
     let dht_settings: DhtSettings = args.dht.into();
-    let dht: Arc<Dht> = Arc::new(Dht::new(&node_id_, &dht_settings, p2p.clone(), ex.clone()).await);
-    let fud = Arc::new(Fud {
-        seeders_router,
-        p2p: p2p.clone(),
-        geode,
-        downloads_path,
-        chunk_timeout: args.chunk_timeout,
-        dht: dht.clone(),
-        path_tree: sled_db.open_tree("path")?,
-        resources: Arc::new(RwLock::new(HashMap::new())),
-        get_tx,
-        get_rx,
-        file_fetch_tx,
-        file_fetch_rx,
-        file_fetch_end_tx,
-        file_fetch_end_rx,
-        rpc_connections: Mutex::new(HashSet::new()),
-        dnet_sub,
-        event_sub: event_sub.clone(),
-        event_publisher: Publisher::new(),
-    });
-    fud.init().await?;
+    let dht: Arc<Dht> = Arc::new(Dht::new(&node_id, &dht_settings, p2p.clone(), ex.clone()).await);
+    let fud: Arc<Fud> = Arc::new(
+        Fud::new(
+            p2p.clone(),
+            basedir,
+            downloads_path,
+            args.chunk_timeout,
+            dht.clone(),
+            sled_db.open_tree("path")?,
+            event_pub.clone(),
+        )
+        .await?,
+    );
 
     info!(target: "fud", "Starting download subs task");
+    let event_sub = JsonSubscriber::new("event");
     let event_sub_ = event_sub.clone();
-    let fud_ = fud.clone();
     let event_task = StoppableTask::new();
     event_task.clone().start(
         async move {
-            let event_sub = fud_.event_publisher.clone().subscribe().await;
+            let event_sub = event_pub.clone().subscribe().await;
             loop {
                 let event = event_sub.receive().await;
                 debug!(target: "fud", "Got event: {:?}", event);
@@ -878,7 +182,7 @@ async fn realmain(args: Args, ex: Arc<Executor<'static>>) -> Result<()> {
     info!(target: "fud", "Starting fetch file task");
     let file_task = StoppableTask::new();
     file_task.clone().start(
-        tasks::fetch_file_task(fud.clone()),
+        fetch_file_task(fud.clone()),
         |res| async {
             match res {
                 Ok(()) | Err(Error::DetachedTaskStopped) => { /* Do nothing */ }
@@ -892,7 +196,7 @@ async fn realmain(args: Args, ex: Arc<Executor<'static>>) -> Result<()> {
     info!(target: "fud", "Starting get task");
     let get_task_ = StoppableTask::new();
     get_task_.clone().start(
-        tasks::get_task(fud.clone()),
+        get_task(fud.clone()),
         |res| async {
             match res {
                 Ok(()) | Err(Error::DetachedTaskStopped) => { /* Do nothing */ }
@@ -905,13 +209,14 @@ async fn realmain(args: Args, ex: Arc<Executor<'static>>) -> Result<()> {
 
     let rpc_settings: RpcSettings = args.rpc.into();
     info!(target: "fud", "Starting JSON-RPC server on {}", rpc_settings.listen);
+    let rpc_interface = Arc::new(JsonRpcInterface::new(fud.clone(), dnet_sub, event_sub));
     let rpc_task = StoppableTask::new();
-    let fud_ = fud.clone();
+    let rpc_interface_ = rpc_interface.clone();
     rpc_task.clone().start(
-        listen_and_serve(rpc_settings, fud.clone(), None, ex.clone()),
+        listen_and_serve(rpc_settings, rpc_interface, None, ex.clone()),
         |res| async move {
             match res {
-                Ok(()) | Err(Error::RpcServerStopped) => fud_.stop_connections().await,
+                Ok(()) | Err(Error::RpcServerStopped) => rpc_interface_.stop_connections().await,
                 Err(e) => error!(target: "fud", "Failed starting sync JSON-RPC server: {}", e),
             }
         },
@@ -967,7 +272,7 @@ async fn realmain(args: Args, ex: Arc<Executor<'static>>) -> Result<()> {
     let announce_task = StoppableTask::new();
     let fud_ = fud.clone();
     announce_task.clone().start(
-        async move { tasks::announce_seed_task(fud_.clone()).await },
+        async move { announce_seed_task(fud_.clone()).await },
         |res| async {
             match res {
                 Ok(()) | Err(Error::DetachedTaskStopped) => { /* Do nothing */ }

bin/fud/fud/src/rpc.rs

@@ -17,28 +17,30 @@
  */
 
 use async_trait::async_trait;
-use log::{error, info};
+use log::error;
 use smol::{
     fs::{self, File},
-    lock::MutexGuard,
+    lock::{Mutex, MutexGuard},
 };
 use std::{
     collections::{HashMap, HashSet},
     path::PathBuf,
+    sync::Arc,
 };
 use tinyjson::JsonValue;
 
 use darkfi::{
     dht::DhtHandler,
     geode::hash_to_string,
+    net::P2pPtr,
     rpc::{
-        jsonrpc::{ErrorCode, JsonError, JsonRequest, JsonResponse, JsonResult},
+        jsonrpc::{ErrorCode, JsonError, JsonRequest, JsonResponse, JsonResult, JsonSubscriber},
         p2p_method::HandlerP2p,
         server::RequestHandler,
     },
     system::StoppableTaskPtr,
     util::path::expand_path,
-    Error, Result,
+    Result,
 };
 
 use crate::{
@@ -48,8 +50,15 @@ use crate::{
     Fud,
 };
 
+pub struct JsonRpcInterface {
+    fud: Arc<Fud>,
+    rpc_connections: Mutex<HashSet<StoppableTaskPtr>>,
+    dnet_sub: JsonSubscriber,
+    event_sub: JsonSubscriber,
+}
+
 #[async_trait]
-impl RequestHandler<()> for Fud {
+impl RequestHandler<()> for JsonRpcInterface {
     async fn handle_request(&self, req: JsonRequest) -> JsonResult {
         return match req.method.as_str() {
             "ping" => self.pong(req.id, req.params).await,
@@ -75,8 +84,18 @@ impl RequestHandler<()> for Fud {
     }
 }
 
+impl HandlerP2p for JsonRpcInterface {
+    fn p2p(&self) -> P2pPtr {
+        self.fud.p2p.clone()
+    }
+}
+
 /// Fud RPC methods
-impl Fud {
+impl JsonRpcInterface {
+    pub fn new(fud: Arc<Fud>, dnet_sub: JsonSubscriber, event_sub: JsonSubscriber) -> Self {
+        Self { fud, rpc_connections: Mutex::new(HashSet::new()), dnet_sub, event_sub }
+    }
+
     // 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.
@@ -84,7 +103,7 @@ impl Fud {
     // --> {"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 self_node = self.dht.node().await;
+        let self_node = self.fud.dht.node().await;
 
         if self_node.addresses.is_empty() {
             error!(target: "fud::put()", "Cannot put file, you don't have any external address");
@@ -117,7 +136,7 @@ impl Fud {
             }
         };
 
-        let (file_hash, chunk_hashes) = match self.geode.insert(fd).await {
+        let (file_hash, chunk_hashes) = match self.fud.geode.insert(fd).await {
             Ok(v) => v,
             Err(e) => {
                 let error_str = format!("Failed inserting file {:?} to geode: {}", path, e);
@@ -128,6 +147,7 @@ impl Fud {
 
         // Add path to the sled db
         if let Err(e) = self
+            .fud
             .path_tree
             .insert(file_hash.as_bytes(), path.to_string_lossy().to_string().as_bytes())
         {
@@ -136,7 +156,7 @@ impl Fud {
         }
 
         // Add resource
-        let mut resources_write = self.resources.write().await;
+        let mut resources_write = self.fud.resources.write().await;
         resources_write.insert(
             file_hash,
             Resource {
@@ -151,7 +171,7 @@ impl Fud {
 
         // Announce file
         let fud_announce = FudAnnounce { key: file_hash, seeders: vec![self_node.into()] };
-        let _ = self.announce(&file_hash, &fud_announce, self.seeders_router.clone()).await;
+        let _ = self.fud.announce(&file_hash, &fud_announce, self.fud.seeders_router.clone()).await;
 
         JsonResponse::new(JsonValue::String(hash_to_string(&file_hash)), id).into()
     }
@@ -186,13 +206,13 @@ impl Fud {
 
         let file_path = match params[1].get::<String>() {
             Some(path) => match path.is_empty() {
-                true => self.downloads_path.join(&file_hash_str).join(&file_hash_str),
+                true => self.fud.downloads_path.join(&file_hash_str).join(&file_hash_str),
                 false => match PathBuf::from(path).is_absolute() {
                     true => PathBuf::from(path),
-                    false => self.downloads_path.join(&file_hash_str).join(path),
+                    false => self.fud.downloads_path.join(&file_hash_str).join(path),
                 },
             },
-            None => self.downloads_path.join(&file_hash_str).join(&file_hash_str),
+            None => self.fud.downloads_path.join(&file_hash_str).join(&file_hash_str),
         };
 
         // Get the parent directory of the file
@@ -201,7 +221,7 @@ impl Fud {
             let _ = fs::create_dir_all(parent).await;
         }
 
-        let _ = self.get_tx.send((id, file_hash, file_path.clone(), Ok(()))).await;
+        let _ = self.fud.get_tx.send((id, file_hash, file_path.clone(), Ok(()))).await;
 
         JsonResponse::new(JsonValue::String(file_path.to_string_lossy().to_string()), id).into()
     }
@@ -231,9 +251,9 @@ impl Fud {
         let switch = params[0].get::<bool>().unwrap();
 
         if *switch {
-            self.p2p.dnet_enable();
+            self.fud.p2p.dnet_enable();
         } else {
-            self.p2p.dnet_disable();
+            self.fud.p2p.dnet_disable();
         }
 
         JsonResponse::new(JsonValue::Boolean(true), id).into()
@@ -266,7 +286,7 @@ impl Fud {
             return JsonError::new(ErrorCode::InvalidParams, None, id).into()
         }
 
-        let resources_read = self.resources.read().await;
+        let resources_read = self.fud.resources.read().await;
         let mut resources: Vec<JsonValue> = vec![];
         for (_, resource) in resources_read.iter() {
             resources.push(resource.clone().into());
@@ -286,7 +306,7 @@ impl Fud {
             return JsonError::new(ErrorCode::InvalidParams, None, id).into()
         }
         let mut buckets = vec![];
-        for bucket in self.dht.buckets.read().await.iter() {
+        for bucket in self.fud.dht.buckets.read().await.iter() {
             let mut nodes = vec![];
             for node in bucket.nodes.clone() {
                 let mut addresses = vec![];
@@ -315,7 +335,7 @@ impl Fud {
             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() {
+        for (hash, items) in self.fud.seeders_router.read().await.iter() {
             let mut node_ids = vec![];
             for item in items {
                 node_ids.push(JsonValue::String(hash_to_string(&item.node.id)));
@@ -345,11 +365,12 @@ impl Fud {
         }
 
         let hash = blake3::Hash::from_bytes(hash_buf);
-        let mut resources_write = self.resources.write().await;
+        let mut resources_write = self.fud.resources.write().await;
         resources_write.remove(&hash);
         drop(resources_write);
 
-        self.event_publisher
+        self.fud
+            .event_publisher
             .notify(FudEvent::ResourceRemoved(event::ResourceRemoved { hash }))
             .await;
 
@@ -387,7 +408,7 @@ impl Fud {
             Some(hashes.unwrap())
         };
 
-        if let Err(e) = self.verify_resources(hashes).await {
+        if let Err(e) = self.fud.verify_resources(hashes).await {
             error!(target: "fud::verify()", "Could not verify resources: {}", e);
             return JsonError::new(ErrorCode::InternalError, None, id).into();
         }
@@ -395,224 +416,3 @@ impl Fud {
         JsonResponse::new(JsonValue::Array(vec![]), id).into()
     }
 }
-
-impl Fud {
-    /// Handle `get` RPC request
-    pub async fn handle_get(&self, file_hash: &blake3::Hash, file_path: &PathBuf) -> Result<()> {
-        let self_node = self.dht().node().await;
-        let mut closest_nodes = vec![];
-
-        // Add path to the sled db
-        self.path_tree
-            .insert(file_hash.as_bytes(), file_path.to_string_lossy().to_string().as_bytes())?;
-
-        // Add resource to `self.resources`
-        let resource = Resource {
-            hash: *file_hash,
-            path: file_path.clone(),
-            status: ResourceStatus::Discovering,
-            chunks_total: 0,
-            chunks_downloaded: 0,
-        };
-        let mut resources_write = self.resources.write().await;
-        resources_write.insert(*file_hash, resource.clone());
-        drop(resources_write);
-
-        // Send a DownloadStarted event
-        self.event_publisher
-            .notify(FudEvent::DownloadStarted(event::DownloadStarted {
-                hash: *file_hash,
-                resource,
-            }))
-            .await;
-
-        // Try to get the chunked file from geode
-        let chunked_file = match self.geode.get(file_hash, file_path).await {
-            // We already know the list of chunk hashes for this file
-            Ok(v) => v,
-            // The metadata in geode is invalid or corrupted
-            Err(Error::GeodeNeedsGc) => todo!(),
-            // If we could not find the file in geode, get the file metadata from the network
-            Err(Error::GeodeFileNotFound) => {
-                // Find nodes close to the file hash
-                info!(target: "self::get()", "Requested file {} not found in Geode, triggering fetch", hash_to_string(file_hash));
-                closest_nodes = self.lookup_nodes(file_hash).await.unwrap_or_default();
-
-                // Fetch file metadata (list of chunk hashes)
-                self.file_fetch_tx
-                    .send((closest_nodes.clone(), *file_hash, file_path.clone(), 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 {
-                    // The file metadata was found and inserted into geode
-                    Ok(()) => self.geode.get(&i_file_hash, file_path).await.unwrap(),
-                    // We could not find the file metadata
-                    Err(Error::GeodeFileRouteNotFound) => {
-                        // Set resource status to `Incomplete` and send FudEvent::FileNotFound
-                        let mut resources_write = self.resources.write().await;
-                        if let Some(resource) = resources_write.get_mut(file_hash) {
-                            resource.status = ResourceStatus::Incomplete;
-
-                            self.event_publisher
-                                .notify(FudEvent::FileNotFound(event::FileNotFound {
-                                    hash: *file_hash,
-                                    resource: resource.clone(),
-                                }))
-                                .await;
-                        }
-                        drop(resources_write);
-                        return Err(Error::GeodeFileRouteNotFound);
-                    }
-
-                    Err(e) => {
-                        error!(target: "fud::handle_get()", "{}", e);
-                        return Err(e);
-                    }
-                }
-            }
-
-            Err(e) => {
-                error!(target: "fud::handle_get()", "{}", e);
-                return Err(e);
-            }
-        };
-
-        // Set resource status to `Downloading`
-        let mut resources_write = self.resources.write().await;
-        let resource = match resources_write.get_mut(file_hash) {
-            Some(resource) => {
-                resource.status = ResourceStatus::Downloading;
-                resource.chunks_downloaded = chunked_file.local_chunks() as u64;
-                resource.chunks_total = chunked_file.len() as u64;
-                resource.clone()
-            }
-            None => return Ok(()), // Resource was removed, abort
-        };
-        drop(resources_write);
-
-        // Send a FileDownloadCompleted event
-        self.event_publisher
-            .notify(FudEvent::FileDownloadCompleted(event::FileDownloadCompleted {
-                hash: *file_hash,
-                resource: resource.clone(),
-            }))
-            .await;
-
-        // If the file is already complete, we don't need to download any chunk
-        if chunked_file.is_complete() {
-            // Announce the file
-            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;
-
-            // Set resource status to `Seeding`
-            let mut resources_write = self.resources.write().await;
-            let resource = match resources_write.get_mut(file_hash) {
-                Some(resource) => {
-                    resource.status = ResourceStatus::Seeding;
-                    resource.chunks_downloaded = chunked_file.len() as u64;
-                    resource.clone()
-                }
-                None => return Ok(()), // Resource was removed, abort
-            };
-            drop(resources_write);
-
-            // Send a DownloadCompleted event
-            self.event_publisher
-                .notify(FudEvent::DownloadCompleted(event::DownloadCompleted {
-                    hash: *file_hash,
-                    resource,
-                }))
-                .await;
-
-            return Ok(());
-        }
-
-        // Find nodes close to the file hash if we didn't previously fetched them
-        if closest_nodes.is_empty() {
-            closest_nodes = self.lookup_nodes(file_hash).await.unwrap_or_default();
-        }
-
-        // Find seeders and remove ourselves from the result
-        let seeders = self
-            .fetch_seeders(&closest_nodes, file_hash)
-            .await
-            .iter()
-            .filter(|seeder| seeder.node.id != self_node.id)
-            .cloned()
-            .collect();
-
-        // List missing chunks
-        let mut missing_chunks = HashSet::new();
-        for (chunk, path) in chunked_file.iter() {
-            if path.is_none() {
-                missing_chunks.insert(*chunk);
-            }
-        }
-
-        // Fetch missing chunks from seeders
-        self.fetch_chunks(file_path, file_hash, &missing_chunks, &seeders).await?;
-
-        // Get chunked file from geode
-        let chunked_file = match self.geode.get(file_hash, file_path).await {
-            Ok(v) => v,
-            Err(e) => {
-                error!(target: "fud::handle_get()", "{}", e);
-                return Err(e);
-            }
-        };
-
-        // We fetched all chunks, but the file is not complete
-        // (some chunks were missing from all seeders)
-        if !chunked_file.is_complete() {
-            // Set resource status to `Incomplete`
-            let mut resources_write = self.resources.write().await;
-            let resource = match resources_write.get_mut(file_hash) {
-                Some(resource) => {
-                    resource.status = ResourceStatus::Incomplete;
-                    resource.clone()
-                }
-                None => return Ok(()), // Resource was removed, abort
-            };
-            drop(resources_write);
-
-            // Send a MissingChunks event
-            self.event_publisher
-                .notify(FudEvent::MissingChunks(event::MissingChunks {
-                    hash: *file_hash,
-                    resource,
-                }))
-                .await;
-            return Ok(());
-        }
-
-        // Announce the file
-        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;
-
-        // Set resource status to `Seeding`
-        let mut resources_write = self.resources.write().await;
-        let resource = match resources_write.get_mut(file_hash) {
-            Some(resource) => {
-                resource.status = ResourceStatus::Seeding;
-                resource.chunks_downloaded = chunked_file.len() as u64;
-                resource.clone()
-            }
-            None => return Ok(()), // Resource was removed, abort
-        };
-        drop(resources_write);
-
-        // Send a DownloadCompleted event
-        self.event_publisher
-            .notify(FudEvent::DownloadCompleted(event::DownloadCompleted {
-                hash: *file_hash,
-                resource,
-            }))
-            .await;
-
-        Ok(())
-    }
-}

bin/fud/fud/src/tasks.rs

@@ -26,20 +26,20 @@ use crate::{
     Fud,
 };
 
+pub enum FetchReply {
+    File(FudFileReply),
+    Chunk(FudChunkReply),
+}
+
 /// Triggered when calling the `get` RPC method
 pub async fn get_task(fud: Arc<Fud>) -> Result<()> {
     loop {
         let (_, file_hash, file_path, _) = fud.get_rx.recv().await.unwrap();
 
-        let _ = fud.handle_get(&file_hash, &file_path).await;
+        let _ = fud.get(&file_hash, &file_path).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.
@@ -93,7 +93,7 @@ pub async fn fetch_file_task(fud: Arc<Fud>) -> Result<()> {
     }
 }
 
-/// Background task that announces our files and chunks once every hour.
+/// Background task that announces our files once every hour.
 /// Also removes seeders that did not announce for too long.
 pub async fn announce_seed_task(fud: Arc<Fud>) -> Result<()> {
     let interval = 3600; // TODO: Make a setting

src/dht/handler.rs

@@ -47,7 +47,7 @@ pub trait DhtHandler {
     /// Send FIND NODES request to a peer to get nodes close to `key`
     async fn fetch_nodes(&self, node: &DhtNode, key: &blake3::Hash) -> Result<Vec<DhtNode>>;
 
-    /// Announce message `m` for a key, and add ourselves to router
+    /// Announce message for a key, and add ourselves to router
     async fn announce<M: Message>(
         &self,
         key: &blake3::Hash,
@@ -85,27 +85,30 @@ pub trait DhtHandler {
             let channel = res.unwrap();
             let channel_cache_lock = self.dht().channel_cache.clone();
             let mut channel_cache = channel_cache_lock.write().await;
-            if !channel.is_stopped() && !channel_cache.values().any(|&v| v == channel.info.id) {
-                // Skip this channel is it's a seed or refine session.
-                if channel.session_type_id() & (SESSION_SEED | SESSION_REFINE) != 0 {
-                    continue;
-                }
 
-                let node = self.ping(channel.clone()).await;
+            // Skip this channel if it's stopped or not new.
+            if channel.is_stopped() || channel_cache.values().any(|&v| v == 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 {
+                continue;
+            }
 
-                if let Ok(n) = node {
-                    channel_cache.insert(n.id, channel.info.id);
-                    drop(channel_cache);
+            let node = self.ping(channel.clone()).await;
 
-                    let node_cache_lock = self.dht().node_cache.clone();
-                    let mut node_cache = node_cache_lock.write().await;
-                    node_cache.insert(channel.info.id, n.clone());
-                    drop(node_cache);
+            if let Ok(n) = node {
+                channel_cache.insert(n.id, channel.info.id);
+                drop(channel_cache);
 
-                    if !n.addresses.is_empty() {
-                        self.add_node(n.clone()).await;
-                        let _ = self.on_new_node(&n.clone()).await;
-                    }
+                let node_cache_lock = self.dht().node_cache.clone();
+                let mut node_cache = node_cache_lock.write().await;
+                node_cache.insert(channel.info.id, n.clone());
+                drop(node_cache);
+
+                if !n.addresses.is_empty() {
+                    self.add_node(n.clone()).await;
+                    let _ = self.on_new_node(&n.clone()).await;
                 }
             }
         }

src/geode/mod.rs

@@ -239,7 +239,7 @@ impl Geode {
         let mut chunk_hashes = vec![];
 
         loop {
-            let mut buf = [0u8; MAX_CHUNK_SIZE];
+            let mut buf = vec![0u8; MAX_CHUNK_SIZE];
             let bytes_read = read_until_filled(&mut stream, &mut buf).await?;
             if bytes_read == 0 {
                 break
@@ -306,16 +306,32 @@ impl Geode {
         info!(target: "geode::write_chunk()", "[Geode] Writing single chunk");
 
         let mut cursor = Cursor::new(&stream);
-        let mut chunk = [0u8; MAX_CHUNK_SIZE];
+        let mut chunk = vec![0u8; MAX_CHUNK_SIZE];
 
         let bytes_read = read_until_filled(&mut cursor, &mut chunk).await?;
         let chunk_slice = &chunk[..bytes_read];
         let chunk_hash = blake3::hash(chunk_slice);
 
-        let chunked_file = self.get(file_hash, file_path).await?;
+        let file_hash_str = hash_to_string(file_hash);
+        let mut file_metadata_path = self.files_path.clone();
+        file_metadata_path.push(file_hash_str);
+
+        // Try to read the file metadata. If it's corrupt, return an error signalling
+        // that garbage collection needs to run.
+        let chunk_hashes = match Self::read_metadata(&file_metadata_path).await {
+            Ok(v) => v,
+            Err(e) => {
+                return match e {
+                    // If the file is not found, return according error.
+                    Error::Io(std::io::ErrorKind::NotFound) => Err(Error::GeodeFileNotFound),
+                    // Anything else should tell the client to do garbage collection
+                    _ => Err(Error::GeodeNeedsGc),
+                }
+            }
+        };
 
         // Get the chunk index in the file from the chunk hash
-        let chunk_index = match chunked_file.iter().position(|c| c.0 == chunk_hash) {
+        let chunk_index = match chunk_hashes.iter().position(|h| *h == chunk_hash) {
             Some(index) => index,
             None => {
                 return Err(Error::GeodeNeedsGc);
@@ -447,7 +463,7 @@ impl Geode {
         chunk_index: &usize,
     ) -> Result<Vec<u8>> {
         let position = (*chunk_index as u64) * (MAX_CHUNK_SIZE as u64);
-        let mut buf = [0u8; MAX_CHUNK_SIZE];
+        let mut buf = vec![0u8; MAX_CHUNK_SIZE];
         stream.seek(SeekFrom::Start(position)).await?;
         let bytes_read = read_until_filled(stream, &mut buf).await?;
         Ok(buf[..bytes_read].to_vec())