fix: properly check local availability before downloading

Author: Frando

iroh-blobs/src/downloader.rs

@@ -51,10 +51,7 @@ use tokio_util::{sync::CancellationToken, time::delay_queue};
 use tracing::{debug, error_span, trace, warn, Instrument};
 
 use crate::{
-    get::{
-        db::{BlobId, DownloadProgress},
-        Stats,
-    },
+    get::{db::DownloadProgress, Stats},
     metrics::Metrics,
     store::Store,
     util::{local_pool::LocalPoolHandle, progress::ProgressSender},
@@ -119,7 +116,13 @@ pub trait Getter {
     ) -> GetFut;
 
     /// Checks if a blob is available in the local store.
-    fn has_complete(&mut self, kind: DownloadKind) -> impl Future<Output = Option<u64>>;
+    ///
+    /// If it exists and is fully complete, emit progress events as if we downloaded the blob.
+    fn check_local(
+        &mut self,
+        kind: DownloadKind,
+        progress: Option<ProgressSubscriber>,
+    ) -> impl Future<Output = anyhow::Result<bool>>;
 }
 
 /// Concurrency limits for the [`Downloader`].
@@ -675,18 +678,13 @@ impl<G: Getter<Connection = D::Connection>, D: Dialer> Service<G, D> {
             on_progress: progress,
         };
 
-        // early exit if the hash is already available.
-        if let Some(local_size) = self.getter.has_complete(kind).await {
-            intent_handlers
-                .on_progress
-                .send(DownloadProgress::FoundLocal {
-                    child: BlobId::Root,
-                    hash: kind.hash(),
-                    size: crate::store::BaoBlobSize::Verified(local_size),
-                    valid_ranges: crate::protocol::RangeSpec::all(),
-                })
-                .await
-                .ok();
+        // early exit if the blob/collection is already complete locally.
+        if self
+            .getter
+            .check_local(kind, intent_handlers.on_progress.clone())
+            .await
+            .unwrap_or(false)
+        {
             self.finalize_download(
                 kind,
                 [(intent_id, intent_handlers)].into(),

iroh-blobs/src/downloader/get.rs

@@ -3,8 +3,12 @@
 //! [`Connection`]: iroh_net::endpoint::Connection
 
 use crate::{
-    get::{db::get_to_db, error::GetError},
-    store::{MapEntry, Store},
+    downloader::progress::ProgressSubscriber,
+    get::{
+        db::{check_local_with_progress_if_complete, get_to_db},
+        error::GetError,
+    },
+    store::Store,
 };
 use futures_lite::FutureExt;
 #[cfg(feature = "metrics")]
@@ -82,12 +86,11 @@ impl<S: Store> Getter for IoGetter<S> {
         fut.boxed_local()
     }
 
-    async fn has_complete(&mut self, kind: DownloadKind) -> Option<u64> {
-        let entry = self.store.get(&kind.hash()).await.ok().flatten()?;
-        if entry.is_complete() {
-            Some(entry.size().value())
-        } else {
-            None
-        }
+    async fn check_local(
+        &mut self,
+        kind: DownloadKind,
+        progress: Option<ProgressSubscriber>,
+    ) -> anyhow::Result<bool> {
+        check_local_with_progress_if_complete(&self.store, &kind.hash_and_format(), progress).await
     }
 }

iroh-blobs/src/downloader/test/getter.rs

@@ -56,8 +56,12 @@ impl Getter for TestingGetter {
         .boxed_local()
     }
 
-    async fn has_complete(&mut self, _kind: DownloadKind) -> Option<u64> {
-        None
+    async fn check_local(
+        &mut self,
+        _kind: DownloadKind,
+        _progress: Option<ProgressSubscriber>,
+    ) -> anyhow::Result<bool> {
+        Ok(false)
     }
 }
 

iroh-blobs/src/get/db.rs

@@ -59,6 +59,73 @@ pub async fn get_to_db<
     }
 }
 
+/// Checks if a blob or collection exists fully in the local store.
+///
+/// Returns `true` if the blob is complete (and for hashseqs if all children are complete too).
+///
+/// Emits the same sequence of progress events as if we were downloading the blob, but only if the
+/// return value is `true`, i.e. if the blob (and all children for hashseqs) are fully available in
+/// the store. If the return value is `false`, no events will be emitted at all.
+pub async fn check_local_with_progress_if_complete<D: BaoStore>(
+    db: &D,
+    hash_and_format: &HashAndFormat,
+    progress: Option<impl ProgressSender<Msg = DownloadProgress> + IdGenerator>,
+) -> anyhow::Result<bool> {
+    // We collect all progress events that should be emitted if the blob/hashseq is complete.
+    // We do not emit them right away because we don't want to emit any events in case the
+    // blob/hashseq is not complete.
+    let mut progress_events = vec![];
+
+    // Check if the root blob is fully available.
+    let HashAndFormat { hash, format } = *hash_and_format;
+    let entry = match db.get(&hash).await? {
+        Some(entry) if entry.is_complete() => entry,
+        _ => return Ok(false),
+    };
+    progress_events.push(DownloadProgress::FoundLocal {
+        child: BlobId::Root,
+        hash,
+        size: entry.size(),
+        valid_ranges: RangeSpec::all(),
+    });
+
+    match format {
+        BlobFormat::Raw => {
+            // For a raw blob, we're done.
+        }
+        BlobFormat::HashSeq => {
+            // For a hashseq, check if all children are complete.
+            let reader = entry.data_reader().await?;
+            let (mut hash_seq, children) = parse_hash_seq(reader).await.map_err(|err| {
+                GetError::NoncompliantNode(anyhow!("Failed to parse downloaded HashSeq: {err}"))
+            })?;
+            progress_events.push(DownloadProgress::FoundHashSeq { hash, children });
+
+            let mut children: Vec<Hash> = vec![];
+            while let Some(hash) = hash_seq.next().await? {
+                children.push(hash);
+            }
+            for hash in children {
+                let entry = match db.get(&hash).await? {
+                    Some(entry) if entry.is_complete() => entry,
+                    _ => return Ok(false),
+                };
+                progress_events.push(DownloadProgress::FoundLocal {
+                    child: BlobId::Root,
+                    hash,
+                    size: entry.size(),
+                    valid_ranges: RangeSpec::all(),
+                });
+            }
+        }
+    }
+
+    for event in progress_events {
+        progress.send(event).await?;
+    }
+    Ok(true)
+}
+
 /// Get a blob that was requested completely.
 ///
 /// We need to create our own files and handle the case where an outboard

iroh/src/node/rpc.rs

@@ -9,7 +9,6 @@ use futures_lite::{Stream, StreamExt};
 use genawaiter::sync::{Co, Gen};
 use iroh_base::rpc::{RpcError, RpcResult};
 use iroh_blobs::downloader::{DownloadRequest, Downloader};
-use iroh_blobs::export::ExportProgress;
 use iroh_blobs::format::collection::Collection;
 use iroh_blobs::get::db::DownloadProgress;
 use iroh_blobs::get::Stats;
@@ -18,6 +17,7 @@ use iroh_blobs::util::local_pool::LocalPoolHandle;
 use iroh_blobs::util::progress::{AsyncChannelProgressSender, ProgressSender};
 use iroh_blobs::util::SetTagOption;
 use iroh_blobs::BlobFormat;
+use iroh_blobs::{export::ExportProgress, get::db::check_local_with_progress_if_complete};
 use iroh_blobs::{
     provider::AddProgress,
     store::{Store as BaoStore, ValidateProgress},
@@ -1201,6 +1201,9 @@ async fn download_direct_from_nodes<D>(
 where
     D: BaoStore,
 {
+    if check_local_with_progress_if_complete(db, &hash_and_format, Some(progress.clone())).await? {
+        return Ok(Default::default());
+    }
     let mut last_err = None;
     for node in nodes {
         let node_id = node.node_id;
@@ -1226,7 +1229,7 @@ where
     }
     match last_err {
         Some(err) => Err(err),
-        None => Err(anyhow!("No nodes to download from provided"))
+        None => Err(anyhow!("No nodes to download from provided")),
     }
 }