From 210fd568cce92f703d2a5e4b3d675c78ed2589ee Mon Sep 17 00:00:00 2001 From: Denis Babchuk Date: Sat, 7 Nov 2026 11:12:39 +0300 Subject: [PATCH 1/2] 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/2] 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 }