fix: remove partial blob output

Author: St4NNi

blob/src/bao_tree.rs

@@ -151,6 +151,8 @@ impl AsyncStreamReader for RecvStreamWrapper<'_> {
 // ----- FuturesAsyncReader/opendal Writer impls for bao_tree ----------
 pub struct OpenDalWriter {
     writer: opendal::Writer,
+    operator: Operator,
+    storage_path: String,
     pub hasher: Hasher,
     idle_timeout: Duration,
 }
@@ -208,6 +210,8 @@ impl OpenDalWriter {
     ) -> Result<Self, BlobLibError> {
         Ok(Self {
             writer: operator.writer(storage_path).await?,
+            operator: operator.clone(),
+            storage_path: storage_path.to_string(),
             hasher: Hasher::new(),
             idle_timeout,
         })
@@ -227,14 +231,14 @@ impl OpenDalWriter {
         )
         .await;
         if let Err(err) = close_result {
-            abort_partial_writer(&mut self.writer).await;
+            abort_partial_writer(&mut self.writer, &self.operator, &self.storage_path).await;
             return Err(err);
         }
         Ok(())
     }
 
     pub async fn abort(mut self) {
-        abort_partial_writer(&mut self.writer).await;
+        abort_partial_writer(&mut self.writer, &self.operator, &self.storage_path).await;
     }
 }
 

blob/src/blob/io.rs

@@ -39,20 +39,20 @@ impl BlobHandler {
             let bytes = match chunk {
                 Ok(bytes) => bytes,
                 Err(err) => {
-                    abort_partial_writer(&mut writer).await;
+                    abort_partial_writer(&mut writer, &operator, &storage_path).await;
                     return BlobEvent::Error(BlobError::WriteError(err.to_string()));
                 }
             };
             hasher.update(&bytes);
             if let Err(err) = writer.write(bytes.to_vec()).await {
-                abort_partial_writer(&mut writer).await;
+                abort_partial_writer(&mut writer, &operator, &storage_path).await;
                 return BlobEvent::Error(BlobError::WriteError(err.to_string()));
             }
             bytes_written += bytes.len() as u64;
         }
 
         if let Err(err) = writer.close().await {
-            abort_partial_writer(&mut writer).await;
+            abort_partial_writer(&mut writer, &operator, &storage_path).await;
             return BlobEvent::Error(BlobError::WriteError(err.to_string()));
         }
         location.blob_size = bytes_written;
@@ -247,7 +247,7 @@ impl BlobHandler {
         let bytes_written = match compose_result {
             Ok(bytes_written) => bytes_written,
             Err(err) => {
-                abort_partial_writer(&mut writer).await;
+                abort_partial_writer(&mut writer, &operator, &storage_path).await;
                 return BlobEvent::Error(err);
             }
         };

blob/src/blob/tests.rs

@@ -666,6 +666,50 @@ async fn write_finalization_failure_emits_no_success_or_load() {
     assert_eq!(aborts.load(std::sync::atomic::Ordering::SeqCst), 1);
 }
 
+#[tokio::test]
+async fn failed_write_cleans() {
+    let context = setup_blob_handle(1).await;
+    let root = tempdir().unwrap();
+    let location = BackendLocation {
+        root: root.path().to_str().unwrap().to_string(),
+        storage_bucket: "bucket".to_string(),
+        backend_path: "partial.bin".to_string(),
+        ulid: Ulid::new(),
+        compressed: false,
+        encrypted: false,
+        created_by: test_user_id(),
+        created_at: SystemTime::now(),
+        staging: false,
+        partial: false,
+        blob_size: 0,
+        hashes: HashMap::new(),
+    };
+    let operator = crate::opendal::init_operator(
+        Backend::FileSystem,
+        HashMap::from([(
+            "root".to_string(),
+            root.path().to_str().unwrap().to_string(),
+        )]),
+    )
+    .unwrap();
+    let blob = BackendStream::new(futures::stream::iter([
+        Ok(bytes::Bytes::from_static(b"partial")),
+        Err(std::io::Error::other("injected stream failure")),
+    ]));
+
+    let event = context
+        .blob_handle
+        .handler
+        .write_stream_to_location(location.clone(), operator, blob)
+        .await;
+
+    assert!(matches!(event, BlobEvent::Error(BlobError::WriteError(_))));
+    assert!(
+        !std::path::Path::new(&location.get_full_path().unwrap()).exists(),
+        "partial filesystem target remains"
+    );
+}
+
 #[tokio::test]
 async fn compose_close_fails() {
     let context = setup_blob_handle(5).await;

blob/src/opendal.rs

@@ -8,9 +8,16 @@ use opendal::layers::{LoggingLayer, RetryLayer};
 use opendal::{Builder, Operator, services};
 use std::collections::HashMap;
 
-pub(crate) async fn abort_partial_writer(writer: &mut opendal::Writer) {
+pub(crate) async fn abort_partial_writer(
+    writer: &mut opendal::Writer,
+    operator: &Operator,
+    storage_path: &str,
+) {
     if let Err(err) = writer.abort().await {
-        tracing::warn!(error = %err, "failed to abort partial blob writer");
+        tracing::warn!(error = %err, "failed to abort partial blob writer; deleting output");
+        if let Err(delete_err) = operator.delete(storage_path).await {
+            tracing::warn!(error = %delete_err, "failed to delete partial blob output");
+        }
     }
 }