Start dealing with the case where a file is partial but really complete.

This can happen if the the store crashes right a the transition from partial to complete

Author: rklaehn

proptest-regressions/protocol/range_spec.txt

@@ -6,3 +6,4 @@
 # everyone who runs the test benefits from these saved cases.
 cc 5ff4de8531a81c637b4d202c97b724a41a989bc6894464e84db5ac2a519c08a9 # shrinks to ranges = [RangeSet{1..2}]
 cc 50cb338763aa276705bb970d57d3d87e834f31a7e57bba810f46690c6d1e9955 # shrinks to ranges = [RangeSet{7..98}, RangeSet{7..98}]
+cc 8579821a8bde7872fed2cfa38e8a5923706b9915f3920e9c2d101a06bc789309 # shrinks to ranges = []

src/api/downloader.rs

@@ -23,7 +23,10 @@ use tracing::instrument::Instrument;
 
 use super::{Store, remote::GetConnection};
 use crate::{
-    protocol::{GetManyRequest, GetRequest}, util::sink::{Drain, IrpcSenderRefSink, Sink, TokioMpscSenderSink}, BlobFormat, Hash, HashAndFormat
+    BlobFormat, Hash, HashAndFormat,
+    get::request,
+    protocol::{GetManyRequest, GetRequest},
+    util::sink::{Drain, IrpcSenderRefSink, Sink, TokioMpscSenderSink},
 };
 
 #[derive(Debug, Clone)]
@@ -134,12 +137,16 @@ async fn handle_download_split_impl(
     request: DownloadRequest,
     tx: &mut spsc::Sender<DownloadProgessItem>,
 ) -> anyhow::Result<()> {
+    println!("Handling download split");
     let providers = request.providers;
     let requests = split_request(&request.request, &providers, &pool, &store, Drain).await?;
+    let requests = requests.collect::<Vec<_>>();
+    println!("Split into {:?} requests", requests);
     // todo: this is it's own mini actor, we should probably refactor this out
     let (progress_tx, progress_rx) = mpsc::channel(32);
     let mut futs = stream::iter(requests.into_iter().enumerate())
         .map(|(id, request)| {
+            println!("Spawning download fut for request {id} {request:?}");
             let pool = pool.clone();
             let providers = providers.clone();
             let store = store.clone();
@@ -178,7 +185,6 @@ async fn handle_download_split_impl(
                 tx.send(item).await?;
             },
             res = futs.next() => {
-                println!("Got result: {:?}", res);
                 match res {
                     Some((hash, Ok(()))) => {
                     }
@@ -245,7 +251,8 @@ impl SupportedRequest for HashAndFormat {
         (match self.format {
             BlobFormat::Raw => GetRequest::blob(self.hash),
             BlobFormat::HashSeq => GetRequest::all(self.hash),
-        }).into()
+        })
+        .into()
     }
 }
 
@@ -387,7 +394,7 @@ async fn split_request<'a>(
 ) -> anyhow::Result<Box<dyn Iterator<Item = GetRequest> + Send + 'a>> {
     Ok(match request {
         FiniteRequest::Get(req) => {
-            let Some(first) = req.ranges.iter().next() else {
+            let Some(first) = req.ranges.iter_infinite().next() else {
                 return Ok(Box::new(std::iter::empty()));
             };
             let first = GetRequest::blob(req.hash);
@@ -397,7 +404,7 @@ async fn split_request<'a>(
             let n = size / 32;
             Box::new(
                 req.ranges
-                    .iter()
+                    .iter_infinite()
                     .take(n as usize + 1)
                     .enumerate()
                     .filter_map(|(i, ranges)| {

src/store/fs.rs

@@ -320,12 +320,13 @@ impl HashContext {
                         self.ctx.options.clone(),
                     )),
                 };
-                res
+                Ok((res?, ()))
             })
             .await
             .map_err(api::Error::from);
         trace!("{res:?}");
-        res
+        let (res, _) = res?;
+        Ok(res)
     }
 }
 
@@ -384,19 +385,20 @@ impl Slot {
     /// If there is nothing in the database, create a new in-memory handle.
     ///
     /// `make` will be called if the a live handle does not exist.
-    pub async fn get_or_create<F, Fut>(&self, make: F) -> io::Result<BaoFileHandle>
+    pub async fn get_or_create<F, Fut, T>(&self, make: F) -> io::Result<(BaoFileHandle, T)>
     where
         F: FnOnce() -> Fut,
-        Fut: std::future::Future<Output = io::Result<BaoFileHandle>>,
+        Fut: std::future::Future<Output = io::Result<(BaoFileHandle, T)>>,
+        T: Default,
     {
         let mut slot = self.0.lock().await;
         if let Some(weak) = &*slot {
             if let Some(handle) = weak.upgrade() {
-                return Ok(handle);
+                return Ok((handle, Default::default()));
             }
         }
         let handle = make().await;
-        if let Ok(handle) = &handle {
+        if let Ok((handle, _)) = &handle {
             *slot = Some(handle.downgrade());
         }
         handle