diff --git a/CHANGELOG.md b/CHANGELOG.md index 6272f5c50..8dad9a0c4 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,11 +1,11 @@ # Changelog -## Unreleased — main as of 2026-09-13 +## Unreleased -The coordinated source is merged through `5d955b5b7444f8a3ab550ce92713607998f89c0d`. +The entries below describe source changes on main since the latest published Server. **The latest published Server remains 20260903.** These changes are not in its binaries, packages or images. See the [component matrix](https://silo.pgsty.com/compatibility/versions/) -and [complete commit range](https://github.com/pgsty/silo/compare/RELEASE.2026-09-03T13-18-01Z...5d955b5b7444f8a3ab550ce92713607998f89c0d). +and [complete commit range](https://github.com/pgsty/silo/compare/RELEASE.2026-09-03T13-18-01Z...main). ### Authorization and security @@ -22,6 +22,17 @@ and [complete commit range](https://github.com/pgsty/silo/compare/RELEASE.2026-0 ### Object storage and replication +- Evaluate conditional multipart completion against the logical current object + across all pools while holding the existing object lock. A stale `If-Match` + can no longer replace newer data in another pool, and the current ETag is no + longer rejected because the upload resides next to an older copy. Conditions + are evaluated once; a current delete marker counts as an absent object. + **Availability change:** if metadata cannot be read from any pool, conditional + completion fails even when another pool can still serve GET/HEAD. This also + applies when the unreadable pool may not hold the object: absence cannot be + verified. Retry after the pool recovers. Unconditional completion and the + single-pool path retain their existing behavior. + - Reconcile ordinary single-object version DELETE across all pools, including null versions, delete markers and unqualified directory-marker DELETE. This applies the deletion to every resolved pool copy under existing quorum diff --git a/cmd/erasure-server-pool-multipart-cond_test.go b/cmd/erasure-server-pool-multipart-cond_test.go new file mode 100644 index 000000000..21b72a791 --- /dev/null +++ b/cmd/erasure-server-pool-multipart-cond_test.go @@ -0,0 +1,172 @@ +// 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" + "testing" + "time" +) + +// A multipart completion's If-Match must be evaluated against the logical +// latest object across pools, not against the copy local to a pool. +// +// Multi-pool write placement is not sticky (getPoolIdx picks by available +// space even for existing objects), so an upload and a newer overwrite of +// the same name routinely end up in different pools. Uploads are pinned to +// their pools directly: routing through z.NewMultipartUpload would make the +// placement depend on the space-weighted random choice. +func TestPoolsMultipartConditionalUsesLogicalLatest(t *testing.T) { + z, bucket := consistencyPools(t) + ctx := t.Context() + + ifMatch := func(etag string) (opts ObjectOptions) { + return ObjectOptions{ + HasIfMatch: true, + CheckPrecondFn: func(oi ObjectInfo) bool { + return oi.ETag != etag + }, + } + } + + uploadPart := func(t *testing.T, bucket, object, uploadID string) []CompletePart { + t.Helper() + pi, err := z.PutObjectPart(ctx, bucket, object, uploadID, 1, + mustGetPutObjReader(t, bytes.NewBufferString("part"), 4, "", ""), ObjectOptions{}) + if err != nil { + t.Fatal(err) + } + return []CompletePart{{PartNumber: 1, ETag: pi.ETag}} + } + + // Scenario A: the uploaded If-Match carries the stale ETag of the pool-0 + // copy while the logical latest object lives in pool 1. The completion + // must fail with 412 instead of shadowing the newer logical state. + objectA := "cond-mp-stale-etag" + base := time.Now() + oldA := putConsistencyObject(t, z, bucket, objectA, 0, "old", ObjectOptions{MTime: base.Add(-2 * time.Minute)}) + mpA, err := z.serverPools[0].NewMultipartUpload(ctx, bucket, objectA, ObjectOptions{}) + if err != nil { + t.Fatal(err) + } + newerA := putConsistencyObject(t, z, bucket, objectA, 1, "new", ObjectOptions{ + MTime: base.Add(-time.Minute), + }) + + latest, _, err := z.getLatestObjectInfoWithIdx(ctx, bucket, objectA, ObjectOptions{}) + if err != nil { + t.Fatal(err) + } + if latest.ETag != newerA.ETag { + t.Fatalf("logical latest should be the pool-1 copy: got %s want %s", latest.ETag, newerA.ETag) + } + + if _, err = z.CompleteMultipartUpload(ctx, bucket, objectA, mpA.UploadID, + uploadPart(t, bucket, objectA, mpA.UploadID), ifMatch(oldA.ETag)); err == nil { + t.Fatal("If-Match with the stale pool-0 ETag must not complete over the newer pool-1 object") + } else if _, ok := err.(PreConditionFailed); !ok { + t.Fatalf("expected PreconditionFailed, got %v", err) + } + + if latest, _, err = z.getLatestObjectInfoWithIdx(ctx, bucket, objectA, ObjectOptions{}); err != nil { + t.Fatal(err) + } + if latest.ETag != newerA.ETag { + t.Fatalf("the newer pool-1 object must remain the logical latest, got %s", latest.ETag) + } + + // Scenario B: the uploaded If-Match carries the logical latest ETag (the + // pool-1 copy) while the upload sits next to the stale pool-0 copy. The + // precondition is satisfied, so completion must succeed and its result + // must become the logical latest. + objectB := "cond-mp-latest-etag" + putConsistencyObject(t, z, bucket, objectB, 0, "old", ObjectOptions{MTime: base.Add(-2 * time.Minute)}) + mpB, err := z.serverPools[0].NewMultipartUpload(ctx, bucket, objectB, ObjectOptions{}) + if err != nil { + t.Fatal(err) + } + newerB := putConsistencyObject(t, z, bucket, objectB, 1, "new", ObjectOptions{ + MTime: base.Add(-time.Minute), + }) + oiB, err := z.CompleteMultipartUpload(ctx, bucket, objectB, mpB.UploadID, + uploadPart(t, bucket, objectB, mpB.UploadID), ifMatch(newerB.ETag)) + if err != nil { + t.Fatalf("If-Match with the logical latest ETag must complete, got %v", err) + } + if oiB.ETag == "" { + t.Fatal("completion returned an empty ETag") + } + if latest, _, err = z.getLatestObjectInfoWithIdx(ctx, bucket, objectB, ObjectOptions{}); err != nil { + t.Fatal(err) + } + if latest.ETag != oiB.ETag { + t.Fatalf("the completed object must be the logical latest: got %s want %s", latest.ETag, oiB.ETag) + } + + // Scenario C: the upload lives in pool 1 with the newer copy while pool 0 + // holds the stale one. A set-local evaluation order would let pool 0's + // stale copy fail the request before pool 1 is reached; the logical + // latest ETag must complete. + objectC := "cond-mp-upload-other-pool" + putConsistencyObject(t, z, bucket, objectC, 0, "old", ObjectOptions{MTime: base.Add(-2 * time.Minute)}) + mpC, err := z.serverPools[1].NewMultipartUpload(ctx, bucket, objectC, ObjectOptions{}) + if err != nil { + t.Fatal(err) + } + newerC := putConsistencyObject(t, z, bucket, objectC, 1, "new", ObjectOptions{ + MTime: base.Add(-time.Minute), + }) + if _, err = z.CompleteMultipartUpload(ctx, bucket, objectC, mpC.UploadID, + uploadPart(t, bucket, objectC, mpC.UploadID), ifMatch(newerC.ETag)); err != nil { + t.Fatalf("If-Match with the logical latest ETag must complete regardless of upload pool, got %v", err) + } + + // Scenario D: pool 1 holds the newer copy but cannot be read. An + // unreadable pool may contain the newest state, so the unverifiable + // condition must fail the request rather than pass it against pool 0's + // stale ETag. + objectD := "cond-mp-unreadable-pool" + oldD := putConsistencyObject(t, z, bucket, objectD, 0, "old", ObjectOptions{MTime: base.Add(-2 * time.Minute)}) + mpD, err := z.serverPools[0].NewMultipartUpload(ctx, bucket, objectD, ObjectOptions{}) + if err != nil { + t.Fatal(err) + } + newerD := putConsistencyObject(t, z, bucket, objectD, 1, "new", ObjectOptions{ + MTime: base.Add(-time.Minute), + }) + if latest, _, err = z.getLatestObjectInfoWithIdx(ctx, bucket, objectD, ObjectOptions{}); err != nil { + t.Fatal(err) + } + if latest.ETag != newerD.ETag { + t.Fatalf("logical latest before faulting pool 1 should be its copy: got %s want %s", latest.ETag, newerD.ETag) + } + set := z.serverPools[1].getHashedSet(objectD) + getDisks := set.getDisks + faulty := append([]StorageAPI(nil), getDisks()...) + for i := range faulty { + faulty[i] = consistencyReadFaultDisk{StorageAPI: faulty[i], bucket: bucket, object: objectD} + } + set.getDisks = func() []StorageAPI { return faulty } + defer func() { set.getDisks = getDisks }() + + _, err = z.CompleteMultipartUpload(ctx, bucket, objectD, mpD.UploadID, + uploadPart(t, bucket, objectD, mpD.UploadID), ifMatch(oldD.ETag)) + if !isErrReadQuorum(err) { + t.Fatalf("expected an insufficient read quorum error, got %v", err) + } +} diff --git a/cmd/erasure-server-pool-multipart-http_test.go b/cmd/erasure-server-pool-multipart-http_test.go new file mode 100644 index 000000000..ed2e771e1 --- /dev/null +++ b/cmd/erasure-server-pool-multipart-http_test.go @@ -0,0 +1,341 @@ +// 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" + "encoding/xml" + "errors" + "fmt" + "io" + "net/http" + "net/http/httptest" + "strings" + "sync" + "testing" + "time" + + xhttp "github.com/minio/minio/internal/http" +) + +func multipartConditionRequest(t *testing.T, router http.Handler, method, target, body string, headers map[string]string) *httptest.ResponseRecorder { + t.Helper() + req, err := newTestSignedRequestV4(method, target, int64(len(body)), strings.NewReader(body), globalActiveCred.AccessKey, globalActiveCred.SecretKey, headers) + if err != nil { + t.Fatal(err) + } + rec := httptest.NewRecorder() + router.ServeHTTP(rec, req) + return rec +} + +// The server writes the wire spelling ETag directly into Header; unlike a +// network response, httptest's map has not canonicalized it to Etag. +func multipartConditionResponseETag(rec *httptest.ResponseRecorder) string { + for key, values := range rec.Header() { + if strings.EqualFold(key, xhttp.ETag) && len(values) > 0 { + return strings.Trim(values[0], "\"") + } + } + return "" +} + +func multipartConditionUpload(t *testing.T, z *erasureServerPools, bucket, object string, owner int, opts ObjectOptions) (string, []CompletePart) { + t.Helper() + mp, err := z.serverPools[owner].NewMultipartUpload(t.Context(), bucket, object, opts) + if err != nil { + t.Fatal(err) + } + part, err := z.serverPools[owner].PutObjectPart(t.Context(), bucket, object, mp.UploadID, 1, mustGetPutObjReader(t, bytes.NewBufferString("replacement"), 11, "", ""), ObjectOptions{}) + if err != nil { + t.Fatal(err) + } + return mp.UploadID, []CompletePart{{PartNumber: 1, ETag: part.ETag}} +} + +func multipartConditionCompleteBody(parts []CompletePart) string { + data, err := xml.Marshal(CompleteMultipartUpload{Parts: parts}) + if err != nil { + panic(err) + } + return string(data) +} + +func multipartConditionError(t *testing.T, rec *httptest.ResponseRecorder, code string) { + t.Helper() + decoder := xml.NewDecoder(strings.NewReader(rec.Body.String())) + var response APIErrorResponse + if err := decoder.Decode(&response); err != nil || response.Code != code { + t.Fatalf("expected %s error, got %q: %v", code, rec.Body.String(), err) + } + if err := decoder.Decode(&response); err != io.EOF { + t.Fatalf("expected exactly one error response, got %q: %v", rec.Body.String(), err) + } +} + +// State is deliberately placed per pool; the final request uses signed HTTP +// and the real handler, precondition callback, erasure metadata and rename. +func TestPoolsMultipartConditionalHTTPMatrix(t *testing.T) { + z, _ := consistencyPools(t) + bucket, router, err := initAPIHandlerTest(t.Context(), z, nil, MakeBucketOptions{}) + if err != nil { + t.Fatal(err) + } + for owner := range 2 { + for _, localOld := range []bool{false, true} { + for _, condition := range []string{"match-old", "match-current", "none-match"} { + t.Run(fmt.Sprintf("owner=%d/old=%t/%s", owner, localOld, condition), func(t *testing.T) { + object := fmt.Sprintf("http-%d-%t-%s", owner, localOld, condition) + oldETag := "old-does-not-exist" + if localOld { + oldETag = putConsistencyObject(t, z, bucket, object, owner, "old", ObjectOptions{MTime: UTCNow().Add(-time.Hour)}).ETag + } + id, parts := multipartConditionUpload(t, z, bucket, object, owner, ObjectOptions{}) + current := putConsistencyObject(t, z, bucket, object, 1-owner, "current", ObjectOptions{MTime: UTCNow().Add(-time.Minute)}) + head := multipartConditionRequest(t, router, http.MethodHead, getGetObjectURL("", bucket, object), "", nil) + if head.Code != 200 || multipartConditionResponseETag(head) != current.ETag { + t.Fatalf("bad HEAD: %d %v", head.Code, head.Header()) + } + h := map[string]string{xhttp.IfMatch: "\"" + oldETag + "\""} + want := 412 + if condition == "match-current" { + h[xhttp.IfMatch] = "\"" + current.ETag + "\"" + want = 200 + } + if condition == "none-match" { + h = map[string]string{xhttp.IfNoneMatch: "*"} + } + rec := multipartConditionRequest(t, router, http.MethodPost, getCompleteMultipartUploadURL("", bucket, object, id), multipartConditionCompleteBody(parts), h) + t.Logf("HEAD current=%s, requested=%v, complete HTTP=%d", current.ETag, h, rec.Code) + if rec.Code != want { + t.Errorf("want HTTP %d, got %d: %s", want, rec.Code, rec.Body.String()) + } + if want == 412 { + multipartConditionError(t, rec, "PreconditionFailed") + got := multipartConditionRequest(t, router, http.MethodGet, getGetObjectURL("", bucket, object), "", nil) + if got.Code != 200 || got.Body.String() != "current" { + t.Errorf("rejected request must preserve current data: %d %q", got.Code, got.Body.String()) + } + if _, err := z.serverPools[owner].ListObjectParts(t.Context(), bucket, object, id, 0, 10, ObjectOptions{}); err != nil { + t.Errorf("rejected request consumed upload: %v", err) + } + } + }) + } + } + } +} + +func TestPoolsMultipartConditionalHTTPAbsentObject(t *testing.T) { + for _, deleted := range []bool{false, true} { + for _, match := range []bool{false, true} { + t.Run(fmt.Sprintf("delete-marker=%t/if-match=%t", deleted, match), func(t *testing.T) { + z, _ := consistencyPools(t) + bucket, router, err := initAPIHandlerTest(t.Context(), z, nil, MakeBucketOptions{VersioningEnabled: deleted}) + if err != nil { + t.Fatal(err) + } + object := "http-absent" + if deleted { + putConsistencyObject(t, z, bucket, object, 0, "old", ObjectOptions{Versioned: true, MTime: UTCNow().Add(-time.Hour)}) + _, err = z.serverPools[1].DeleteObject(t.Context(), bucket, object, ObjectOptions{Versioned: true, VersionID: mustGetUUID(), DeleteMarker: true, MTime: UTCNow().Add(-time.Minute)}) + if err != nil { + t.Fatal(err) + } + } + id, parts := multipartConditionUpload(t, z, bucket, object, 0, ObjectOptions{Versioned: deleted}) + headers := map[string]string{xhttp.IfNoneMatch: "*"} + want := http.StatusOK + if match { + headers = map[string]string{xhttp.IfMatch: "\"missing\""} + want = http.StatusNotFound + } + rec := multipartConditionRequest(t, router, http.MethodPost, getCompleteMultipartUploadURL("", bucket, object, id), multipartConditionCompleteBody(parts), headers) + if rec.Code != want { + t.Fatalf("expected HTTP %d, got %d: %s", want, rec.Code, rec.Body.String()) + } + if match { + multipartConditionError(t, rec, "NoSuchKey") + if _, err := z.serverPools[0].ListObjectParts(t.Context(), bucket, object, id, 0, 10, ObjectOptions{}); err != nil { + t.Errorf("rejected request consumed upload: %v", err) + } + } + }) + } + } +} + +// All writes below use ordinary signed S3 requests. The result must be correct +// for every placement; the matrix above deterministically covers split pools. +func TestPoolsMultipartConditionalHTTPNormalRouting(t *testing.T) { + z, _ := consistencyPools(t) + bucket, router, err := initAPIHandlerTest(t.Context(), z, nil, MakeBucketOptions{}) + if err != nil { + t.Fatal(err) + } + object := "normal-routing" + url := getPutObjectURL("", bucket, object) + old := multipartConditionRequest(t, router, http.MethodPut, url, "old-data", nil) + if old.Code != http.StatusOK { + t.Fatalf("initial PUT %d: %s", old.Code, old.Body.String()) + } + init := multipartConditionRequest(t, router, http.MethodPost, url+"?uploads", "", nil) + if init.Code != http.StatusOK { + t.Fatalf("init %d: %s", init.Code, init.Body.String()) + } + var mp InitiateMultipartUploadResponse + if err := xml.Unmarshal(init.Body.Bytes(), &mp); err != nil { + t.Fatal(err) + } + part := multipartConditionRequest(t, router, http.MethodPut, getPutObjectPartURL("", bucket, object, mp.UploadID, "1"), "replacement", nil) + if part.Code != http.StatusOK { + t.Fatalf("part %d: %s", part.Code, part.Body.String()) + } + parts := []CompletePart{{PartNumber: 1, ETag: multipartConditionResponseETag(part)}} + newer := multipartConditionRequest(t, router, http.MethodPut, url, "newer-data", nil) + if newer.Code != http.StatusOK { + t.Fatalf("new PUT %d: %s", newer.Code, newer.Body.String()) + } + head := multipartConditionRequest(t, router, http.MethodHead, url, "", nil) + if head.Code != http.StatusOK || multipartConditionResponseETag(head) != multipartConditionResponseETag(newer) { + t.Fatalf("HEAD did not pick new object: %d %v", head.Code, head.Header()) + } + rec := multipartConditionRequest(t, router, http.MethodPost, getCompleteMultipartUploadURL("", bucket, object, mp.UploadID), multipartConditionCompleteBody(parts), map[string]string{xhttp.IfMatch: "\"" + multipartConditionResponseETag(old) + "\""}) + if rec.Code != http.StatusPreconditionFailed { + t.Errorf("stale If-Match should be HTTP 412, got %d: %s", rec.Code, rec.Body.String()) + } + get := multipartConditionRequest(t, router, http.MethodGet, url, "", nil) + if get.Code != http.StatusOK || get.Body.String() != "newer-data" { + t.Errorf("conditional completion changed newer data: %d %q", get.Code, get.Body.String()) + } +} + +func TestPoolsMultipartConditionalUnreadablePool(t *testing.T) { + z, bucket := consistencyPools(t) + object := "quorum-with-readable-copy" + old := putConsistencyObject(t, z, bucket, object, 0, "readable", ObjectOptions{MTime: UTCNow().Add(-time.Hour)}) + putConsistencyObject(t, z, bucket, object, 1, "hidden-newer", ObjectOptions{MTime: UTCNow().Add(-time.Minute)}) + id, parts := multipartConditionUpload(t, z, bucket, object, 0, ObjectOptions{}) + 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 }() + read, readErr := z.GetObjectInfo(t.Context(), bucket, object, ObjectOptions{}) + t.Logf("ordinary GET lookup: etag=%s err=%v", read.ETag, readErr) + called := 0 + _, err := z.CompleteMultipartUpload(t.Context(), bucket, object, id, parts, ObjectOptions{HasIfMatch: true, CheckPrecondFn: func(oi ObjectInfo) bool { called++; return oi.ETag != old.ETag }}) + if !isErrReadQuorum(err) { + t.Errorf("conditional write must fail on unreadable pool, got %v", err) + } + if called != 0 { + t.Errorf("callback evaluated without complete state: %d calls", called) + } +} + +func TestPoolsMultipartConditionalLatestVersionAndCallbackOnce(t *testing.T) { + for _, explicit := range []bool{false, true} { + t.Run(fmt.Sprintf("explicit=%t", explicit), func(t *testing.T) { + z, bucket := consistencyPools(t) + object := "tie-version" + opts := ObjectOptions{MTime: UTCNow().Add(-time.Hour), Versioned: explicit} + if explicit { + opts.VersionID = mustGetUUID() + } + current := putConsistencyObject(t, z, bucket, object, 0, "first", opts) + putConsistencyObject(t, z, bucket, object, 1, "second", opts) + if explicit { + current = putConsistencyObject(t, z, bucket, object, 1, "latest-other-version", ObjectOptions{Versioned: true, MTime: UTCNow().Add(-time.Minute)}) + } + id, parts := multipartConditionUpload(t, z, bucket, object, 1, opts) + called := 0 + opts.CheckPrecondFn = func(oi ObjectInfo) bool { called++; return oi.ETag != current.ETag } + opts.MTime = time.Time{} + opts.HasIfMatch = true + _, err := z.CompleteMultipartUpload(t.Context(), bucket, object, id, parts, opts) + if err != nil { + t.Errorf("logical current object should match: %v", err) + } + if called != 1 { + t.Errorf("condition evaluated %d times; want exactly once", called) + } + }) + } +} + +func TestPoolsMultipartConditionalConcurrentCompletes(t *testing.T) { + z, bucket := consistencyPools(t) + object := "concurrent-completes" + old := putConsistencyObject(t, z, bucket, object, 0, "old", ObjectOptions{MTime: UTCNow().Add(-time.Hour)}) + putConsistencyObject(t, z, bucket, object, 1, "old", ObjectOptions{MTime: old.ModTime}) + ids := make([]string, 2) + parts := make([][]CompletePart, 2) + for i := range 2 { + ids[i], parts[i] = multipartConditionUpload(t, z, bucket, object, i, ObjectOptions{}) + } + entered := make(chan struct{}) + release := make(chan struct{}) + var gate, releaseOnce sync.Once + defer releaseOnce.Do(func() { close(release) }) + errs := make([]error, 2) + var wg sync.WaitGroup + // Let pool 1's completion hold the object lock before pool 0's starts. + // Once pool 1 commits, a set-local read in pool 0 would still see the + // old ETag and incorrectly accept the second completion. + wg.Add(1) + go func() { + defer wg.Done() + _, errs[1] = z.CompleteMultipartUpload(t.Context(), bucket, object, ids[1], parts[1], ObjectOptions{HasIfMatch: true, CheckPrecondFn: func(oi ObjectInfo) bool { + gate.Do(func() { close(entered); <-release }) + return oi.ETag != old.ETag + }}) + }() + select { + case <-entered: + case <-time.After(10 * time.Second): + t.Fatal("first completion did not enter its condition callback") + } + started := make(chan struct{}) + wg.Add(1) + go func() { + defer wg.Done() + close(started) + _, errs[0] = z.CompleteMultipartUpload(t.Context(), bucket, object, ids[0], parts[0], ObjectOptions{HasIfMatch: true, CheckPrecondFn: func(oi ObjectInfo) bool { return oi.ETag != old.ETag }}) + }() + <-started + releaseOnce.Do(func() { close(release) }) + wg.Wait() + success, failed := 0, 0 + for _, err := range errs { + var p PreConditionFailed + switch { + case err == nil: + success++ + case errors.As(err, &p): + failed++ + default: + t.Errorf("unexpected completion error %v", err) + } + } + if success != 1 || failed != 1 { + t.Errorf("CAS writers: success=%d conditional failures=%d errors=%v", success, failed, errs) + } +} diff --git a/cmd/erasure-server-pool-multipart-preconditions_test.go b/cmd/erasure-server-pool-multipart-preconditions_test.go new file mode 100644 index 000000000..b72392afc --- /dev/null +++ b/cmd/erasure-server-pool-multipart-preconditions_test.go @@ -0,0 +1,135 @@ +// 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" + "errors" + "fmt" + "testing" + "time" +) + +func TestPoolsMultipartConditionMatrix(t *testing.T) { + for owner := range 2 { + for _, withOld := range []bool{false, true} { + for _, condition := range []string{"match-current", "match-old", "none-match-any"} { + t.Run(fmt.Sprintf("upload-pool=%d/old-copy=%t/%s", owner, withOld, condition), func(t *testing.T) { + z, bucket := consistencyPools(t) + object := "conditional-multipart" + oldETag := "arbitrary-old" + if withOld { + old := putConsistencyObject(t, z, bucket, object, owner, "old", ObjectOptions{MTime: UTCNow().Add(-time.Hour)}) + oldETag = old.ETag + } + mp, err := z.serverPools[owner].NewMultipartUpload(t.Context(), bucket, object, ObjectOptions{}) + if err != nil { + t.Fatal(err) + } + part, err := z.serverPools[owner].PutObjectPart(t.Context(), bucket, object, mp.UploadID, 1, mustGetPutObjReader(t, bytes.NewBufferString("replacement"), 11, "", ""), ObjectOptions{}) + if err != nil { + t.Fatal(err) + } + current := putConsistencyObject(t, z, bucket, object, 1-owner, "current", ObjectOptions{MTime: UTCNow().Add(-time.Minute)}) + visible, err := z.GetObjectInfo(t.Context(), bucket, object, ObjectOptions{}) + if err != nil || visible.ETag != current.ETag { + t.Fatalf("invalid current state: %v", err) + } + opts := ObjectOptions{HasIfMatch: condition != "none-match-any", CheckPrecondFn: func(oi ObjectInfo) bool { + switch condition { + case "match-current": + return oi.ETag != current.ETag + case "match-old": + return oi.ETag != oldETag + default: + return true + } + }} + _, err = z.CompleteMultipartUpload(t.Context(), bucket, object, mp.UploadID, []CompletePart{{PartNumber: 1, ETag: part.ETag}}, opts) + if condition == "match-current" { + if err != nil { + t.Errorf("correct logical If-Match rejected: %v", err) + } + } else { + var expected PreConditionFailed + if !errors.As(err, &expected) { + t.Errorf("logical condition must fail, got %v", err) + } + } + }) + } + } + } +} + +func TestPoolsMultipartConditionBoundaries(t *testing.T) { + for _, state := range []string{"missing", "latest-delete-marker", "unreadable-other-pool"} { + for _, match := range []bool{false, true} { + t.Run(fmt.Sprintf("%s/if-match=%t", state, match), func(t *testing.T) { + z, bucket := consistencyPools(t) + object := "conditional-boundary" + versioned := state == "latest-delete-marker" + if versioned { + putConsistencyObject(t, z, bucket, object, 0, "old", ObjectOptions{Versioned: true, MTime: UTCNow().Add(-time.Hour)}) + if _, err := z.serverPools[1].DeleteObject(t.Context(), bucket, object, ObjectOptions{Versioned: true, VersionID: mustGetUUID(), DeleteMarker: true, MTime: UTCNow().Add(-time.Minute)}); err != nil { + t.Fatal(err) + } + } + mp, err := z.serverPools[0].NewMultipartUpload(t.Context(), bucket, object, ObjectOptions{Versioned: versioned}) + if err != nil { + t.Fatal(err) + } + part, err := z.serverPools[0].PutObjectPart(t.Context(), bucket, object, mp.UploadID, 1, mustGetPutObjReader(t, bytes.NewBufferString("new"), 3, "", ""), ObjectOptions{}) + if err != nil { + t.Fatal(err) + } + if state == "unreadable-other-pool" { + set := z.serverPools[1].getHashedSet(object) + getDisks := set.getDisks + faulty := append([]StorageAPI(nil), getDisks()...) + for i := range faulty { + faulty[i] = consistencyReadFaultDisk{StorageAPI: faulty[i], bucket: bucket, object: object} + } + set.getDisks = func() []StorageAPI { return faulty } + defer func() { set.getDisks = getDisks }() + } + opts := ObjectOptions{Versioned: versioned, HasIfMatch: match, CheckPrecondFn: func(ObjectInfo) bool { return true }} + _, err = z.CompleteMultipartUpload(t.Context(), bucket, object, mp.UploadID, []CompletePart{{PartNumber: 1, ETag: part.ETag}}, opts) + switch { + case state == "unreadable-other-pool": + if !isErrReadQuorum(err) { + t.Errorf("unreadable pool must not mean absence: %v", err) + } + case match: + if !isErrObjectNotFound(err) { + t.Errorf("If-Match against logical absence should report absence: %v", err) + } + default: + if err != nil { + t.Errorf("If-None-Match against logical absence must succeed: %v", err) + } + } + if err != nil { + if _, lerr := z.serverPools[0].ListObjectParts(t.Context(), bucket, object, mp.UploadID, 0, 10, ObjectOptions{}); lerr != nil { + t.Errorf("failed condition consumed upload: %v", lerr) + } + } + }) + } + } +} diff --git a/cmd/erasure-server-pool.go b/cmd/erasure-server-pool.go index cb71abff8..f5f612c2a 100644 --- a/cmd/erasure-server-pool.go +++ b/cmd/erasure-server-pool.go @@ -2154,6 +2154,47 @@ func (z *erasureServerPools) CompleteMultipartUpload(ctx context.Context, bucket defer lk.Unlock(lkctx) } opts.NoLock = true + + // A conditional completion must be evaluated against the logical + // latest object across pools, under the object write lock held for + // this operation. The pool hosting the upload may only hold a stale + // duplicate, so its set-local check would both accept an outdated + // ETag and reject the current one. An unreadable pool is not + // absence: it may hold the newest copy, so a read that cannot be + // verified fails the request instead of passing the condition. + // 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 + } + opts.CheckPrecondFn = nil + } } // Hold write locks to verify uploaded parts, also disallows any diff --git a/docs/investigations/multipart-conditional-completion.md b/docs/investigations/multipart-conditional-completion.md new file mode 100644 index 000000000..3a2148ce3 --- /dev/null +++ b/docs/investigations/multipart-conditional-completion.md @@ -0,0 +1,57 @@ +# Conditional multipart completion across pools + +## Defect and scope + +An unfinished multipart upload can remain in one pool while another pool holds +the logical current object. Evaluating `If-Match` against the upload pool's +local copy can accept a stale ETag or reject the current ETag. The pools object +lock serializes writes, but a local read still does not identify the logical +current object. + +This layout does not require rebalance. Commit +`83b2ad418b15ff0fa78175e2014d78d02046edd8` (upstream #21115) made `getPoolIdx` +choose an available pool even when `pinfo.Err == nil`: normal overwrites and +upload initiation can select different pools. This change predates the SILO +multi-pool consistency work. The deterministic regression fixtures place copies +and uploads directly in real erasure pools; they do not claim to run rebalance. + +## Minimal correction + +For multi-pool conditional completion, retain the existing object lock and use +`objectPoolInfos` to read the logical current object before completing the +upload. An unreadable pool is an error, not proof of absence. The first sorted +copy supplies the ETag and encryption metadata used by the existing callback. +A current delete marker is treated as an absent key. `If-Match` then fails for +an absent object; `If-None-Match: *` may proceed. + +Use explicit read options with an empty `VersionID` and `NoAuditLog: true`. +The precondition concerns the logical current object, independently of an +internal completion's destination version. After a successful check, clear +the callback before entering the set layer, so it is evaluated only once. +No new lock, storage format, replica cleanup algorithm or distributed protocol +is introduced. Single-pool and unconditional completion retain their existing +paths. + +## Availability and validation + +If any pool cannot supply the required metadata, conditional completion fails, +even when GET/HEAD can still read a copy from another pool. The unreadable pool +might hold a newer object, a delete marker, or no copy at all; none of these +possibilities can be assumed. Retry after recovery. This behavior is recorded +in the unreleased changelog. + +The regression suite covers both upload-pool directions, stale/current ETags, +`If-None-Match: *`, absent objects and delete markers, read-quorum errors, +explicit destination versions, tied modification times, callback counts, +upload preservation, signed HTTP error bodies and concurrent completions. +The ordinary HTTP routing control accepts every valid placement; deterministic +fixtures provide the cross-pool regression gate. + +## Separate follow-up scope + +The pool-placement change and conditional checks in `PutObject` and +`NewMultipartUpload` require separate assessment. This completion fix does not +repair those paths. In particular, PUT has live destination-version and +preserved-ETag semantics, so its repair must not copy this completion-specific +empty-VersionID rule without examining that contract. Parallelizing the shared +pool metadata reader is also outside this correctness fix.