mirrors/versitygw
1
0
Fork 0
mirror of https://github.com/versity/versitygw.git synced 2026-09-03 14:46:55 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix: retry link on ENOENT caused by racing DeleteObject

Browse Source 
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
This commit is contained in:
Ben McClelland
2026-04-02 08:14:44 -07:00
parent e0209ebab4
commit c26012905c
5 changed files with 279 additions and 73 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/posix/otmpfile_common.go
+48
View File
@@ -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)
}
backend/posix/with_otmpfile.go
+112 -55
View File
@@ -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
}
backend/posix/without_otmpfile.go
+39 -18
View File
@@ -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
}
tests/integration/group-tests.go
+2
View File
@@ -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,
tests/integration/posix.go
+78
View File
@@ -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()
})
}