Merge pull request #1991 from versity/ben/scoutfs-fail-upload

fix: abort scoutfs multipart uploads on error after successful moveblocks

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
 		})
 }