fix(storage): evaluate multipart preconditions across pools

Signed-off-by: Feng Ruohang <rh@vonng.com>
This commit is contained in:
Feng Ruohang
2026-09-15 21:13:37 +08:00
parent d848fb52b5
commit e069fe9d92
6 changed files with 760 additions and 3 deletions
@@ -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 <http://www.gnu.org/licenses/>.
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)
}
}
@@ -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 <http://www.gnu.org/licenses/>.
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)
}
}
@@ -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 <http://www.gnu.org/licenses/>.
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)
}
}
})
}
}
}
+41
View File
@@ -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