mirror of
https://github.com/versity/versitygw.git
synced 2026-09-21 23:44:14 +00:00
fix(posix): make conditional PUT evaluation and publication atomic per key
S3 conditional writes (If-Match / If-None-Match: *) were evaluated with a check-then-act pattern: PutObjectWithPostFunc and CompleteMultipartUploadWithCopy read the current etag, evaluated the precondition, and only later published the replacement via link/rename. Under concurrent conditional PUTs to the same key, many writers could read the same old state, all pass the check, and all succeed, breaking compare-and-swap coordination built on conditional writes (observed by Buzz's git object-store A3 conformance probe: 32-way races returned up to 32 winners instead of exactly 1x2xx + 31x412). Introduce a per-object publish lock (lockObjectPublish) that every object publication path holds across condition re-evaluation, metadata stores, and the final link: - Exclusion is an advisory file lock (flock on unix, LockFileEx on windows) on bucket/.sgwtmp/objlock/<shard>, where shard is the first byte of sha256(key). The gateway is stateless and multiple gateway processes may share one filesystem, so a process-local mutex alone is not sufficient; file locks provide cross-process and (via NFSv4 lock semantics) cross-client exclusion. A process-local striped mutex is taken alongside so in-process contention never thrashes the filesystem lock. Lock files are empty, bounded (max 256 per bucket), and never unlinked to avoid the unlink/recreate flock race that cannot be detected reliably on NFS. - Request bodies are staged to the temp file before the lock is taken; the lock covers only the short commit phase. The kernel releases the lock on close or process death, so failures and crashes cannot leave a stale lock. If the filesystem does not support advisory locking (e.g. NFS mounted with -o nolock), the gateway falls back to process-local exclusion and warns once. - The pre-staging precondition check is kept as an advisory fast-fail; the authoritative check runs under the lock. Unconditional PUTs, directory-object PUTs, CopyObject (via PutObject), scoutfs (via PutObjectWithPostFunc/CompleteMultipartUploadWithCopy), and multipart completion all participate. - For conditional writes on versioned buckets the version snapshot is deferred until the precondition is confirmed under the lock, so losing writers no longer create spurious version snapshots. - The postprocess hook now runs before any path-based metadata is written, so a failing hook no longer leaves stale sidecar attributes (previously the new etag was visible in sidecar mode even when publication failed). Add regression tests covering 32-way If-Match and If-None-Match: * races for both ordinary PUT and multipart completion, mixed conditional/unconditional races, sequential semantics, per-key independence, and cleanup after failed publication, for both xattr and sidecar metadata modes. Add an external SigV4 test (tests/conditionalrace) that replicates the Buzz A3 probe against a live gateway, optionally across two gateway processes sharing one backend directory. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
co-authored by
Claude Fable 5
parent
8bd73a45ee
commit
f7e13d71ce
@@ -0,0 +1,148 @@
|
||||
// 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 (
|
||||
"context"
|
||||
"crypto/sha256"
|
||||
"fmt"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"time"
|
||||
|
||||
"github.com/versity/versitygw/backend"
|
||||
"github.com/versity/versitygw/debuglogger"
|
||||
)
|
||||
|
||||
// Object publish locking
|
||||
//
|
||||
// S3 conditional writes (If-Match / If-None-Match: *) require that reading the
|
||||
// current object state, evaluating the condition, and publishing the
|
||||
// replacement happen as one atomic step per bucket/key. The gateway is
|
||||
// stateless and multiple gateway processes may share the same backend
|
||||
// filesystem, so an in-process mutex alone is not sufficient: exclusion is
|
||||
// provided by an advisory lock (flock on unix, LockFileEx on windows) on a
|
||||
// shared lock file, combined with a process-local striped mutex so that
|
||||
// contention within one process is resolved cheaply and each process presents
|
||||
// at most one waiter to the filesystem lock.
|
||||
//
|
||||
// Lock identity: bucket/.sgwtmp/objlock/<shard>, where shard is the first
|
||||
// byte of sha256(object key) rendered as two hex characters. Hashing the key
|
||||
// gives a stable, traversal-safe, fixed-length name; sharding (256 slots per
|
||||
// bucket) keeps the number of lock files bounded while still letting writes
|
||||
// to unrelated keys proceed concurrently in the common case. Lock files are
|
||||
// empty, created on demand, and never unlinked: unlinking an flock file opens
|
||||
// a classic race where a waiter holds a lock on an unlinked inode while a new
|
||||
// file takes its place, which is unsafe to detect reliably on NFS due to
|
||||
// attribute caching.
|
||||
//
|
||||
// The lock is held only for the commit phase (condition re-check, metadata
|
||||
// stores, final link/rename) — request bodies are staged to a temp file
|
||||
// before the lock is taken. The OS releases advisory locks automatically when
|
||||
// the file handle is closed or the process exits, so failures, cancellation,
|
||||
// or crashes cannot leave a permanently stale lock.
|
||||
//
|
||||
// NFS notes: flock on Linux NFS clients is mapped to NFSv4 byte-range locks
|
||||
// (or NLM on NFSv3), giving cross-client exclusion. Mounting with
|
||||
// "-o nolock" or "-o local_lock=flock"/"local_lock=all" disables server-side
|
||||
// locking and reduces exclusion to a single client; conditional-write
|
||||
// atomicity across gateways requires server-backed locking. If the filesystem
|
||||
// does not support advisory locking at all, the gateway falls back to
|
||||
// process-local exclusion and logs a warning once.
|
||||
|
||||
const (
|
||||
// objLockDir is the per-bucket directory holding object publish lock files
|
||||
objLockDir = MetaTmpDir + "/objlock"
|
||||
// objLockShards is the number of lock shards per bucket
|
||||
objLockShards = 256
|
||||
)
|
||||
|
||||
// objLockShard returns the shard index for an object key.
|
||||
func objLockShard(object string) uint8 {
|
||||
sum := sha256.Sum256([]byte(object))
|
||||
return sum[0]
|
||||
}
|
||||
|
||||
// lockObjectPublish acquires the publish lock for bucket/object. It returns a
|
||||
// release function that must be called (typically deferred) once the new
|
||||
// object state is visible. All code paths that create or replace an object at
|
||||
// its final key must hold this lock across condition evaluation and
|
||||
// publication.
|
||||
func (p *Posix) lockObjectPublish(ctx context.Context, bucket, object string) (func(), error) {
|
||||
shard := objLockShard(object)
|
||||
|
||||
mu := &p.objLockMus[shard]
|
||||
mu.Lock()
|
||||
|
||||
f, err := p.openObjLockFile(bucket, shard)
|
||||
if err != nil {
|
||||
mu.Unlock()
|
||||
return nil, err
|
||||
}
|
||||
|
||||
err = lockFileExclusive(ctx, f)
|
||||
if err != nil {
|
||||
f.Close()
|
||||
if ctx.Err() != nil {
|
||||
mu.Unlock()
|
||||
return nil, ctx.Err()
|
||||
}
|
||||
// The filesystem does not support advisory locking (e.g. NFS
|
||||
// mounted with -o nolock). Fall back to process-local exclusion
|
||||
// and warn once: conditional writes are then only atomic within
|
||||
// this gateway process.
|
||||
p.objLockWarn.Do(func() {
|
||||
debuglogger.Logf("object lock file locking unavailable (%v): "+
|
||||
"conditional write atomicity limited to this process", err)
|
||||
})
|
||||
return mu.Unlock, nil
|
||||
}
|
||||
|
||||
return func() {
|
||||
// closing the file releases the advisory lock
|
||||
f.Close()
|
||||
mu.Unlock()
|
||||
}, nil
|
||||
}
|
||||
|
||||
// openObjLockFile opens (creating as needed) the lock file for the shard in
|
||||
// the given bucket.
|
||||
func (p *Posix) openObjLockFile(bucket string, shard uint8) (*os.File, error) {
|
||||
name := filepath.Join(bucket, objLockDir, fmt.Sprintf("%02x", shard))
|
||||
|
||||
f, err := os.OpenFile(name, os.O_RDWR|os.O_CREATE, os.FileMode(defaultFilePerm))
|
||||
if err == nil {
|
||||
return f, nil
|
||||
}
|
||||
if !os.IsNotExist(err) {
|
||||
return nil, fmt.Errorf("open object lock file: %w", err)
|
||||
}
|
||||
|
||||
// lock dir not created yet
|
||||
err = backend.MkdirAll(filepath.Join(bucket, objLockDir), 0, 0, false, p.newDirPerm)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("make object lock dir: %w", err)
|
||||
}
|
||||
f, err = os.OpenFile(name, os.O_RDWR|os.O_CREATE, os.FileMode(defaultFilePerm))
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("open object lock file: %w", err)
|
||||
}
|
||||
return f, nil
|
||||
}
|
||||
|
||||
const (
|
||||
objLockInitialBackoff = time.Millisecond
|
||||
objLockMaxBackoff = 16 * time.Millisecond
|
||||
)
|
||||
@@ -0,0 +1,51 @@
|
||||
// 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.
|
||||
|
||||
//go:build !windows
|
||||
|
||||
package posix
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"os"
|
||||
"time"
|
||||
|
||||
"golang.org/x/sys/unix"
|
||||
)
|
||||
|
||||
// lockFileExclusive takes an exclusive advisory flock on f, polling with
|
||||
// backoff so that context cancellation is honored while waiting. The lock is
|
||||
// released by closing f or on process exit, so no cleanup beyond Close is
|
||||
// required on any failure path.
|
||||
func lockFileExclusive(ctx context.Context, f *os.File) error {
|
||||
backoff := objLockInitialBackoff
|
||||
for {
|
||||
err := unix.Flock(int(f.Fd()), unix.LOCK_EX|unix.LOCK_NB)
|
||||
if err == nil {
|
||||
return nil
|
||||
}
|
||||
if !errors.Is(err, unix.EWOULDBLOCK) && !errors.Is(err, unix.EAGAIN) &&
|
||||
!errors.Is(err, unix.EINTR) {
|
||||
return err
|
||||
}
|
||||
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return ctx.Err()
|
||||
case <-time.After(backoff):
|
||||
}
|
||||
backoff = min(backoff*2, objLockMaxBackoff)
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,52 @@
|
||||
// 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.
|
||||
|
||||
//go:build windows
|
||||
|
||||
package posix
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"os"
|
||||
"time"
|
||||
|
||||
"golang.org/x/sys/windows"
|
||||
)
|
||||
|
||||
// lockFileExclusive takes an exclusive lock on f via LockFileEx, polling with
|
||||
// backoff so that context cancellation is honored while waiting. The lock is
|
||||
// released when f is closed or the process exits.
|
||||
func lockFileExclusive(ctx context.Context, f *os.File) error {
|
||||
backoff := objLockInitialBackoff
|
||||
for {
|
||||
ol := new(windows.Overlapped)
|
||||
err := windows.LockFileEx(windows.Handle(f.Fd()),
|
||||
windows.LOCKFILE_EXCLUSIVE_LOCK|windows.LOCKFILE_FAIL_IMMEDIATELY,
|
||||
0, 1, 0, ol)
|
||||
if err == nil {
|
||||
return nil
|
||||
}
|
||||
if !errors.Is(err, windows.ERROR_LOCK_VIOLATION) {
|
||||
return err
|
||||
}
|
||||
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return ctx.Err()
|
||||
case <-time.After(backoff):
|
||||
}
|
||||
backoff = min(backoff*2, objLockMaxBackoff)
|
||||
}
|
||||
}
|
||||
+149
-41
@@ -128,6 +128,15 @@ type Posix struct {
|
||||
// multipart CRC64NVME: "CRC64NVME-<whole-file-checksum>"
|
||||
// multipart composite: "ALGO-<composite-checksum>-<part-count>"
|
||||
dataIntegrityEtag bool
|
||||
|
||||
// objLockMus are process-local striped mutexes taken alongside the
|
||||
// shared per-bucket lock files by lockObjectPublish so that in-process
|
||||
// contention on an object's publish lock is resolved without filesystem
|
||||
// lock thrash. See objlock.go.
|
||||
objLockMus [objLockShards]sync.Mutex
|
||||
// objLockWarn ensures the advisory-locking-unavailable warning is
|
||||
// logged at most once
|
||||
objLockWarn sync.Once
|
||||
}
|
||||
|
||||
var _ backend.Backend = &Posix{}
|
||||
@@ -2076,12 +2085,12 @@ func (p *Posix) CompleteMultipartUploadWithCopy(ctx context.Context, input *s3.C
|
||||
defer os.Rename(uploadIDInProgress, uploadIDDir)
|
||||
defer p.meta.RenameObject(bucket, newMetaObj, oldMetaObj)
|
||||
|
||||
b, err := p.meta.RetrieveAttribute(nil, bucket, object, etagkey)
|
||||
if err == nil || errors.Is(err, fs.ErrNotExist) || errors.Is(err, meta.ErrNoSuchKey) {
|
||||
err = backend.EvaluateObjectPutPreconditions(string(b), input.IfMatch, input.IfNoneMatch, err == nil)
|
||||
if err != nil {
|
||||
return res, "", err
|
||||
}
|
||||
// Fast-fail precondition check before the parts are assembled. This is
|
||||
// only advisory: the authoritative check is repeated while holding the
|
||||
// object publish lock just before the final link.
|
||||
err = p.checkPutPreconditions(bucket, object, input.IfMatch, input.IfNoneMatch)
|
||||
if err != nil {
|
||||
return res, "", err
|
||||
}
|
||||
|
||||
checksums, err := p.retrieveChecksums(nil, bucket, filepath.Join(objdir, activeUploadName))
|
||||
@@ -2370,6 +2379,21 @@ func (p *Posix) CompleteMultipartUploadWithCopy(ctx context.Context, input *s3.C
|
||||
}
|
||||
}
|
||||
|
||||
// The parts are fully assembled into the staging file; serialize the
|
||||
// commit phase (condition evaluation, metadata stores, and final link)
|
||||
// with competing writers to the same key.
|
||||
unlock, err := p.lockObjectPublish(ctx, bucket, object)
|
||||
if err != nil {
|
||||
return res, "", err
|
||||
}
|
||||
defer unlock()
|
||||
|
||||
// authoritative precondition check under the publish lock
|
||||
err = p.checkPutPreconditions(bucket, object, input.IfMatch, input.IfNoneMatch)
|
||||
if err != nil {
|
||||
return res, "", err
|
||||
}
|
||||
|
||||
upiddir := filepath.Join(objdir, activeUploadName)
|
||||
|
||||
objMeta := p.loadObjectMetaProperties(nil, bucket, upiddir, nil)
|
||||
@@ -3842,6 +3866,62 @@ func getEmptyChecksumValue(algo types.ChecksumAlgorithm) string {
|
||||
}
|
||||
}
|
||||
|
||||
// checkPutPreconditions evaluates the conditional write headers against the
|
||||
// object's current etag. Callers that publish an object must repeat this
|
||||
// check while holding the object publish lock (lockObjectPublish) so that
|
||||
// evaluation and publication are atomic per bucket/key.
|
||||
func (p *Posix) checkPutPreconditions(bucket, object string, ifMatch, ifNoneMatch *string) error {
|
||||
if ifMatch == nil && ifNoneMatch == nil {
|
||||
return nil
|
||||
}
|
||||
|
||||
etagBytes, err := p.meta.RetrieveAttribute(nil, bucket, object, etagkey)
|
||||
if err == nil || errors.Is(err, fs.ErrNotExist) || errors.Is(err, meta.ErrNoSuchKey) {
|
||||
return backend.EvaluateObjectPutPreconditions(string(etagBytes), ifMatch, ifNoneMatch, err == nil)
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
// snapshotObjVersion copies the current object at bucket/key into the
|
||||
// versioning directory before the object is replaced, if versioning is
|
||||
// configured for the bucket and there is an object to snapshot.
|
||||
func (p *Posix) snapshotObjVersion(bucket, key string, vStatus types.BucketVersioningStatus, acct auth.Account) error {
|
||||
if !p.versioningEnabled() || vStatus == "" {
|
||||
return nil
|
||||
}
|
||||
|
||||
d, err := os.Stat(filepath.Join(bucket, key))
|
||||
if err != nil || d.IsDir() {
|
||||
// nothing to snapshot
|
||||
return nil
|
||||
}
|
||||
|
||||
var isVersionIdMissing bool
|
||||
if p.isBucketVersioningSuspended(vStatus) {
|
||||
vIdBytes, err := p.meta.RetrieveAttribute(nil, bucket, key, versionIdKey)
|
||||
if err != nil && !errors.Is(err, meta.ErrNoSuchKey) {
|
||||
return fmt.Errorf("get object versionId: %w", err)
|
||||
}
|
||||
isVersionIdMissing = len(vIdBytes) == 0
|
||||
}
|
||||
if !isVersionIdMissing {
|
||||
_, err := p.createObjVersion(bucket, key, d.Size(), acct, false)
|
||||
if err != nil {
|
||||
return fmt.Errorf("create object version: %w", err)
|
||||
}
|
||||
// With path-based metadata backends (e.g. sidecar), object-lock
|
||||
// attributes written on the previous version persist at this path
|
||||
// after createObjVersion because metadata is not replaced atomically
|
||||
// the way xattrs are on file rename. Delete them so they do not
|
||||
// bleed into the new version.
|
||||
_ = p.meta.DeleteAttribute(bucket, key, objectLegalHoldKey)
|
||||
_ = p.meta.DeleteAttribute(bucket, key, objectRetentionKey)
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
func (p *Posix) PutObject(ctx context.Context, po s3response.PutObjectInput) (s3response.PutObjectOutput, error) {
|
||||
release, err := p.acquireActionSlot(ctx)
|
||||
if err != nil {
|
||||
@@ -3879,13 +3959,13 @@ func (p *Posix) PutObjectWithPostFunc(ctx context.Context, po s3response.PutObje
|
||||
|
||||
name := filepath.Join(*po.Bucket, *po.Key)
|
||||
|
||||
// evaluate preconditions
|
||||
etagBytes, err := p.meta.RetrieveAttribute(nil, *po.Bucket, *po.Key, etagkey)
|
||||
if err == nil || errors.Is(err, fs.ErrNotExist) || errors.Is(err, meta.ErrNoSuchKey) {
|
||||
err = backend.EvaluateObjectPutPreconditions(string(etagBytes), po.IfMatch, po.IfNoneMatch, err == nil)
|
||||
if err != nil {
|
||||
return s3response.PutObjectOutput{}, err
|
||||
}
|
||||
// Fast-fail precondition check before the request body is staged. This
|
||||
// is only advisory: the authoritative check is repeated while holding
|
||||
// the object publish lock at commit time so that evaluation and
|
||||
// publication are atomic under concurrent writers.
|
||||
err = p.checkPutPreconditions(*po.Bucket, *po.Key, po.IfMatch, po.IfNoneMatch)
|
||||
if err != nil {
|
||||
return s3response.PutObjectOutput{}, err
|
||||
}
|
||||
|
||||
uid, gid, doChown := p.getChownIDs(acct)
|
||||
@@ -3941,6 +4021,19 @@ func (p *Posix) PutObjectWithPostFunc(ctx context.Context, po s3response.PutObje
|
||||
return s3response.PutObjectOutput{}, s3err.GetAPIError(s3err.ErrDirectoryObjectContainsData)
|
||||
}
|
||||
|
||||
// serialize with competing writers to the same key and repeat the
|
||||
// precondition check authoritatively under the publish lock
|
||||
unlock, err := p.lockObjectPublish(ctx, *po.Bucket, *po.Key)
|
||||
if err != nil {
|
||||
return s3response.PutObjectOutput{}, err
|
||||
}
|
||||
defer unlock()
|
||||
|
||||
err = p.checkPutPreconditions(*po.Bucket, *po.Key, po.IfMatch, po.IfNoneMatch)
|
||||
if err != nil {
|
||||
return s3response.PutObjectOutput{}, err
|
||||
}
|
||||
|
||||
err = backend.MkdirAll(name, uid, gid, doChown, p.newDirPerm)
|
||||
if err != nil {
|
||||
if errors.Is(err, syscall.EDQUOT) {
|
||||
@@ -4041,28 +4134,15 @@ func (p *Posix) PutObjectWithPostFunc(ctx context.Context, po s3response.PutObje
|
||||
return s3response.PutObjectOutput{}, s3err.GetAPIError(s3err.ErrExistingObjectIsDirectory)
|
||||
}
|
||||
|
||||
// if the versioning is enabled first create the file object version
|
||||
if p.versioningEnabled() && vStatus != "" && err == nil {
|
||||
var isVersionIdMissing bool
|
||||
if p.isBucketVersioningSuspended(vStatus) {
|
||||
vIdBytes, err := p.meta.RetrieveAttribute(nil, *po.Bucket, *po.Key, versionIdKey)
|
||||
if err != nil && !errors.Is(err, meta.ErrNoSuchKey) {
|
||||
return s3response.PutObjectOutput{}, fmt.Errorf("get object versionId: %w", err)
|
||||
}
|
||||
isVersionIdMissing = len(vIdBytes) == 0
|
||||
}
|
||||
if !isVersionIdMissing {
|
||||
_, err := p.createObjVersion(*po.Bucket, *po.Key, d.Size(), acct, false)
|
||||
if err != nil {
|
||||
return s3response.PutObjectOutput{}, fmt.Errorf("create object version: %w", err)
|
||||
}
|
||||
// With path-based metadata backends (e.g. sidecar), object-lock
|
||||
// attributes written on the previous version persist at this path
|
||||
// after createObjVersion because metadata is not replaced atomically
|
||||
// the way xattrs are on file rename. Delete them so they do not
|
||||
// bleed into the new version.
|
||||
_ = p.meta.DeleteAttribute(*po.Bucket, *po.Key, objectLegalHoldKey)
|
||||
_ = p.meta.DeleteAttribute(*po.Bucket, *po.Key, objectRetentionKey)
|
||||
// If versioning is enabled, first create the file object version.
|
||||
// Conditional writes defer the snapshot until the precondition has been
|
||||
// confirmed under the object publish lock, so that a losing conditional
|
||||
// write does not create a spurious version snapshot.
|
||||
conditional := po.IfMatch != nil || po.IfNoneMatch != nil
|
||||
if err == nil && !conditional {
|
||||
verr := p.snapshotObjVersion(*po.Bucket, *po.Key, vStatus, acct)
|
||||
if verr != nil {
|
||||
return s3response.PutObjectOutput{}, verr
|
||||
}
|
||||
}
|
||||
if isErrNameTooLong(err) {
|
||||
@@ -4161,6 +4241,40 @@ func (p *Posix) PutObjectWithPostFunc(ctx context.Context, po s3response.PutObje
|
||||
versionID = ulid.Make().String()
|
||||
}
|
||||
|
||||
// Run the backend post-process hook on the staged file before taking
|
||||
// the publish lock and before any path-based metadata is written, so
|
||||
// that a failing hook neither holds up other writers nor leaves
|
||||
// partial metadata behind.
|
||||
err = postprocess(f.File())
|
||||
if err != nil {
|
||||
return s3response.PutObjectOutput{},
|
||||
fmt.Errorf("put object post process failed: %w", err)
|
||||
}
|
||||
|
||||
// The body is fully staged; serialize the commit phase (condition
|
||||
// evaluation, metadata stores, and final link) with competing writers
|
||||
// to the same key.
|
||||
unlock, err := p.lockObjectPublish(ctx, *po.Bucket, *po.Key)
|
||||
if err != nil {
|
||||
return s3response.PutObjectOutput{}, err
|
||||
}
|
||||
defer unlock()
|
||||
|
||||
if conditional {
|
||||
// authoritative precondition check under the publish lock
|
||||
err = p.checkPutPreconditions(*po.Bucket, *po.Key, po.IfMatch, po.IfNoneMatch)
|
||||
if err != nil {
|
||||
return s3response.PutObjectOutput{}, err
|
||||
}
|
||||
|
||||
// snapshot the object version that is about to be replaced, now
|
||||
// that this writer is known to win the conditional race
|
||||
verr := p.snapshotObjVersion(*po.Bucket, *po.Key, vStatus, acct)
|
||||
if verr != nil {
|
||||
return s3response.PutObjectOutput{}, verr
|
||||
}
|
||||
}
|
||||
|
||||
// Before finalizing the object creation remove
|
||||
// null versionId object from versioning directory
|
||||
// if it exists and the versioning status is Suspended
|
||||
@@ -4241,12 +4355,6 @@ func (p *Posix) PutObjectWithPostFunc(ctx context.Context, po s3response.PutObje
|
||||
versionID = ""
|
||||
}
|
||||
|
||||
err = postprocess(f.File())
|
||||
if err != nil {
|
||||
return s3response.PutObjectOutput{},
|
||||
fmt.Errorf("put object post process failed: %w", err)
|
||||
}
|
||||
|
||||
err = f.link()
|
||||
if errors.Is(err, syscall.EEXIST) {
|
||||
return s3response.PutObjectOutput{
|
||||
|
||||
@@ -0,0 +1,575 @@
|
||||
// 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 (
|
||||
"bytes"
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
"io"
|
||||
"io/fs"
|
||||
"net/http"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"strings"
|
||||
"sync"
|
||||
"testing"
|
||||
|
||||
"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/backend/meta"
|
||||
"github.com/versity/versitygw/s3err"
|
||||
"github.com/versity/versitygw/s3response"
|
||||
)
|
||||
|
||||
const condRaceWriters = 32
|
||||
|
||||
// metaModes returns the metadata backends supported by the posix backend
|
||||
// that should both be exercised by the conditional-put tests.
|
||||
func metaModes(t *testing.T) map[string]func(t *testing.T) (meta.MetadataStorer, PosixOpts) {
|
||||
t.Helper()
|
||||
return map[string]func(t *testing.T) (meta.MetadataStorer, PosixOpts){
|
||||
"xattr": func(t *testing.T) (meta.MetadataStorer, PosixOpts) {
|
||||
return meta.XattrMeta{}, PosixOpts{NewDirPerm: 0755}
|
||||
},
|
||||
"sidecar": func(t *testing.T) (meta.MetadataStorer, PosixOpts) {
|
||||
dir := t.TempDir()
|
||||
sc, err := meta.NewSideCar(dir)
|
||||
if err != nil {
|
||||
t.Fatalf("new sidecar: %v", err)
|
||||
}
|
||||
return sc, PosixOpts{NewDirPerm: 0755, SideCarDir: dir}
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
func newTestPosix(t *testing.T, mkMeta func(t *testing.T) (meta.MetadataStorer, PosixOpts)) *Posix {
|
||||
t.Helper()
|
||||
root := t.TempDir()
|
||||
storer, opts := mkMeta(t)
|
||||
p, err := New(root, storer, opts)
|
||||
if err != nil {
|
||||
t.Fatalf("new posix: %v", err)
|
||||
}
|
||||
return p
|
||||
}
|
||||
|
||||
func createTestBucket(t *testing.T, p *Posix, bucket string) {
|
||||
t.Helper()
|
||||
err := p.CreateBucket(context.Background(), &s3.CreateBucketInput{
|
||||
Bucket: &bucket,
|
||||
CreateBucketConfiguration: &types.CreateBucketConfiguration{},
|
||||
}, []byte{})
|
||||
if err != nil {
|
||||
t.Fatalf("create bucket: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
func testPut(p *Posix, bucket, key string, body []byte, ifMatch, ifNoneMatch *string) (s3response.PutObjectOutput, error) {
|
||||
return p.PutObject(context.Background(), s3response.PutObjectInput{
|
||||
Bucket: &bucket,
|
||||
Key: &key,
|
||||
Body: bytes.NewReader(body),
|
||||
ContentLength: aws.Int64(int64(len(body))),
|
||||
IfMatch: ifMatch,
|
||||
IfNoneMatch: ifNoneMatch,
|
||||
})
|
||||
}
|
||||
|
||||
func getTestObject(t *testing.T, p *Posix, bucket, key string) ([]byte, string) {
|
||||
t.Helper()
|
||||
out, err := p.GetObject(context.Background(), &s3.GetObjectInput{
|
||||
Bucket: &bucket,
|
||||
Key: &key,
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatalf("get object: %v", err)
|
||||
}
|
||||
defer out.Body.Close()
|
||||
data, err := io.ReadAll(out.Body)
|
||||
if err != nil {
|
||||
t.Fatalf("read object body: %v", err)
|
||||
}
|
||||
return data, *out.ETag
|
||||
}
|
||||
|
||||
// trimEtag strips the surrounding quotes the same way the s3api layer does
|
||||
// before passing If-Match values to the backend.
|
||||
func trimEtag(etag string) string {
|
||||
return strings.Trim(etag, "\"")
|
||||
}
|
||||
|
||||
func isPreconditionFailed(err error) bool {
|
||||
var pe s3err.PreconditionFailedError
|
||||
if errors.As(err, &pe) {
|
||||
return pe.HTTPStatusCode == http.StatusPreconditionFailed
|
||||
}
|
||||
var ae s3err.APIError
|
||||
return errors.As(err, &ae) && ae.HTTPStatusCode == http.StatusPreconditionFailed
|
||||
}
|
||||
|
||||
type raceOutcome struct {
|
||||
idx int
|
||||
etag string
|
||||
err error
|
||||
}
|
||||
|
||||
// runRace launches one goroutine per writer, releases them all through a
|
||||
// common start barrier, and collects the outcome of each writer.
|
||||
func runRace(t *testing.T, n int, writer func(i int) (string, error)) []raceOutcome {
|
||||
t.Helper()
|
||||
start := make(chan struct{})
|
||||
var ready, done sync.WaitGroup
|
||||
results := make([]raceOutcome, n)
|
||||
for i := range n {
|
||||
ready.Add(1)
|
||||
done.Add(1)
|
||||
go func(i int) {
|
||||
defer done.Done()
|
||||
ready.Done()
|
||||
<-start
|
||||
etag, err := writer(i)
|
||||
results[i] = raceOutcome{idx: i, etag: etag, err: err}
|
||||
}(i)
|
||||
}
|
||||
ready.Wait()
|
||||
close(start)
|
||||
done.Wait()
|
||||
return results
|
||||
}
|
||||
|
||||
// classifyRace asserts exactly one winner and all losers 412, returning the
|
||||
// winner outcome.
|
||||
func classifyRace(t *testing.T, results []raceOutcome) raceOutcome {
|
||||
t.Helper()
|
||||
var winners []raceOutcome
|
||||
for _, r := range results {
|
||||
switch {
|
||||
case r.err == nil:
|
||||
winners = append(winners, r)
|
||||
case isPreconditionFailed(r.err):
|
||||
default:
|
||||
t.Errorf("writer %d: unexpected error (want nil or 412): %v", r.idx, r.err)
|
||||
}
|
||||
}
|
||||
if len(winners) != 1 {
|
||||
t.Fatalf("expected exactly 1 winner among %d classified writers, got %d",
|
||||
len(results), len(winners))
|
||||
}
|
||||
return winners[0]
|
||||
}
|
||||
|
||||
func raceBody(i int) []byte {
|
||||
return fmt.Appendf(nil, "conditional-race-writer-%02d-%s", i,
|
||||
string(bytes.Repeat([]byte{byte('a' + i%26)}, 64)))
|
||||
}
|
||||
|
||||
func TestPosixConditionalPutIfNoneMatchRace(t *testing.T) {
|
||||
for mode, mkMeta := range metaModes(t) {
|
||||
t.Run(mode, func(t *testing.T) {
|
||||
p := newTestPosix(t, mkMeta)
|
||||
bucket := "testbucket"
|
||||
createTestBucket(t, p, bucket)
|
||||
|
||||
for round := range 3 {
|
||||
key := fmt.Sprintf("obj-inm-%d", round)
|
||||
results := runRace(t, condRaceWriters, func(i int) (string, error) {
|
||||
res, err := testPut(p, bucket, key, raceBody(i), nil, aws.String("*"))
|
||||
return res.ETag, err
|
||||
})
|
||||
winner := classifyRace(t, results)
|
||||
|
||||
data, etag := getTestObject(t, p, bucket, key)
|
||||
if !bytes.Equal(data, raceBody(winner.idx)) {
|
||||
t.Errorf("round %d: final object bytes do not match winner %d",
|
||||
round, winner.idx)
|
||||
}
|
||||
if etag != winner.etag {
|
||||
t.Errorf("round %d: final etag %q does not match winner etag %q",
|
||||
round, etag, winner.etag)
|
||||
}
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestPosixConditionalPutIfMatchRace(t *testing.T) {
|
||||
for mode, mkMeta := range metaModes(t) {
|
||||
t.Run(mode, func(t *testing.T) {
|
||||
p := newTestPosix(t, mkMeta)
|
||||
bucket := "testbucket"
|
||||
key := "obj-ifmatch"
|
||||
createTestBucket(t, p, bucket)
|
||||
|
||||
seed, err := testPut(p, bucket, key, []byte("seed"), nil, nil)
|
||||
if err != nil {
|
||||
t.Fatalf("seed put: %v", err)
|
||||
}
|
||||
curEtag := seed.ETag
|
||||
|
||||
// several rounds to make timing-sensitive failures visible;
|
||||
// each round races against the previous round's winner etag.
|
||||
// bodies are unique per round and per writer so that every
|
||||
// successful write necessarily changes the etag
|
||||
for round := range 5 {
|
||||
raceBody := func(i int) []byte {
|
||||
return fmt.Appendf(nil, "round-%d-%s", round, raceBody(i))
|
||||
}
|
||||
results := runRace(t, condRaceWriters, func(i int) (string, error) {
|
||||
res, err := testPut(p, bucket, key, raceBody(i), aws.String(trimEtag(curEtag)), nil)
|
||||
return res.ETag, err
|
||||
})
|
||||
winner := classifyRace(t, results)
|
||||
|
||||
data, etag := getTestObject(t, p, bucket, key)
|
||||
if !bytes.Equal(data, raceBody(winner.idx)) {
|
||||
t.Errorf("round %d: final object bytes do not match winner %d",
|
||||
round, winner.idx)
|
||||
}
|
||||
if etag != winner.etag {
|
||||
t.Errorf("round %d: final etag %q does not match winner etag %q",
|
||||
round, etag, winner.etag)
|
||||
}
|
||||
curEtag = winner.etag
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestPosixConditionalPutSequential(t *testing.T) {
|
||||
for mode, mkMeta := range metaModes(t) {
|
||||
t.Run(mode, func(t *testing.T) {
|
||||
p := newTestPosix(t, mkMeta)
|
||||
bucket := "testbucket"
|
||||
createTestBucket(t, p, bucket)
|
||||
|
||||
// if-none-match on absent object succeeds
|
||||
res, err := testPut(p, bucket, "obj", []byte("first"), nil, aws.String("*"))
|
||||
if err != nil {
|
||||
t.Fatalf("if-none-match create: %v", err)
|
||||
}
|
||||
|
||||
// if-none-match on existing object fails 412
|
||||
_, err = testPut(p, bucket, "obj", []byte("second"), nil, aws.String("*"))
|
||||
if !isPreconditionFailed(err) {
|
||||
t.Errorf("if-none-match on existing object: want 412, got %v", err)
|
||||
}
|
||||
|
||||
// stale if-match fails 412
|
||||
_, err = testPut(p, bucket, "obj", []byte("third"),
|
||||
aws.String("\"deadbeefdeadbeefdeadbeefdeadbeef\""), nil)
|
||||
if !isPreconditionFailed(err) {
|
||||
t.Errorf("stale if-match: want 412, got %v", err)
|
||||
}
|
||||
|
||||
// correct if-match succeeds
|
||||
res2, err := testPut(p, bucket, "obj", []byte("fourth"), aws.String(trimEtag(res.ETag)), nil)
|
||||
if err != nil {
|
||||
t.Fatalf("valid if-match: %v", err)
|
||||
}
|
||||
data, etag := getTestObject(t, p, bucket, "obj")
|
||||
if string(data) != "fourth" || etag != res2.ETag {
|
||||
t.Errorf("unexpected final state after if-match put")
|
||||
}
|
||||
|
||||
// if-match on missing object keeps existing project behavior (NoSuchKey)
|
||||
_, err = testPut(p, bucket, "missing", []byte("x"), aws.String(trimEtag(res2.ETag)), nil)
|
||||
if !errors.Is(err, s3err.GetAPIError(s3err.ErrNoSuchKey)) {
|
||||
t.Errorf("if-match on missing object: want NoSuchKey, got %v", err)
|
||||
}
|
||||
|
||||
// object content untouched by the failed attempts
|
||||
data, _ = getTestObject(t, p, bucket, "obj")
|
||||
if string(data) != "fourth" {
|
||||
t.Errorf("failed conditional puts modified the object: %q", data)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
// TestPosixConditionalPutUnconditionalRace races an unconditional PUT with an
|
||||
// If-None-Match create. Per S3 semantics some serial order must exist: the
|
||||
// conditional writer may win or lose, but since the unconditional write always
|
||||
// succeeds and any valid serialization orders the unconditional write after a
|
||||
// successful conditional create (otherwise the conditional write would have
|
||||
// observed the object and failed), the final object must always be the
|
||||
// unconditional writer's.
|
||||
func TestPosixConditionalPutUnconditionalRace(t *testing.T) {
|
||||
for mode, mkMeta := range metaModes(t) {
|
||||
t.Run(mode, func(t *testing.T) {
|
||||
p := newTestPosix(t, mkMeta)
|
||||
bucket := "testbucket"
|
||||
createTestBucket(t, p, bucket)
|
||||
|
||||
for round := range 10 {
|
||||
key := fmt.Sprintf("obj-mixed-%d", round)
|
||||
uncondBody := fmt.Appendf(nil, "unconditional-%d", round)
|
||||
condBody := fmt.Appendf(nil, "conditional-%d", round)
|
||||
|
||||
var uncondEtag string
|
||||
results := runRace(t, 2, func(i int) (string, error) {
|
||||
if i == 0 {
|
||||
res, err := testPut(p, bucket, key, uncondBody, nil, nil)
|
||||
uncondEtag = res.ETag
|
||||
return res.ETag, err
|
||||
}
|
||||
res, err := testPut(p, bucket, key, condBody, nil, aws.String("*"))
|
||||
return res.ETag, err
|
||||
})
|
||||
|
||||
if results[0].err != nil {
|
||||
t.Fatalf("round %d: unconditional put failed: %v", round, results[0].err)
|
||||
}
|
||||
if results[1].err != nil && !isPreconditionFailed(results[1].err) {
|
||||
t.Fatalf("round %d: conditional put unexpected error: %v", round, results[1].err)
|
||||
}
|
||||
|
||||
data, etag := getTestObject(t, p, bucket, key)
|
||||
if !bytes.Equal(data, uncondBody) || etag != uncondEtag {
|
||||
t.Errorf("round %d: final object is not the unconditional writer's "+
|
||||
"(cond err=%v)", round, results[1].err)
|
||||
}
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestPosixConditionalPutDistinctKeysConcurrent(t *testing.T) {
|
||||
for mode, mkMeta := range metaModes(t) {
|
||||
t.Run(mode, func(t *testing.T) {
|
||||
p := newTestPosix(t, mkMeta)
|
||||
bucket := "testbucket"
|
||||
createTestBucket(t, p, bucket)
|
||||
|
||||
results := runRace(t, condRaceWriters, func(i int) (string, error) {
|
||||
key := fmt.Sprintf("distinct-key-%02d", i)
|
||||
res, err := testPut(p, bucket, key, raceBody(i), nil, aws.String("*"))
|
||||
return res.ETag, err
|
||||
})
|
||||
for _, r := range results {
|
||||
if r.err != nil {
|
||||
t.Errorf("writer %d to its own key failed: %v", r.idx, r.err)
|
||||
}
|
||||
}
|
||||
for i := range condRaceWriters {
|
||||
key := fmt.Sprintf("distinct-key-%02d", i)
|
||||
data, _ := getTestObject(t, p, bucket, key)
|
||||
if !bytes.Equal(data, raceBody(i)) {
|
||||
t.Errorf("key %s has wrong content", key)
|
||||
}
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func testCompleteMultipart(t *testing.T, p *Posix, bucket, key string, body []byte, ifMatch, ifNoneMatch *string) (string, func() (string, error)) {
|
||||
t.Helper()
|
||||
ctx := context.Background()
|
||||
mp, err := p.CreateMultipartUpload(ctx, s3response.CreateMultipartUploadInput{
|
||||
Bucket: &bucket,
|
||||
Key: &key,
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatalf("create multipart upload: %v", err)
|
||||
}
|
||||
uploadID := mp.UploadId
|
||||
part, err := p.UploadPart(ctx, &s3.UploadPartInput{
|
||||
Bucket: &bucket,
|
||||
Key: &key,
|
||||
UploadId: &uploadID,
|
||||
PartNumber: aws.Int32(1),
|
||||
ContentLength: aws.Int64(int64(len(body))),
|
||||
Body: bytes.NewReader(body),
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatalf("upload part: %v", err)
|
||||
}
|
||||
return uploadID, func() (string, error) {
|
||||
res, _, err := p.CompleteMultipartUpload(ctx, &s3.CompleteMultipartUploadInput{
|
||||
Bucket: &bucket,
|
||||
Key: &key,
|
||||
UploadId: &uploadID,
|
||||
MultipartUpload: &types.CompletedMultipartUpload{
|
||||
Parts: []types.CompletedPart{
|
||||
{ETag: part.ETag, PartNumber: aws.Int32(1)},
|
||||
},
|
||||
},
|
||||
IfMatch: ifMatch,
|
||||
IfNoneMatch: ifNoneMatch,
|
||||
})
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
return *res.ETag, nil
|
||||
}
|
||||
}
|
||||
|
||||
func TestPosixCompleteMultipartUploadIfNoneMatchRace(t *testing.T) {
|
||||
for mode, mkMeta := range metaModes(t) {
|
||||
t.Run(mode, func(t *testing.T) {
|
||||
p := newTestPosix(t, mkMeta)
|
||||
bucket := "testbucket"
|
||||
key := "obj-mp-inm"
|
||||
createTestBucket(t, p, bucket)
|
||||
|
||||
completes := make([]func() (string, error), condRaceWriters)
|
||||
bodies := make([][]byte, condRaceWriters)
|
||||
for i := range condRaceWriters {
|
||||
bodies[i] = raceBody(i)
|
||||
_, completes[i] = testCompleteMultipart(t, p, bucket, key,
|
||||
bodies[i], nil, aws.String("*"))
|
||||
}
|
||||
|
||||
results := runRace(t, condRaceWriters, func(i int) (string, error) {
|
||||
return completes[i]()
|
||||
})
|
||||
winner := classifyRace(t, results)
|
||||
|
||||
data, etag := getTestObject(t, p, bucket, key)
|
||||
if !bytes.Equal(data, bodies[winner.idx]) {
|
||||
t.Errorf("final object bytes do not match winning completer %d", winner.idx)
|
||||
}
|
||||
if etag != winner.etag {
|
||||
t.Errorf("final etag %q does not match winner etag %q", etag, winner.etag)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestPosixCompleteMultipartUploadIfMatchRace(t *testing.T) {
|
||||
for mode, mkMeta := range metaModes(t) {
|
||||
t.Run(mode, func(t *testing.T) {
|
||||
p := newTestPosix(t, mkMeta)
|
||||
bucket := "testbucket"
|
||||
key := "obj-mp-ifmatch"
|
||||
createTestBucket(t, p, bucket)
|
||||
|
||||
seed, err := testPut(p, bucket, key, []byte("seed"), nil, nil)
|
||||
if err != nil {
|
||||
t.Fatalf("seed put: %v", err)
|
||||
}
|
||||
|
||||
completes := make([]func() (string, error), condRaceWriters)
|
||||
bodies := make([][]byte, condRaceWriters)
|
||||
for i := range condRaceWriters {
|
||||
bodies[i] = raceBody(i)
|
||||
_, completes[i] = testCompleteMultipart(t, p, bucket, key,
|
||||
bodies[i], aws.String(trimEtag(seed.ETag)), nil)
|
||||
}
|
||||
|
||||
results := runRace(t, condRaceWriters, func(i int) (string, error) {
|
||||
return completes[i]()
|
||||
})
|
||||
winner := classifyRace(t, results)
|
||||
|
||||
data, etag := getTestObject(t, p, bucket, key)
|
||||
if !bytes.Equal(data, bodies[winner.idx]) {
|
||||
t.Errorf("final object bytes do not match winning completer %d", winner.idx)
|
||||
}
|
||||
if etag != winner.etag {
|
||||
t.Errorf("final etag %q does not match winner etag %q", etag, winner.etag)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
// strayTmpFiles returns any regular files under the bucket's temp dir that are
|
||||
// not part of the multipart staging area or the object lock directory. Any
|
||||
// leftover file here indicates a leaked staging tmpfile.
|
||||
func strayTmpFiles(t *testing.T, bucket string) []string {
|
||||
t.Helper()
|
||||
var stray []string
|
||||
root := filepath.Join(bucket, MetaTmpDir)
|
||||
err := filepath.WalkDir(root, func(path string, d fs.DirEntry, err error) error {
|
||||
if err != nil {
|
||||
if errors.Is(err, fs.ErrNotExist) {
|
||||
return nil
|
||||
}
|
||||
return err
|
||||
}
|
||||
if d.IsDir() {
|
||||
base := filepath.Base(path)
|
||||
if base == "multipart" || base == "objlock" {
|
||||
return filepath.SkipDir
|
||||
}
|
||||
return nil
|
||||
}
|
||||
stray = append(stray, path)
|
||||
return nil
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatalf("walk tmp dir: %v", err)
|
||||
}
|
||||
return stray
|
||||
}
|
||||
|
||||
func TestPosixConditionalPutFailureCleanup(t *testing.T) {
|
||||
for mode, mkMeta := range metaModes(t) {
|
||||
t.Run(mode, func(t *testing.T) {
|
||||
p := newTestPosix(t, mkMeta)
|
||||
bucket := "testbucket"
|
||||
key := "obj-cleanup"
|
||||
createTestBucket(t, p, bucket)
|
||||
|
||||
res, err := testPut(p, bucket, key, []byte("base"), nil, nil)
|
||||
if err != nil {
|
||||
t.Fatalf("seed put: %v", err)
|
||||
}
|
||||
|
||||
// failed precondition leaves no staging garbage behind
|
||||
_, err = testPut(p, bucket, key, []byte("cond"), nil, aws.String("*"))
|
||||
if !isPreconditionFailed(err) {
|
||||
t.Fatalf("want 412, got %v", err)
|
||||
}
|
||||
if stray := strayTmpFiles(t, bucket); len(stray) != 0 {
|
||||
t.Errorf("stray tmp files after 412: %v", stray)
|
||||
}
|
||||
|
||||
// injected publication failure: the post-process hook fails after
|
||||
// staging, before publication. The object must be unchanged, no
|
||||
// garbage left, and the key must still be writable afterwards
|
||||
// (i.e. no stale lock is left behind).
|
||||
_, err = p.PutObjectWithPostFunc(context.Background(), s3response.PutObjectInput{
|
||||
Bucket: &bucket,
|
||||
Key: &key,
|
||||
Body: bytes.NewReader([]byte("injected")),
|
||||
ContentLength: aws.Int64(int64(len("injected"))),
|
||||
IfMatch: aws.String(trimEtag(res.ETag)),
|
||||
}, func(*os.File) error { return errors.New("injected publication failure") })
|
||||
if err == nil {
|
||||
t.Fatalf("expected injected publication failure to propagate")
|
||||
}
|
||||
data, _ := getTestObject(t, p, bucket, key)
|
||||
if string(data) != "base" {
|
||||
t.Errorf("object modified by failed publication: %q", data)
|
||||
}
|
||||
if stray := strayTmpFiles(t, bucket); len(stray) != 0 {
|
||||
t.Errorf("stray tmp files after injected failure: %v", stray)
|
||||
}
|
||||
|
||||
// key still writable, conditional protocol still functional
|
||||
res2, err := testPut(p, bucket, key, []byte("after"), aws.String(trimEtag(res.ETag)), nil)
|
||||
if err != nil {
|
||||
t.Fatalf("put after injected failure: %v", err)
|
||||
}
|
||||
data, etag := getTestObject(t, p, bucket, key)
|
||||
if string(data) != "after" || etag != res2.ETag {
|
||||
t.Errorf("unexpected final state after recovery put")
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,414 @@
|
||||
// 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 conditionalrace is an external, S3-level verification of
|
||||
// conditional-write atomicity against a live versitygw process. It sends real
|
||||
// SigV4-authenticated requests through the standard AWS SDK and replicates
|
||||
// the algorithm of Buzz's git object-store A3 conformance probe
|
||||
// (https://github.com/block/buzz, crates/buzz-relay/src/api/git/store.rs):
|
||||
//
|
||||
// - sequential object round-trip with etag consistency checks
|
||||
// - 32-way concurrent If-Match races, expecting exactly 1×2xx + 31×412
|
||||
// - 32-way concurrent If-None-Match: * races, expecting 1×2xx + 31×412
|
||||
// - three rounds of each race
|
||||
//
|
||||
// It also verifies ordinary operations: bucket create/stat, PUT, HEAD, GET,
|
||||
// DELETE, and a 20 MiB multipart upload with SHA-256 download comparison.
|
||||
//
|
||||
// The test is skipped unless the following environment variables are set
|
||||
// (credentials are never logged or persisted):
|
||||
//
|
||||
// VERSITY_ENDPOINT e.g. http://127.0.0.1:7070
|
||||
// VERSITY_ACCESS_KEY
|
||||
// VERSITY_SECRET_KEY
|
||||
// VERSITY_BUCKET
|
||||
//
|
||||
// Optionally, VERSITY_ENDPOINT2 may point to a second gateway process
|
||||
// sharing the same backend directory; racing writers are then spread across
|
||||
// both processes, which verifies cross-process (filesystem lock based)
|
||||
// conditional-write exclusion.
|
||||
package conditionalrace
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"crypto/rand"
|
||||
"crypto/sha256"
|
||||
"errors"
|
||||
"fmt"
|
||||
"io"
|
||||
"net/http"
|
||||
"os"
|
||||
"sync"
|
||||
"testing"
|
||||
|
||||
"github.com/aws/aws-sdk-go-v2/aws"
|
||||
awshttp "github.com/aws/aws-sdk-go-v2/aws/transport/http"
|
||||
"github.com/aws/aws-sdk-go-v2/credentials"
|
||||
"github.com/aws/aws-sdk-go-v2/service/s3"
|
||||
"github.com/aws/aws-sdk-go-v2/service/s3/types"
|
||||
)
|
||||
|
||||
const (
|
||||
raceWriters = 32
|
||||
raceRounds = 3
|
||||
)
|
||||
|
||||
type liveEnv struct {
|
||||
client *s3.Client
|
||||
clients []*s3.Client
|
||||
bucket string
|
||||
}
|
||||
|
||||
// raceClient returns the client a racing writer should use, spreading
|
||||
// writers across all configured gateway endpoints.
|
||||
func (e *liveEnv) raceClient(i int) *s3.Client {
|
||||
return e.clients[i%len(e.clients)]
|
||||
}
|
||||
|
||||
func newLiveEnv(t *testing.T) *liveEnv {
|
||||
t.Helper()
|
||||
endpoint := os.Getenv("VERSITY_ENDPOINT")
|
||||
access := os.Getenv("VERSITY_ACCESS_KEY")
|
||||
secret := os.Getenv("VERSITY_SECRET_KEY")
|
||||
bucket := os.Getenv("VERSITY_BUCKET")
|
||||
if endpoint == "" || access == "" || secret == "" || bucket == "" {
|
||||
t.Skip("VERSITY_ENDPOINT, VERSITY_ACCESS_KEY, VERSITY_SECRET_KEY, " +
|
||||
"and VERSITY_BUCKET must be set for the live S3 race test")
|
||||
}
|
||||
|
||||
newClient := func(endpoint string) *s3.Client {
|
||||
return s3.New(s3.Options{
|
||||
BaseEndpoint: aws.String(endpoint),
|
||||
Region: "us-east-1",
|
||||
UsePathStyle: true,
|
||||
Credentials: credentials.NewStaticCredentialsProvider(access, secret, ""),
|
||||
})
|
||||
}
|
||||
|
||||
client := newClient(endpoint)
|
||||
clients := []*s3.Client{client}
|
||||
if endpoint2 := os.Getenv("VERSITY_ENDPOINT2"); endpoint2 != "" {
|
||||
clients = append(clients, newClient(endpoint2))
|
||||
}
|
||||
|
||||
env := &liveEnv{client: client, clients: clients, bucket: bucket}
|
||||
env.ensureBucket(t)
|
||||
return env
|
||||
}
|
||||
|
||||
func (e *liveEnv) ensureBucket(t *testing.T) {
|
||||
t.Helper()
|
||||
ctx := context.Background()
|
||||
_, err := e.client.HeadBucket(ctx, &s3.HeadBucketInput{Bucket: &e.bucket})
|
||||
if err == nil {
|
||||
return
|
||||
}
|
||||
_, err = e.client.CreateBucket(ctx, &s3.CreateBucketInput{Bucket: &e.bucket})
|
||||
if err != nil {
|
||||
t.Fatalf("create bucket: %v", err)
|
||||
}
|
||||
_, err = e.client.HeadBucket(ctx, &s3.HeadBucketInput{Bucket: &e.bucket})
|
||||
if err != nil {
|
||||
t.Fatalf("stat bucket after create: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
func httpStatus(err error) int {
|
||||
var re *awshttp.ResponseError
|
||||
if errors.As(err, &re) {
|
||||
return re.HTTPStatusCode()
|
||||
}
|
||||
return 0
|
||||
}
|
||||
|
||||
func (e *liveEnv) get(t *testing.T, key string) ([]byte, string) {
|
||||
t.Helper()
|
||||
out, err := e.client.GetObject(context.Background(), &s3.GetObjectInput{
|
||||
Bucket: &e.bucket,
|
||||
Key: &key,
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatalf("get %q: %v", key, err)
|
||||
}
|
||||
defer out.Body.Close()
|
||||
data, err := io.ReadAll(out.Body)
|
||||
if err != nil {
|
||||
t.Fatalf("read %q: %v", key, err)
|
||||
}
|
||||
return data, aws.ToString(out.ETag)
|
||||
}
|
||||
|
||||
type outcome struct {
|
||||
idx int
|
||||
etag string
|
||||
err error
|
||||
status int
|
||||
}
|
||||
|
||||
// racePut launches n concurrent conditional PUTs released through a common
|
||||
// start barrier.
|
||||
func racePut(t *testing.T, n int, put func(i int) (string, error)) []outcome {
|
||||
t.Helper()
|
||||
start := make(chan struct{})
|
||||
var ready, done sync.WaitGroup
|
||||
results := make([]outcome, n)
|
||||
for i := range n {
|
||||
ready.Add(1)
|
||||
done.Add(1)
|
||||
go func(i int) {
|
||||
defer done.Done()
|
||||
ready.Done()
|
||||
<-start
|
||||
etag, err := put(i)
|
||||
results[i] = outcome{idx: i, etag: etag, err: err, status: httpStatus(err)}
|
||||
}(i)
|
||||
}
|
||||
ready.Wait()
|
||||
close(start)
|
||||
done.Wait()
|
||||
return results
|
||||
}
|
||||
|
||||
func classify(t *testing.T, results []outcome) outcome {
|
||||
t.Helper()
|
||||
var winners []outcome
|
||||
failed := 0
|
||||
for _, r := range results {
|
||||
switch {
|
||||
case r.err == nil:
|
||||
winners = append(winners, r)
|
||||
case r.status == http.StatusPreconditionFailed:
|
||||
failed++
|
||||
default:
|
||||
t.Errorf("writer %d: unexpected error (want 2xx or 412): %v", r.idx, r.err)
|
||||
}
|
||||
}
|
||||
if len(winners) != 1 || failed != len(results)-1 {
|
||||
t.Fatalf("expected exactly 1×2xx + %d×412 among %d classified observers, got %d×2xx + %d×412",
|
||||
len(results)-1, len(results), len(winners), failed)
|
||||
}
|
||||
return winners[0]
|
||||
}
|
||||
|
||||
func body(round, i int) []byte {
|
||||
return fmt.Appendf(nil, "buzz-a3-replica-round-%d-writer-%02d-%s",
|
||||
round, i, string(bytes.Repeat([]byte{byte('a' + i%26)}, 128)))
|
||||
}
|
||||
|
||||
// TestLiveSequentialRoundTrip is the sequential portion of the Buzz probe:
|
||||
// create-only PUT, GET with etag consistency, and delete.
|
||||
func TestLiveSequentialRoundTrip(t *testing.T) {
|
||||
e := newLiveEnv(t)
|
||||
ctx := context.Background()
|
||||
key := "cond-race/sequential-roundtrip"
|
||||
content := []byte("sequential-round-trip-payload")
|
||||
|
||||
_, _ = e.client.DeleteObject(ctx, &s3.DeleteObjectInput{Bucket: &e.bucket, Key: &key})
|
||||
|
||||
put, err := e.client.PutObject(ctx, &s3.PutObjectInput{
|
||||
Bucket: &e.bucket,
|
||||
Key: &key,
|
||||
Body: bytes.NewReader(content),
|
||||
IfNoneMatch: aws.String("*"),
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatalf("create-only put: %v", err)
|
||||
}
|
||||
|
||||
// duplicate create-only put must fail 412
|
||||
_, err = e.client.PutObject(ctx, &s3.PutObjectInput{
|
||||
Bucket: &e.bucket,
|
||||
Key: &key,
|
||||
Body: bytes.NewReader([]byte("other")),
|
||||
IfNoneMatch: aws.String("*"),
|
||||
})
|
||||
if httpStatus(err) != http.StatusPreconditionFailed {
|
||||
t.Fatalf("duplicate create-only put: want 412, got %v", err)
|
||||
}
|
||||
|
||||
// stale if-match must fail 412
|
||||
_, err = e.client.PutObject(ctx, &s3.PutObjectInput{
|
||||
Bucket: &e.bucket,
|
||||
Key: &key,
|
||||
Body: bytes.NewReader([]byte("other")),
|
||||
IfMatch: aws.String("\"deadbeefdeadbeefdeadbeefdeadbeef\""),
|
||||
})
|
||||
if httpStatus(err) != http.StatusPreconditionFailed {
|
||||
t.Fatalf("stale if-match put: want 412, got %v", err)
|
||||
}
|
||||
|
||||
data, etag := e.get(t, key)
|
||||
if !bytes.Equal(data, content) {
|
||||
t.Fatalf("round-trip content mismatch")
|
||||
}
|
||||
if etag != aws.ToString(put.ETag) {
|
||||
t.Fatalf("round-trip etag mismatch: put %q get %q", aws.ToString(put.ETag), etag)
|
||||
}
|
||||
|
||||
head, err := e.client.HeadObject(ctx, &s3.HeadObjectInput{Bucket: &e.bucket, Key: &key})
|
||||
if err != nil {
|
||||
t.Fatalf("head object: %v", err)
|
||||
}
|
||||
if aws.ToString(head.ETag) != etag {
|
||||
t.Fatalf("head etag mismatch")
|
||||
}
|
||||
|
||||
_, err = e.client.DeleteObject(ctx, &s3.DeleteObjectInput{Bucket: &e.bucket, Key: &key})
|
||||
if err != nil {
|
||||
t.Fatalf("delete object: %v", err)
|
||||
}
|
||||
_, err = e.client.HeadObject(ctx, &s3.HeadObjectInput{Bucket: &e.bucket, Key: &key})
|
||||
if httpStatus(err) != http.StatusNotFound {
|
||||
t.Fatalf("head after delete: want 404, got %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
// TestLiveIfMatchRace replicates the Buzz probe's if_match_race phase.
|
||||
func TestLiveIfMatchRace(t *testing.T) {
|
||||
e := newLiveEnv(t)
|
||||
ctx := context.Background()
|
||||
key := "cond-race/if-match-race"
|
||||
|
||||
for round := range raceRounds {
|
||||
seedBody := fmt.Appendf(nil, "if-match-seed-round-%d", round)
|
||||
seed, err := e.client.PutObject(ctx, &s3.PutObjectInput{
|
||||
Bucket: &e.bucket,
|
||||
Key: &key,
|
||||
Body: bytes.NewReader(seedBody),
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatalf("round %d: seed put: %v", round, err)
|
||||
}
|
||||
|
||||
results := racePut(t, raceWriters, func(i int) (string, error) {
|
||||
res, err := e.raceClient(i).PutObject(ctx, &s3.PutObjectInput{
|
||||
Bucket: &e.bucket,
|
||||
Key: &key,
|
||||
Body: bytes.NewReader(body(round, i)),
|
||||
IfMatch: seed.ETag,
|
||||
})
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
return aws.ToString(res.ETag), nil
|
||||
})
|
||||
winner := classify(t, results)
|
||||
|
||||
data, etag := e.get(t, key)
|
||||
if !bytes.Equal(data, body(round, winner.idx)) {
|
||||
t.Errorf("round %d: final bytes are not the winner's", round)
|
||||
}
|
||||
if etag != winner.etag {
|
||||
t.Errorf("round %d: final etag %q != winner etag %q", round, etag, winner.etag)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// TestLiveIfNoneMatchRace replicates the Buzz probe's if_none_match_race phase.
|
||||
func TestLiveIfNoneMatchRace(t *testing.T) {
|
||||
e := newLiveEnv(t)
|
||||
ctx := context.Background()
|
||||
|
||||
for round := range raceRounds {
|
||||
key := fmt.Sprintf("cond-race/if-none-match-race-%d", round)
|
||||
_, _ = e.client.DeleteObject(ctx, &s3.DeleteObjectInput{Bucket: &e.bucket, Key: &key})
|
||||
|
||||
results := racePut(t, raceWriters, func(i int) (string, error) {
|
||||
res, err := e.raceClient(i).PutObject(ctx, &s3.PutObjectInput{
|
||||
Bucket: &e.bucket,
|
||||
Key: &key,
|
||||
Body: bytes.NewReader(body(round, i)),
|
||||
IfNoneMatch: aws.String("*"),
|
||||
})
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
return aws.ToString(res.ETag), nil
|
||||
})
|
||||
winner := classify(t, results)
|
||||
|
||||
data, etag := e.get(t, key)
|
||||
if !bytes.Equal(data, body(round, winner.idx)) {
|
||||
t.Errorf("round %d: final bytes are not the winner's", round)
|
||||
}
|
||||
if etag != winner.etag {
|
||||
t.Errorf("round %d: final etag %q != winner etag %q", round, etag, winner.etag)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// TestLiveBasicOpsAndMultipart verifies ordinary operations still work:
|
||||
// PUT/HEAD/GET/DELETE and a 20 MiB multipart upload with SHA-256 comparison.
|
||||
func TestLiveBasicOpsAndMultipart(t *testing.T) {
|
||||
e := newLiveEnv(t)
|
||||
ctx := context.Background()
|
||||
|
||||
key := "cond-race/multipart-20mib"
|
||||
const partSize = 5 * 1024 * 1024
|
||||
const numParts = 4
|
||||
|
||||
payload := make([]byte, partSize*numParts)
|
||||
if _, err := rand.Read(payload); err != nil {
|
||||
t.Fatalf("generate payload: %v", err)
|
||||
}
|
||||
wantSum := sha256.Sum256(payload)
|
||||
|
||||
mp, err := e.client.CreateMultipartUpload(ctx, &s3.CreateMultipartUploadInput{
|
||||
Bucket: &e.bucket,
|
||||
Key: &key,
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatalf("create multipart upload: %v", err)
|
||||
}
|
||||
|
||||
var completed []types.CompletedPart
|
||||
for part := 1; part <= numParts; part++ {
|
||||
chunk := payload[(part-1)*partSize : part*partSize]
|
||||
res, err := e.client.UploadPart(ctx, &s3.UploadPartInput{
|
||||
Bucket: &e.bucket,
|
||||
Key: &key,
|
||||
UploadId: mp.UploadId,
|
||||
PartNumber: aws.Int32(int32(part)),
|
||||
Body: bytes.NewReader(chunk),
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatalf("upload part %d: %v", part, err)
|
||||
}
|
||||
completed = append(completed, types.CompletedPart{
|
||||
ETag: res.ETag,
|
||||
PartNumber: aws.Int32(int32(part)),
|
||||
})
|
||||
}
|
||||
|
||||
_, err = e.client.CompleteMultipartUpload(ctx, &s3.CompleteMultipartUploadInput{
|
||||
Bucket: &e.bucket,
|
||||
Key: &key,
|
||||
UploadId: mp.UploadId,
|
||||
MultipartUpload: &types.CompletedMultipartUpload{Parts: completed},
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatalf("complete multipart upload: %v", err)
|
||||
}
|
||||
|
||||
data, _ := e.get(t, key)
|
||||
if gotSum := sha256.Sum256(data); gotSum != wantSum {
|
||||
t.Fatalf("multipart download sha256 mismatch (%d bytes)", len(data))
|
||||
}
|
||||
|
||||
_, err = e.client.DeleteObject(ctx, &s3.DeleteObjectInput{Bucket: &e.bucket, Key: &key})
|
||||
if err != nil {
|
||||
t.Fatalf("delete multipart object: %v", err)
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user