Fix internal failure on atomic upload (#1733)

Improve handling of errors on `CreateMultiPartUpload` in the atomic
upload code path. Similarly to the change in #1728, the issue only
manifests when attempting to further write or complete an upload after
an error and it does not affect Mountpoint file system users, since
that's already prevented at that level.

### Does this change impact existing behavior?

No, user-visible behavior not impacted.

### Does this change need a changelog entry? Does it require a version
change?

No.

---

By submitting this pull request, I confirm that my contribution is made
under the terms of the Apache 2.0 license and I agree to the terms of
the [Developer Certificate of Origin
(DCO)](https://developercertificate.org/).

Signed-off-by: Alessandro Passaro <[email protected]>

Author: passaro

mountpoint-s3-client/src/failure_client.rs

@@ -435,8 +435,10 @@ pub fn countdown_failure_client<Client: ObjectClient>(
                 },
                 result_fn: |state| {
                     state.count += 1;
-                    if state.count >= state.fail_count {
-                        Err(state.error.take().unwrap())
+                    if state.count >= state.fail_count
+                        && let Some(error) = state.error.take()
+                    {
+                        Err(error)
                     } else {
                         Ok(())
                     }

mountpoint-s3-fs/src/upload/atomic.rs

@@ -109,7 +109,12 @@ where
         }
 
         self.hasher.update(data);
-        self.request.get_mut().await?.unwrap().write(data).await?;
+        self.request
+            .get_mut()
+            .await?
+            .ok_or(UploadError::UploadAlreadyTerminated)?
+            .write(data)
+            .await?;
 
         self.next_request_offset += data.len() as u64;
         Ok(data.len())
@@ -122,7 +127,7 @@ where
             .request
             .into_inner()
             .await?
-            .unwrap()
+            .ok_or(UploadError::UploadAlreadyTerminated)?
             .review_and_complete(move |review| verify_checksums(review, size, checksum))
             .await?;
         if let Err(err) = self
@@ -306,6 +311,7 @@ mod tests {
         let mut put_failures = HashMap::new();
         put_failures.insert(1, Ok((1, MockClientError("error".to_owned().into()))));
         put_failures.insert(2, Ok((2, MockClientError("error".to_owned().into()))));
+        put_failures.insert(3, Err(MockClientError("error".to_owned().into()).into()));
 
         let failure_client = Arc::new(countdown_failure_client(
             client.clone(),
@@ -338,6 +344,25 @@ mod tests {
         }
         assert!(!client.is_upload_in_progress(key));
         assert!(!client.contains_key(key));
+
+        // Third request fails on first write (because CreateMPU returns an error).
+        {
+            let mut request = uploader.start_atomic_upload(bucket.to_owned(), key.to_owned()).unwrap();
+
+            let data = b"foo";
+            request.write(0, data).await.expect_err("first write should fail");
+
+            let err = request
+                .write(0, data)
+                .await
+                .expect_err("subsequent writes should also fail");
+            assert!(matches!(err, UploadError::UploadAlreadyTerminated));
+
+            let err = request.complete().await.expect_err("complete should also fail");
+            assert!(matches!(err, UploadError::UploadAlreadyTerminated));
+        }
+        assert!(!client.is_upload_in_progress(key));
+        assert!(!client.contains_key(key));
     }
 
     #[test_case(8000; "divisible by max size")]