fud: add put task, add `InsertError` and `InsertCompleted` events

bin/fud/fud/src/event.rs

@@ -16,6 +16,7 @@
  * along with this program.  If not, see <https://www.gnu.org/licenses/>.
  */
 
+use std::path::PathBuf;
 use tinyjson::JsonValue;
 
 use darkfi::{
@@ -75,6 +76,16 @@ pub struct DownloadError {
     pub hash: blake3::Hash,
     pub error: String,
 }
+#[derive(Clone, Debug)]
+pub struct InsertCompleted {
+    pub hash: blake3::Hash,
+    pub path: PathBuf,
+}
+#[derive(Clone, Debug)]
+pub struct InsertError {
+    pub path: PathBuf,
+    pub error: String,
+}
 
 #[derive(Clone, Debug)]
 pub enum FudEvent {
@@ -88,6 +99,8 @@ pub enum FudEvent {
     MetadataNotFound(MetadataNotFound),
     MissingChunks(MissingChunks),
     DownloadError(DownloadError),
+    InsertCompleted(InsertCompleted),
+    InsertError(InsertError),
 }
 
 impl From<DownloadStarted> for JsonValue {
@@ -168,6 +181,22 @@ impl From<DownloadError> for JsonValue {
         ])
     }
 }
+impl From<InsertCompleted> for JsonValue {
+    fn from(info: InsertCompleted) -> JsonValue {
+        json_map([
+            ("hash", JsonValue::String(hash_to_string(&info.hash))),
+            ("path", JsonValue::String(info.path.to_string_lossy().to_string())),
+        ])
+    }
+}
+impl From<InsertError> for JsonValue {
+    fn from(info: InsertError) -> JsonValue {
+        json_map([
+            ("path", JsonValue::String(info.path.to_string_lossy().to_string())),
+            ("error", JsonValue::String(info.error)),
+        ])
+    }
+}
 impl From<FudEvent> for JsonValue {
     fn from(event: FudEvent) -> JsonValue {
         match event {
@@ -202,6 +231,12 @@ impl From<FudEvent> for JsonValue {
             FudEvent::DownloadError(info) => {
                 json_map([("event", json_str("download_error")), ("info", info.into())])
             }
+            FudEvent::InsertCompleted(info) => {
+                json_map([("event", json_str("insert_completed")), ("info", info.into())])
+            }
+            FudEvent::InsertError(info) => {
+                json_map([("event", json_str("insert_error")), ("info", info.into())])
+            }
         }
     }
 }

bin/fud/fud/src/lib.rs

@@ -163,9 +163,15 @@ pub struct Fud {
     get_tx: channel::Sender<(blake3::Hash, PathBuf, FileSelection)>,
     get_rx: channel::Receiver<(blake3::Hash, PathBuf, FileSelection)>,
 
-    /// Currently active includingdownloading tasks (running the `fud.fetch_resource()` method)
+    put_tx: channel::Sender<PathBuf>,
+    put_rx: channel::Receiver<PathBuf>,
+
+    /// Currently active downloading tasks (running the `fud.fetch_resource()` method)
     fetch_tasks: Arc<RwLock<HashMap<blake3::Hash, Arc<StoppableTask>>>>,
 
+    /// Currently active put tasks (running the `fud.insert_resource()` method)
+    put_tasks: Arc<RwLock<HashMap<PathBuf, Arc<StoppableTask>>>>,
+
     /// Used to send events to fud clients
     event_publisher: PublisherPtr<FudEvent>,
 }
@@ -301,6 +307,7 @@ impl Fud {
             Arc::new(Dht::<FudNode>::new(&dht_settings, p2p.clone(), executor.clone()).await);
 
         let (get_tx, get_rx) = smol::channel::unbounded();
+        let (put_tx, put_rx) = smol::channel::unbounded();
         let fud = Self {
             node_data: Arc::new(RwLock::new(node_data)),
             secret_key: Arc::new(RwLock::new(secret_key)),
@@ -317,7 +324,10 @@ impl Fud {
             resources: Arc::new(RwLock::new(HashMap::new())),
             get_tx,
             get_rx,
+            put_tx,
+            put_rx,
             fetch_tasks: Arc::new(RwLock::new(HashMap::new())),
+            put_tasks: Arc::new(RwLock::new(HashMap::new())),
             event_publisher,
         };
 
@@ -878,7 +888,8 @@ impl Fud {
                 }
                 queried_seeders.insert(seeder.node.id());
 
-                if let Ok(channel) = self.get_channel(&seeder.node, Some(*hash)).await {
+                let channel = self.get_channel(&seeder.node, Some(*hash)).await;
+                if let Ok(channel) = channel {
                     let msg_subsystem = channel.message_subsystem();
                     msg_subsystem.add_dispatch::<FudChunkReply>().await;
                     msg_subsystem.add_dispatch::<FudFileReply>().await;
@@ -1078,7 +1089,7 @@ impl Fud {
             // If we could not find the metadata in geode, get it from the network
             Err(Error::GeodeFileNotFound) => {
                 // Find nodes close to the file hash
-                info!(target: "fud::fetch_resource()", "Requested metadata {} not found in Geode, triggering fetch", hash_to_string(hash));
+                info!(target: "fud::get_metadata()", "Requested metadata {} not found in Geode, triggering fetch", hash_to_string(hash));
                 let closest_nodes = self.lookup_nodes(hash).await.unwrap_or_default();
 
                 // Fetch file or directory metadata
@@ -1091,7 +1102,7 @@ impl Fud {
             }
 
             Err(e) => {
-                error!(target: "fud::fetch_resource()", "{e}");
+                error!(target: "fud::get_metadata()", "{e}");
                 Err(e)
             }
         }
@@ -1519,12 +1530,23 @@ impl Fud {
     }
 
     /// Add a resource from the file system.
-    pub async fn put(&self, path: &PathBuf) -> Result<blake3::Hash> {
+    pub async fn put(&self, path: &Path) -> Result<()> {
+        let put_tasks = self.put_tasks.read().await;
+        drop(put_tasks);
+
+        self.put_tx.send(path.to_path_buf()).await?;
+
+        Ok(())
+    }
+
+    /// Insert a file or directory from the file system.
+    /// Called when `put()` creates a new put task.
+    pub async fn insert_resource(&self, path: &PathBuf) -> Result<()> {
         let self_node = self.node().await;
 
         if self_node.addresses.is_empty() {
             return Err(Error::Custom(
-                "Cannot put file, you don't have any external address".to_string(),
+                "Cannot put resource, you don't have any external address".to_string(),
             ))
         }
 
@@ -1609,7 +1631,13 @@ impl Fud {
         let fud_announce = FudAnnounce { key: hash, seeders: vec![self_node.into()] };
         let _ = self.announce(&hash, &fud_announce, self.seeders_router.clone()).await;
 
-        Ok(hash)
+        // Send InsertCompleted event
+        notify_event!(self, InsertCompleted, {
+            hash,
+            path: path.to_path_buf()
+        });
+
+        Ok(())
     }
 
     /// Removes:
@@ -1652,7 +1680,7 @@ impl Fud {
         notify_event!(self, ResourceRemoved, { hash: *hash });
     }
 
-    /// Stop all tasks in `fetch_tasks`.
+    /// Stop all tasks in `fetch_tasks` and `put_tasks.
     pub async fn stop(&self) {
         // Create a clone of fetch_tasks because `task.stop()` needs a write lock
         let fetch_tasks = self.fetch_tasks.read().await;
@@ -1660,9 +1688,20 @@ impl Fud {
             fetch_tasks.iter().map(|(key, value)| (*key, value.clone())).collect();
         drop(fetch_tasks);
 
-        // Stop all tasks
+        // Stop all fetch tasks
         for task in cloned_fetch_tasks.values() {
             task.stop().await;
         }
+
+        // Create a clone of put_tasks because `task.stop()` needs a write lock
+        let put_tasks = self.put_tasks.read().await;
+        let cloned_put_tasks: HashMap<PathBuf, Arc<StoppableTask>> =
+            put_tasks.iter().map(|(key, value)| (key.clone(), value.clone())).collect();
+        drop(put_tasks);
+
+        // Stop all put tasks
+        for task in cloned_put_tasks.values() {
+            task.stop().await;
+        }
     }
 }

bin/fud/fud/src/main.rs

@@ -39,7 +39,7 @@ use fud::{
     proto::{FudFindNodesReply, ProtocolFud},
     rpc::JsonRpcInterface,
     settings::{Args, CONFIG_FILE, CONFIG_FILE_CONTENTS},
-    tasks::{announce_seed_task, get_task, node_id_task},
+    tasks::{announce_seed_task, get_task, node_id_task, put_task},
     Fud,
 };
 
@@ -126,6 +126,20 @@ async fn realmain(args: Args, ex: Arc<Executor<'static>>) -> Result<()> {
         ex.clone(),
     );
 
+    info!(target: "fud", "Starting put task");
+    let put_task_ = StoppableTask::new();
+    put_task_.clone().start(
+        put_task(fud.clone(), ex.clone()),
+        |res| async {
+            match res {
+                Ok(()) | Err(Error::DetachedTaskStopped) => { /* Do nothing */ }
+                Err(e) => error!(target: "fud", "Failed starting put task: {e}"),
+            }
+        },
+        Error::DetachedTaskStopped,
+        ex.clone(),
+    );
+
     let rpc_settings: RpcSettings = args.rpc.into();
     info!(target: "fud", "Starting JSON-RPC server on {}", rpc_settings.listen);
     let rpc_interface = Arc::new(JsonRpcInterface::new(fud.clone(), dnet_sub, event_sub));
@@ -215,6 +229,9 @@ async fn realmain(args: Args, ex: Arc<Executor<'static>>) -> Result<()> {
     info!(target: "fud", "Stopping get task...");
     get_task_.stop().await;
 
+    info!(target: "fud", "Stopping put task...");
+    put_task_.stop().await;
+
     info!(target: "fud", "Stopping JSON-RPC server...");
     rpc_task.stop().await;
 

bin/fud/fud/src/rpc.rs

@@ -113,7 +113,7 @@ impl JsonRpcInterface {
             return JsonError::new(ErrorCode::InternalError, Some(format!("{e}")), id).into()
         }
 
-        JsonResponse::new(JsonValue::String(hash_to_string(&res.unwrap())), id).into()
+        JsonResponse::new(JsonValue::String(path.to_string_lossy().to_string()), id).into()
     }
 
     // RPCAPI:
@@ -183,7 +183,7 @@ impl JsonRpcInterface {
     }
 
     // RPCAPI:
-    // Subscribe to download events.
+    // Subscribe to fud events.
     //
     // --> {"jsonrpc": "2.0", "method": "get", "params": [], "id": 42}
     // <-- {"jsonrpc": "2.0", "result": `event`, "id": 42}

bin/fud/fud/src/tasks.rs

@@ -83,6 +83,47 @@ pub async fn get_task(fud: Arc<Fud>, executor: ExecutorPtr) -> Result<()> {
     }
 }
 
+/// Triggered when calling the `fud.put()` method.
+pub async fn put_task(fud: Arc<Fud>, executor: ExecutorPtr) -> Result<()> {
+    loop {
+        let path = fud.put_rx.recv().await.unwrap();
+
+        // Create the new task
+        let mut put_tasks = fud.put_tasks.write().await;
+        let task = StoppableTask::new();
+        put_tasks.insert(path.clone(), task.clone());
+        drop(put_tasks);
+
+        // Start the new task
+        let fud_1 = fud.clone();
+        let fud_2 = fud.clone();
+        let path_ = path.clone();
+        task.start(
+            async move { fud_1.insert_resource(&path_).await },
+            move |res| async move {
+                // Remove the task from the `fud.put_tasks` hashmap once it is
+                // stopped (error, manually, or just done).
+                let mut put_tasks = fud_2.put_tasks.write().await;
+                put_tasks.remove(&path);
+                match res {
+                    Ok(()) | Err(Error::DetachedTaskStopped) => { /* Do nothing */ }
+                    Err(e) => {
+                        error!(target: "fud::put_task()", "Error while inserting resource: {e}");
+
+                        // Send a InsertError for any error that stopped the fetch task
+                        notify_event!(fud_2, InsertError, {
+                            path,
+                            error: e.to_string(),
+                        });
+                    }
+                }
+            },
+            Error::DetachedTaskStopped,
+            executor.clone(),
+        );
+    }
+}
+
 /// 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<()> {