fud, fu, geode: allow downloading a subset of files from a directory

This also introduces "scraps" in the sled database, which are chunks that contains both data the user wants and data the user does not want. Those chunks happen because files (in fud directories) are not aligned with chunks, so they can contain the data of multiple files.

After you download a chunk, fud checks that all bytes of it was written to the filesystem, if not that chunk will be saved as a scrap in sled. It's useful for a few things:

1. You don't download that chunk again later
2. You can still verify integrity, even if you did not write the full chunk into your downloaded files
3. You can seed that chunk (once partial seeding is implemented, which is currently not the case)

They are automatically removed from sled once they are not needed anymore (the fud resource is removed, or the chunk was rewritten to the filesystem, and this time it was fully written).

Those scraps are a concept in fud, not in geode.

bin/fud/fu/src/main.rs

@@ -61,12 +61,15 @@ struct Args {
 
 #[derive(Subcommand)]
 enum Subcmd {
-    /// Retrieve provided file name from the fud network
+    /// Retrieve provided resource from the fud network
     Get {
-        /// File hash
-        file: String,
-        /// File name
-        name: Option<String>,
+        /// Resource hash
+        hash: String,
+        /// Download path (relative or absolute)
+        path: Option<String>,
+        /// Optional list of files you want to download (only used for directories)
+        #[arg(short, long, num_args = 1..)]
+        files: Option<Vec<String>>,
     },
 
     /// Put a file onto the fud network
@@ -109,7 +112,8 @@ impl Fu {
     async fn get(
         &self,
         file_hash: String,
-        file_name: Option<String>,
+        file_path: Option<String>,
+        files: Option<Vec<String>>,
         ex: ExecutorPtr,
     ) -> Result<()> {
         let publisher = Publisher::new();
@@ -153,8 +157,8 @@ impl Fu {
             let chunks_downloaded =
                 *resource.get("chunks_downloaded").unwrap().get::<f64>().unwrap() as usize;
             let chunks_total =
-                *resource.get("chunks_total").unwrap().get::<f64>().unwrap() as usize;
-            let status = resource.get("status").unwrap().get::<String>().unwrap();
+                *resource.get("chunks_target").unwrap().get::<f64>().unwrap() as usize;
+            let mut status = resource.get("status").unwrap().get::<String>().unwrap().clone();
             let percent = match chunks_total {
                 0 => 0f64,
                 _ => chunks_downloaded as f64 / chunks_total as f64,
@@ -166,7 +170,10 @@ impl Fu {
                 "\x1B[2K\r[{bar}] {:.1}% | {chunks_downloaded}/{chunks_total} chunks | ",
                 percent * 100.0
             );
-            tstdout.set_color(&status_to_colorspec(status)).unwrap();
+            if remaining == 0 {
+                status = "seeding".to_string();
+            }
+            tstdout.set_color(&status_to_colorspec(&status)).unwrap();
             print!(
                 "{}",
                 match status.as_str() {
@@ -182,7 +189,13 @@ impl Fu {
             "get",
             JsonValue::Array(vec![
                 JsonValue::String(file_hash_.clone()),
-                JsonValue::String(file_name.unwrap_or_default()),
+                JsonValue::String(file_path.unwrap_or_default()),
+                match files {
+                    Some(files) => {
+                        JsonValue::Array(files.into_iter().map(JsonValue::String).collect())
+                    }
+                    None => JsonValue::Null,
+                },
             ]),
         );
         // Create a RPC client to send the `get` request
@@ -564,7 +577,7 @@ fn main() -> Result<()> {
             let fu = Fu { rpc_client, endpoint: args.endpoint.clone() };
 
             match args.command {
-                Subcmd::Get { file, name } => fu.get(file, name, ex.clone()).await,
+                Subcmd::Get { hash, path, files } => fu.get(hash, path, files, ex.clone()).await,
                 Subcmd::Put { file } => fu.put(file).await,
                 Subcmd::Ls {} => fu.list_resources().await,
                 Subcmd::Watch {} => fu.watch(ex.clone()).await,

bin/fud/fud/src/lib.rs

@@ -71,7 +71,10 @@ use tasks::FetchReply;
 
 /// Utils
 pub mod util;
-use util::get_all_files;
+use util::{get_all_files, FileSelection};
+
+const SLED_PATH_TREE: &[u8] = b"_fud_paths";
+const SLED_SCRAP_TREE: &[u8] = b"_fud_scraps";
 
 // TODO: This is not Sybil-resistant
 fn generate_node_id() -> Result<blake3::Hash> {
@@ -130,8 +133,14 @@ pub struct Fud {
     /// Sled tree containing "resource hash -> path on the filesystem"
     path_tree: sled::Tree,
 
-    get_tx: channel::Sender<(blake3::Hash, PathBuf)>,
-    get_rx: channel::Receiver<(blake3::Hash, PathBuf)>,
+    /// Sled tree containing scraps which are chunks containing data the user
+    /// did not want to save to files. They also contain data the user wanted
+    /// otherwise we would not have downloaded the chunk at all.
+    /// "chunk/scrap hash -> chunk content"
+    scrap_tree: sled::Tree,
+
+    get_tx: channel::Sender<(blake3::Hash, PathBuf, FileSelection)>,
+    get_rx: channel::Receiver<(blake3::Hash, PathBuf, FileSelection)>,
 
     /// Currently active downloading tasks (running the `fud.fetch_resource()` method)
     fetch_tasks: Arc<RwLock<HashMap<blake3::Hash, Arc<StoppableTask>>>>,
@@ -222,7 +231,7 @@ impl Fud {
         downloads_path: PathBuf,
         chunk_timeout: u64,
         dht: Arc<Dht>,
-        path_tree: sled::Tree,
+        sled_db: &sled::Db,
         event_publisher: PublisherPtr<FudEvent>,
     ) -> Result<Self> {
         let (get_tx, get_rx) = smol::channel::unbounded();
@@ -242,7 +251,8 @@ impl Fud {
             downloads_path,
             chunk_timeout,
             dht,
-            path_tree,
+            path_tree: sled_db.open_tree(SLED_PATH_TREE)?,
+            scrap_tree: sled_db.open_tree(SLED_SCRAP_TREE)?,
             resources: Arc::new(RwLock::new(HashMap::new())),
             get_tx,
             get_rx,
@@ -294,6 +304,7 @@ impl Fud {
                     status: ResourceStatus::Incomplete,
                     chunks_total: 0,
                     chunks_downloaded: 0,
+                    chunks_target: 0,
                 },
             );
         }
@@ -403,7 +414,7 @@ impl Fud {
                     continue;
                 }
             };
-            if let Err(e) = self.geode.verify_chunks(&mut chunked).await {
+            if let Err(e) = self.verify_chunks(&mut chunked).await {
                 error!(target: "fud::verify_resources()", "Error while verifying chunks of {}: {e}", hash_to_string(&resource.hash));
                 update_resource(&mut resource, ResourceStatus::Incomplete, None).await;
                 continue;
@@ -477,24 +488,15 @@ impl Fud {
         seeders
     }
 
-    /// Fetch chunks for `chunked` (file or directory) from `seeders`.
-    async fn fetch_missing_chunks(
+    /// Fetch `chunks` for `chunked` (file or directory) from `seeders`.
+    async fn fetch_chunks(
         &self,
         hash: &blake3::Hash,
         chunked: &mut ChunkedStorage,
         seeders: &HashSet<DhtRouterItem>,
+        chunks: &HashSet<blake3::Hash>,
     ) -> Result<()> {
-        let missing_chunks: HashSet<blake3::Hash> = {
-            let mut missing_chunks = HashSet::new();
-            for (chunk, available) in chunked.iter() {
-                if !available {
-                    missing_chunks.insert(*chunk);
-                }
-            }
-            missing_chunks
-        };
-
-        let mut remaining_chunks = missing_chunks.clone();
+        let mut remaining_chunks = chunks.clone();
         let mut shuffled_seeders = {
             let mut vec: Vec<_> = seeders.iter().cloned().collect();
             vec.shuffle(&mut OsRng);
@@ -505,7 +507,7 @@ impl Fud {
             let channel = match self.get_channel(&seeder.node, Some(*hash)).await {
                 Ok(channel) => channel,
                 Err(e) => {
-                    warn!(target: "fud::fetch_missing_chunks()", "Could not get a channel for node {}: {e}", hash_to_string(&seeder.node.id));
+                    warn!(target: "fud::fetch_chunks()", "Could not get a channel for node {}: {e}", hash_to_string(&seeder.node.id));
                     continue;
                 }
             };
@@ -534,7 +536,7 @@ impl Fud {
                 let send_res =
                     channel.send(&FudFindRequest { info: Some(*hash), key: chunk_hash }).await;
                 if let Err(e) = send_res {
-                    warn!(target: "fud::fetch_missing_chunks()", "Error while sending FudFindRequest: {e}");
+                    warn!(target: "fud::fetch_chunks()", "Error while sending FudFindRequest: {e}");
                     break; // Switch to another seeder
                 }
 
@@ -549,20 +551,31 @@ impl Fud {
                 select! {
                     chunk_reply = chunk_recv => {
                         if let Err(e) = chunk_reply {
-                            warn!(target: "fud::fetch_missing_chunks()", "Error waiting for chunk reply: {e}");
+                            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(chunked, &reply.chunk).await {
-                            Ok(inserted_hash) => {
+                            Ok((inserted_hash, bytes_written)) => {
                                 if inserted_hash != chunk_hash {
-                                    warn!(target: "fud::fetch_missing_chunks()", "Received chunk does not match requested chunk");
+                                    warn!(target: "fud::fetch_chunks()", "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
                                 }
 
+                                info!(target: "fud::fetch_chunks()", "Received chunk {} from seeder {}", hash_to_string(&chunk_hash), hash_to_string(&seeder.node.id));
+
+                                // If we did not write the whole chunk to the filesystem,
+                                // save the chunk in the scraps.
+                                if bytes_written < reply.chunk.len() {
+                                    info!(target: "fud::fetch_chunks()", "Saving chunk {} as a scrap", hash_to_string(&chunk_hash));
+                                    if let Err(e) = self.scrap_tree.insert(chunk_hash.as_bytes(), reply.chunk.clone()) {
+                                        error!(target: "fud::fetch_chunks()", "Failed to save chunk {} as a scrap: {e}", hash_to_string(&chunk_hash))
+                                    }
+                                }
+
                                 // Update resource `chunks_downloaded`
                                 let mut resources_write = self.resources.write().await;
                                 let resource = match resources_write.get_mut(hash) {
@@ -575,7 +588,6 @@ impl Fud {
                                 };
                                 drop(resources_write);
 
-                                info!(target: "fud::fetch_missing_chunks()", "Received chunk {} from seeder {}", hash_to_string(&chunk_hash), hash_to_string(&seeder.node.id));
                                 self.event_publisher
                                     .notify(FudEvent::ChunkDownloadCompleted(ChunkDownloadCompleted {
                                         hash: *hash,
@@ -586,18 +598,18 @@ impl Fud {
                                 remaining_chunks.remove(&chunk_hash);
                             }
                             Err(e) => {
-                                error!(target: "fud::fetch_missing_chunks()", "Failed inserting chunk {} to Geode: {e}", hash_to_string(&chunk_hash));
+                                error!(target: "fud::fetch_chunks()", "Failed inserting chunk {} to Geode: {e}", hash_to_string(&chunk_hash));
                             }
                         };
                     }
                     notfound_reply = notfound_recv => {
                         if let Err(e) = notfound_reply {
-                            warn!(target: "fud::fetch_missing_chunks()", "Error waiting for NOTFOUND reply: {e}");
+                            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_missing_chunks()", "Received NOTFOUND {} from seeder {}", hash_to_string(&chunk_hash), hash_to_string(&seeder.node.id));
+                        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: *hash,
@@ -858,7 +870,9 @@ impl Fud {
 
     /// Start downloading a file or directory from the network to `path`.
     /// This creates a new task in `fetch_tasks` calling `fetch_resource()`.
-    pub async fn get(&self, hash: &blake3::Hash, path: &Path) -> Result<()> {
+    /// `files` is the list of files (relative paths) you want to download
+    /// (if the resource is a directory), None means you want all files.
+    pub async fn get(&self, hash: &blake3::Hash, path: &Path, files: FileSelection) -> Result<()> {
         let fetch_tasks = self.fetch_tasks.read().await;
         if fetch_tasks.contains_key(hash) {
             return Err(Error::Custom(format!(
@@ -868,14 +882,19 @@ impl Fud {
         }
         drop(fetch_tasks);
 
-        self.get_tx.send((*hash, path.to_path_buf())).await?;
+        self.get_tx.send((*hash, path.to_path_buf(), files)).await?;
 
         Ok(())
     }
 
     /// Download a file or directory from the network to `path`.
     /// Called when `get()` creates a new fetch task.
-    pub async fn fetch_resource(&self, hash: &blake3::Hash, path: &Path) -> Result<()> {
+    pub async fn fetch_resource(
+        &self,
+        hash: &blake3::Hash,
+        path: &Path,
+        files: &FileSelection,
+    ) -> Result<()> {
         let self_node = self.dht().node().await;
         let mut closest_nodes = vec![];
 
@@ -909,6 +928,7 @@ impl Fud {
             status: ResourceStatus::Discovering,
             chunks_total: 0,
             chunks_downloaded: 0,
+            chunks_target: 0,
         };
         let mut resources_write = self.resources.write().await;
         resources_write.insert(*hash, resource.clone());
@@ -961,10 +981,19 @@ impl Fud {
             }
         };
 
+        // Get a list of all file paths
+        let files_to_create: Vec<PathBuf> = match files {
+            FileSelection::Set(files) => files
+                .iter()
+                .map(|file| path.join(file))
+                .filter(|abs| chunked.get_files().iter().any(|(f, _)| f == abs))
+                .collect(),
+            FileSelection::All => chunked.get_files().iter().map(|(f, _)| f.clone()).collect(),
+        };
         // Create all files (and all necessary directories)
-        for (file_path, _) in chunked.get_files().iter() {
+        for file_path in files_to_create.iter() {
             if !file_path.exists() {
-                if let Some(dir) = path.join(file_path).parent() {
+                if let Some(dir) = file_path.parent() {
                     fs::create_dir_all(dir).await?;
                 }
                 File::create(&file_path).await?;
@@ -991,24 +1020,26 @@ impl Fud {
         drop(resources_write);
 
         // Mark locally available chunks as such
-        if let Err(e) = self.geode.verify_chunks(&mut chunked).await {
-            error!(target: "self::get()", "Error while verifying chunks: {e}");
+        let scraps = self.verify_chunks(&mut chunked).await;
+        if let Err(e) = scraps {
+            error!(target: "fud::get()", "Error while verifying chunks: {e}");
             return Err(e);
         }
+        let scraps = scraps.unwrap();
 
-        // Set resource.chunks_downloaded and send FudEvent::ResourceUpdated
-        let mut resources_write = self.resources.write().await;
-        if let Some(resource) = resources_write.get_mut(hash) {
-            resource.chunks_downloaded = chunked.local_chunks() as u64;
+        // Write all scraps to make sure the data on the filesystem is correct
+        if !scraps.is_empty() {
+            info!(target: "fud::get()", "Writing {} scraps...", scraps.len());
+        }
+        for (scrap_hash, scrap) in scraps {
+            let len = scrap.len();
+            let (_, bytes_written) = self.geode.write_chunk(&mut chunked, scrap).await?;
 
-            self.event_publisher
-                .notify(FudEvent::ResourceUpdated(ResourceUpdated {
-                    hash: *hash,
-                    resource: resource.clone(),
-                }))
-                .await;
+            // If the whole scrap was written, we can remove it from sled
+            if bytes_written == len {
+                self.scrap_tree.remove(scrap_hash.as_bytes())?;
+            }
         }
-        drop(resources_write);
 
         // If `chunked` is a file that is bigger than the all its chunks,
         // truncate the file to the chunks.
@@ -1023,24 +1054,55 @@ impl Fud {
             }
         }
 
-        // If the resource is already complete, we don't need to download any chunk
-        if chunked.is_complete() {
-            // Announce the file
-            let self_announce = FudAnnounce { key: *hash, seeders: vec![self_node.clone().into()] };
-            let _ = self.announce(hash, &self_announce, self.seeders_router.clone()).await;
+        // Set of all chunks we need locally (including the ones we already have)
+        let chunks: HashSet<(blake3::Hash, bool)> = match files {
+            FileSelection::Set(files) => {
+                let mut chunks = HashSet::new();
+                for file in files {
+                    chunks.extend(chunked.get_chunks_of_file(&path.join(file)));
+                }
+                chunks
+            }
+            FileSelection::All => chunked.iter().cloned().collect(),
+        };
+        let chunk_hashes: HashSet<_> = chunks.iter().map(|(hash, _)| hash).collect();
 
-            // Set resource status to `Seeding`
+        // Set of the chunks we need to download
+        let missing_chunks: HashSet<blake3::Hash> = {
+            let mut missing_chunks = HashSet::new();
+            for (chunk, available) in chunks.clone() {
+                if !available {
+                    missing_chunks.insert(chunk);
+                }
+            }
+            missing_chunks
+        };
+
+        // If we don't need to download any chunk
+        if missing_chunks.is_empty() {
+            // Set resource status to `Seeding` or `Incomplete`
             let mut resources_write = self.resources.write().await;
             let resource = match resources_write.get_mut(hash) {
                 Some(resource) => {
-                    resource.status = ResourceStatus::Seeding;
-                    resource.chunks_downloaded = chunked.len() as u64;
+                    resource.status = match chunked.is_complete() {
+                        true => ResourceStatus::Seeding,
+                        false => ResourceStatus::Incomplete,
+                    };
+                    resource.chunks_downloaded = chunks.len() as u64;
+                    resource.chunks_target = chunks.len() as u64;
                     resource.clone()
                 }
                 None => return Ok(()), // Resource was removed, abort
             };
             drop(resources_write);
 
+            // Announce the resource if we have all chunks
+            if chunked.is_complete() {
+                let self_announce =
+                    FudAnnounce { key: *hash, seeders: vec![self_node.clone().into()] };
+                let _ = self.announce(hash, &self_announce, self.seeders_router.clone()).await;
+            }
+
             // Send a DownloadCompleted event
             self.event_publisher
                 .notify(FudEvent::DownloadCompleted(event::DownloadCompleted {
@@ -1057,6 +1119,8 @@ impl Fud {
         let resource = match resources_write.get_mut(hash) {
             Some(resource) => {
                 resource.status = ResourceStatus::Downloading;
+                resource.chunks_target = chunks.len() as u64;
+                resource.chunks_downloaded = (chunks.len() - missing_chunks.len()) as u64;
                 resource.clone()
             }
             None => return Ok(()), // Resource was removed, abort
@@ -1086,7 +1150,7 @@ impl Fud {
             .collect();
 
         // Fetch missing chunks from seeders
-        self.fetch_missing_chunks(hash, &mut chunked, &seeders).await?;
+        self.fetch_chunks(hash, &mut chunked, &seeders, &missing_chunks).await?;
 
         // Get chunked file from geode
         let mut chunked = match self.geode.get(hash, path).await {
@@ -1112,11 +1176,16 @@ impl Fud {
         drop(resources_write);
 
         // Verify all chunks
-        self.geode.verify_chunks(&mut chunked).await?;
+        self.verify_chunks(&mut chunked).await?;
 
-        // We fetched all chunks, but the file is not complete
+        let is_complete = chunked
+            .iter()
+            .filter(|(hash, _)| chunk_hashes.contains(hash))
+            .all(|(_, available)| *available);
+
+        // We fetched all chunks, but the resource is not complete
         // (some chunks were missing from all seeders)
-        if !chunked.is_complete() {
+        if !is_complete {
             // Set resource status to `Incomplete`
             let mut resources_write = self.resources.write().await;
             let resource = match resources_write.get_mut(hash) {
@@ -1135,22 +1204,27 @@ impl Fud {
             return Ok(());
         }
 
-        // Announce the file
-        let self_announce = FudAnnounce { key: *hash, seeders: vec![self_node.clone().into()] };
-        let _ = self.announce(hash, &self_announce, self.seeders_router.clone()).await;
-
-        // Set resource status to `Seeding`
+        // Set resource status to `Seeding` or `Incomplete`
         let mut resources_write = self.resources.write().await;
         let resource = match resources_write.get_mut(hash) {
             Some(resource) => {
-                resource.status = ResourceStatus::Seeding;
-                resource.chunks_downloaded = chunked.len() as u64;
+                resource.status = match chunked.is_complete() {
+                    true => ResourceStatus::Seeding,
+                    false => ResourceStatus::Incomplete,
+                };
+                resource.chunks_downloaded = chunks.len() as u64;
                 resource.clone()
             }
             None => return Ok(()), // Resource was removed, abort
         };
         drop(resources_write);
 
+        // Announce the resource if we have all chunks
+        if chunked.is_complete() {
+            let self_announce = FudAnnounce { key: *hash, seeders: vec![self_node.clone().into()] };
+            let _ = self.announce(hash, &self_announce, self.seeders_router.clone()).await;
+        }
+
         // Send a DownloadCompleted event
         self.event_publisher
             .notify(FudEvent::DownloadCompleted(event::DownloadCompleted { hash: *hash, resource }))
@@ -1159,6 +1233,38 @@ impl Fud {
         Ok(())
     }
 
+    /// Iterate over chunks and find which chunks are available locally,
+    /// either in the filesystem (using geode::verify_chunks()) or in scraps.
+    /// Return the scraps in a HashMap.
+    pub async fn verify_chunks(
+        &self,
+        chunked: &mut ChunkedStorage,
+    ) -> Result<HashMap<blake3::Hash, Vec<u8>>> {
+        self.geode.verify_chunks(chunked).await?;
+
+        // Look for the chunks that are not on the filesystem in the scraps
+        let chunks = chunked.get_chunks().clone();
+        let missing_on_fs: Vec<_> =
+            chunks.iter().enumerate().filter(|(_, (_, available))| !available).collect();
+        let mut scraps = HashMap::new();
+        for (chunk_index, (chunk_hash, _)) in missing_on_fs {
+            let chunk = self.scrap_tree.get(chunk_hash.as_bytes())?;
+            if chunk.is_none() {
+                continue;
+            }
+
+            // Verify the scrap we found
+            let chunk = chunk.unwrap();
+            if self.geode.verify_chunk(chunk_hash, &chunk) {
+                // Mark it as available if it's valid
+                chunked.get_chunk_mut(chunk_index).1 = true;
+                scraps.insert(*chunk_hash, chunk.to_vec());
+            }
+        }
+
+        Ok(scraps)
+    }
+
     /// Add a resource from the file system.
     pub async fn put(&self, path: &PathBuf) -> Result<blake3::Hash> {
         let self_node = self.dht.node().await;
@@ -1233,6 +1339,7 @@ impl Fud {
                 status: ResourceStatus::Seeding,
                 chunks_total: chunk_hashes.len() as u64,
                 chunks_downloaded: chunk_hashes.len() as u64,
+                chunks_target: chunk_hashes.len() as u64,
             },
         );
         drop(resources_write);
@@ -1244,18 +1351,39 @@ impl Fud {
         Ok(hash)
     }
 
-    /// Remove a resource, its metadata in geode, and its path in the sled path tree.
+    /// Removes:
+    /// - a resource
+    /// - its metadata in geode
+    /// - its path in the sled path tree
+    /// - and any related scrap in the sled scrap tree,
+    ///
+    /// then sends a `ResourceRemoved` fud event.
     pub async fn remove(&self, hash: &blake3::Hash) {
+        // Remove the resource
         let mut resources_write = self.resources.write().await;
         resources_write.remove(hash);
         drop(resources_write);
 
+        // Remove the scraps in sled
+        if let Ok(Some(path)) = self.hash_to_path(hash) {
+            let chunked = self.geode.get(hash, &path).await;
+
+            if let Ok(chunked) = chunked {
+                for (chunk_hash, _) in chunked.iter() {
+                    let _ = self.scrap_tree.remove(chunk_hash.as_bytes());
+                }
+            }
+        }
+
+        // Remove the metadata in geode
         let hash_str = hash_to_string(hash);
         let _ = fs::remove_file(self.geode.files_path.join(&hash_str)).await;
         let _ = fs::remove_file(self.geode.dirs_path.join(&hash_str)).await;
 
+        // Remove the path in sled
         let _ = self.path_tree.remove(hash.as_bytes());
 
+        // Send a `ResourceRemoved` event
         self.event_publisher
             .notify(FudEvent::ResourceRemoved(event::ResourceRemoved { hash: *hash }))
             .await;

bin/fud/fud/src/main.rs

@@ -150,7 +150,7 @@ async fn realmain(args: Args, ex: Arc<Executor<'static>>) -> Result<()> {
             downloads_path,
             args.chunk_timeout,
             dht.clone(),
-            sled_db.open_tree("path")?,
+            &sled_db,
             event_pub.clone(),
         )
         .await?,

bin/fud/fud/src/resource.rs

@@ -44,6 +44,7 @@ pub struct Resource {
     pub status: ResourceStatus,
     pub chunks_total: u64,
     pub chunks_downloaded: u64,
+    pub chunks_target: u64,
 }
 
 impl From<Resource> for JsonValue {
@@ -83,6 +84,7 @@ impl From<Resource> for JsonValue {
             ),
             ("chunks_total", JsonValue::Number(rs.chunks_total as f64)),
             ("chunks_downloaded", JsonValue::Number(rs.chunks_downloaded as f64)),
+            ("chunks_target", JsonValue::Number(rs.chunks_target as f64)),
         ])
     }
 }

bin/fud/fud/src/rpc.rs

@@ -39,7 +39,7 @@ use darkfi::{
     Result,
 };
 
-use crate::Fud;
+use crate::{util::FileSelection, Fud};
 
 pub struct JsonRpcInterface {
     fud: Arc<Fud>,
@@ -116,14 +116,15 @@ impl JsonRpcInterface {
     }
 
     // RPCAPI:
-    // Fetch a resource from the network. Takes a hash and path (absolute or relative) as parameters.
+    // Fetch a resource from the network. Takes a hash, path (absolute or relative), and an
+    // optional list of file paths (only used for directories) as parameters.
     // Returns the path where the resource will be located once downloaded.
     //
-    // --> {"jsonrpc": "2.0", "method": "get", "params": ["1211...abfd", "~/myfile.jpg"], "id": 42}
+    // --> {"jsonrpc": "2.0", "method": "get", "params": ["1211...abfd", "~/myfile.jpg", null], "id": 42}
     // <-- {"jsonrpc": "2.0", "result": "/home/user/myfile.jpg", "id": 42}
     async fn get(&self, id: u16, params: JsonValue) -> JsonResult {
         let params = params.get::<Vec<JsonValue>>().unwrap();
-        if params.len() != 2 || !params[0].is_string() || !params[1].is_string() {
+        if params.len() != 3 || !params[0].is_string() || !params[1].is_string() {
             return JsonError::new(ErrorCode::InvalidParams, None, id).into()
         }
 
@@ -157,8 +158,23 @@ impl JsonRpcInterface {
             None => self.fud.downloads_path.join(&hash_str),
         };
 
+        let files: FileSelection = match &params[2] {
+            JsonValue::Array(files) => files
+                .iter()
+                .filter_map(|v| {
+                    if let JsonValue::String(file) = v {
+                        Some(PathBuf::from(file.clone()))
+                    } else {
+                        None
+                    }
+                })
+                .collect(),
+            JsonValue::Null => FileSelection::All,
+            _ => return JsonError::new(ErrorCode::InvalidParams, None, id).into(),
+        };
+
         // Start downloading the resource
-        if let Err(e) = self.fud.get(&hash, &path).await {
+        if let Err(e) = self.fud.get(&hash, &path, files).await {
             return JsonError::new(ErrorCode::InternalError, Some(e.to_string()), id).into()
         }
 

bin/fud/fud/src/tasks.rs

@@ -42,7 +42,7 @@ pub enum FetchReply {
 /// removed from the hashmap.
 pub async fn get_task(fud: Arc<Fud>, executor: ExecutorPtr) -> Result<()> {
     loop {
-        let (hash, path) = fud.get_rx.recv().await.unwrap();
+        let (hash, path, files) = fud.get_rx.recv().await.unwrap();
 
         // Create the new task
         let mut fetch_tasks = fud.fetch_tasks.write().await;
@@ -54,7 +54,7 @@ pub async fn get_task(fud: Arc<Fud>, executor: ExecutorPtr) -> Result<()> {
         let fud_1 = fud.clone();
         let fud_2 = fud.clone();
         task.start(
-            async move { fud_1.fetch_resource(&hash, &path).await },
+            async move { fud_1.fetch_resource(&hash, &path, &files).await },
             move |res| async move {
                 // Remove the task from the `fud.fetch_tasks` hashmap once it is
                 // stopped (error, manually, or just done).

bin/fud/fud/src/util.rs

@@ -18,7 +18,10 @@
 
 use darkfi::Result;
 use smol::{fs, stream::StreamExt};
-use std::path::{Path, PathBuf};
+use std::{
+    collections::HashSet,
+    path::{Path, PathBuf},
+};
 
 pub async fn get_all_files(dir: &Path) -> Result<Vec<(PathBuf, u64)>> {
     let mut files = Vec::new();
@@ -39,3 +42,18 @@ pub async fn get_all_files(dir: &Path) -> Result<Vec<(PathBuf, u64)>> {
 
     Ok(files)
 }
+
+/// An enum to represent a set of files, where you can use `All` if you want
+/// all files without having to specify all of them.
+/// We could use an Option<HashSet<PathBuf>>, but this is more explicit.
+pub enum FileSelection {
+    All,
+    Set(HashSet<PathBuf>),
+}
+
+impl FromIterator<PathBuf> for FileSelection {
+    fn from_iter<I: IntoIterator<Item = PathBuf>>(iter: I) -> Self {
+        let paths: HashSet<PathBuf> = iter.into_iter().collect();
+        FileSelection::Set(paths)
+    }
+}

src/geode/chunked_storage.rs

@@ -16,8 +16,9 @@
  * along with this program.  If not, see <https://www.gnu.org/licenses/>.
  */
 
-use crate::geode::file_sequence::FileSequence;
-use std::path::PathBuf;
+use std::path::{Path, PathBuf};
+
+use crate::geode::{file_sequence::FileSequence, MAX_CHUNK_SIZE};
 
 /// `ChunkedStorage` is a representation of a file or directory we're trying to
 /// retrieve from `Geode`.
@@ -97,4 +98,25 @@ impl ChunkedStorage {
     pub fn is_dir(&self) -> bool {
         self.is_dir
     }
+
+    /// Return all chunks that contain parts of `file`.
+    pub fn get_chunks_of_file(&self, file: &Path) -> Vec<(blake3::Hash, bool)> {
+        let files = self.fileseq.get_files();
+        let file_index = files.iter().position(|(f, _)| f == file);
+        if file_index.is_none() {
+            return vec![];
+        }
+        let file_index = file_index.unwrap();
+
+        let start_pos = self.fileseq.get_file_position(file_index);
+
+        let end_pos = start_pos + files[file_index].1;
+
+        let start_index = (start_pos as f64 / MAX_CHUNK_SIZE as f64).floor();
+        let end_index = (end_pos as f64 / MAX_CHUNK_SIZE as f64).floor();
+
+        let chunk_indexes: Vec<usize> = (start_index as usize..=end_index as usize).collect();
+
+        chunk_indexes.iter().filter_map(|&index| self.chunks.get(index)).cloned().collect()
+    }
 }

src/geode/file_sequence.rs

@@ -35,6 +35,10 @@ use std::{path::PathBuf, pin::Pin};
 /// This allows seamless handling of multiple files as if they were a single
 /// continuous file. It automatically opens the next file in the list when the
 /// current file is exhausted.
+///
+/// It's also made so that files in `files` that do not exist on the filesystem
+/// will get skipped, without returning an Error. All files you want to read,
+/// write, and seek to should be created before using the FileSequence.
 #[derive(Debug)]
 pub struct FileSequence {
     /// List of (file path, file size). File sizes are not the sizes of the
@@ -44,6 +48,8 @@ pub struct FileSequence {
     current_file: Option<File>,
     /// Index of the currently opened file in the `files` vector
     current_file_index: Option<usize>,
+
+    position: u64,
     /// Set to `true` to automatically set the length of the file on the
     /// filesystem to it's size as defined in the `files` vector, after a write
     auto_set_len: bool,
@@ -51,7 +57,13 @@ pub struct FileSequence {
 
 impl FileSequence {
     pub fn new(files: &[(PathBuf, u64)], auto_set_len: bool) -> Self {
-        Self { files: files.to_vec(), current_file: None, current_file_index: None, auto_set_len }
+        Self {
+            files: files.to_vec(),
+            current_file: None,
+            current_file_index: None,
+            position: 0,
+            auto_set_len,
+        }
     }
 
     /// Update a single file size.
@@ -69,10 +81,21 @@ impl FileSequence {
         &self.files
     }
 
+    /// Compute the starting position of the file (in bytes) by suming up
+    /// the size of the previous files.
+    pub fn get_file_position(&self, file_index: usize) -> u64 {
+        let mut pos = 0;
+        for i in 0..file_index {
+            pos += self.files[i].1;
+        }
+        pos
+    }
+
     /// Open the file at (`current_file_index` + 1).
     /// If no file is currently open (`current_file_index` is None), it opens
     /// the first file.
     async fn open_next_file(&mut self) -> io::Result<()> {
+        self.current_file = None;
         self.current_file_index = match self.current_file_index {
             Some(i) => Some(i + 1),
             None => Some(0),
@@ -83,7 +106,7 @@ impl FileSequence {
         let file = OpenOptions::new()
             .read(true)
             .write(true)
-            .create(true)
+            .create(false)
             .open(self.files[self.current_file_index.unwrap()].0.clone())
             .await?;
         self.current_file = Some(file);
@@ -92,14 +115,15 @@ impl FileSequence {
 
     /// Open the file at `file_index`.
     async fn open_file(&mut self, file_index: usize) -> io::Result<()> {
+        self.current_file = None;
+        self.current_file_index = Some(file_index);
         let file = OpenOptions::new()
             .read(true)
             .write(true)
-            .create(true)
+            .create(false)
             .open(self.files[file_index].0.clone())
             .await?;
         self.current_file = Some(file);
-        self.current_file_index = Some(file_index);
         Ok(())
     }
 }
@@ -177,13 +201,20 @@ impl AsyncSeek for FileSequence {
             )))
         }
 
+        this.position = abs_pos; // Update FileSequence position
+
         // Open the file
         if this.current_file.is_none() ||
             this.current_file_index.is_some() && this.current_file_index.unwrap() != file_index
         {
-            if let Err(e) = smol::block_on(this.open_file(file_index)) {
-                return Poll::Ready(Err(e));
-            }
+            match smol::block_on(this.open_file(file_index)) {
+                Ok(_) => {}
+                Err(e) if e.kind() == io::ErrorKind::NotFound => {
+                    // If the file does not exist, return without actually seeking it
+                    return Poll::Ready(Ok(this.position));
+                }
+                Err(e) => return Poll::Ready(Err(e)),
+            };
         }
 
         let file = this.current_file.as_mut().unwrap();
@@ -191,7 +222,7 @@ impl AsyncSeek for FileSequence {
 
         // Seek in the current file
         match smol::block_on(file.seek(SeekFrom::Start(file_pos))) {
-            Ok(new_position) => Poll::Ready(Ok(new_position)),
+            Ok(_) => Poll::Ready(Ok(this.position)),
             Err(e) => Poll::Ready(Err(e)),
         }
     }
@@ -223,9 +254,26 @@ impl AsyncWrite for FileSequence {
                     if file_index >= this.files.len() - 1 {
                         break; // No more files
                     }
+                    if remaining_buf.is_empty() {
+                        break; // No more data to write
+                    }
+                    let start_pos = this.get_file_position(file_index);
+                    let file_size = this.files[file_index].1 as usize;
+                    let file_pos = this.position - start_pos;
+                    let space_left = file_size - file_pos as usize;
+                    let skip_bytes = remaining_buf.len().min(space_left);
+                    this.position += skip_bytes as u64;
+                    remaining_buf = &remaining_buf[skip_bytes..]; // Update the remaining buffer
                 }
-                if let Err(e) = smol::block_on(this.open_next_file()) {
-                    return Poll::Ready(Err(e));
+
+                // Switch to the next file
+                match smol::block_on(this.open_next_file()) {
+                    Ok(_) => {}
+                    Err(e) if e.kind() == io::ErrorKind::NotFound => {
+                        this.current_file = None;
+                        continue; // Skip to next file
+                    }
+                    Err(e) => return Poll::Ready(Err(e)),
                 }
             }
 
@@ -250,6 +298,7 @@ impl AsyncWrite for FileSequence {
             match smol::block_on(file.write(&remaining_buf[..bytes_to_write])) {
                 Ok(bytes_written) => {
                     total_bytes_written += bytes_written;
+                    this.position += bytes_written as u64;
                     remaining_buf = &remaining_buf[bytes_written..]; // Update the remaining buffer
                     if remaining_buf.is_empty() {
                         if let Err(e) = finalize_current_file(file, max_size) {

src/geode/mod.rs

@@ -82,7 +82,10 @@ use futures::{AsyncRead, AsyncSeek};
 use log::{debug, info, warn};
 use smol::{
     fs::{self, File},
-    io::{AsyncBufReadExt, AsyncReadExt, AsyncSeekExt, AsyncWriteExt, BufReader, Cursor, SeekFrom},
+    io::{
+        AsyncBufReadExt, AsyncReadExt, AsyncSeekExt, AsyncWriteExt, BufReader, Cursor, ErrorKind,
+        SeekFrom,
+    },
     stream::StreamExt,
 };
 use std::path::Path;
@@ -330,12 +333,13 @@ impl Geode {
 
     /// Write a single chunk given a stream.
     /// The file must be inserted into Geode before calling this method.
-    /// Always overwrites any existing chunk. Returns the chunk hash once inserted.
+    /// Always overwrites any existing chunk. Returns the chunk hash and
+    /// the number of bytes written to the file system.
     pub async fn write_chunk(
         &self,
         chunked: &mut ChunkedStorage,
         stream: impl AsRef<[u8]>,
-    ) -> Result<blake3::Hash> {
+    ) -> Result<(blake3::Hash, usize)> {
         info!(target: "geode::write_chunk()", "[Geode] Writing single chunk");
 
         let mut cursor = Cursor::new(&stream);
@@ -363,7 +367,7 @@ impl Geode {
         fileseq.seek(SeekFrom::Start(position)).await?;
 
         // This will write the chunk, and truncate files if `chunked` is a directory.
-        fileseq.write_all(chunk_slice).await?;
+        let bytes_written = fileseq.write(chunk_slice).await?;
 
         // If it's the last chunk of a file (and it's *not* a directory),
         // truncate the file to the correct length.
@@ -379,7 +383,7 @@ impl Geode {
             }
         }
 
-        Ok(chunk_hash)
+        Ok((chunk_hash, bytes_written))
     }
 
     /// Iterate over chunks and find which chunks are available locally.
@@ -392,8 +396,9 @@ impl Geode {
             // Read the chunk using the FileSequence
             let chunk = match self.read_chunk(&mut chunked_file.get_fileseq(), &chunk_index).await {
                 Ok(c) => c,
+                Err(Error::Io(ErrorKind::NotFound)) => continue,
                 Err(e) => {
-                    warn!("Error while verifying chunks: {e}");
+                    warn!(target: "geode::verify_chunks()", "Error while verifying chunks: {e}");
                     break
                 }
             };
@@ -427,7 +432,7 @@ impl Geode {
                     return self.create_chunked_storage(hash, path, &chunk_hashes, &files).await
                 }
                 Err(e) => {
-                    if !matches!(e, Error::Io(std::io::ErrorKind::NotFound)) {
+                    if !matches!(e, Error::Io(ErrorKind::NotFound)) {
                         return Err(Error::GeodeNeedsGc)
                     }
                 }