diff --git a/backend/posix/otmpfile_common.go b/backend/posix/otmpfile_common.go new file mode 100644 index 00000000..a76876dd --- /dev/null +++ b/backend/posix/otmpfile_common.go @@ -0,0 +1,48 @@ +// Copyright 2026 Versity Software +// This file is licensed under the Apache License, Version 2.0 +// (the "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +package posix + +import ( + "fmt" + "math/rand" + "os" + "time" +) + +func (tmp *tmpfile) Write(b []byte) (int, error) { + if int64(len(b)) > tmp.size { + return 0, fmt.Errorf("write exceeds content length %v", tmp.size) + } + + n, err := tmp.f.Write(b) + tmp.size -= int64(n) + return n, err +} + +func (tmp *tmpfile) File() *os.File { + return tmp.f +} + +func sleepWithJitter(backoffMs int) { + if backoffMs <= 1 { + time.Sleep(1 * time.Millisecond) + return + } + + maxJitter := max(1, backoffMs/4) + jitter := rand.Intn((maxJitter*2)+1) - maxJitter + sleepMs := max(backoffMs+jitter, 1) + time.Sleep(time.Duration(sleepMs) * time.Millisecond) +} diff --git a/backend/posix/with_otmpfile.go b/backend/posix/with_otmpfile.go index 17abd9bd..6554f023 100644 --- a/backend/posix/with_otmpfile.go +++ b/backend/posix/with_otmpfile.go @@ -43,7 +43,7 @@ type tmpfile struct { objname string isOTmp bool size int64 - needsChown bool + doChown bool uid int gid int newDirPerm fs.FileMode @@ -87,7 +87,7 @@ func (p *Posix) openTmpFile(dir, bucket, obj string, size int64, acct auth.Accou objname: obj, isOTmp: true, size: size, - needsChown: doChown, + doChown: doChown, uid: uid, gid: gid, newDirPerm: p.newDirPerm, @@ -126,13 +126,13 @@ func (p *Posix) openMkTemp(dir, bucket, obj string, size int64, dofalloc bool, u return nil, err } tmp := &tmpfile{ - f: f, - bucket: bucket, - objname: obj, - size: size, - needsChown: doChown, - uid: uid, - gid: gid, + f: f, + bucket: bucket, + objname: obj, + size: size, + doChown: doChown, + uid: uid, + gid: gid, } // falloc is best effort, its fine if this fails if size > 0 && dofalloc { @@ -159,6 +159,40 @@ func (tmp *tmpfile) falloc() error { return nil } +const ( + maxTmpFileNameRetries = 3 + maxDirRecreateRetries = 3 + initialBackoffMs = 1 + maxBackoffMs = 1024 // ~1 second +) + +// linkatOTmpfile links the O_TMPFILE identified by procdir/fdName into dir/basename. +// Handles EEXIST by linking to a temporary name and atomically renaming it into place. +func linkatOTmpfile(procdirFd, dirFd int, fdName, basename string) error { + err := unix.Linkat(procdirFd, fdName, dirFd, basename, unix.AT_SYMLINK_FOLLOW) + if !errors.Is(err, syscall.EEXIST) { + return err + } + // Linkat cannot overwrite an existing file; link to a temp name then rename atomically. + for retries := 1; ; retries++ { + tmpName := fmt.Sprintf(".%s.sgwtmp.%d", basename, time.Now().UnixNano()) + err := unix.Linkat(procdirFd, fdName, dirFd, tmpName, unix.AT_SYMLINK_FOLLOW) + if errors.Is(err, syscall.EEXIST) && retries < maxTmpFileNameRetries { + continue + } + if err != nil { + return fmt.Errorf("cannot find free temporary file: %w", err) + } + err = unix.Renameat(dirFd, tmpName, dirFd, basename) + if err != nil { + // cleanup temp name previously linked into namespace + _ = unix.Unlinkat(dirFd, tmpName, 0) + return fmt.Errorf("overwriting renameat failed: %w", err) + } + return nil + } +} + func (tmp *tmpfile) link() error { // make sure this is cleaned up in all error cases defer tmp.f.Close() @@ -173,7 +207,7 @@ func (tmp *tmpfile) link() error { dir := filepath.Dir(objPath) - err := backend.MkdirAll(dir, tmp.uid, tmp.gid, tmp.needsChown, tmp.newDirPerm) + err := backend.MkdirAll(dir, tmp.uid, tmp.gid, tmp.doChown, tmp.newDirPerm) if err != nil { return fmt.Errorf("make parent dir: %w", err) } @@ -189,39 +223,40 @@ func (tmp *tmpfile) link() error { } defer procdir.Close() - dirf, err := os.Open(dir) - if err != nil { - return fmt.Errorf("open parent dir: %w", err) - } - defer dirf.Close() - - err = unix.Linkat(int(procdir.Fd()), filepath.Base(tmp.f.Name()), - int(dirf.Fd()), filepath.Base(objPath), unix.AT_SYMLINK_FOLLOW) - if errors.Is(err, syscall.EEXIST) { - // Linkat cannot overwrite files; we will allocate a temporary file, Linkat to it and then Renameat it - // to avoid potential race condition - retries := 1 - for { - tmpName := fmt.Sprintf(".%s.sgwtmp.%d", filepath.Base(objPath), time.Now().UnixNano()) - err := unix.Linkat(int(procdir.Fd()), filepath.Base(tmp.f.Name()), - int(dirf.Fd()), tmpName, unix.AT_SYMLINK_FOLLOW) - if errors.Is(err, syscall.EEXIST) && retries < 3 { - retries += 1 - continue - } + backoffMs := initialBackoffMs + var dirf *os.File + for { + dirf, err = os.Open(dir) + if errors.Is(err, fs.ErrNotExist) { + err := backend.MkdirAll(dir, tmp.uid, tmp.gid, tmp.doChown, tmp.newDirPerm) if err != nil { - return fmt.Errorf("cannot find free temporary file: %w", err) + return fmt.Errorf("make parent dir: %w", err) } - - err = unix.Renameat(int(dirf.Fd()), tmpName, int(dirf.Fd()), filepath.Base(objPath)) - if err != nil { - return fmt.Errorf("overwriting renameat failed: %w", err) - } - break + continue } - } else if err != nil { - return fmt.Errorf("link tmpfile (fd %q as %q): %w", - filepath.Base(tmp.f.Name()), objPath, err) + if err != nil { + return fmt.Errorf("open parent dir: %w", err) + } + err = linkatOTmpfile(int(procdir.Fd()), int(dirf.Fd()), + filepath.Base(tmp.f.Name()), filepath.Base(objPath)) + dirf.Close() + if errors.Is(err, syscall.ENOENT) { + // The directory was removed between open and linkat; backoff and retry. + // Add jitter to avoid synchronized retry waves. + sleepWithJitter(backoffMs) + backoffMs = min((backoffMs * 2), maxBackoffMs) + + mkErr := backend.MkdirAll(dir, tmp.uid, tmp.gid, tmp.doChown, tmp.newDirPerm) + if mkErr != nil { + return fmt.Errorf("make parent dir: %w", mkErr) + } + continue + } + if err != nil { + return fmt.Errorf("link tmpfile (fd %q as %q): %w", + filepath.Base(tmp.f.Name()), objPath, err) + } + break } err = tmp.f.Close() @@ -244,33 +279,55 @@ func (tmp *tmpfile) fallbackLink() error { } objPath := filepath.Join(tmp.bucket, tmp.objname) + dir := filepath.Dir(objPath) err = os.Rename(tempname, objPath) + if errors.Is(err, syscall.ENOENT) { + // The parent directory was concurrently removed; backoff and retry. + backoffMs := initialBackoffMs + for range maxDirRecreateRetries { + // Add jitter to avoid synchronized retry waves. + sleepWithJitter(backoffMs) + backoffMs = min((backoffMs * 2), maxBackoffMs) + + err = backend.MkdirAll(dir, tmp.uid, tmp.gid, tmp.doChown, tmp.newDirPerm) + if err != nil { + return fmt.Errorf("recreate parent dir: %w", err) + } + err = os.Rename(tempname, objPath) + if !errors.Is(err, syscall.ENOENT) { + break + } + } + } if err != nil { // rename only works for files within the same filesystem // if this fails fallback to copy - return backend.MoveFile(tempname, objPath, fs.FileMode(defaultFilePerm)) + backoffMs := initialBackoffMs + for range maxDirRecreateRetries { + err = backend.MoveFile(tempname, objPath, fs.FileMode(defaultFilePerm)) + if !errors.Is(err, syscall.ENOENT) { + break + } + + // Add jitter to avoid synchronized retry waves. + sleepWithJitter(backoffMs) + backoffMs = min((backoffMs * 2), maxBackoffMs) + + // The parent directory was concurrently removed; recreate and retry. + mkErr := backend.MkdirAll(dir, tmp.uid, tmp.gid, tmp.doChown, tmp.newDirPerm) + if mkErr != nil { + return fmt.Errorf("recreate parent dir: %w", mkErr) + } + } + return err } return nil } -func (tmp *tmpfile) Write(b []byte) (int, error) { - if int64(len(b)) > tmp.size { - return 0, fmt.Errorf("write exceeds content length %v", tmp.size) - } - - n, err := tmp.f.Write(b) - tmp.size -= int64(n) - return n, err -} - func (tmp *tmpfile) cleanup() { tmp.f.Close() if !strings.HasPrefix(tmp.f.Name(), procfddir) { os.Remove(tmp.f.Name()) } } - -func (tmp *tmpfile) File() *os.File { - return tmp.f -} diff --git a/backend/posix/without_otmpfile.go b/backend/posix/without_otmpfile.go index 7fb35e9d..7b22a5f6 100644 --- a/backend/posix/without_otmpfile.go +++ b/backend/posix/without_otmpfile.go @@ -30,11 +30,20 @@ import ( "github.com/versity/versitygw/s3err" ) +const ( + initialBackoffMs = 1 + maxBackoffMs = 1024 // ~1 second +) + type tmpfile struct { - f *os.File - bucket string - objname string - size int64 + f *os.File + bucket string + objname string + size int64 + newDirPerm fs.FileMode + uid int + gid int + doChown bool } func (p *Posix) openTmpFile(dir, bucket, obj string, size int64, acct auth.Account, _ bool, _ bool) (*tmpfile, error) { @@ -67,7 +76,16 @@ func (p *Posix) openTmpFile(dir, bucket, obj string, size int64, acct auth.Accou } } - return &tmpfile{f: f, bucket: bucket, objname: obj, size: size}, nil + return &tmpfile{ + f: f, + bucket: bucket, + objname: obj, + size: size, + newDirPerm: p.newDirPerm, + uid: uid, + gid: gid, + doChown: doChown, + }, nil } var ( @@ -88,24 +106,27 @@ func (tmp *tmpfile) link() error { return fmt.Errorf("close tmpfile: %w", err) } - return backend.MoveFile(tempname, objPath, defaultFilePerm) -} + backoffMs := initialBackoffMs + for { + err = backend.MoveFile(tempname, objPath, defaultFilePerm) + if !errors.Is(err, syscall.ENOENT) { + break + } + // The parent directory may have been concurrently removed; backoff and retry. + // Add jitter to avoid synchronized retry waves. + sleepWithJitter(backoffMs) + backoffMs = min((backoffMs * 2), maxBackoffMs) -func (tmp *tmpfile) Write(b []byte) (int, error) { - if int64(len(b)) > tmp.size { - return 0, fmt.Errorf("write exceeds content length %v", tmp.size) + err = backend.MkdirAll(filepath.Dir(objPath), tmp.uid, tmp.gid, + tmp.doChown, tmp.newDirPerm) + if err != nil { + return fmt.Errorf("recreate parent dir: %w", err) + } } - - n, err := tmp.f.Write(b) - tmp.size -= int64(n) - return n, err + return err } func (tmp *tmpfile) cleanup() { tmp.f.Close() os.Remove(tmp.f.Name()) } - -func (tmp *tmpfile) File() *os.File { - return tmp.f -} diff --git a/tests/integration/group-tests.go b/tests/integration/group-tests.go index 3659a380..14bc0d1f 100644 --- a/tests/integration/group-tests.go +++ b/tests/integration/group-tests.go @@ -874,6 +874,7 @@ func TestPosix(ts *TestState) { ts.Run(CopyObject_overwrite_same_dir_object) ts.Run(CopyObject_overwrite_same_file_object) ts.Run(DeleteObject_directory_not_empty) + ts.Run(PutObject_race_with_delete) // posix specific versioning tests if !ts.conf.versioningEnabled { TestVersioningDisabled(ts) @@ -1794,6 +1795,7 @@ func GetIntTests() IntTests { "PutObject_overwrite_file_obj_with_nested_obj": PutObject_overwrite_file_obj_with_nested_obj, "PutObject_dir_obj_with_data": PutObject_dir_obj_with_data, "PutObject_with_slashes": PutObject_with_slashes, + "PutObject_race_with_delete": PutObject_race_with_delete, "CreateMultipartUpload_dir_obj": CreateMultipartUpload_dir_obj, "IAM_user_access_denied": IAM_user_access_denied, "IAM_userplus_access_denied": IAM_userplus_access_denied, diff --git a/tests/integration/posix.go b/tests/integration/posix.go index 8b50602f..b9401a77 100644 --- a/tests/integration/posix.go +++ b/tests/integration/posix.go @@ -18,8 +18,11 @@ import ( "context" "fmt" + "github.com/aws/aws-sdk-go-v2/aws" "github.com/aws/aws-sdk-go-v2/service/s3" + "github.com/aws/aws-sdk-go-v2/service/s3/types" "github.com/versity/versitygw/s3err" + "golang.org/x/sync/errgroup" ) func PutObject_overwrite_dir_obj(s *S3Conf) error { @@ -218,3 +221,78 @@ func CopyObject_overwrite_same_file_object(s *S3Conf) error { return nil }) } + +// PutObject_race_with_delete tests the race between PutObject and DeleteObject +// in the same subdirectory. +// One goroutine sequentially puts "race-dir/0.txt" … "race-dir/N-1.txt". +// A second goroutine loops for the same number of iterations: it lists all +// objects under "race-dir/" and bulk-deletes them. When the batch delete +// removes the last visible object in the directory, removeParents() rmdir's +// "race-dir/", which can race with the uploader's final link() step. +// The test asserts that no error is returned from either goroutine. +func PutObject_race_with_delete(s *S3Conf) error { + testName := "PutObject_race_with_delete" + const iterations = 100 + return actionHandler(s, testName, func(s3client *s3.Client, bucket string) error { + eg := errgroup.Group{} + + // Upload goroutine: sequentially puts objects into race-dir/ + eg.Go(func() error { + for i := range iterations { + key := fmt.Sprintf("race-dir/%d.txt", i) + ctx, cancel := context.WithTimeout(context.Background(), shortTimeout) + _, err := s3client.PutObject(ctx, &s3.PutObjectInput{ + Bucket: &bucket, + Key: &key, + }) + cancel() + if err != nil { + return fmt.Errorf("put %d: %w", i, err) + } + } + return nil + }) + + // Delete goroutine: repeatedly lists race-dir/ and bulk-deletes everything. + // When the last object is removed, removeParents() rmdir's the directory, + // racing with the concurrent link() call in the upload goroutine. + eg.Go(func() error { + for range iterations { + ctx, cancel := context.WithTimeout(context.Background(), shortTimeout) + res, err := s3client.ListObjectsV2(ctx, &s3.ListObjectsV2Input{ + Bucket: &bucket, + Prefix: getPtr("race-dir/"), + }) + cancel() + if err != nil { + return fmt.Errorf("list: %w", err) + } + if len(res.Contents) == 0 { + continue + } + + objs := make([]types.ObjectIdentifier, 0, len(res.Contents)) + for _, obj := range res.Contents { + objs = append(objs, types.ObjectIdentifier{Key: obj.Key}) + } + + ctx, cancel = context.WithTimeout(context.Background(), shortTimeout) + out, err := s3client.DeleteObjects(ctx, &s3.DeleteObjectsInput{ + Bucket: &bucket, + Delete: &types.Delete{Objects: objs}, + }) + cancel() + if err != nil { + return fmt.Errorf("delete objects: %w", err) + } + if len(out.Errors) > 0 { + return fmt.Errorf("delete error: key=%v code=%v msg=%v", + aws.ToString(out.Errors[0].Key), aws.ToString(out.Errors[0].Code), aws.ToString(out.Errors[0].Message)) + } + } + return nil + }) + + return eg.Wait() + }) +}