fix: abort scoutfs multipart uploads on error after successful moveblocks

The scoutfs backend uses the move blocks ioctl when combining parts into the
final multipart upload object. Once a move blocks from any part is successful,
the original data is no longer in the part file. If the multipart upload fails
and retries, future complete multipart upload calls will not have the correct
data within the part files anymore.

To prevent this case, once a move blocks call is successful for an upload, any
future failure for the complete upload is set to auto-abort the upload to force
clients to re-upload the part data again.

backend/posix/posix.go

@@ -1623,13 +1623,20 @@ type onlyRead struct {
 	io.Reader
 }
 
-func (p *Posix) CompleteMultipartUploadWithCopy(ctx context.Context, input *s3.CompleteMultipartUploadInput, customMove func(from *os.File, to *os.File) error) (s3response.CompleteMultipartUploadResult, string, error) {
+// CustomCopyFunc implements copying/moving data from one file descriptor
+// to another. The bool return signifies if function is idempotent. If true
+// the system will allow clients to retry CompleteMultipartUpload. If
+// false, then once a copy function is called successfully the upload will
+// be completely aborted on any subsequent failure.
+type CustomCopyFunc func(from *os.File, to *os.File) (bool, error)
+
+func (p *Posix) CompleteMultipartUploadWithCopy(ctx context.Context, input *s3.CompleteMultipartUploadInput, customCopy CustomCopyFunc) (s3response.CompleteMultipartUploadResult, string, error) {
 	acct, ok := ctx.Value("account").(auth.Account)
 	if !ok {
 		acct = auth.Account{}
 	}
 
-	var res s3response.CompleteMultipartUploadResult
+	res := s3response.CompleteMultipartUploadResult{}
 
 	if input.Key == nil {
 		return res, "", s3err.GetAPIError(s3err.ErrNoSuchKey)
@@ -1775,6 +1782,7 @@ func (p *Posix) CompleteMultipartUploadWithCopy(ctx context.Context, input *s3.C
 	defer f.cleanup()
 
 	var composableCsum string
+	var abortOnErrSet bool
 	for i, part := range parts {
 		partObjPath := filepath.Join(objdir, uploadID, fmt.Sprintf("%v", *part.PartNumber))
 		fullPartPath := filepath.Join(bucket, partObjPath)
@@ -1824,8 +1832,8 @@ func (p *Posix) CompleteMultipartUploadWithCopy(ctx context.Context, input *s3.C
 			}
 		}
 
-		if customMove != nil {
-			err = customMove(pf, f.File())
+		if customCopy != nil {
+			idemp, err := customCopy(pf, f.File())
 			if err != nil {
 				// Fail back to standard copy
 				debuglogger.Logf("custom data block move failed (%q/%q): %v, failing back to io.Copy()",
@@ -1838,6 +1846,22 @@ func (p *Posix) CompleteMultipartUploadWithCopy(ctx context.Context, input *s3.C
 					_, err = io.Copy(fw, pf)
 				}
 			}
+			if !idemp && err == nil {
+				// a successful non-idempotent call means we can no longer
+				// retry this upload. any failure needs to abort the complete
+				// upload.
+				if !abortOnErrSet {
+					defer func() {
+						// cleanup tmp dirs
+						os.RemoveAll(filepath.Join(bucket, objdir, uploadID))
+						// use Remove for objdir in case there are still other
+						// uploads for same object name outstanding, this will
+						// fail if there are any
+						os.Remove(filepath.Join(bucket, objdir))
+					}()
+				}
+				abortOnErrSet = true
+			}
 		} else {
 			if p.forceNoCopyFileRange {
 				_, err = io.Copy(f.File(), &onlyRead{pf})
@@ -2009,7 +2033,7 @@ func (p *Posix) CompleteMultipartUploadWithCopy(ctx context.Context, input *s3.C
 	// cleanup tmp dirs
 	os.RemoveAll(filepath.Join(bucket, objdir, uploadID))
 	// use Remove for objdir in case there are still other uploads
-	// for same object name outstanding, this will fail if there are
+	// for same object name outstanding, this will fail if there are any
 	os.Remove(filepath.Join(bucket, objdir))
 
 	return s3response.CompleteMultipartUploadResult{

backend/scoutfs/scoutfs_compat.go

@@ -282,18 +282,18 @@ func (s *ScoutFS) CompleteMultipartUpload(ctx context.Context, input *s3.Complet
 	}
 
 	return s.Posix.CompleteMultipartUploadWithCopy(ctx, input,
-		func(from *os.File, to *os.File) error {
+		func(from *os.File, to *os.File) (bool, error) {
 			// May fail if the files are not 4K aligned; check for alignment
 			ffi, err := from.Stat()
 			if err != nil {
-				return fmt.Errorf("complete-mpu stat from: %w", err)
+				return true, fmt.Errorf("complete-mpu stat from: %w", err)
 			}
 			tfi, err := to.Stat()
 			if err != nil {
-				return fmt.Errorf("complete-mpu stat to: %w", err)
+				return true, fmt.Errorf("complete-mpu stat to: %w", err)
 			}
 			if ffi.Size()%4096 != 0 || tfi.Size()%4096 != 0 {
-				return os.ErrInvalid
+				return true, os.ErrInvalid
 			}
 
 			err = s.setProjectID(to, acct.ProjectID)
@@ -304,10 +304,13 @@ func (s *ScoutFS) CompleteMultipartUpload(ctx context.Context, input *s3.Complet
 
 			err = scoutfs.MoveData(from, to)
 			if err != nil {
-				return fmt.Errorf("complete-mpu movedata: %w", err)
+				return true, fmt.Errorf("complete-mpu movedata: %w", err)
 			}
 
-			return nil
+			// once scoutfs.MoveData is successful, we are no longer
+			// idempotent since we have moved the extents from the
+			// source file
+			return false, nil
 		})
 }