From 210fd568cce92f703d2a5e4b3d675c78ed2589ee Mon Sep 17 00:00:00 2001 From: Denis Babchuk Date: Sat, 7 Nov 2026 11:12:39 +0300 Subject: [PATCH 1/4] test(pools): cover conditional NewMultipartUpload across pools Signed-off-by: Denis Babchuk --- ...ure-server-pool-multipart-initiate_test.go | 194 ++++++++++++++++++ 1 file changed, 194 insertions(+) create mode 100644 cmd/erasure-server-pool-multipart-initiate_test.go diff --git a/cmd/erasure-server-pool-multipart-initiate_test.go b/cmd/erasure-server-pool-multipart-initiate_test.go new file mode 100644 index 000000000..c4f81c893 --- /dev/null +++ b/cmd/erasure-server-pool-multipart-initiate_test.go @@ -0,0 +1,194 @@ +// Copyright (c) 2026 Denis Babchuk +// +// This file is part of Silo Object Storage stack +// +// This program is free software: you can redistribute it and/or modify +// it under the terms of the GNU Affero General Public License as published by +// the Free Software Foundation, either version 3 of the License, or +// (at your option) any later version. +// +// This program is distributed in the hope that it will be useful, +// but WITHOUT ANY WARRANTY; without even the implied warranty of +// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +// GNU Affero General Public License for more details. +// +// You should have received a copy of the GNU Affero General Public License +// along with this program. If not, see . + +package cmd + +import ( + "fmt" + "net/http" + "slices" + "strings" + "testing" + "time" + + xhttp "github.com/minio/minio/internal/http" +) + +// multipartInitiateUploads returns the number of pending uploads of object in +// each pool. +func multipartInitiateUploads(t *testing.T, z *erasureServerPools, bucket, object string) []int { + t.Helper() + counts := make([]int, len(z.serverPools)) + for i, pool := range z.serverPools { + res, err := pool.listMultipartUploadsExact(t.Context(), bucket, object) + if err != nil { + t.Fatal(err) + } + counts[i] = len(res.Uploads) + } + return counts +} + +// A multi-pool NewMultipartUpload goes to the pool of a pending upload of the +// key or, without one, to the pool chosen by capacity. Neither has to hold the +// current object, so a condition sent with the initiate must be evaluated +// against the current object across pools, as for PUT and completion. +// Routing itself does not change: an accepted upload lands in the same pool. +// Allocation follows a current delete marker to its pool, so a stale version +// under a marker in another pool is reached only through a pending upload. +func TestPoolsMultipartConditionalInitiateHTTP(t *testing.T) { + const wrongETag = "ffffffffffffffffffffffffffffffff" + for _, mode := range []string{"unversioned", "versioned"} { + t.Run(mode, func(t *testing.T) { + z, _ := consistencyPools(t) + bucket, router := conditionalPutBucket(t, z, mode) + versioned := mode == "versioned" + for _, route := range []string{"pending-upload", "allocation"} { + for target := range 2 { + for _, layout := range []string{"current-in-other-pool", "stale-copy-in-target", "marker-in-other-pool", "control-current-in-target"} { + if layout == "marker-in-other-pool" && (!versioned || route == "allocation") { + continue + } + for _, cond := range []string{"if-match-current", "if-match-any", "if-match-stale", "if-none-match-any"} { + t.Run(fmt.Sprintf("%s/target=%d/%s/%s", route, target, layout, cond), func(t *testing.T) { + object := fmt.Sprintf("%s-%d-%s-%s", route, target, layout, cond) + other := 1 - target + opts := func(age time.Duration) ObjectOptions { + return ObjectOptions{Versioned: versioned, MTime: UTCNow().Add(-age)} + } + present, staleETag := true, wrongETag + var current ObjectInfo + switch layout { + case "current-in-other-pool": + current = putConsistencyObject(t, z, bucket, object, other, "current", opts(time.Minute)) + case "stale-copy-in-target": + staleETag = putConsistencyObject(t, z, bucket, object, target, "stale", opts(time.Hour)).ETag + current = putConsistencyObject(t, z, bucket, object, other, "current", opts(time.Minute)) + case "marker-in-other-pool": + // The client knows the version under the marker. + current = putConsistencyObject(t, z, bucket, object, target, "deleted", opts(time.Hour)) + if _, err := z.serverPools[other].DeleteObject(t.Context(), bucket, object, ObjectOptions{Versioned: true, VersionID: mustGetUUID(), DeleteMarker: true, MTime: UTCNow().Add(-time.Minute)}); err != nil { + t.Fatal(err) + } + present = false + case "control-current-in-target": + staleETag = putConsistencyObject(t, z, bucket, object, other, "stale", opts(time.Hour)).ETag + current = putConsistencyObject(t, z, bucket, object, target, "current", opts(time.Minute)) + } + + before := make([]int, len(z.serverPools)) + if route == "pending-upload" { + if _, err := z.serverPools[target].NewMultipartUpload(t.Context(), bucket, object, ObjectOptions{Versioned: versioned}); err != nil { + t.Fatal(err) + } + before[target] = 1 + } else { + defer conditionalPutPool(t, z, object, target)() + if idx, err := z.getWritePoolIdx(t.Context(), bucket, object, -1, false); err != nil || idx != target { + t.Fatalf("allocation target=%d: idx=%d err=%v", target, idx, err) + } + } + + var headers map[string]string + switch cond { + case "if-match-current": + headers = map[string]string{xhttp.IfMatch: fmt.Sprintf("%q", current.ETag)} + case "if-match-any": + headers = map[string]string{xhttp.IfMatch: "*"} + case "if-match-stale": + headers = map[string]string{xhttp.IfMatch: fmt.Sprintf("%q", staleETag)} + case "if-none-match-any": + headers = map[string]string{xhttp.IfNoneMatch: "*"} + } + want := http.StatusOK + switch { + case !present && strings.HasPrefix(cond, "if-match"): + want = http.StatusNotFound + case !present: + case cond == "if-match-stale", cond == "if-none-match-any": + want = http.StatusPreconditionFailed + } + + rec := multipartConditionRequest(t, router, http.MethodPost, getNewMultipartURL("", bucket, object), "", headers) + after := multipartInitiateUploads(t, z, bucket, object) + t.Logf("initiate %v: HTTP %d, uploads per pool %v -> %v", headers, rec.Code, before, after) + if rec.Code != want { + t.Errorf("initiate HTTP %d, want %d: %s", rec.Code, want, rec.Body.String()) + } + wantAfter := slices.Clone(before) + if want == http.StatusOK { + wantAfter[target]++ + } + if !slices.Equal(after, wantAfter) { + t.Errorf("uploads per pool %v, want %v", after, wantAfter) + } + if rec.Code != want { + return + } + switch want { + case http.StatusPreconditionFailed: + multipartConditionError(t, rec, "PreconditionFailed") + if etag := multipartConditionResponseETag(rec); etag != current.ETag { + t.Errorf("412 describes ETag %s, want the current %s", etag, current.ETag) + } + case http.StatusNotFound: + multipartConditionError(t, rec, "NoSuchKey") + } + }) + } + } + } + } + }) + } +} + +// A pool that cannot be read may hold the newest object, so the condition +// cannot be evaluated even when the upload goes to a readable pool. +func TestPoolsMultipartConditionalInitiateUnreadable(t *testing.T) { + z, _ := consistencyPools(t) + bucket, router := conditionalPutBucket(t, z, "unversioned") + for faultPool := range 2 { + for _, match := range []bool{false, true} { + t.Run(fmt.Sprintf("fault-pool=%d/match=%t", faultPool, match), func(t *testing.T) { + object := fmt.Sprintf("unreadable-%d-%t", faultPool, match) + putConsistencyObject(t, z, bucket, object, 0, "stale", ObjectOptions{MTime: UTCNow().Add(-time.Hour)}) + putConsistencyObject(t, z, bucket, object, 1, "current", ObjectOptions{MTime: UTCNow().Add(-time.Minute)}) + defer conditionalPutPool(t, z, object, 0)() + defer conditionalPutSwapDisks(z.serverPools[faultPool], object, func(disk StorageAPI) StorageAPI { + return consistencyReadFaultDisk{StorageAPI: disk, bucket: bucket, object: object} + })() + headers := map[string]string{xhttp.IfNoneMatch: "*"} + if match { + headers = map[string]string{xhttp.IfMatch: "*"} + } + rec := multipartConditionRequest(t, router, http.MethodPost, getNewMultipartURL("", bucket, object), "", headers) + if rec.Code != http.StatusServiceUnavailable { + t.Errorf("unverified initiate must fail: HTTP %d: %s", rec.Code, rec.Body.String()) + } + called := 0 + _, err := z.NewMultipartUpload(t.Context(), bucket, object, ObjectOptions{HasIfMatch: match, CheckPrecondFn: func(ObjectInfo) bool { called++; return false }}) + if !isErrReadQuorum(err) || called != 0 { + t.Errorf("initiate error=%v, callback calls=%d", err, called) + } + if n := multipartInitiateUploads(t, z, bucket, object); !slices.Equal(n, []int{0, 0}) { + t.Errorf("uploads per pool %v after failed initiates", n) + } + }) + } + } +} From 1516273162424ac44d906459a0fc2c260b49c4a9 Mon Sep 17 00:00:00 2001 From: Denis Babchuk Date: Sat, 7 Nov 2026 11:16:38 +0300 Subject: [PATCH 2/4] fix(pools): evaluate cross-pool NewMultipartUpload conditions against the current object Signed-off-by: Denis Babchuk --- cmd/erasure-server-pool-consistency.go | 35 +++++++++++ cmd/erasure-server-pool.go | 85 +++++++++++--------------- 2 files changed, 69 insertions(+), 51 deletions(-) diff --git a/cmd/erasure-server-pool-consistency.go b/cmd/erasure-server-pool-consistency.go index ad3e24d38..4840ddde7 100644 --- a/cmd/erasure-server-pool-consistency.go +++ b/cmd/erasure-server-pool-consistency.go @@ -63,6 +63,41 @@ func (z *erasureServerPools) objectPoolInfos(ctx context.Context, bucket, object return copies, nil } +// checkCurrentPreconditions evaluates the public write conditions in +// opts.CheckPrecondFn against the logical current object across pools, +// including draining pools. The pool that receives a write may be empty or +// hold a stale copy, so its local state cannot decide the condition. A +// current delete marker reads as an absent key: If-Match then fails with +// ObjectNotFound and If-None-Match may proceed. An unreadable pool may hold +// the newest object; it is not absence, and its error is returned without +// evaluating the condition. The caller must hold the pools-layer object lock +// and, once the condition is accepted, clear the callback so the destination +// does not evaluate it again against its local copy. +func (z *erasureServerPools) checkCurrentPreconditions(ctx context.Context, bucket, object string, opts ObjectOptions) error { + copies, err := z.objectPoolInfos(ctx, bucket, object, ObjectOptions{ + VersionID: "", // Compare the current object, not the write's version. + Versioned: opts.Versioned, + VersionSuspended: opts.VersionSuspended, + NoAuditLog: true, + }) + if err == nil && copies[0].ObjInfo.DeleteMarker { + err = toObjectErr(errFileNotFound, bucket, decodeDirObject(object)) + } + switch { + case err == nil: + if opts.CheckPrecondFn(copies[0].ObjInfo) { + return PreConditionFailed{} + } + return nil + case !isErrObjectNotFound(err) && !isErrVersionNotFound(err): + return err + case opts.HasIfMatch: + return err + default: + return nil + } +} + // checkDataMovementCopies rejects moving a version when another pool holds a // newer copy of it, which reads return instead. A cross-pool overwrite can // leave such a copy, and a client write can add one while the move copies. diff --git a/cmd/erasure-server-pool.go b/cmd/erasure-server-pool.go index d1e83bb71..946496ea3 100644 --- a/cmd/erasure-server-pool.go +++ b/cmd/erasure-server-pool.go @@ -1157,28 +1157,8 @@ func (z *erasureServerPools) PutObject(ctx context.Context, bucket string, objec // metadata reconciliation at the set layer. if opts.CheckPrecondFn != nil && !opts.ReplicationRequest && !opts.ReplicaLockReconcile && !opts.DataMovement { - copies, lerr := z.objectPoolInfos(ctx, bucket, object, ObjectOptions{ - VersionID: "", // Compare the current object, not the write's version. - Versioned: opts.Versioned, - VersionSuspended: opts.VersionSuspended, - NoAuditLog: true, - }) - var latest ObjectInfo - if lerr == nil { - latest = copies[0].ObjInfo - if latest.DeleteMarker { - lerr = toObjectErr(errFileNotFound, bucket, object) - } - } - // An unreadable pool may hold the newest object; it is not absence. - if lerr != nil && !isErrObjectNotFound(lerr) && !isErrVersionNotFound(lerr) { - return ObjectInfo{}, lerr - } - if lerr == nil && opts.CheckPrecondFn(latest) { - return ObjectInfo{}, PreConditionFailed{} - } - if lerr != nil && opts.HasIfMatch { - return ObjectInfo{}, lerr + if err := z.checkCurrentPreconditions(ctx, bucket, object, opts); err != nil { + return ObjectInfo{}, err } // Do not repeat an accepted condition against the destination's copy. opts.CheckPrecondFn = nil @@ -2094,6 +2074,33 @@ func (z *erasureServerPools) NewMultipartUpload(ctx context.Context, bucket, obj return z.serverPools[0].NewMultipartUpload(ctx, bucket, object, opts) } + // The upload goes to the pool of a pending upload of the key or to the + // pool selected by capacity; neither has to hold the current object. As + // for PutObject, public write conditions compare the logical current + // object under the pools-layer write lock, which is kept while the upload + // is placed. Replica and data-movement callbacks keep their set-layer + // semantics. + locked := false + if opts.CheckPrecondFn != nil && !opts.ReplicationRequest && + !opts.ReplicaLockReconcile && !opts.DataMovement { + if !opts.NoLock { + lk := z.NewNSLock(bucket, encodeDirObject(object)) + lkctx, err := lk.GetLock(ctx, globalOperationTimeout) + if err != nil { + return nil, err + } + ctx = lkctx.Context() + defer lk.Unlock(lkctx) + opts.NoLock = true + } + locked = true + if err := z.checkCurrentPreconditions(ctx, bucket, encodeDirObject(object), opts); err != nil { + return nil, err + } + // Do not repeat an accepted condition against the destination's copy. + opts.CheckPrecondFn = nil + } + for idx, pool := range z.serverPools { if z.IsSuspended(idx) || z.IsPoolRebalancing(idx) { continue @@ -2112,8 +2119,9 @@ func (z *erasureServerPools) NewMultipartUpload(ctx context.Context, bucket, obj } // any parallel writes on the object will block for this poolIdx - // to return since this holds a read lock on the namespace. - idx, err := z.getPoolIdx(ctx, bucket, object, -1) + // to return since this holds a read lock on the namespace, or the + // write lock taken above for a conditional upload. + idx, err := z.getWritePoolIdx(ctx, bucket, object, -1, locked) if err != nil { return nil, err } @@ -2334,33 +2342,8 @@ func (z *erasureServerPools) CompleteMultipartUpload(ctx context.Context, bucket // Once satisfied, the callback is cleared so the set layer does not // re-evaluate it against its local copy. if opts.CheckPrecondFn != nil { - copies, lerr := z.objectPoolInfos(ctx, bucket, encodeDirObject(object), ObjectOptions{ - // Conditions always compare the logical current object, - // independently of the completion's destination version. - VersionID: "", - Versioned: opts.Versioned, - VersionSuspended: opts.VersionSuspended, - NoAuditLog: true, - }) - var latest ObjectInfo - if lerr == nil { - latest = copies[0].ObjInfo - if latest.DeleteMarker { - // A delete-marker latest reads as an absent key, matching - // the set layer's getObjectInfo. - lerr = toObjectErr(errFileNotFound, bucket, object) - } - } - if lerr == nil && opts.CheckPrecondFn(latest) { - return ObjectInfo{}, PreConditionFailed{} - } - if lerr != nil && !isErrVersionNotFound(lerr) && !isErrObjectNotFound(lerr) { - return ObjectInfo{}, lerr - } - // if object doesn't exist return error for If-Match conditional requests - // If-None-Match should be allowed to proceed for non-existent objects - if lerr != nil && opts.HasIfMatch && (isErrObjectNotFound(lerr) || isErrVersionNotFound(lerr)) { - return ObjectInfo{}, lerr + if err := z.checkCurrentPreconditions(ctx, bucket, encodeDirObject(object), opts); err != nil { + return ObjectInfo{}, err } opts.CheckPrecondFn = nil } From 9d2be5daaa57c3502081b9b4d8de051343d9a2b1 Mon Sep 17 00:00:00 2001 From: Feng Ruohang Date: Sun, 8 Nov 2026 11:38:59 +0800 Subject: [PATCH 3/4] chore(batch): keep the policy snapshot out of msgp generation batchPolicySnapshot holds the IAM store, a context and a mutex. Without an ignore directive, go generate emitted msgp encoders for it that do not compile, so make check-gen (and the check-gen step of CI) failed on the generated batch-authz files. The committed generated files are unchanged. Co-Authored-By: Claude Opus 5.5 Signed-off-by: Feng Ruohang (cherry picked from commit a6571a53e9be790da34bc85cc737832c223c1d7e) --- cmd/batch-authz.go | 2 ++ 1 file changed, 2 insertions(+) diff --git a/cmd/batch-authz.go b/cmd/batch-authz.go index 7b7163b8d..59c74262d 100644 --- a/cmd/batch-authz.go +++ b/cmd/batch-authz.go @@ -572,6 +572,8 @@ const batchIAMLoadTimeout = 5 * time.Second // waiting for the cache lock, which taking the write lock back after each // read would cause. It writes nothing to the cache but the lazily loaded // mappings, which the request path stores under the read lock as well. +// +//msgp:ignore batchPolicySnapshot type batchPolicySnapshot struct { *IAMStoreSys ctx context.Context From af3064f2b54fe90c4b0f14a0c54c2507bf7b29d4 Mon Sep 17 00:00:00 2001 From: Feng Ruohang Date: Mon, 9 Nov 2026 13:16:12 +0800 Subject: [PATCH 4/4] test(pools): cover initiate locking and completion conditions Exercise concurrent overwrite and delete while initiation persists metadata, and completion after intervening object mutations. Check that failed conditions preserve uploaded parts for retry. Include conditional multipart regressions in the CI race set and record the multipart fix from Denis Babchuk in the contributor history. Signed-off-by: Feng Ruohang --- .github/workflows/go.yml | 4 +- CONTRIBUTORS.md | 4 +- ...ure-server-pool-multipart-initiate_test.go | 195 ++++++++++++++++++ 3 files changed, 199 insertions(+), 4 deletions(-) diff --git a/.github/workflows/go.yml b/.github/workflows/go.yml index 97e519ee8..de1742adf 100644 --- a/.github/workflows/go.yml +++ b/.github/workflows/go.yml @@ -90,8 +90,8 @@ jobs: - name: Run S3 Select tests under race detector run: go test -race ./internal/s3select/... -count=1 - - name: Run conditional PUT tests under race detector - run: go test -race ./cmd -run '^Test(PoolsConditionalPut|SinglePoolConditionalPutHTTP)' -count=1 -timeout=5m + - name: Run conditional PUT and multipart tests under race detector + run: go test -race ./cmd -run '^Test(PoolsConditionalPut|SinglePoolConditionalPutHTTP|PoolsMultipartCondition)' -count=1 -timeout=5m - name: Run multi-pool delete consistency tests under race detector run: go test -race ./cmd -run '^Test(Pools(Delete|BulkDelete|RetiringCopyPreservesSharedTierObject)|BulkDeleteQuorumEvidence)' -count=1 -timeout=5m diff --git a/CONTRIBUTORS.md b/CONTRIBUTORS.md index 31d94f30b..ce0c5d89a 100644 --- a/CONTRIBUTORS.md +++ b/CONTRIBUTORS.md @@ -25,7 +25,7 @@ When a proposal was incorporated through a later repair, its adoption is credite @AleksaMCode @yupoet @clown15 -@botdeniska +@botdeniska @910patin @magicxor @davinkevin @@ -98,7 +98,7 @@ When a proposal was incorporated through a later repair, its adoption is credite | [@AleksaMCode](https://github.com/AleksaMCode) | Fixed the Lambda target-set emptiness check and added a regression test | Merged PRs: [pgsty/silo#220](https://github.com/pgsty/silo/pull/220) | | [@yupoet](https://github.com/yupoet) | Improved checksum mismatch messages, verified erasure-coding documentation, and contributed a single-node migration case | Merged PRs: [pgsty/silo#228](https://github.com/pgsty/silo/pull/228), [pgsty/silo#229](https://github.com/pgsty/silo/pull/229)
PRs closed without merging: [pgsty/silo#230](https://github.com/pgsty/silo/pull/230)
Also [cross-checked the two-drive EC:1 documentation](https://github.com/pgsty/silo/issues/197#issuecomment-5846341939). The single-node migration report in #228 was merged on 2026-10-23; the registry-mirror proposal in #230 was closed without merging. | | [@clown15](https://github.com/clown15) | Fixed native multipart-upload timestamp parsing during stale-upload cleanup | Merged PRs: [pgsty/silo#236](https://github.com/pgsty/silo/pull/236) | -| [@botdeniska](https://github.com/botdeniska) | Contributed multi-pool migration and deletion safety, tiered expiry and replication checksum repairs, and faster policy evaluation in silo-pkg | Merged PRs: [pgsty/silo#253](https://github.com/pgsty/silo/pull/253), [pgsty/silo#255](https://github.com/pgsty/silo/pull/255), [pgsty/silo#257](https://github.com/pgsty/silo/pull/257)
PRs closed without merging: [pgsty/silo#245](https://github.com/pgsty/silo/pull/245), [pgsty/silo#248](https://github.com/pgsty/silo/pull/248)
Closed issues: [pgsty/silo#252](https://github.com/pgsty/silo/issues/252), [pgsty/silo#254](https://github.com/pgsty/silo/issues/254), [pgsty/silo#256](https://github.com/pgsty/silo/issues/256)
Merged PRs: [pgsty/silo-pkg#11](https://github.com/pgsty/silo-pkg/pull/11)
The migration repair from [#248](https://github.com/pgsty/silo/pull/248) was incorporated in [60c7179dd](https://github.com/pgsty/silo/commit/60c7179ddbfcbd41a1fe67707230f5b360a43020), retaining the original authorship and DCO. | +| [@botdeniska](https://github.com/botdeniska) | Contributed multi-pool migration, deletion safety and conditional multipart checks, tiered expiry and replication checksum repairs, and faster policy evaluation in silo-pkg | Merged PRs: [pgsty/silo#253](https://github.com/pgsty/silo/pull/253), [pgsty/silo#255](https://github.com/pgsty/silo/pull/255), [pgsty/silo#257](https://github.com/pgsty/silo/pull/257), [pgsty/silo#292](https://github.com/pgsty/silo/pull/292)
PRs closed without merging: [pgsty/silo#245](https://github.com/pgsty/silo/pull/245), [pgsty/silo#248](https://github.com/pgsty/silo/pull/248)
Closed issues: [pgsty/silo#252](https://github.com/pgsty/silo/issues/252), [pgsty/silo#254](https://github.com/pgsty/silo/issues/254), [pgsty/silo#256](https://github.com/pgsty/silo/issues/256), [pgsty/silo#291](https://github.com/pgsty/silo/issues/291)
Merged PRs: [pgsty/silo-pkg#11](https://github.com/pgsty/silo-pkg/pull/11)
The migration repair from [#248](https://github.com/pgsty/silo/pull/248) was incorporated in [60c7179dd](https://github.com/pgsty/silo/commit/60c7179ddbfcbd41a1fe67707230f5b360a43020), retaining the original authorship and DCO. | | [@910patin](https://github.com/910patin) | Fixed long-DN LDAP login while preserving IAM revocation-record errors | Merged PRs: [pgsty/silo#250](https://github.com/pgsty/silo/pull/250) | ## Other pull-request authors diff --git a/cmd/erasure-server-pool-multipart-initiate_test.go b/cmd/erasure-server-pool-multipart-initiate_test.go index c4f81c893..80726d97f 100644 --- a/cmd/erasure-server-pool-multipart-initiate_test.go +++ b/cmd/erasure-server-pool-multipart-initiate_test.go @@ -1,4 +1,5 @@ // Copyright (c) 2026 Denis Babchuk +// Copyright (c) 2026 Feng Ruohang // // This file is part of Silo Object Storage stack // @@ -18,10 +19,13 @@ package cmd import ( + "context" + "encoding/xml" "fmt" "net/http" "slices" "strings" + "sync" "testing" "time" @@ -192,3 +196,194 @@ func TestPoolsMultipartConditionalInitiateUnreadable(t *testing.T) { } } } + +type multipartInitiateMetadataGate struct { + entered, resume chan struct{} + once sync.Once +} + +type multipartInitiateMetadataDisk struct { + StorageAPI + gate *multipartInitiateMetadataGate +} + +func (d multipartInitiateMetadataDisk) WriteMetadata(ctx context.Context, orig, volume, path string, fi FileInfo) error { + if volume == minioMetaMultipartBucket { + d.gate.once.Do(func() { close(d.gate.entered) }) + select { + case <-d.gate.resume: + case <-ctx.Done(): + return ctx.Err() + } + } + return d.StorageAPI.WriteMetadata(ctx, orig, volume, path, fi) +} + +// The pools-layer lock must cover placement of the upload, not just reading +// the current object. Pause the metadata write after the condition passed and +// require a concurrent public overwrite or delete to wait for that placement. +func TestPoolsMultipartConditionalInitiateHoldsLockThroughPersistence(t *testing.T) { + for _, mutation := range []string{"overwrite", "delete"} { + t.Run(mutation, func(t *testing.T) { + z, bucket := consistencyPools(t) + object := "locked-initiate" + current := putConsistencyObject(t, z, bucket, object, 1, "current", ObjectOptions{}) + if _, err := z.serverPools[0].NewMultipartUpload(t.Context(), bucket, object, ObjectOptions{}); err != nil { + t.Fatal(err) + } + gate := &multipartInitiateMetadataGate{entered: make(chan struct{}), resume: make(chan struct{})} + restoreDisks := conditionalPutSwapDisks(z.serverPools[0], object, func(d StorageAPI) StorageAPI { + return multipartInitiateMetadataDisk{StorageAPI: d, gate: gate} + }) + ctx, cancel := context.WithTimeout(t.Context(), 10*time.Second) + var workers sync.WaitGroup + var releaseOnce sync.Once + release := func() { releaseOnce.Do(func() { close(gate.resume) }) } + defer func() { + release() + cancel() + workers.Wait() + restoreDisks() + }() + + initiated := make(chan error, 1) + workers.Add(1) + go func() { + defer workers.Done() + _, err := z.NewMultipartUpload(ctx, bucket, object, ObjectOptions{ + HasIfMatch: true, + CheckPrecondFn: func(oi ObjectInfo) bool { + return oi.ETag != current.ETag + }, + }) + initiated <- err + }() + select { + case <-gate.entered: + case err := <-initiated: + t.Fatalf("initiate returned before persistence: %v", err) + case <-ctx.Done(): + t.Fatal(ctx.Err()) + } + + mutated := make(chan error, 1) + started := make(chan struct{}) + reader := mustGetPutObjReader(t, strings.NewReader("new"), 3, "", "") + workers.Add(1) + go func() { + defer workers.Done() + close(started) + var err error + if mutation == "delete" { + _, err = z.DeleteObject(ctx, bucket, object, ObjectOptions{}) + } else { + _, err = z.PutObject(ctx, bucket, object, reader, ObjectOptions{}) + } + mutated <- err + }() + <-started + select { + case err := <-mutated: + t.Fatalf("mutation escaped initiate lock: %v", err) + case <-time.After(100 * time.Millisecond): + } + release() + for _, result := range []<-chan error{initiated, mutated} { + select { + case err := <-result: + if err != nil { + t.Fatal(err) + } + case <-ctx.Done(): + t.Fatal(ctx.Err()) + } + } + }) + } +} + +// Initiate only checks the current state; it does not reserve the object or +// persist a condition for completion. Completion must carry its own condition, +// compares ETags rather than object generations, and leaves parts available +// for retry when a condition fails. +func TestPoolsMultipartConditionalInitiateLifecycle(t *testing.T) { + z, _ := consistencyPools(t) + bucket, router := conditionalPutBucket(t, z, "unversioned") + for _, change := range []string{"overwrite", "delete", "recreate-same-etag", "omit-complete-condition"} { + t.Run(change, func(t *testing.T) { + object := "lifecycle-" + change + current := putConsistencyObject(t, z, bucket, object, 1, "original", ObjectOptions{}) + defer conditionalPutPool(t, z, object, 0)() + headers := map[string]string{xhttp.IfMatch: fmt.Sprintf("%q", current.ETag)} + initiate := multipartConditionRequest(t, router, http.MethodPost, getNewMultipartURL("", bucket, object), "", headers) + if initiate.Code != http.StatusOK { + t.Fatalf("initiate: %d %s", initiate.Code, initiate.Body.String()) + } + var upload InitiateMultipartUploadResponse + if err := xml.Unmarshal(initiate.Body.Bytes(), &upload); err != nil || upload.UploadID == "" { + t.Fatalf("upload ID: %v %s", err, initiate.Body.String()) + } + const finalBody = "multipart-final" + part, err := z.PutObjectPart(t.Context(), bucket, object, upload.UploadID, 1, + mustGetPutObjReader(t, strings.NewReader(finalBody), int64(len(finalBody)), "", ""), ObjectOptions{}) + if err != nil { + t.Fatal(err) + } + + if change == "delete" || change == "recreate-same-etag" { + if _, err := z.DeleteObject(t.Context(), bucket, object, ObjectOptions{}); err != nil { + t.Fatal(err) + } + } + if change != "delete" { + body := "concurrent" + if change == "recreate-same-etag" { + body = "original" + } + if _, err := z.PutObject(t.Context(), bucket, object, + mustGetPutObjReader(t, strings.NewReader(body), int64(len(body)), "", ""), ObjectOptions{}); err != nil { + t.Fatal(err) + } + } + if change == "omit-complete-condition" { + headers = nil + } + url := getCompleteMultipartUploadURL("", bucket, object, upload.UploadID) + body := multipartConditionCompleteBody([]CompletePart{{PartNumber: 1, ETag: part.ETag}}) + complete := multipartConditionRequest(t, router, http.MethodPost, url, body, headers) + want := http.StatusOK + if change == "overwrite" { + want = http.StatusPreconditionFailed + } + if change == "delete" { + want = http.StatusNotFound + } + if complete.Code != want { + t.Fatalf("complete: %d, want %d: %s", complete.Code, want, complete.Body.String()) + } + if want != http.StatusOK { + parts, err := z.ListObjectParts(t.Context(), bucket, object, upload.UploadID, 0, 100, ObjectOptions{}) + if err != nil || len(parts.Parts) != 1 || parts.Parts[0].ETag != part.ETag { + t.Fatalf("failed condition consumed upload: %v %+v", err, parts) + } + if change == "overwrite" { + oi, err := z.GetObjectInfo(t.Context(), bucket, object, ObjectOptions{}) + if err != nil { + t.Fatal(err) + } + headers = map[string]string{xhttp.IfMatch: fmt.Sprintf("%q", oi.ETag)} + } else { + headers = map[string]string{xhttp.IfNoneMatch: "*"} + } + complete = multipartConditionRequest(t, router, http.MethodPost, url, body, headers) + if complete.Code != http.StatusOK { + t.Fatalf("retry with new condition: %d %s", complete.Code, complete.Body.String()) + } + } + get := multipartConditionRequest(t, router, http.MethodGet, getGetObjectURL("", bucket, object), "", nil) + if get.Code != http.StatusOK || get.Body.String() != finalBody { + t.Fatalf("committed body: %d %q", get.Code, get.Body.String()) + } + }) + } +}