diff --git a/backend/posix/objlock.go b/backend/posix/objlock.go new file mode 100644 index 00000000..e0ba2331 --- /dev/null +++ b/backend/posix/objlock.go @@ -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/, 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 +) diff --git a/backend/posix/objlock_unix.go b/backend/posix/objlock_unix.go new file mode 100644 index 00000000..fe667723 --- /dev/null +++ b/backend/posix/objlock_unix.go @@ -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) + } +} diff --git a/backend/posix/objlock_windows.go b/backend/posix/objlock_windows.go new file mode 100644 index 00000000..5a6740ee --- /dev/null +++ b/backend/posix/objlock_windows.go @@ -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) + } +} diff --git a/backend/posix/posix.go b/backend/posix/posix.go index 6259c74c..725ed94d 100644 --- a/backend/posix/posix.go +++ b/backend/posix/posix.go @@ -128,6 +128,15 @@ type Posix struct { // multipart CRC64NVME: "CRC64NVME-" // multipart composite: "ALGO--" 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{ diff --git a/backend/posix/posix_conditional_put_test.go b/backend/posix/posix_conditional_put_test.go new file mode 100644 index 00000000..ad94362e --- /dev/null +++ b/backend/posix/posix_conditional_put_test.go @@ -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") + } + }) + } +} diff --git a/tests/conditionalrace/conditional_race_test.go b/tests/conditionalrace/conditional_race_test.go new file mode 100644 index 00000000..331e6eb4 --- /dev/null +++ b/tests/conditionalrace/conditional_race_test.go @@ -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) + } +}