fud, fu: `subscribe` rpc method, add FudEvent::DownloadStarted and FudEvent::DownloadError, `get` improvements

bin/fud/fu/src/main.rs

@@ -96,13 +96,7 @@ impl Fu {
         let rpc_client_ = self.rpc_client.clone();
         subscriber_task.clone().start(
             async move {
-                let req = JsonRequest::new(
-                    "get",
-                    JsonValue::Array(vec![
-                        JsonValue::String(file_hash_),
-                        JsonValue::String(file_name.unwrap_or_default()),
-                    ]),
-                );
+                let req = JsonRequest::new("subscribe", JsonValue::Array(vec![]));
                 rpc_client_.subscribe(req, publisher).await
             },
             move |res| async move {
@@ -137,17 +131,27 @@ impl Fu {
             stdout().flush().unwrap();
         };
 
+        let req = JsonRequest::new(
+            "get",
+            JsonValue::Array(vec![
+                JsonValue::String(file_hash_.clone()),
+                JsonValue::String(file_name.unwrap_or_default()),
+            ]),
+        );
+        let _ = self.rpc_client.request(req).await;
+
         loop {
             match subscription.receive().await {
                 JsonResult::Notification(n) => {
                     let params = n.params.get::<HashMap<String, JsonValue>>().unwrap();
+                    let info =
+                        params.get("info").unwrap().get::<HashMap<String, JsonValue>>().unwrap();
+                    let hash = info.get("file_hash").unwrap().get::<String>().unwrap();
+                    if *hash != file_hash_ {
+                        continue;
+                    }
                     match params.get("event").unwrap().get::<String>().unwrap().as_str() {
                         "file_download_completed" => {
-                            let info = params
-                                .get("info")
-                                .unwrap()
-                                .get::<HashMap<String, JsonValue>>()
-                                .unwrap();
                             chunks_total =
                                 *info.get("chunk_count").unwrap().get::<f64>().unwrap() as usize;
                             print_progress_bar(chunks_downloaded, chunks_total);
@@ -180,6 +184,18 @@ impl Fu {
                             println!();
                             return Err(Error::Custom("Missing chunks".to_string()));
                         }
+                        "download_error" => {
+                            // An error that caused the download to be unsuccessful
+                            let info = params
+                                .get("info")
+                                .unwrap()
+                                .get::<HashMap<String, JsonValue>>()
+                                .unwrap();
+                            println!();
+                            return Err(Error::Custom(
+                                info.get("error").unwrap().get::<String>().unwrap().to_string(),
+                            ));
+                        }
                         _ => {}
                     }
                 }

bin/fud/fud/src/main.rs

@@ -95,6 +95,10 @@ struct Args {
     /// Base directory for filesystem storage
     base_dir: String,
 
+    #[structopt(short, long)]
+    /// Default path to store downloaded files (defaults to <base_dir>/downloads)
+    downloads_path: Option<String>,
+
     #[structopt(flatten)]
     /// Network settings
     net: SettingsOpt,
@@ -114,11 +118,14 @@ pub struct Fud {
     /// The Geode instance
     geode: Geode,
 
+    /// Default download directory
+    downloads_path: PathBuf,
+
     /// The DHT instance
     dht: Arc<Dht>,
 
-    get_tx: channel::Sender<(u16, blake3::Hash, Option<String>, Result<()>)>,
-    get_rx: channel::Receiver<(u16, blake3::Hash, Option<String>, Result<()>)>,
+    get_tx: channel::Sender<(u16, blake3::Hash, PathBuf, Result<()>)>,
+    get_rx: channel::Receiver<(u16, blake3::Hash, PathBuf, Result<()>)>,
     file_fetch_tx: channel::Sender<(blake3::Hash, Result<()>)>,
     file_fetch_rx: channel::Receiver<(blake3::Hash, Result<()>)>,
     file_fetch_end_tx: channel::Sender<(blake3::Hash, Result<()>)>,
@@ -233,8 +240,8 @@ impl Fud {
     }
 
     /// Query nodes close to `key` to find the seeders
-    async fn fetch_seeders(&self, key: blake3::Hash) -> HashSet<DhtRouterItem> {
-        let closest_nodes = self.lookup_nodes(&key).await; // Find the `k` closest nodes
+    async fn fetch_seeders(&self, key: &blake3::Hash) -> HashSet<DhtRouterItem> {
+        let closest_nodes = self.lookup_nodes(key).await; // Find the `k` closest nodes
         if closest_nodes.is_err() {
             return HashSet::new();
         }
@@ -260,7 +267,7 @@ impl Fud {
                 }
             };
 
-            let send_res = channel.send(&FudFindSeedersRequest { key }).await;
+            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;
@@ -281,7 +288,7 @@ impl Fud {
             seeders.extend(reply.seeders.clone());
         }
 
-        info!(target: "fud::fetch_seeders()", "Found {} seeders for {}", seeders.len(), hash_to_string(&key));
+        info!(target: "fud::fetch_seeders()", "Found {} seeders for {}", seeders.len(), hash_to_string(key));
         seeders
     }
 
@@ -573,6 +580,12 @@ async fn realmain(args: Args, ex: Arc<Executor<'static>>) -> Result<()> {
     // The working directory for this daemon and geode.
     let basedir = expand_path(&args.base_dir)?;
 
+    // The directory to store the downloaded files
+    let downloads_path = match args.downloads_path {
+        Some(downloads_path) => expand_path(&downloads_path)?,
+        None => basedir.join("downloads"),
+    };
+
     // Hashmap used for routing
     let seeders_router = Arc::new(RwLock::new(HashMap::new()));
 
@@ -653,6 +666,7 @@ async fn realmain(args: Args, ex: Arc<Executor<'static>>) -> Result<()> {
         seeders_router,
         p2p: p2p.clone(),
         geode,
+        downloads_path,
         dht: dht.clone(),
         get_tx,
         get_rx,

bin/fud/fud/src/rpc.rs

@@ -36,7 +36,10 @@ use darkfi::{
     Error,
 };
 use log::{error, info};
-use smol::{fs::File, lock::MutexGuard};
+use smol::{
+    fs::{self, File},
+    lock::MutexGuard,
+};
 use tinyjson::JsonValue;
 
 #[async_trait]
@@ -47,6 +50,7 @@ impl RequestHandler<()> for Fud {
 
             "put" => self.put(req.id, req.params).await,
             "get" => self.get(req.id, req.params).await,
+            "subscribe" => self.subscribe(req.id, req.params).await,
 
             "dnet.switch" => self.dnet_switch(req.id, req.params).await,
             "dnet.subscribe_events" => self.dnet_subscribe_events(req.id, req.params).await,
@@ -73,7 +77,12 @@ impl Fud {
     async fn put(&self, id: u16, params: JsonValue) -> JsonResult {
         if self.dht().node.addresses.is_empty() {
             error!(target: "fud::put()", "Cannot put file, you don't have any external address");
-            return JsonError::new(ErrorCode::InternalError, None, id).into()
+            return JsonError::new(
+                ErrorCode::InternalError,
+                Some("You don't have any external address".to_string()),
+                id,
+            )
+            .into()
         }
 
         let params = params.get::<Vec<JsonValue>>().unwrap();
@@ -100,8 +109,9 @@ impl Fud {
         let (file_hash, _) = match self.geode.insert(fd).await {
             Ok(v) => v,
             Err(e) => {
-                error!(target: "fud::put()", "Failed inserting file {:?} to geode: {}", path, e);
-                return JsonError::new(ErrorCode::InternalError, None, id).into()
+                let error_str = format!("Failed inserting file {:?} to geode: {}", path, e);
+                error!(target: "fud::put()", "{}", error_str);
+                return JsonError::new(ErrorCode::InternalError, Some(error_str), id).into()
             }
         };
 
@@ -114,24 +124,17 @@ impl Fud {
     }
 
     // RPCAPI:
-    // Fetch a file from the network, and subscribe to download events. Takes a file hash as parameter.
+    // Fetch a file from the network. Takes a file hash and file path (absolute or relative) as parameters.
+    // Returns the path where the file will be located once downloaded.
     //
-    // --> {"jsonrpc": "2.0", "method": "get", "params": ["1211...abfd"], "id": 42}
-    // <-- {"jsonrpc": "2.0", "method": "get", "params": `event`}
+    // --> {"jsonrpc": "2.0", "method": "get", "params": ["1211...abfd", "~/myfile.jpg"], "id": 42}
+    // <-- {"jsonrpc": "2.0", "method": "get", "params": "/home/user/myfile.jpg"}
     async fn get(&self, id: u16, params: JsonValue) -> JsonResult {
         let params = params.get::<Vec<JsonValue>>().unwrap();
         if params.len() != 2 || !params[0].is_string() || !params[1].is_string() {
             return JsonError::new(ErrorCode::InvalidParams, None, id).into()
         }
 
-        let file_name: Option<String> = match params[1].get::<String>() {
-            Some(name) => match name.is_empty() {
-                true => None,
-                false => Some(name.clone()),
-            },
-            None => None,
-        };
-
         let mut hash_buf = [0u8; 32];
         match bs58::decode(params[0].get::<String>().unwrap().as_str()).onto(&mut hash_buf) {
             Ok(_) => {}
@@ -139,9 +142,36 @@ impl Fud {
         }
 
         let file_hash = blake3::Hash::from_bytes(hash_buf);
+        let file_hash_str = hash_to_string(&file_hash);
+
+        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),
+                false => match PathBuf::from(path).is_absolute() {
+                    true => PathBuf::from(path),
+                    false => self.downloads_path.join(&file_hash_str).join(path),
+                },
+            },
+            None => self.downloads_path.join(&file_hash_str).join(&file_hash_str),
+        };
+
+        // Get the parent directory of the file
+        if let Some(parent) = file_path.parent() {
+            // Create all directories leading up to the file
+            let _ = fs::create_dir_all(parent).await;
+        }
 
-        let _ = self.get_tx.send((id, file_hash, file_name, Ok(()))).await;
+        let _ = self.get_tx.send((id, file_hash, file_path.clone(), Ok(()))).await;
+
+        JsonResponse::new(JsonValue::String(file_path.to_string_lossy().to_string()), id).into()
+    }
 
+    // RPCAPI:
+    // Subscribe to download events.
+    //
+    // --> {"jsonrpc": "2.0", "method": "get", "params": [], "id": 42}
+    // <-- {"jsonrpc": "2.0", "method": "get", "params": `event`}
+    async fn subscribe(&self, _id: u16, _params: JsonValue) -> JsonResult {
         self.download_sub.clone().into()
     }
 
@@ -239,6 +269,11 @@ impl Fud {
     }
 }
 
+#[derive(Clone, Debug)]
+pub struct DownloadStarted {
+    pub file_hash: blake3::Hash,
+    pub file_path: PathBuf,
+}
 #[derive(Clone, Debug)]
 pub struct ChunkDownloadCompleted {
     pub file_hash: blake3::Hash,
@@ -265,17 +300,32 @@ pub struct FileNotFound {
 }
 #[derive(Clone, Debug)]
 pub struct MissingChunks {}
+#[derive(Clone, Debug)]
+pub struct DownloadError {
+    pub file_hash: blake3::Hash,
+    pub error: String,
+}
 
 #[derive(Clone, Debug)]
 pub enum FudEvent {
+    DownloadStarted(DownloadStarted),
     ChunkDownloadCompleted(ChunkDownloadCompleted),
     FileDownloadCompleted(FileDownloadCompleted),
     DownloadCompleted(DownloadCompleted),
     ChunkNotFound(ChunkNotFound),
     FileNotFound(FileNotFound),
     MissingChunks(MissingChunks),
+    DownloadError(DownloadError),
 }
 
+impl From<DownloadStarted> for JsonValue {
+    fn from(info: DownloadStarted) -> JsonValue {
+        json_map([
+            ("file_hash", JsonValue::String(hash_to_string(&info.file_hash))),
+            ("file_path", JsonValue::String(info.file_path.to_string_lossy().to_string())),
+        ])
+    }
+}
 impl From<ChunkDownloadCompleted> for JsonValue {
     fn from(info: ChunkDownloadCompleted) -> JsonValue {
         json_map([
@@ -313,9 +363,20 @@ impl From<FileNotFound> for JsonValue {
         json_map([("file_hash", JsonValue::String(hash_to_string(&info.file_hash)))])
     }
 }
+impl From<DownloadError> for JsonValue {
+    fn from(info: DownloadError) -> JsonValue {
+        json_map([
+            ("file_hash", JsonValue::String(hash_to_string(&info.file_hash))),
+            ("error", JsonValue::String(info.error)),
+        ])
+    }
+}
 impl From<FudEvent> for JsonValue {
     fn from(event: FudEvent) -> JsonValue {
         match event {
+            FudEvent::DownloadStarted(info) => {
+                json_map([("event", json_str("download_started")), ("info", info.into())])
+            }
             FudEvent::ChunkDownloadCompleted(info) => {
                 json_map([("event", json_str("chunk_download_completed")), ("info", info.into())])
             }
@@ -332,21 +393,31 @@ impl From<FudEvent> for JsonValue {
                 json_map([("event", json_str("file_not_found")), ("info", info.into())])
             }
             FudEvent::MissingChunks(_) => json_map([("event", json_str("missing_chunks"))]),
+            FudEvent::DownloadError(info) => {
+                json_map([("event", json_str("download_error")), ("info", info.into())])
+            }
         }
     }
 }
 
 impl Fud {
     /// Handle `get` RPC request
-    pub async fn handle_get(&self, file_hash: blake3::Hash, file_name: Option<String>) {
+    pub async fn handle_get(&self, file_hash: &blake3::Hash, file_path: &PathBuf) {
         let self_node = self.dht().node.clone();
 
-        let chunked_file = match self.geode.get(&file_hash).await {
+        self.download_publisher
+            .notify(FudEvent::DownloadStarted(DownloadStarted {
+                file_hash: *file_hash,
+                file_path: file_path.clone(),
+            }))
+            .await;
+
+        let chunked_file = match self.geode.get(file_hash).await {
             Ok(v) => v,
             Err(Error::GeodeNeedsGc) => todo!(),
             Err(Error::GeodeFileNotFound) => {
-                info!(target: "self::get()", "Requested file {} not found in Geode, triggering fetch", file_hash);
-                self.file_fetch_tx.send((file_hash, Ok(()))).await.unwrap();
+                info!(target: "self::get()", "Requested file {} not found in Geode, triggering fetch", hash_to_string(file_hash));
+                self.file_fetch_tx.send((*file_hash, Ok(()))).await.unwrap();
                 info!(target: "self::get()", "Waiting for background file fetch task...");
                 let (i_file_hash, status) = self.file_fetch_end_rx.recv().await.unwrap();
                 match status {
@@ -354,7 +425,7 @@ impl Fud {
 
                     Err(Error::GeodeFileRouteNotFound) => {
                         self.download_publisher
-                            .notify(FudEvent::FileNotFound(FileNotFound { file_hash }))
+                            .notify(FudEvent::FileNotFound(FileNotFound { file_hash: *file_hash }))
                             .await;
                         return;
                     }
@@ -368,27 +439,33 @@ impl Fud {
 
         self.download_publisher
             .notify(FudEvent::FileDownloadCompleted(FileDownloadCompleted {
-                file_hash,
+                file_hash: *file_hash,
                 chunk_count: chunked_file.len(),
             }))
             .await;
 
         if chunked_file.is_complete() {
             let self_announce =
-                FudAnnounce { key: file_hash, seeders: vec![self_node.clone().into()] };
-            let _ = self.announce(&file_hash, &self_announce, self.seeders_router.clone()).await;
+                FudAnnounce { key: *file_hash, seeders: vec![self_node.clone().into()] };
+            let _ = self.announce(file_hash, &self_announce, self.seeders_router.clone()).await;
 
-            return match self.geode.assemble_file(&file_hash, &chunked_file, file_name).await {
-                Ok(file_path) => {
+            return match self.geode.assemble_file(file_hash, &chunked_file, file_path).await {
+                Ok(_) => {
                     self.download_publisher
                         .notify(FudEvent::DownloadCompleted(DownloadCompleted {
-                            file_hash,
+                            file_hash: *file_hash,
                             file_path: file_path.clone(),
                         }))
                         .await;
                 }
                 Err(e) => {
                     error!(target: "fud::handle_get()", "{}", e);
+                    self.download_publisher
+                        .notify(FudEvent::DownloadError(DownloadError {
+                            file_hash: *file_hash,
+                            error: e.to_string(),
+                        }))
+                        .await;
                 }
             };
         }
@@ -403,7 +480,7 @@ impl Fud {
             } else {
                 self.download_publisher
                     .notify(FudEvent::ChunkDownloadCompleted(ChunkDownloadCompleted {
-                        file_hash,
+                        file_hash: *file_hash,
                         chunk_hash: *chunk,
                     }))
                     .await;
@@ -411,9 +488,9 @@ impl Fud {
         }
 
         // Fetch missing chunks from seeders
-        self.fetch_chunks(&file_hash, &missing_chunks, &seeders).await;
+        self.fetch_chunks(file_hash, &missing_chunks, &seeders).await;
 
-        let chunked_file = match self.geode.get(&file_hash).await {
+        let chunked_file = match self.geode.get(file_hash).await {
             Ok(v) => v,
             Err(e) => panic!("{}", e),
         };
@@ -425,17 +502,23 @@ impl Fud {
             return;
         }
 
-        match self.geode.assemble_file(&file_hash, &chunked_file, file_name).await {
-            Ok(file_path) => {
+        match self.geode.assemble_file(file_hash, &chunked_file, file_path).await {
+            Ok(_) => {
                 self.download_publisher
                     .notify(FudEvent::DownloadCompleted(DownloadCompleted {
-                        file_hash,
+                        file_hash: *file_hash,
                         file_path: file_path.clone(),
                     }))
                     .await;
             }
             Err(e) => {
                 error!(target: "fud::handle_get()", "{}", e);
+                self.download_publisher
+                    .notify(FudEvent::DownloadError(DownloadError {
+                        file_hash: *file_hash,
+                        error: e.to_string(),
+                    }))
+                    .await;
             }
         };
     }

bin/fud/fud/src/tasks.rs

@@ -30,9 +30,9 @@ use log::{error, info};
 /// Triggered when calling the `get` RPC method
 pub async fn get_task(fud: Arc<Fud>) -> Result<()> {
     loop {
-        let (_, file_hash, file_name, _) = fud.get_rx.recv().await.unwrap();
+        let (_, file_hash, file_path, _) = fud.get_rx.recv().await.unwrap();
 
-        let _ = fud.handle_get(file_hash, file_name).await;
+        let _ = fud.handle_get(&file_hash, &file_path).await;
     }
 }
 
@@ -97,8 +97,10 @@ pub async fn fetch_file_task(fud: Arc<Fud>) -> Result<()> {
 /// Background task that removes seeders that did not announce a file/chunk
 /// for more than an hour.
 pub async fn prune_seeders_task(fud: Arc<Fud>) -> Result<()> {
+    sleep(120).await;
+
     loop {
-        sleep(1800).await; // TODO: Make a setting
+        sleep(3600).await; // TODO: Make a setting
 
         info!(target: "fud::prune_seeders_task()", "Pruning seeders...");
         fud.dht().prune_router(fud.seeders_router.clone(), 3600).await;

src/geode/mod.rs

@@ -78,8 +78,6 @@ pub const MAX_CHUNK_SIZE: usize = 262_144;
 const FILES_PATH: &str = "files";
 /// Path prefix where file chunks are stored
 const CHUNKS_PATH: &str = "chunks";
-/// Path prefix where full files are stored
-const DOWNLOADS_PATH: &str = "downloads";
 
 pub fn hash_to_string(hash: &blake3::Hash) -> String {
     bs58::encode(hash.as_bytes()).into_string()
@@ -127,8 +125,6 @@ pub struct Geode {
     files_path: PathBuf,
     /// Path to the filesystem directory where file chunks are stored
     chunks_path: PathBuf,
-    /// Path to the filesystem directory where full files are stored
-    downloads_path: PathBuf,
 }
 
 /// smol::fs::File::read does not guarantee that the buffer will be filled, even if the buffer is
@@ -158,17 +154,14 @@ impl Geode {
     pub async fn new(base_path: &PathBuf) -> Result<Self> {
         let mut files_path: PathBuf = base_path.into();
         let mut chunks_path: PathBuf = base_path.into();
-        let mut downloads_path: PathBuf = base_path.into();
         files_path.push(FILES_PATH);
         chunks_path.push(CHUNKS_PATH);
-        downloads_path.push(DOWNLOADS_PATH);
 
         // Create necessary directory structure if needed
         fs::create_dir_all(&files_path).await?;
         fs::create_dir_all(&chunks_path).await?;
-        fs::create_dir_all(&downloads_path).await?;
 
-        Ok(Self { files_path, chunks_path, downloads_path })
+        Ok(Self { files_path, chunks_path })
     }
 
     /// Attempt to read chunk hashes from a given file path and return
@@ -535,15 +528,14 @@ impl Geode {
         &self,
         file_hash: &blake3::Hash,
         chunked_file: &ChunkedFile,
-        file_name: Option<String>,
-    ) -> Result<PathBuf> {
+        file_path: &PathBuf,
+    ) -> Result<()> {
         let file_hash_str = hash_to_string(file_hash);
         info!(target: "geode::assemble_file()", "[Geode] Assembling file {}", file_hash_str);
 
-        let mut file_path = self.downloads_path.clone();
-        file_path.push(&file_hash_str);
-        fs::create_dir_all(&file_path).await?;
-        file_path.push(file_name.unwrap_or(file_hash_str));
+        if file_path.exists() && file_path.is_dir() {
+            return Err(Error::Custom("File path is an existing directory".to_string())) // TODO
+        }
 
         let mut file_fd = File::create(&file_path).await?;
         for (_, chunk_path) in chunked_file.iter() {
@@ -555,7 +547,7 @@ impl Geode {
             file_fd.flush().await?;
         }
 
-        Ok(file_path)
+        Ok(())
     }
 
     /// List file hashes.