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