update

izihawa · Apr 16, 2024 · c415ec6 · c415ec6
1 parent ea90a36
commit c415ec6
Show file tree

Hide file tree

Showing 3 changed files with 51 additions and 68 deletions.
diff --git a/Cargo.lock b/Cargo.lock
diff --git a/bin/main.rs b/bin/main.rs
@@ -245,15 +245,9 @@ async fn blobs_get(
         return Response::builder()
             .status(StatusCode::NOT_FOUND)
             .body(Body::default())
-            .unwrap()
+            .unwrap();
     };
-    match state
-        .iroh_node
-        .read()
-        .await
-        .blobs_get(hash)
-        .await
-    {
+    match state.iroh_node.read().await.blobs_get(hash).await {
         Ok(Some((reader, file_size))) => match method {
             Method::HEAD => Response::builder()
                 .header("Content-Length", file_size)

diff --git a/src/table.rs b/src/table.rs
@@ -13,7 +13,7 @@ use iroh::client::{Entry, LiveEvent};
 use iroh::net::key::PublicKey;
 use iroh::net::NodeAddr;
 use iroh::node::Node;
-use iroh::rpc_protocol::{BlobDownloadRequest, SetTagOption, ShareMode};
+use iroh::rpc_protocol::{BlobDownloadRequest, DownloadMode, SetTagOption, ShareMode};
 use iroh::sync::store::{DownloadPolicy, Query, SortBy, SortDirection};
 use iroh::sync::{AuthorId, ContentStatus, PeerIdBytes};
 use iroh::ticket::DocTicket;
@@ -310,55 +310,34 @@ impl Table {
 
     async fn download_entry_from_peers(&self, entry: &Entry, peers: &[PeerIdBytes]) -> Result<()> {
         let key = std::str::from_utf8(entry.key()).map_err(Error::incorrect_key)?;
-        for peer in peers {
-            match PublicKey::from_bytes(peer) {
-                Ok(public_key) => {
-                    let progress = match self
-                        .node
-                        .blobs
-                        .download(BlobDownloadRequest {
-                            hash: entry.content_hash(),
-                            format: BlobFormat::Raw,
-                            peer: NodeAddr::new(public_key),
-                            tag: SetTagOption::Auto,
-                        })
-                        .await
-                        .map_err(Error::io_error)
-                    {
-                        Ok(progress) => progress,
-                        Err(error) => {
-                            warn!(error = ?error);
-                            continue;
-                        }
-                    };
-                    match progress.finish().await {
-                        Ok(import_result) => {
-                            info!(
-                                "found local_size {}, downloaded_size {}, content_len {}",
-                                import_result.local_size,
-                                import_result.downloaded_size,
-                                entry.content_len()
-                            );
-                            if import_result.local_size + import_result.downloaded_size
-                                == entry.content_len()
-                            {
-                                self.process_remote_entry(key, entry).await?;
-                                return Ok(());
-                            }
-                        }
-                        Err(error) => {
-                            warn!(error = ?error);
-                            continue;
-                        }
-                    }
-                }
-                Err(_signing_error) => {
-                    warn!("potential db corruption: peers per doc can't be decoded");
-                    continue;
-                }
-            };
-        }
-        Err(Error::failed_download(key))
+        let progress = self
+            .node
+            .blobs
+            .download(BlobDownloadRequest {
+                hash: entry.content_hash(),
+                format: BlobFormat::Raw,
+                nodes: peers
+                    .iter()
+                    .filter_map(|peer| {
+                        PublicKey::from_bytes(peer)
+                            .map(NodeAddr::from)
+                            .ok()
+                    })
+                    .collect(),
+                tag: SetTagOption::Auto,
+                mode: DownloadMode::Direct,
+            })
+            .await
+            .map_err(Error::io_error)?;
+        let import_result = progress.finish().await.map_err(Error::failed_download)?;
+        info!(
+            "found local_size {}, downloaded_size {}, content_len {}",
+            import_result.local_size,
+            import_result.downloaded_size,
+            entry.content_len()
+        );
+        self.process_remote_entry(key, entry).await?;
+        Ok(())
     }
 
     pub async fn download_missing(
@@ -407,7 +386,7 @@ impl Table {
                     })
                     .await
                     .map_err(Error::io_error)?;
-                let _ = self.task_tracker.track_future(join_handle);
+                tokio::spawn(self.task_tracker.track_future(join_handle));
             }
         }
         Ok(())