From 5e7d603083469196bc92da70fe7db701271139e8 Mon Sep 17 00:00:00 2001 From: Feng Ruohang Date: Wed, 16 Sep 2026 08:21:21 +0800 Subject: [PATCH 1/2] fix(pools): evaluate cross-pool PUT conditions against the current object Multi-pool PUT selected a destination by capacity and evaluated If-Match/If-None-Match only against that destination's local object state. An empty or stale destination could accept a stale ETag or If-None-Match:* while another pool held the current object, replacing it; a current ETag could instead be rejected with 412 or 404. Under PUT's existing pools-layer object lock, resolve the comparison object with objectPoolInfos (including draining pools), treat a latest delete marker as absence, fail closed on unreadable pool metadata, and clear an accepted callback before destination dispatch. Replica and data-movement callbacks keep their addressed-version semantics and metadata reconciliation. Reproduced on 40220bd836 and RELEASE.2026-09-03T13-18-01Z with six signed HTTP scenarios: four defect cases failed, two controls passed. Refs #199 Signed-off-by: Feng Ruohang --- ...rasure-server-pool-put-conditional_test.go | 686 ++++++++++++++++++ cmd/erasure-server-pool.go | 34 + 2 files changed, 720 insertions(+) create mode 100644 cmd/erasure-server-pool-put-conditional_test.go diff --git a/cmd/erasure-server-pool-put-conditional_test.go b/cmd/erasure-server-pool-put-conditional_test.go new file mode 100644 index 000000000..cb22bb78c --- /dev/null +++ b/cmd/erasure-server-pool-put-conditional_test.go @@ -0,0 +1,686 @@ +// Copyright (c) 2026 Feng Ruohang +// +// 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 ( + "bytes" + "context" + "crypto/md5" + "encoding/base64" + "fmt" + "maps" + "net/http" + "net/http/httptest" + "strings" + "testing" + "time" + + xhttp "github.com/minio/minio/internal/http" + "github.com/minio/minio/internal/kms" +) + +// Change allocation capacity only; metadata and data use real fixture disks. +// Restoring the adapters also lets encrypted fixtures move the write target +// between requests without changing the production allocation policy. +type conditionalPutCapacityDisk struct { + StorageAPI + full bool +} + +func (d conditionalPutCapacityDisk) DiskInfo(ctx context.Context, opts DiskInfoOptions) (DiskInfo, error) { + info, err := d.StorageAPI.DiskInfo(ctx, opts) + info.Total, info.Used = 1<<40, 0 + if d.full { + info.Used = info.Total - (1 << 20) + } + info.Free = info.Total - info.Used + return info, err +} + +func conditionalPutPool(t *testing.T, z *erasureServerPools, object string, target int) func() { + t.Helper() + var restore []func() + for i, pool := range z.serverPools { + set := pool.getHashedSet(object) + previous := set.getDisks + disks := append([]StorageAPI(nil), previous()...) + for j, disk := range disks { + disks[j] = conditionalPutCapacityDisk{StorageAPI: disk, full: i != target} + } + set.getDisks = func() []StorageAPI { return disks } + restore = append(restore, func() { set.getDisks = previous }) + } + return func() { + for _, fn := range restore { + fn() + } + } +} + +func conditionalPutBucket(t *testing.T, z *erasureServerPools, mode string) (string, http.Handler) { + t.Helper() + bucket, router, err := initAPIHandlerTest(t.Context(), z, nil, MakeBucketOptions{VersioningEnabled: mode != "unversioned"}) + if err != nil { + t.Fatal(err) + } + if mode == "suspended" { + if _, err := globalBucketMetadataSys.Update(t.Context(), bucket, bucketVersioningConfig, + []byte(`Suspended`)); err != nil { + t.Fatal(err) + } + } + return bucket, router +} + +func TestPoolsConditionalPutHTTP(t *testing.T) { + for _, mode := range []string{"unversioned", "versioned", "suspended"} { + t.Run(mode, func(t *testing.T) { + z, _ := consistencyPools(t) + bucket, router := conditionalPutBucket(t, z, mode) + for target := range 2 { + for _, tc := range []struct { + name string + oldCopy, localCurrent bool + condition string + status int + }{ + {"stale-etag-accepted", true, false, "old", http.StatusPreconditionFailed}, + {"current-etag-rejected", true, false, "current", http.StatusOK}, + {"create-only-overwrites-other-pool", false, false, "none", http.StatusPreconditionFailed}, + {"current-etag-missing-in-write-pool", false, false, "current", http.StatusOK}, + {"control-current-in-write-pool", true, true, "current", http.StatusOK}, + {"control-stale-etag-rejected", true, true, "old", http.StatusPreconditionFailed}, + {"none-match-current-etag", true, false, "none-current", http.StatusPreconditionFailed}, + {"none-match-old-etag", true, false, "none-old", http.StatusOK}, + } { + t.Run(fmt.Sprintf("target=%d/%s", target, tc.name), func(t *testing.T) { + object := fmt.Sprintf("%d-%s", target, tc.name) + currentPool := 1 - target + if tc.localCurrent { + currentPool = target + } + opts := ObjectOptions{Versioned: mode == "versioned", VersionSuspended: mode == "suspended", MTime: UTCNow().Add(-time.Hour)} + oldETag := "absent-old" + if tc.oldCopy { + oldETag = putConsistencyObject(t, z, bucket, object, 1-currentPool, "old", opts).ETag + } + opts.MTime = UTCNow().Add(-time.Minute) + current := putConsistencyObject(t, z, bucket, object, currentPool, "current", opts) + defer conditionalPutPool(t, z, object, target)() + idx, err := z.getWritePoolIdx(t.Context(), bucket, object, 11, false) + if err != nil || idx != target { + t.Fatalf("allocation target=%d: idx=%d err=%v", target, idx, err) + } + headers := map[string]string{xhttp.IfMatch: fmt.Sprintf("%q", current.ETag)} + if tc.condition == "old" { + headers[xhttp.IfMatch] = fmt.Sprintf("%q", oldETag) + } + if tc.condition == "none" { + headers = map[string]string{xhttp.IfNoneMatch: "*"} + } + if tc.condition == "none-current" { + headers = map[string]string{xhttp.IfNoneMatch: fmt.Sprintf("%q", current.ETag)} + } + if tc.condition == "none-old" { + headers = map[string]string{xhttp.IfNoneMatch: fmt.Sprintf("%q", oldETag)} + } + url := getPutObjectURL("", bucket, object) + before := multipartConditionRequest(t, router, http.MethodGet, url, "", nil) + if before.Code != http.StatusOK || before.Body.String() != "current" || multipartConditionResponseETag(before) != current.ETag { + t.Fatalf("invalid current object: %d %q %v", before.Code, before.Body.String(), before.Header()) + } + put := multipartConditionRequest(t, router, http.MethodPut, url, "replacement", headers) + if put.Code != tc.status { + t.Errorf("PUT status=%d want=%d: %s", put.Code, tc.status, put.Body.String()) + } + if put.Code == http.StatusPreconditionFailed { + multipartConditionError(t, put, "PreconditionFailed") + if multipartConditionResponseETag(put) != current.ETag || put.Header().Get(xhttp.LastModified) != before.Header().Get(xhttp.LastModified) { + t.Errorf("412 headers do not describe current object: %v", put.Header()) + } + } + get := multipartConditionRequest(t, router, http.MethodGet, url, "", nil) + wantBody, wantETag := "current", current.ETag + if tc.status == http.StatusOK { + wantBody, wantETag = "replacement", fmt.Sprintf("%x", md5.Sum([]byte("replacement"))) + } + t.Logf("PUT %d; GET %d bytes=%q ETag=%s", put.Code, get.Code, get.Body.String(), multipartConditionResponseETag(get)) + if get.Code != http.StatusOK || get.Body.String() != wantBody || multipartConditionResponseETag(get) != wantETag { + t.Errorf("GET=%d bytes=%q ETag=%s; want %q %s", get.Code, get.Body.String(), multipartConditionResponseETag(get), wantBody, wantETag) + } + }) + } + } + }) + } +} + +func TestPoolsConditionalPutHTTPAbsence(t *testing.T) { + for _, state := range []string{"missing", "uuid-marker", "null-marker"} { + for _, match := range []bool{false, true} { + t.Run(fmt.Sprintf("%s/match=%t", state, match), func(t *testing.T) { + z, _ := consistencyPools(t) + mode := "versioned" + if state == "null-marker" { + mode = "suspended" + } + bucket, router := conditionalPutBucket(t, z, mode) + object := "absent-key" + if state != "missing" { + putConsistencyObject(t, z, bucket, object, 0, "old", ObjectOptions{Versioned: true, MTime: UTCNow().Add(-time.Hour)}) + vid := mustGetUUID() + if state == "null-marker" { + vid = nullVersionID + } + if _, err := z.serverPools[1].DeleteObject(t.Context(), bucket, object, ObjectOptions{Versioned: true, VersionID: vid, DeleteMarker: true, MTime: UTCNow().Add(-time.Minute)}); err != nil { + t.Fatal(err) + } + } + defer conditionalPutPool(t, z, object, 0)() + headers := map[string]string{xhttp.IfNoneMatch: "*"} + want := http.StatusOK + if match { + headers = map[string]string{xhttp.IfMatch: "*"} + want = http.StatusNotFound + } + url := getPutObjectURL("", bucket, object) + put := multipartConditionRequest(t, router, http.MethodPut, url, "new", headers) + if put.Code != want { + t.Fatalf("PUT %d want %d: %s", put.Code, want, put.Body.String()) + } + get := multipartConditionRequest(t, router, http.MethodGet, url, "", nil) + if match { + multipartConditionError(t, put, "NoSuchKey") + if get.Code != http.StatusNotFound { + t.Fatalf("failed PUT exposed data: %d %s", get.Code, get.Body.String()) + } + } else if get.Code != http.StatusOK || get.Body.String() != "new" || multipartConditionResponseETag(get) != multipartConditionResponseETag(put) { + t.Fatalf("successful create GET: %d %q %v", get.Code, get.Body.String(), get.Header()) + } + }) + } + } +} + +func TestPoolsConditionalPutUnreadable(t *testing.T) { + for faultPool := range 2 { + for _, present := range []bool{false, true} { + for _, match := range []bool{false, true} { + t.Run(fmt.Sprintf("fault-pool=%d/present=%t/match=%t", faultPool, present, match), func(t *testing.T) { + z, _ := consistencyPools(t) + bucket, router := conditionalPutBucket(t, z, "unversioned") + object := "unreadable-key" + if present { + putConsistencyObject(t, z, bucket, object, 0, "old", 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)() + set := z.serverPools[faultPool].getHashedSet(object) + original := set.getDisks + disks := append([]StorageAPI(nil), original()...) + for i := range disks { + disks[i] = consistencyReadFaultDisk{StorageAPI: disks[i], bucket: bucket, object: object} + } + set.getDisks = func() []StorageAPI { return disks } + defer func() { set.getDisks = original }() + headers := map[string]string{xhttp.IfNoneMatch: "*"} + if match { + headers = map[string]string{xhttp.IfMatch: "*"} + } + url := getPutObjectURL("", bucket, object) + put := multipartConditionRequest(t, router, http.MethodPut, url, "replacement", headers) + if put.Code != http.StatusServiceUnavailable { + t.Errorf("unverified PUT must fail: %d %s", put.Code, put.Body.String()) + } + called := 0 + _, err := z.PutObject(t.Context(), bucket, object, mustGetPutObjReader(t, strings.NewReader("replacement"), 11, "", ""), ObjectOptions{HasIfMatch: match, CheckPrecondFn: func(ObjectInfo) bool { called++; return false }}) + if !isErrReadQuorum(err) || called != 0 { + t.Errorf("lookup error=%v callback calls=%d", err, called) + } + set.getDisks = original + get := multipartConditionRequest(t, router, http.MethodGet, url, "", nil) + if present { + if get.Code != http.StatusOK || get.Body.String() != "current" || multipartConditionResponseETag(get) != fmt.Sprintf("%x", md5.Sum([]byte("current"))) { + t.Fatalf("failed PUT changed object: %d %q", get.Code, get.Body.String()) + } + } else if get.Code != http.StatusNotFound { + t.Fatalf("failed PUT created object: %d %q", get.Code, get.Body.String()) + } + }) + } + } + } +} + +func TestPoolsConditionalPutEncryptedETag(t *testing.T) { + for _, kind := range []string{"SSE-C", "SSE-S3", "SSE-KMS"} { + t.Run(kind, func(t *testing.T) { + z, _ := consistencyPools(t) + bucket, router := conditionalPutBucket(t, z, "unversioned") + oldKMS, oldTLS := GlobalKMS, globalIsTLS + GlobalKMS, globalIsTLS = kms.NewStub("conditional-put-key"), true + defer func() { GlobalKMS, globalIsTLS = oldKMS, oldTLS }() + headers := map[string]string{xhttp.AmzServerSideEncryption: xhttp.AmzEncryptionAES} + readHeaders := map[string]string{} + if kind == "SSE-C" { + key := bytes.Repeat([]byte{0x42}, 32) + digest := md5.Sum(key) + headers = map[string]string{ + xhttp.AmzServerSideEncryptionCustomerAlgorithm: xhttp.AmzEncryptionAES, + xhttp.AmzServerSideEncryptionCustomerKey: base64.StdEncoding.EncodeToString(key), + xhttp.AmzServerSideEncryptionCustomerKeyMD5: base64.StdEncoding.EncodeToString(digest[:]), + } + readHeaders = maps.Clone(headers) + } + if kind == "SSE-KMS" { + headers[xhttp.AmzServerSideEncryption] = xhttp.AmzEncryptionKMS + headers[xhttp.AmzServerSideEncryptionKmsID] = "conditional-put-key" + } + object := "encrypted-key" + url := getPutObjectURL("", bucket, object) + restore := conditionalPutPool(t, z, object, 0) + old := multipartConditionRequest(t, router, http.MethodPut, url, "old", headers) + restore() + restore = conditionalPutPool(t, z, object, 1) + current := multipartConditionRequest(t, router, http.MethodPut, url, "current", headers) + restore() + if old.Code != http.StatusOK || current.Code != http.StatusOK { + t.Fatalf("encrypted setup: %d %s / %d %s", old.Code, old.Body.String(), current.Code, current.Body.String()) + } + defer conditionalPutPool(t, z, object, 0)() + for _, stale := range []bool{true, false} { + h := maps.Clone(headers) + h[xhttp.IfMatch] = fmt.Sprintf("%q", multipartConditionResponseETag(current)) + wantStatus, wantBody, wantETag := http.StatusOK, "replacement", "" + if stale { + h[xhttp.IfMatch] = fmt.Sprintf("%q", multipartConditionResponseETag(old)) + wantStatus, wantBody, wantETag = http.StatusPreconditionFailed, "current", multipartConditionResponseETag(current) + } + put := multipartConditionRequest(t, router, http.MethodPut, url, "replacement", h) + if put.Code != wantStatus { + t.Fatalf("encrypted condition: %d want %d: %s", put.Code, wantStatus, put.Body.String()) + } + if !stale { + wantETag = multipartConditionResponseETag(put) + } + get := multipartConditionRequest(t, router, http.MethodGet, url, "", readHeaders) + if get.Code != http.StatusOK || get.Body.String() != wantBody || multipartConditionResponseETag(get) != wantETag { + t.Fatalf("encrypted GET: %d %q %v", get.Code, get.Body.String(), get.Header()) + } + } + }) + } +} + +func TestPoolsConditionalPutVersionSelection(t *testing.T) { + for _, kind := range []string{"public-version", "replica", "replica-preserve-etag", "movement", "no-lock", "tie", "draining"} { + t.Run(kind, func(t *testing.T) { + z, bucket := consistencyPools(t) + object := "version-selection" + addressed := putConsistencyObject(t, z, bucket, object, 1, "addressed", ObjectOptions{Versioned: true, MTime: UTCNow().Add(-time.Hour)}) + current := putConsistencyObject(t, z, bucket, object, 1, "current", ObjectOptions{Versioned: true, MTime: UTCNow().Add(-time.Minute)}) + opts := ObjectOptions{Versioned: true, VersionID: addressed.VersionID, HasIfMatch: true} + want := current + switch kind { + case "replica": + opts.ReplicaLockReconcile, opts.ReplicationRequest = true, true + want = addressed + case "replica-preserve-etag": + opts.PreserveETag = addressed.ETag + opts.ReplicaLockReconcile, opts.ReplicationRequest = true, true + want = addressed + case "movement": + opts.DataMovement, opts.SrcPoolIdx = true, 1 + want = addressed + case "tie": + want = putConsistencyObject(t, z, bucket, object, 0, "tie-winner", ObjectOptions{Versioned: true, MTime: current.ModTime}) + case "draining": + z.poolMetaMutex.Lock() + z.poolMeta.Pools[1].Decommission = &PoolDecommissionInfo{} + z.poolMetaMutex.Unlock() + } + defer conditionalPutPool(t, z, object, 0)() + ctx := t.Context() + if kind == "no-lock" { + lk := z.NewNSLock(bucket, object) + lkctx, err := lk.GetLock(ctx, globalOperationTimeout) + if err != nil { + t.Fatal(err) + } + defer lk.Unlock(lkctx) + ctx, opts.NoLock = lkctx.Context(), true + } + called := 0 + opts.UserDefined = make(map[string]string) + opts.CheckPrecondFn = func(oi ObjectInfo) bool { + called++ + if oi.ETag != want.ETag || oi.VersionID != want.VersionID { + t.Errorf("comparison ETag/version=%s/%s want %s/%s", oi.ETag, oi.VersionID, want.ETag, want.VersionID) + } + return oi.ETag != want.ETag + } + oi, err := z.PutObject(ctx, bucket, object, mustGetPutObjReader(t, strings.NewReader("replacement"), 11, "", ""), opts) + if err != nil || called != 1 { + t.Fatalf("PUT err=%v callback calls=%d", err, called) + } + if oi.VersionID != addressed.VersionID { + t.Fatalf("destination version changed: %s", oi.VersionID) + } + if opts.PreserveETag != "" && oi.ETag != opts.PreserveETag { + t.Fatalf("PreserveETag changed: %s", oi.ETag) + } + }) + } +} + +func TestPoolsConditionalPutReplicaDuplicateHTTP(t *testing.T) { + for _, null := range []bool{false, true} { + t.Run(fmt.Sprintf("null=%t", null), func(t *testing.T) { + z, _ := consistencyPools(t) + bucket, router := conditionalPutBucket(t, z, "versioned") + object := "replica-duplicate" + vid := mustGetUUID() + if null { + vid = nullVersionID + } + addressed := putConsistencyObject(t, z, bucket, object, 1, "addressed", ObjectOptions{Versioned: true, VersionID: vid, MTime: UTCNow().Add(-time.Hour)}) + current := putConsistencyObject(t, z, bucket, object, 1, "current", ObjectOptions{Versioned: true, MTime: UTCNow().Add(-time.Minute)}) + defer conditionalPutPool(t, z, object, 0)() + headers := map[string]string{ + xhttp.MinIOSourceReplicationRequest: "true", + xhttp.AmzBucketReplicationStatus: "REPLICA", + xhttp.MinIOSourceETag: addressed.ETag, + xhttp.MinIOSourceMTime: addressed.ModTime.Format(time.RFC3339Nano), + } + url := getPutObjectURL("", bucket, object) + put := multipartConditionRequest(t, router, http.MethodPut, url+"?versionId="+vid, "addressed", headers) + if put.Code != http.StatusPreconditionFailed { + t.Fatalf("replica duplicate: %d %s", put.Code, put.Body.String()) + } + multipartConditionError(t, put, "PreconditionFailed") + get := multipartConditionRequest(t, router, http.MethodGet, url, "", nil) + if get.Code != http.StatusOK || get.Body.String() != "current" || multipartConditionResponseETag(get) != current.ETag { + t.Fatalf("duplicate changed current object: %d %q", get.Code, get.Body.String()) + } + get = multipartConditionRequest(t, router, http.MethodGet, url+"?versionId="+vid, "", nil) + if get.Code != http.StatusOK || get.Body.String() != "addressed" || multipartConditionResponseETag(get) != addressed.ETag { + t.Fatalf("duplicate changed addressed version: %d %q", get.Code, get.Body.String()) + } + }) + } +} + +func TestPoolsConditionalPutConcurrentHTTP(t *testing.T) { + z, _ := consistencyPools(t) + bucket, router := conditionalPutBucket(t, z, "unversioned") + for _, match := range []bool{false, true} { + for iteration := range 5 { + t.Run(fmt.Sprintf("match=%t/iteration=%d", match, iteration), func(t *testing.T) { + object := fmt.Sprintf("concurrent-%t-%d", match, iteration) + headers := map[string]string{xhttp.IfNoneMatch: "*"} + if match { + oi := putConsistencyObject(t, z, bucket, object, 1, "old", ObjectOptions{MTime: UTCNow().Add(-time.Minute)}) + headers = map[string]string{xhttp.IfMatch: fmt.Sprintf("%q", oi.ETag)} + } + defer conditionalPutPool(t, z, object, 0)() + start := make(chan struct{}) + results := make(chan *httptest.ResponseRecorder, 2) + url := getPutObjectURL("", bucket, object) + for i := range 2 { + body := fmt.Sprintf("writer-%d", i) + req, err := newTestSignedRequestV4(http.MethodPut, url, int64(len(body)), strings.NewReader(body), globalActiveCred.AccessKey, globalActiveCred.SecretKey, headers) + if err != nil { + t.Fatal(err) + } + go func() { + <-start + rec := httptest.NewRecorder() + router.ServeHTTP(rec, req) + results <- rec + }() + } + close(start) + success, failed, winnerETag := 0, 0, "" + for range 2 { + result := <-results + switch result.Code { + case http.StatusOK: + success++ + winnerETag = multipartConditionResponseETag(result) + case http.StatusPreconditionFailed: + failed++ + multipartConditionError(t, result, "PreconditionFailed") + default: + t.Errorf("unexpected PUT %d: %s", result.Code, result.Body.String()) + } + } + if success != 1 || failed != 1 { + t.Fatalf("success=%d precondition failures=%d", success, failed) + } + get := multipartConditionRequest(t, router, http.MethodGet, url, "", nil) + if get.Code != http.StatusOK || multipartConditionResponseETag(get) != winnerETag || fmt.Sprintf("%x", md5.Sum(get.Body.Bytes())) != winnerETag { + t.Fatalf("winner lost: %d %q %v", get.Code, get.Body.String(), get.Header()) + } + }) + } + } +} + +func TestPoolsConditionalPutSerializesMutation(t *testing.T) { + for _, deletion := range []bool{false, true} { + t.Run(fmt.Sprintf("delete=%t", deletion), func(t *testing.T) { + z, bucket := consistencyPools(t) + object := "conditional-mutation" + current := putConsistencyObject(t, z, bucket, object, 1, "current", ObjectOptions{MTime: UTCNow().Add(-time.Minute)}) + defer conditionalPutPool(t, z, object, 0)() + ctx, cancel := context.WithTimeout(t.Context(), 10*time.Second) + defer cancel() + gate := &consistencyGateReader{Reader: strings.NewReader("replacement"), entered: make(chan struct{}), resume: make(chan struct{})} + release := func() { gate.release.Do(func() { close(gate.resume) }) } + defer release() + reader := mustGetPutObjReader(t, gate, 11, "", "") + written := make(chan error, 1) + go func() { + _, err := z.PutObject(ctx, bucket, object, reader, ObjectOptions{HasIfMatch: true, CheckPrecondFn: func(oi ObjectInfo) bool { return oi.ETag != current.ETag }}) + written <- err + }() + select { + case <-gate.entered: + case err := <-written: + t.Fatalf("PUT failed before body read: %v", err) + case <-ctx.Done(): + t.Fatal(ctx.Err()) + } + mutated := make(chan error, 1) + go func() { + var err error + if deletion { + _, err = z.DeleteObject(ctx, bucket, object, ObjectOptions{}) + } else { + _, err = z.PutObjectMetadata(ctx, bucket, object, ObjectOptions{EvalMetadataFn: func(oi *ObjectInfo, _ error) (ReplicateDecision, error) { + oi.UserDefined["x-amz-meta-after-put"] = "present" + return ReplicateDecision{}, nil + }}) + } + mutated <- err + }() + select { + case err := <-mutated: + t.Fatalf("mutation escaped PUT lock: %v", err) + case <-time.After(100 * time.Millisecond): + } + release() + if err := <-written; err != nil { + t.Fatal(err) + } + if err := <-mutated; err != nil { + t.Fatal(err) + } + oi, err := z.GetObjectInfo(ctx, bucket, object, ObjectOptions{}) + if deletion { + if !isErrObjectNotFound(err) { + t.Fatalf("delete lost: %v", err) + } + } else if err != nil || oi.UserDefined["x-amz-meta-after-put"] != "present" || oi.ETag != fmt.Sprintf("%x", md5.Sum([]byte("replacement"))) { + t.Fatalf("metadata/PUT lost: %+v %v", oi, err) + } + }) + } +} + +func TestSinglePoolConditionalPutHTTP(t *testing.T) { + ctx, cancel := context.WithCancel(t.Context()) + obj, dirs, err := prepareErasure16(ctx) + if err != nil { + cancel() + t.Fatal(err) + } + z := obj.(*erasureServerPools) + t.Cleanup(func() { cancel(); z.Shutdown(context.Background()); removeRoots(dirs) }) + if !z.SinglePool() { + t.Fatal("fixture is not a single pool") + } + bucket, router := conditionalPutBucket(t, z, "unversioned") + object := "single-pool-condition" + defer conditionalPutPool(t, z, object, 0)() + url := getPutObjectURL("", bucket, object) + for _, tc := range []struct { + body, match, none string + status int + }{ + {"missing", "*", "", http.StatusNotFound}, + {"first", "", "*", http.StatusOK}, + {"blocked", "", "*", http.StatusPreconditionFailed}, + {"blocked", "stale", "", http.StatusPreconditionFailed}, + {"second", fmt.Sprintf("%x", md5.Sum([]byte("first"))), "", http.StatusOK}, + } { + rec := multipartConditionRequest(t, router, http.MethodPut, url, tc.body, map[string]string{xhttp.IfMatch: tc.match, xhttp.IfNoneMatch: tc.none}) + if rec.Code != tc.status { + t.Fatalf("PUT %d want %d: %s", rec.Code, tc.status, rec.Body.String()) + } + } + get := multipartConditionRequest(t, router, http.MethodGet, url, "", nil) + if get.Code != http.StatusOK || get.Body.String() != "second" { + t.Fatalf("single-pool GET: %d %q", get.Code, get.Body.String()) + } +} + +// An internal replica callback without an addressed version is not a public +// condition. Preserve its availability when another pool is unreadable. An +// addressed replica already requires all pools for lock/tag reconciliation. +func TestPoolsConditionalPutReplicaAvailability(t *testing.T) { + for _, addressed := range []bool{false, true} { + t.Run(fmt.Sprintf("addressed=%t", addressed), func(t *testing.T) { + z, _ := consistencyPools(t) + mode := "unversioned" + if addressed { + mode = "versioned" + } + bucket, router := conditionalPutBucket(t, z, mode) + object := "replica-availability" + oi := putConsistencyObject(t, z, bucket, object, 0, "old", ObjectOptions{Versioned: addressed, MTime: UTCNow().Add(-time.Minute)}) + defer conditionalPutPool(t, z, object, 0)() + set := z.serverPools[1].getHashedSet(object) + original := set.getDisks + disks := append([]StorageAPI(nil), original()...) + for i := range disks { + disks[i] = consistencyReadFaultDisk{StorageAPI: disks[i], bucket: bucket, object: object} + } + set.getDisks = func() []StorageAPI { return disks } + defer func() { set.getDisks = original }() + headers := map[string]string{ + xhttp.MinIOSourceReplicationRequest: "true", + xhttp.AmzBucketReplicationStatus: "REPLICA", + xhttp.MinIOSourceETag: fmt.Sprintf("%x", md5.Sum([]byte("replacement"))), + } + url := getPutObjectURL("", bucket, object) + wantStatus, wantBody := http.StatusOK, "replacement" + if addressed { + url += "?versionId=" + oi.VersionID + wantStatus, wantBody = http.StatusServiceUnavailable, "old" + } + put := multipartConditionRequest(t, router, http.MethodPut, url, "replacement", headers) + if put.Code != wantStatus { + t.Fatalf("replica PUT: %d want %d: %s", put.Code, wantStatus, put.Body.String()) + } + set.getDisks = original + get := multipartConditionRequest(t, router, http.MethodGet, url, "", nil) + if get.Code != http.StatusOK || get.Body.String() != wantBody { + t.Fatalf("replica GET: %d %q", get.Code, get.Body.String()) + } + }) + } +} + +func TestPoolsConditionalPutDestinationVersionHTTP(t *testing.T) { + z, _ := consistencyPools(t) + bucket, router := conditionalPutBucket(t, z, "versioned") + object := "client-version" + old := putConsistencyObject(t, z, bucket, object, 0, "old", ObjectOptions{Versioned: true, MTime: UTCNow().Add(-time.Hour)}) + current := putConsistencyObject(t, z, bucket, object, 1, "current", ObjectOptions{Versioned: true, MTime: UTCNow().Add(-time.Minute)}) + defer conditionalPutPool(t, z, object, 0)() + url := getPutObjectURL("", bucket, object) + "?versionId=" + old.VersionID + for _, stale := range []bool{true, false} { + tag, status := current.ETag, http.StatusOK + if stale { + tag, status = old.ETag, http.StatusPreconditionFailed + } + put := multipartConditionRequest(t, router, http.MethodPut, url, "replacement", map[string]string{xhttp.IfMatch: fmt.Sprintf("%q", tag)}) + if put.Code != status { + t.Fatalf("version-addressed public PUT: %d want %d: %s", put.Code, status, put.Body.String()) + } + if got := put.Header()[xhttp.AmzVersionID]; !stale && (len(got) != 1 || got[0] != old.VersionID) { + t.Fatalf("write version changed: %v", put.Header()) + } + } + get := multipartConditionRequest(t, router, http.MethodGet, url, "", nil) + if get.Code != http.StatusOK || get.Body.String() != "replacement" { + t.Fatalf("addressed GET: %d %q", get.Code, get.Body.String()) + } +} + +func TestPoolsConditionalPutDeleteMarkerTie(t *testing.T) { + for markerPool := range 2 { + t.Run(fmt.Sprintf("marker-pool=%d", markerPool), func(t *testing.T) { + z, _ := consistencyPools(t) + bucket, router := conditionalPutBucket(t, z, "versioned") + object := "marker-tie" + mtime := UTCNow().Add(-time.Minute) + putConsistencyObject(t, z, bucket, object, 1-markerPool, "live", ObjectOptions{Versioned: true, MTime: mtime}) + if _, err := z.serverPools[markerPool].DeleteObject(t.Context(), bucket, object, ObjectOptions{Versioned: true, VersionID: mustGetUUID(), DeleteMarker: true, MTime: mtime}); err != nil { + t.Fatal(err) + } + defer conditionalPutPool(t, z, object, 0)() + url := getPutObjectURL("", bucket, object) + before := multipartConditionRequest(t, router, http.MethodGet, url, "", nil) + want := http.StatusPreconditionFailed + if markerPool == 0 { + want = http.StatusOK + if before.Code != http.StatusNotFound { + t.Fatalf("GET tie: %d", before.Code) + } + } else if before.Code != http.StatusOK { + t.Fatalf("GET tie: %d", before.Code) + } + put := multipartConditionRequest(t, router, http.MethodPut, url, "replacement", map[string]string{xhttp.IfNoneMatch: "*"}) + if put.Code != want { + t.Fatalf("PUT tie: %d want %d: %s", put.Code, want, put.Body.String()) + } + }) + } +} diff --git a/cmd/erasure-server-pool.go b/cmd/erasure-server-pool.go index 555986bc0..344800e56 100644 --- a/cmd/erasure-server-pool.go +++ b/cmd/erasure-server-pool.go @@ -1149,6 +1149,40 @@ func (z *erasureServerPools) PutObject(ctx context.Context, bucket string, objec } opts.NoLock = true + // Public write conditions compare the logical current object while the + // pools-layer write lock is held. The destination selected by capacity may + // be empty or stale, and draining pools can still hold the current object. + // Replica callbacks retain their existing addressed-version semantics and + // 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 + } + // Do not repeat an accepted condition against the destination's copy. + opts.CheckPrecondFn = nil + } + idx, err := z.getWritePoolIdx(ctx, bucket, object, data.Size(), true) if err != nil { return ObjectInfo{}, err From 4620be394b523601285098be7b0fa3547dc27b8a Mon Sep 17 00:00:00 2001 From: Feng Ruohang Date: Wed, 16 Sep 2026 10:53:24 +0800 Subject: [PATCH 2/2] test: synchronize conditional PUT disk fixtures Replace unsynchronized getDisks swaps with backing disk-list updates under erasureDisksMu, matching the existing GetDisks reader lock. Apply the same helper to capacity and read-fault adapters while preserving nested restore ordering. Add a regression that overlaps fixture changes with the real IAM Walk reader, and run the conditional PUT suite under the race detector in CI. The regression reproduces the old fixture race; ten fixed race iterations pass without warnings. Production conditional PUT behavior is unchanged. Refs #199 Signed-off-by: Feng Ruohang --- .github/workflows/go.yml | 3 + ...rasure-server-pool-put-conditional_test.go | 87 +++++++++++++------ 2 files changed, 64 insertions(+), 26 deletions(-) diff --git a/.github/workflows/go.yml b/.github/workflows/go.yml index e5f4660fb..2b5ef6bdc 100644 --- a/.github/workflows/go.yml +++ b/.github/workflows/go.yml @@ -90,6 +90,9 @@ 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 + crosscompile: name: Cross Compile runs-on: ubuntu-latest diff --git a/cmd/erasure-server-pool-put-conditional_test.go b/cmd/erasure-server-pool-put-conditional_test.go index cb22bb78c..4bd576037 100644 --- a/cmd/erasure-server-pool-put-conditional_test.go +++ b/cmd/erasure-server-pool-put-conditional_test.go @@ -52,18 +52,32 @@ func (d conditionalPutCapacityDisk) DiskInfo(ctx context.Context, opts DiskInfoO return info, err } +// Keep getDisks immutable while background IAM and storage readers use it. +// GetDisks takes this same mutex when it copies the backing disk list. +func conditionalPutSwapDisks(pool *erasureSets, object string, wrap func(StorageAPI) StorageAPI) func() { + setIndex := pool.getHashedSet(object).setIndex + pool.erasureDisksMu.Lock() + previous := pool.erasureDisks[setIndex] + disks := append([]StorageAPI(nil), previous...) + for i, disk := range disks { + disks[i] = wrap(disk) + } + pool.erasureDisks[setIndex] = disks + pool.erasureDisksMu.Unlock() + return func() { + pool.erasureDisksMu.Lock() + pool.erasureDisks[setIndex] = previous + pool.erasureDisksMu.Unlock() + } +} + func conditionalPutPool(t *testing.T, z *erasureServerPools, object string, target int) func() { t.Helper() var restore []func() for i, pool := range z.serverPools { - set := pool.getHashedSet(object) - previous := set.getDisks - disks := append([]StorageAPI(nil), previous()...) - for j, disk := range disks { - disks[j] = conditionalPutCapacityDisk{StorageAPI: disk, full: i != target} - } - set.getDisks = func() []StorageAPI { return disks } - restore = append(restore, func() { set.getDisks = previous }) + restore = append(restore, conditionalPutSwapDisks(pool, object, func(disk StorageAPI) StorageAPI { + return conditionalPutCapacityDisk{StorageAPI: disk, full: i != target} + })) } return func() { for _, fn := range restore { @@ -230,14 +244,10 @@ func TestPoolsConditionalPutUnreadable(t *testing.T) { putConsistencyObject(t, z, bucket, object, 1, "current", ObjectOptions{MTime: UTCNow().Add(-time.Minute)}) } defer conditionalPutPool(t, z, object, 0)() - set := z.serverPools[faultPool].getHashedSet(object) - original := set.getDisks - disks := append([]StorageAPI(nil), original()...) - for i := range disks { - disks[i] = consistencyReadFaultDisk{StorageAPI: disks[i], bucket: bucket, object: object} - } - set.getDisks = func() []StorageAPI { return disks } - defer func() { set.getDisks = original }() + restoreFault := conditionalPutSwapDisks(z.serverPools[faultPool], object, func(disk StorageAPI) StorageAPI { + return consistencyReadFaultDisk{StorageAPI: disk, bucket: bucket, object: object} + }) + defer restoreFault() headers := map[string]string{xhttp.IfNoneMatch: "*"} if match { headers = map[string]string{xhttp.IfMatch: "*"} @@ -252,7 +262,7 @@ func TestPoolsConditionalPutUnreadable(t *testing.T) { if !isErrReadQuorum(err) || called != 0 { t.Errorf("lookup error=%v callback calls=%d", err, called) } - set.getDisks = original + restoreFault() get := multipartConditionRequest(t, router, http.MethodGet, url, "", nil) if present { if get.Code != http.StatusOK || get.Body.String() != "current" || multipartConditionResponseETag(get) != fmt.Sprintf("%x", md5.Sum([]byte("current"))) { @@ -595,14 +605,10 @@ func TestPoolsConditionalPutReplicaAvailability(t *testing.T) { object := "replica-availability" oi := putConsistencyObject(t, z, bucket, object, 0, "old", ObjectOptions{Versioned: addressed, MTime: UTCNow().Add(-time.Minute)}) defer conditionalPutPool(t, z, object, 0)() - set := z.serverPools[1].getHashedSet(object) - original := set.getDisks - disks := append([]StorageAPI(nil), original()...) - for i := range disks { - disks[i] = consistencyReadFaultDisk{StorageAPI: disks[i], bucket: bucket, object: object} - } - set.getDisks = func() []StorageAPI { return disks } - defer func() { set.getDisks = original }() + restoreFault := conditionalPutSwapDisks(z.serverPools[1], object, func(disk StorageAPI) StorageAPI { + return consistencyReadFaultDisk{StorageAPI: disk, bucket: bucket, object: object} + }) + defer restoreFault() headers := map[string]string{ xhttp.MinIOSourceReplicationRequest: "true", xhttp.AmzBucketReplicationStatus: "REPLICA", @@ -618,7 +624,7 @@ func TestPoolsConditionalPutReplicaAvailability(t *testing.T) { if put.Code != wantStatus { t.Fatalf("replica PUT: %d want %d: %s", put.Code, wantStatus, put.Body.String()) } - set.getDisks = original + restoreFault() get := multipartConditionRequest(t, router, http.MethodGet, url, "", nil) if get.Code != http.StatusOK || get.Body.String() != wantBody { t.Fatalf("replica GET: %d %q", get.Code, get.Body.String()) @@ -684,3 +690,32 @@ func TestPoolsConditionalPutDeleteMarkerTie(t *testing.T) { }) } } + +// Overlap fixture changes with the real IAM Walk reader instead of relying on +// its periodic refresh timer to expose an unsynchronized disk-adapter swap. +func TestPoolsConditionalPutFixtureConcurrentIAM(t *testing.T) { + z, _ := consistencyPools(t) + conditionalPutBucket(t, z, "unversioned") + ctx, cancel := context.WithTimeout(t.Context(), 30*time.Second) + defer cancel() + started, finished := make(chan struct{}), make(chan error, 1) + iam := globalIAMSys + go func() { + close(started) + for range 20 { + if err := iam.Load(ctx, false); err != nil { + finished <- err + return + } + } + finished <- nil + }() + <-started + for i := range 5000 { + restore := conditionalPutPool(t, z, "fixture-concurrent-iam", i%2) + restore() + } + if err := <-finished; err != nil { + t.Fatal(err) + } +}