fud: Implement background tasks for fetching file metadata and file chunks.

bin/fud/fud/src/main.rs

@@ -16,15 +16,19 @@
  * along with this program.  If not, see <https://www.gnu.org/licenses/>.
  */
 
-use std::collections::{HashMap, HashSet};
+use std::{
+    collections::{HashMap, HashSet},
+    ffi::OsString,
+};
 
 use async_std::{
+    channel,
     fs::File,
     stream::StreamExt,
     sync::{Arc, RwLock},
 };
 use async_trait::async_trait;
-use log::{error, info};
+use log::{debug, error, info, warn};
 use serde_json::{json, Value};
 use smol::Executor;
 use structopt_toml::{structopt::StructOpt, StructOptToml};
@@ -33,18 +37,23 @@ use url::Url;
 use darkfi::{
     async_daemonize, cli_desc,
     geode::Geode,
-    net::{self, settings::SettingsOpt, P2p, P2pPtr},
+    net::{
+        self, connector::Connector, protocol::ProtocolVersion, session::Session,
+        settings::SettingsOpt, P2p, P2pPtr,
+    },
     rpc::{
         jsonrpc::{ErrorCode, JsonError, JsonRequest, JsonResponse, JsonResult},
         server::{listen_and_serve, RequestHandler},
     },
     util::path::expand_path,
-    Result,
+    Error, Result,
 };
 
 /// P2P protocols
 mod proto;
-use proto::{FudFilePut, ProtocolFud};
+use proto::{
+    FudChunkReply, FudChunkRequest, FudFilePut, FudFileReply, FudFileRequest, ProtocolFud,
+};
 
 const CONFIG_FILE: &str = "fud_config.toml";
 const CONFIG_FILE_CONTENTS: &str = include_str!("../fud_config.toml");
@@ -87,6 +96,11 @@ pub struct Fud {
     p2p: P2pPtr,
     /// The Geode instance
     geode: Geode,
+
+    file_fetch_tx: channel::Sender<(blake3::Hash, Result<()>)>,
+    file_fetch_rx: channel::Receiver<(blake3::Hash, Result<()>)>,
+    chunk_fetch_tx: channel::Sender<(blake3::Hash, Result<()>)>,
+    chunk_fetch_rx: channel::Receiver<(blake3::Hash, Result<()>)>,
 }
 
 #[async_trait]
@@ -113,7 +127,7 @@ impl RequestHandler for Fud {
 impl Fud {
     // RPCAPI:
     // Put a file onto the network. Takes a local filesystem path as a parameter.
-    // Returns the fil hashe that serves as a pointer to the uploaded file.
+    // Returns the file hash that serves as a pointer to the uploaded file.
     //
     // --> {"jsonrpc": "2.0", "method": "put", "params": ["/foo.txt"], "id": 42}
     // <-- {"jsonrpc": "2.0", "result: "df4...3db7", "id": 42}
@@ -141,8 +155,7 @@ impl Fud {
             Ok(v) => v,
             Err(e) => {
                 error!("Failed inserting file {:?} to geode: {}", path, e);
-                // FIXME: Custom error here
-                return JsonError::new(ErrorCode::InvalidParams, None, id).into()
+                return JsonError::new(ErrorCode::InternalError, None, id).into()
             }
         };
 
@@ -154,12 +167,67 @@ impl Fud {
 
     // RPCAPI:
     // Fetch a file from the network. Takes a file hash as parameter.
-    // Returns the path to the local file containing the metadata.
+    // Returns the paths to the local chunks of the file, if found/fetched.
     //
     // --> {"jsonrpc": "2.0", "method": "get", "params": ["1211...abfd"], "id": 42}
-    // <-- {"jsonrpc": "2.0", "result: "~/.local/share/fud/files/1211...abfd", "id": 42}
-    async fn get(&self, id: Value, _params: &[Value]) -> JsonResult {
-        JsonResponse::new(json!([]), id).into()
+    // <-- {"jsonrpc": "2.0", "result: ["~/.local/share/fud/chunks/fab1...2314", ...], "id": 42}
+    async fn get(&self, id: Value, params: &[Value]) -> JsonResult {
+        if params.len() != 1 || !params[0].is_string() {
+            return JsonError::new(ErrorCode::InvalidParams, None, id).into()
+        }
+
+        let file_hash = match blake3::Hash::from_hex(params[0].as_str().unwrap()) {
+            Ok(v) => v,
+            Err(_) => return JsonError::new(ErrorCode::InvalidParams, None, id).into(),
+        };
+
+        let chunked_file = match self.geode.get(&file_hash).await {
+            Ok(v) => v,
+            Err(e) => todo!(), // fetch file
+        };
+
+        if chunked_file.is_complete() {
+            let chunks: Vec<OsString> = chunked_file
+                .iter()
+                .map(|(_, path)| path.as_ref().unwrap().clone().into_os_string())
+                .collect();
+
+            return JsonResponse::new(json!(chunks), id).into()
+        }
+
+        // Fetch any missing chunks
+        let mut missing_chunks = vec![];
+        for (chunk, path) in chunked_file.iter() {
+            if path.is_none() {
+                missing_chunks.push(*chunk);
+            }
+        }
+
+        for chunk in missing_chunks {
+            self.chunk_fetch_tx.send((chunk, Ok(()))).await.unwrap();
+            let (_, result) = self.chunk_fetch_rx.recv().await.unwrap();
+            match result {
+                Ok(()) => continue,
+                Err(e) => todo!(),
+            }
+        }
+
+        let chunked_file = match self.geode.get(&file_hash).await {
+            Ok(v) => v,
+            Err(e) => todo!(),
+        };
+
+        if !chunked_file.is_complete() {
+            todo!();
+            // Return JsonError missing chunks
+        }
+
+        let chunks: Vec<OsString> = chunked_file
+            .iter()
+            .map(|(_, path)| path.as_ref().unwrap().clone().into_os_string())
+            .collect();
+
+        JsonResponse::new(json!(chunks), id).into()
     }
 
     // RPCAPI:
@@ -202,6 +270,226 @@ impl Fud {
     }
 }
 
+/// Background task that receives file fetch requests and tries to
+/// fetch objects from the network using the routing table.
+/// TODO: This can be optimised a lot for connection reuse, etc.
+async fn fetch_file_task(fud: Arc<Fud>, executor: Arc<Executor<'_>>) {
+    info!("Started background file fetch task");
+    loop {
+        let (file_hash, _) = fud.file_fetch_rx.recv().await.unwrap();
+        info!("fetch_file_task: Received {}", file_hash);
+
+        let mut metadata_router = fud.metadata_router.write().await;
+        let peers = metadata_router.get_mut(&file_hash);
+
+        if peers.is_none() {
+            warn!("File {} not in routing table, cannot fetch", file_hash);
+            fud.file_fetch_tx.send((file_hash, Err(Error::GeodeFileRouteNotFound))).await.unwrap();
+            continue
+        }
+
+        let mut found = false;
+        let peers = peers.unwrap();
+        let mut invalid_file_routes = vec![];
+
+        for peer in peers.iter() {
+            let session_out = fud.p2p.session_outbound().await;
+            let session_weak = Arc::downgrade(&fud.p2p.session_outbound().await);
+
+            info!("Connecting to {} to fetch {}", peer, file_hash);
+            let connector = Connector::new(fud.p2p.settings(), Arc::new(session_weak));
+            match connector.connect(peer).await {
+                Ok((url, channel)) => {
+                    let proto_ver = ProtocolVersion::new(
+                        channel.clone(),
+                        fud.p2p.settings().clone(),
+                        fud.p2p.hosts().clone(),
+                    )
+                    .await;
+
+                    let handshake_task = session_out.perform_handshake_protocols(
+                        proto_ver,
+                        channel.clone(),
+                        executor.clone(),
+                    );
+
+                    channel.clone().start(executor.clone());
+
+                    if let Err(e) = handshake_task.await {
+                        error!("Handshake with {} failed: {}", url, e);
+                        // Delete peer from router
+                        invalid_file_routes.push(peer.clone());
+                        continue
+                    }
+
+                    let msg_subscriber = channel.subscribe_msg::<FudFileReply>().await.unwrap();
+                    let request = FudFileRequest { file_hash };
+
+                    if let Err(e) = channel.send(&request).await {
+                        error!("Failed sending FudFileRequest({}) to {}: {}", file_hash, url, e);
+                        continue
+                    }
+
+                    // TODO: With timeout!
+                    let reply = match msg_subscriber.receive().await {
+                        Ok(v) => v,
+                        Err(e) => {
+                            error!("Error receiving FudFileReply from subscriber: {}", e);
+                            continue
+                        }
+                    };
+
+                    msg_subscriber.unsubscribe().await;
+                    channel.stop().await;
+
+                    if let Err(e) = fud.geode.insert_file(&file_hash, &reply.chunk_hashes).await {
+                        error!("Failed inserting file {} to Geode: {}", file_hash, e);
+                        continue
+                    }
+
+                    found = true;
+                    break
+                }
+
+                Err(e) => {
+                    error!("Failed to connect to {}: {}", peer, e);
+                    continue
+                }
+            }
+        }
+
+        for peer in invalid_file_routes {
+            debug!("Removing peer {} from {} file router", peer, file_hash);
+            peers.remove(&peer);
+        }
+
+        if !found {
+            warn!("Did not manage to fetch {} file metadata", file_hash);
+            fud.file_fetch_tx.send((file_hash, Err(Error::GeodeFileRouteNotFound))).await.unwrap();
+            continue
+        }
+
+        info!("Successfully fetched {} file metadata", file_hash);
+        fud.file_fetch_tx.send((file_hash, Ok(()))).await.unwrap();
+    }
+}
+
+/// Background task that receives chunk fetch requests and tries to
+/// fetch objects from the network using the routing table.
+/// TODO: This can be optimised a lot for connection reuse, etc.
+async fn fetch_chunk_task(fud: Arc<Fud>, executor: Arc<Executor<'_>>) {
+    info!("Started background chunk fetch task");
+    loop {
+        let (chunk_hash, _) = fud.chunk_fetch_rx.recv().await.unwrap();
+        info!("fetch_chunk_task: Received {}", chunk_hash);
+
+        let mut chunk_router = fud.chunks_router.write().await;
+        let peers = chunk_router.get_mut(&chunk_hash);
+
+        if peers.is_none() {
+            warn!("Chunk {} not in routing table, cannot fetch", chunk_hash);
+            fud.chunk_fetch_tx
+                .send((chunk_hash, Err(Error::GeodeChunkRouteNotFound)))
+                .await
+                .unwrap();
+            continue
+        }
+
+        let mut found = false;
+        let peers = peers.unwrap();
+        let mut invalid_chunk_routes = vec![];
+
+        for peer in peers.iter() {
+            let session_out = fud.p2p.session_outbound().await;
+            let session_weak = Arc::downgrade(&fud.p2p.session_outbound().await);
+
+            info!("Connecting to {} to fetch {}", peer, chunk_hash);
+            let connector = Connector::new(fud.p2p.settings(), Arc::new(session_weak));
+            match connector.connect(peer).await {
+                Ok((url, channel)) => {
+                    let proto_ver = ProtocolVersion::new(
+                        channel.clone(),
+                        fud.p2p.settings().clone(),
+                        fud.p2p.hosts().clone(),
+                    )
+                    .await;
+
+                    let handshake_task = session_out.perform_handshake_protocols(
+                        proto_ver,
+                        channel.clone(),
+                        executor.clone(),
+                    );
+
+                    channel.clone().start(executor.clone());
+
+                    if let Err(e) = handshake_task.await {
+                        error!("Handshake with {} failed: {}", url, e);
+                        // Delete peer from router
+                        invalid_chunk_routes.push(peer.clone());
+                        continue
+                    }
+
+                    let msg_subscriber = channel.subscribe_msg::<FudChunkReply>().await.unwrap();
+                    let request = FudChunkRequest { chunk_hash };
+
+                    if let Err(e) = channel.send(&request).await {
+                        error!("Failed sending FudChunkRequest({}) to {}: {}", chunk_hash, url, e);
+                        continue
+                    }
+
+                    // TODO: With timeout!
+                    let reply = match msg_subscriber.receive().await {
+                        Ok(v) => v,
+                        Err(e) => {
+                            error!("Error receiving FudChunkReply from subscriber: {}", e);
+                            continue
+                        }
+                    };
+
+                    msg_subscriber.unsubscribe().await;
+                    channel.stop().await;
+
+                    match fud.geode.insert_chunk(&reply.chunk).await {
+                        Ok(inserted_hash) => {
+                            if inserted_hash != chunk_hash {
+                                warn!("Received chunk does not match requested chunk");
+                                invalid_chunk_routes.push(peer.clone());
+                                continue
+                            }
+                        }
+                        Err(e) => {
+                            error!("Failed inserting chunk {} to Geode: {}", chunk_hash, e);
+                            continue
+                        }
+                    }
+
+                    found = true;
+                    break
+                }
+
+                Err(e) => {
+                    error!("Failed to connect to {}: {}", peer, e);
+                    continue
+                }
+            }
+        }
+
+        for peer in invalid_chunk_routes {
+            debug!("Removing peer {} from {} chunk router", peer, chunk_hash);
+            peers.remove(&peer);
+        }
+
+        if !found {
+            warn!("Did not manage to fetch {} chunk", chunk_hash);
+            fud.chunk_fetch_tx.send((chunk_hash, Err(Error::GeodeChunkNotFound))).await.unwrap();
+            continue
+        }
+
+        info!("Successfully fetched {} chunk", chunk_hash);
+        fud.chunk_fetch_tx.send((chunk_hash, Ok(()))).await.unwrap();
+    }
+}
+
 async_daemonize!(realmain);
 async fn realmain(args: Args, ex: Arc<Executor<'_>>) -> Result<()> {
     // The working directory for this daemon and geode.
@@ -218,8 +506,21 @@ async fn realmain(args: Args, ex: Arc<Executor<'_>>) -> Result<()> {
     let p2p = P2p::new(args.net.into()).await;
 
     // Daemon instantiation
-    let fud = Arc::new(Fud { metadata_router, chunks_router, p2p: p2p.clone(), geode });
+    let (file_fetch_tx, file_fetch_rx) = smol::channel::unbounded();
+    let (chunk_fetch_tx, chunk_fetch_rx) = smol::channel::unbounded();
+    let fud = Arc::new(Fud {
+        metadata_router,
+        chunks_router,
+        p2p: p2p.clone(),
+        geode,
+        file_fetch_tx,
+        file_fetch_rx,
+        chunk_fetch_tx,
+        chunk_fetch_rx,
+    });
     let _fud = fud.clone();
+    ex.spawn(fetch_file_task(fud.clone(), ex.clone())).detach();
+    ex.spawn(fetch_chunk_task(fud.clone(), ex.clone())).detach();
 
     info!("Starting JSON-RPC server on {}", args.rpc_listen);
     let _ex = ex.clone();

src/error.rs

@@ -472,6 +472,12 @@ pub enum Error {
     #[error("Geode chunk not found")]
     GeodeChunkNotFound,
 
+    #[error("Geode file route not found")]
+    GeodeFileRouteNotFound,
+
+    #[error("Geode chunk route not found")]
+    GeodeChunkRouteNotFound,
+
     // =========
     // Catch-all
     // =========

src/geode/mod.rs

@@ -326,9 +326,29 @@ impl Geode {
         Ok((file_hash, chunk_hashes))
     }
 
+    /// Create and insert file metadata into Geode given a list of hashes.
+    /// Always overwrites any existing file.
+    pub async fn insert_file(
+        &self,
+        file_hash: &blake3::Hash,
+        chunk_hashes: &[blake3::Hash],
+    ) -> Result<()> {
+        info!(target: "geode::insert_file()", "[Geode] Inserting file metadata");
+
+        let mut file_path = self.files_path.clone();
+        file_path.push(file_hash.to_hex().as_str());
+        let mut file_fd = File::create(&file_path).await?;
+
+        for ch in chunk_hashes {
+            file_fd.write(format!("{}\n", ch.to_hex().as_str()).as_bytes()).await?;
+        }
+
+        Ok(())
+    }
+
     /// Create and insert a single chunk into Geode given a stream.
     /// Always overwrites any existing chunk. Returns the chunk hash once inserted.
-    pub async fn insert_chunk(&mut self, stream: impl AsRef<[u8]>) -> Result<blake3::Hash> {
+    pub async fn insert_chunk(&self, stream: impl AsRef<[u8]>) -> Result<blake3::Hash> {
         info!(target: "geode::insert_chunk()", "[Geode] Inserting single chunk");
 
         let mut cursor = Cursor::new(&stream);