From 495b38a89940307efcb042a4c4c0f745a14ccc53 Mon Sep 17 00:00:00 2001 From: Ben McClelland Date: Mon, 23 Mar 2026 10:40:30 -0700 Subject: [PATCH] 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 | 34 ++++++++++++++++++++++++++----- backend/scoutfs/scoutfs_compat.go | 15 ++++++++------ 2 files changed, 38 insertions(+), 11 deletions(-) diff --git a/backend/posix/posix.go b/backend/posix/posix.go index 29698816..c35f8624 100644 --- a/backend/posix/posix.go +++ b/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{ diff --git a/backend/scoutfs/scoutfs_compat.go b/backend/scoutfs/scoutfs_compat.go index a3fa703f..b792aa5f 100644 --- a/backend/scoutfs/scoutfs_compat.go +++ b/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 }) }