From c26012905c277f8fc9126836a2e959eb47fe12b9 Mon Sep 17 00:00:00 2001 From: Ben McClelland Date: Fri, 27 Mar 2026 08:30:42 -0700 Subject: [PATCH] fix: retry link on ENOENT caused by racing DeleteObject MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A concurrent PutObject and DeleteObject on the same prefix directory can race: PutObject opens an O_TMPFILE in MetaTmpDir (not yet visible in the fs) DeleteObject removes the last visible object in the prefix directory and calls removeParents(), which rmdir's the now-empty prefix directory PutObject's link() tries to link the fd into a parent directory that no longer exists Fix by detecting ENOENT in the final link step (Linkat, Rename, and MoveFile) and retrying after recreating the parent directory. Also extract linkatOTmpfile() to consolidate the Linkat+EEXIST→Renameat logic that was previously inline in link(). Fixes #1988 --- backend/posix/otmpfile_common.go | 48 +++++++++ backend/posix/with_otmpfile.go | 167 ++++++++++++++++++++---------- backend/posix/without_otmpfile.go | 57 ++++++---- tests/integration/group-tests.go | 2 + tests/integration/posix.go | 78 ++++++++++++++ 5 files changed, 279 insertions(+), 73 deletions(-) create mode 100644 backend/posix/otmpfile_common.go 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 588643fd..1d5e38f8 100644 --- a/tests/integration/group-tests.go +++ b/tests/integration/group-tests.go @@ -872,6 +872,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) @@ -1789,6 +1790,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() + }) +}