diff --git a/src/iroh.rs b/src/iroh.rs index a179b68..1b2366b 100644 --- a/src/iroh.rs +++ b/src/iroh.rs @@ -2,7 +2,6 @@ use std::{ collections::{BTreeMap, BTreeSet}, - future::IntoFuture, path::Path, sync::{Arc, Mutex}, }; @@ -12,6 +11,7 @@ use anyhow::Context; use futures_util::{StreamExt, stream}; use iroh_blobs::{ BlobsProtocol, Hash, + api::remote::{GetProgress, GetProgressItem}, protocol::GetManyRequest, provider::events::{EventMask, EventSender, ProviderMessage, RequestMode}, store::fs::FsStore, @@ -38,7 +38,7 @@ const MAX_CONTROL_MESSAGE_BYTES: usize = 16 * 1024 * 1024; const MAX_CONCURRENT_ANNOUNCEMENTS: usize = 4; const ANNOUNCEMENT_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(5); pub(super) const CONTROL_REQUEST_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(10); -const BLOB_OPERATION_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(5 * 60); +const BLOB_TRANSFER_IDLE_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(5 * 60); const ONLINE_WAIT_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(30); const BLOB_ACTIVITY_EVENT_BUFFER: usize = 32; @@ -193,11 +193,7 @@ impl NodeHost { } pub async fn import_file(&self, file_path: &Path) -> anyhow::Result { - Ok( - wait_for_blob_operation("import", self.store.blobs().add_path(file_path)) - .await? - .hash, - ) + Ok(self.store.blobs().add_path(file_path).await?.hash) } pub async fn download_blob( @@ -218,17 +214,12 @@ impl NodeHost { } let connection = self.connect_to_blob_provider(provider).await?; let request = blob_hashes.into_iter().collect::(); - wait_for_blob_operation( - "batch download", - self.store.remote().execute_get_many(connection, request), - ) - .await?; + wait_for_blob_download(self.store.remote().execute_get_many(connection, request)).await?; Ok(()) } pub async fn export_blob(&self, blob_hash: Hash, destination: &Path) -> anyhow::Result<()> { - wait_for_blob_operation("export", self.store.blobs().export(blob_hash, destination)) - .await?; + self.store.blobs().export(blob_hash, destination).await?; Ok(()) } @@ -479,15 +470,21 @@ impl NodeHost { } } -async fn wait_for_blob_operation(operation: &str, future: F) -> anyhow::Result -where - F: IntoFuture>, - E: Into, -{ - tokio::time::timeout(BLOB_OPERATION_TIMEOUT, future) - .await - .map_err(|_| anyhow::anyhow!("blob {operation} timed out"))? - .map_err(Into::into) +async fn wait_for_blob_download(progress: GetProgress) -> anyhow::Result<()> { + let updates = progress.stream(); + tokio::pin!(updates); + + loop { + let update = tokio::time::timeout(BLOB_TRANSFER_IDLE_TIMEOUT, updates.next()) + .await + .map_err(|_| anyhow::anyhow!("blob download made no progress for 5 minutes"))? + .ok_or_else(|| anyhow::anyhow!("blob download ended before completing"))?; + match update { + GetProgressItem::Progress(_) => {} + GetProgressItem::Done(_) => return Ok(()), + GetProgressItem::Error(error) => return Err(error.into()), + } + } } fn verify_connected_peer(