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

Merge pull request #2006 from versity/ben/racing-put-delete

Browse Source 
fix: retry link on ENOENT caused by racing DeleteObject
This commit is contained in:
Ben McClelland
2026-04-02 09:57:48 -07:00
committed by GitHub
parent 98a186cc4e c26012905c
commit 1fca33e738
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
@@ -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,
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()
})
}