This is an automated email from the ASF dual-hosted git repository.
alamb pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/arrow-rs-object-store.git
The following commit(s) were added to refs/heads/main by this push:
new 7cbbeaa Fix #818 - WriteMultipart::finish should abort after part
upload failure (#819)
7cbbeaa is described below
commit 7cbbeaacd17ceba9a625d7119b98276ddb2f5cfc
Author: Anson VanDoren <[email protected]>
AuthorDate: Mon Aug 17 13:38:06 2026 -0700
Fix #818 - WriteMultipart::finish should abort after part upload failure
(#819)
---
src/upload.rs | 44 +++++++++++++++++++++++++++++++++++++++++++-
1 file changed, 43 insertions(+), 1 deletion(-)
diff --git a/src/upload.rs b/src/upload.rs
index b5bc940..d7d50b1 100644
--- a/src/upload.rs
+++ b/src/upload.rs
@@ -229,7 +229,10 @@ impl WriteMultipart {
self.put_part(part.into())
}
- self.wait_for_capacity(0).await?;
+ if let Err(e) = self.wait_for_capacity(0).await {
+ self.abort().await?;
+ return Err(e);
+ }
match self.upload.complete().await {
Err(e) => {
@@ -303,6 +306,45 @@ mod tests {
}
}
+ #[tokio::test]
+ async fn test_finish_aborts_on_part_failure() {
+ #[derive(Debug)]
+ struct FailingUpload {
+ aborted: Arc<Mutex<bool>>,
+ }
+
+ #[async_trait]
+ impl MultipartUpload for FailingUpload {
+ fn put_part(&mut self, _data: PutPayload) -> UploadPart {
+ futures_util::future::ready(Err(crate::Error::Generic {
+ store: "Test",
+ source: std::io::Error::other("part upload failed").into(),
+ }))
+ .boxed()
+ }
+
+ async fn complete(&mut self) -> Result<PutResult> {
+ panic!("complete should not be called after a part failure")
+ }
+
+ async fn abort(&mut self) -> Result<()> {
+ *self.aborted.lock() = true;
+ Ok(())
+ }
+ }
+
+ let aborted = Arc::new(Mutex::new(false));
+ let upload = Box::new(FailingUpload {
+ aborted: Arc::clone(&aborted),
+ });
+ let mut write = WriteMultipart::new_with_chunk_size(upload, 1);
+ write.write(&[0]);
+
+ let err = write.finish().await.unwrap_err();
+ assert!(err.to_string().contains("part upload failed"));
+ assert!(*aborted.lock());
+ }
+
#[tokio::test]
async fn test_write_multipart() {
let mut rng = StdRng::seed_from_u64(42);
