Resume incomplete downloads

HIRO-MicroDataCenters-BV · Nov 16, 2024 · 8c52612 · 8c52612
1 parent 89a4a4f
commit 8c52612
Show file tree

Hide file tree

Showing 3 changed files with 27 additions and 13 deletions.
diff --git a/rhio/src/blobs/actor.rs b/rhio/src/blobs/actor.rs
@@ -27,7 +27,7 @@ pub enum ToBlobsActor {
     },
     DownloadBlob {
         hash: BlobHash,
-        bucket: ScopedBucket,
+        bucket_name: BucketName,
         key: ObjectKey,
         size: ObjectSize,
         reply: oneshot::Sender<Result<()>>,
@@ -118,12 +118,12 @@ impl BlobsActor {
             }
             ToBlobsActor::DownloadBlob {
                 hash,
-                bucket,
+                bucket_name,
                 key,
                 size,
                 reply,
             } => {
-                let result = self.on_download_blob(hash, bucket, key, size).await;
+                let result = self.on_download_blob(hash, bucket_name, key, size).await;
                 reply.send(result).ok();
             }
             ToBlobsActor::CompleteBlobs { reply } => {
@@ -145,12 +145,12 @@ impl BlobsActor {
     async fn on_download_blob(
         &mut self,
         hash: BlobHash,
-        bucket: ScopedBucket,
+        bucket_name: BucketName,
         key: ObjectKey,
         size: ObjectSize,
     ) -> Result<()> {
         self.store
-            .blob_discovered(hash, &bucket.bucket_name(), key.clone(), size)
+            .blob_discovered(hash, &bucket_name, key.clone(), size)
             .await?;
 
         let hash = Hash::from_bytes(*hash.as_bytes());
@@ -161,7 +161,7 @@ impl BlobsActor {
                     error!(%err, "failed downloading blob");
                 }
                 DownloadBlobEvent::Done => {
-                    debug!(%hash, bucket = %bucket.bucket_name(), %key, %size, "finished downloading blob");
+                    debug!(%hash, %bucket_name, %key, %size, "finished downloading blob");
                 }
             }
         }

diff --git a/rhio/src/blobs/mod.rs b/rhio/src/blobs/mod.rs
@@ -88,15 +88,15 @@ impl Blobs {
     pub async fn download(
         &self,
         hash: BlobHash,
-        bucket: ScopedBucket,
+        bucket_name: BucketName,
         key: ObjectKey,
         size: ObjectSize,
     ) -> Result<()> {
         let (reply, reply_rx) = oneshot::channel();
         self.blobs_actor_tx
             .send(ToBlobsActor::DownloadBlob {
                 hash,
-                bucket,
+                bucket_name,
                 key,
                 size,
                 reply,

diff --git a/rhio/src/node/actor.rs b/rhio/src/node/actor.rs
@@ -306,7 +306,9 @@ impl NodeActor {
         match network_message.payload {
             NetworkPayload::BlobAnnouncement(hash, bucket, key, size) => {
                 if is_bucket_matching(&self.subscriptions, &bucket) {
-                    self.blobs.download(hash, bucket, key, size).await?;
+                    self.blobs
+                        .download(hash, bucket.bucket_name(), key, size)
+                        .await?;
                 }
             }
             NetworkPayload::NatsMessage(message) => {
@@ -346,10 +348,9 @@ impl NodeActor {
                 self.on_import_finished(hash, bucket_name, key, size)
                     .await?;
             }
-            // We've detected an uncomplete blob, probably the process was exited before the
-            // download finished.
-            S3Event::DetectedIncompleteBlob(_hash, _bucket_name, _key, _size) => {
-                // @TODO: Resume download!
+            S3Event::DetectedIncompleteBlob(hash, bucket_name, key, size) => {
+                self.on_incomplete_blob_detected(hash, bucket_name, key, size)
+                    .await?;
             }
         }
 
@@ -396,6 +397,19 @@ impl NodeActor {
         Ok(())
     }
 
+    /// Handler when incomplete blob was detected, probably the process was exited before the
+    /// download hash finished.
+    async fn on_incomplete_blob_detected(
+        &self,
+        hash: BlobHash,
+        bucket_name: BucketName,
+        key: ObjectKey,
+        size: ObjectSize,
+    ) -> Result<()> {
+        self.blobs.download(hash, bucket_name, key, size).await?;
+        Ok(())
+    }
+
     async fn shutdown(&self) -> Result<()> {
         self.nats.shutdown().await?;
         self.panda.shutdown().await?;