mirror of
https://github.com/pgsty/minio.git
synced 2026-10-03 16:25:58 +03:00
Compare commits
7 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| dff81f293b | |||
| f653a6ea03 | |||
| 47d239f84f | |||
| e069fe9d92 | |||
| d848fb52b5 | |||
| 3ce8319251 | |||
| 9df0f4abaf |
+14
-3
@@ -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
|
||||
|
||||
@@ -1179,7 +1179,7 @@ func (er erasureObjects) CompleteMultipartUpload(ctx context.Context, bucket str
|
||||
switch {
|
||||
case gerr == nil:
|
||||
reconcileStoredObjectLock(fi.Metadata, storedObjectLockState(curr.UserDefined))
|
||||
reconcileStoredObjectTags(fi.Metadata, curr.UserDefined)
|
||||
reconcileStoredObjectTags(fi.Metadata, curr.UserTags, curr.UserDefined[ReservedMetadataPrefixLower+TaggingTimestamp])
|
||||
case isErrVersionNotFound(gerr) || isErrObjectNotFound(gerr):
|
||||
// No existing version to order against: keep the upload's own accepted
|
||||
// lock, including a pre-upgrade upload that persisted values without
|
||||
|
||||
@@ -135,7 +135,7 @@ func (er erasureObjects) CopyObject(ctx context.Context, srcBucket, srcObject, d
|
||||
|
||||
if dstOpts.ReplicaLockReconcile {
|
||||
reconcileStoredObjectLock(srcInfo.UserDefined, storedObjectLockState(fi.Metadata))
|
||||
reconcileStoredObjectTags(srcInfo.UserDefined, fi.Metadata)
|
||||
reconcileStoredObjectTags(srcInfo.UserDefined, fi.Metadata[xhttp.AmzObjectTagging], fi.Metadata[ReservedMetadataPrefixLower+TaggingTimestamp])
|
||||
}
|
||||
|
||||
filterOnlineDisksInplace(fi, metaArr, onlineDisks)
|
||||
@@ -1311,7 +1311,7 @@ func (er erasureObjects) putObject(ctx context.Context, bucket string, object st
|
||||
// existing version contributes independently ordered lock and tags.
|
||||
if opts.ReplicaLockReconcile && err == nil {
|
||||
reconcileStoredObjectLock(opts.UserDefined, storedObjectLockState(obj.UserDefined))
|
||||
reconcileStoredObjectTags(opts.UserDefined, obj.UserDefined)
|
||||
reconcileStoredObjectTags(opts.UserDefined, obj.UserTags, obj.UserDefined[ReservedMetadataPrefixLower+TaggingTimestamp])
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -126,10 +126,9 @@ func mergedPoolObjectInfo(copies []PoolObjInfo) ObjectInfo {
|
||||
stamp, _ := time.Parse(time.RFC3339Nano, ts)
|
||||
if olderThan(oi.UserDefined[ReservedMetadataPrefixLower+TaggingTimestamp], stamp) {
|
||||
oi.UserDefined[ReservedMetadataPrefixLower+TaggingTimestamp] = ts
|
||||
oi.UserDefined[xhttp.AmzObjectTagging] = copy.ObjInfo.UserDefined[xhttp.AmzObjectTagging]
|
||||
oi.UserTags = copy.ObjInfo.UserTags
|
||||
}
|
||||
}
|
||||
oi.UserTags = oi.UserDefined[xhttp.AmzObjectTagging]
|
||||
return oi
|
||||
}
|
||||
|
||||
@@ -190,6 +189,12 @@ func (z *erasureServerPools) updatePoolMetadata(ctx context.Context, bucket, obj
|
||||
changes[key] = ""
|
||||
}
|
||||
}
|
||||
// Metadata callbacks may explicitly replace tags in the raw write map.
|
||||
// Otherwise retain the merged value from the ObjectInfo read model.
|
||||
tags, ok := updated.UserDefined[xhttp.AmzObjectTagging]
|
||||
if !ok {
|
||||
tags = updated.UserTags
|
||||
}
|
||||
state := storedObjectLockState(updated.UserDefined)
|
||||
opts.VersionID = updated.VersionID
|
||||
if opts.VersionID == "" {
|
||||
@@ -199,11 +204,13 @@ func (z *erasureServerPools) updatePoolMetadata(ctx context.Context, bucket, obj
|
||||
opts.EvalMetadataFn = func(oi *ObjectInfo, _ error) (ReplicateDecision, error) {
|
||||
maps.Copy(oi.UserDefined, changes)
|
||||
replaceObjectLockMetadata(oi.UserDefined, state)
|
||||
for _, key := range []string{xhttp.AmzObjectTagging, ReservedMetadataPrefixLower + TaggingTimestamp} {
|
||||
value, exists := updated.UserDefined[key]
|
||||
if exists || oi.UserDefined[key] != "" {
|
||||
oi.UserDefined[key] = value
|
||||
}
|
||||
// Reassemble the tag value and its ordering timestamp for storage.
|
||||
if tags != "" || oi.UserTags != "" {
|
||||
oi.UserDefined[xhttp.AmzObjectTagging] = tags
|
||||
}
|
||||
key := ReservedMetadataPrefixLower + TaggingTimestamp
|
||||
if value, exists := updated.UserDefined[key]; exists || oi.UserDefined[key] != "" {
|
||||
oi.UserDefined[key] = value
|
||||
}
|
||||
return ReplicateDecision{}, nil
|
||||
}
|
||||
@@ -220,16 +227,18 @@ func (z *erasureServerPools) updatePoolMetadata(ctx context.Context, bucket, obj
|
||||
return primary, nil
|
||||
}
|
||||
|
||||
func reconcileStoredObjectTags(metadata, stored map[string]string) {
|
||||
// Pass the stored tag value explicitly: ObjectInfo.UserDefined excludes it,
|
||||
// whereas FileInfo.Metadata retains the raw storage key.
|
||||
func reconcileStoredObjectTags(metadata map[string]string, storedTags, storedTimestamp string) {
|
||||
key := ReservedMetadataPrefixLower + TaggingTimestamp
|
||||
stamp, err := time.Parse(time.RFC3339Nano, stored[key])
|
||||
stamp, err := time.Parse(time.RFC3339Nano, storedTimestamp)
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
incoming, err := time.Parse(time.RFC3339Nano, metadata[key])
|
||||
if err != nil || !stamp.Before(incoming) {
|
||||
metadata[key] = stored[key]
|
||||
metadata[xhttp.AmzObjectTagging] = stored[xhttp.AmzObjectTagging]
|
||||
metadata[key] = storedTimestamp
|
||||
metadata[xhttp.AmzObjectTagging] = storedTags
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,459 @@
|
||||
// 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 (
|
||||
"context"
|
||||
"fmt"
|
||||
"maps"
|
||||
"strings"
|
||||
"testing"
|
||||
|
||||
xhttp "github.com/minio/minio/internal/http"
|
||||
)
|
||||
|
||||
// The fixture places copies directly in real erasure pools. It models a
|
||||
// duplicated version; it does not claim to exercise a rebalance workflow.
|
||||
func TestPoolsMetadataUpdatePreservesTags(t *testing.T) {
|
||||
z, bucket := consistencyPools(t)
|
||||
const (
|
||||
old = "2026-09-09T09:00:00Z"
|
||||
recent = "2026-09-09T10:00:00Z"
|
||||
timestamp = ReservedMetadataPrefixLower + TaggingTimestamp
|
||||
)
|
||||
for _, test := range []struct {
|
||||
name string
|
||||
tags, stamps [2]string
|
||||
winner int
|
||||
single bool
|
||||
}{
|
||||
{name: "newer-secondary", tags: [2]string{"key=old", "key=new"}, stamps: [2]string{old, recent}, winner: 1},
|
||||
{name: "newer-primary", tags: [2]string{"key=new", "key=old"}, stamps: [2]string{recent, old}},
|
||||
{name: "empty-secondary", tags: [2]string{"key=old", ""}, stamps: [2]string{old, recent}, winner: 1},
|
||||
{name: "empty-primary", tags: [2]string{"", "key=old"}, stamps: [2]string{recent, old}},
|
||||
{name: "single-copy-primary", tags: [2]string{"key=only", ""}, stamps: [2]string{recent, ""}, single: true},
|
||||
{name: "single-copy-secondary", tags: [2]string{"", "key=only"}, stamps: [2]string{"", recent}, winner: 1, single: true},
|
||||
{name: "legacy-single-copy", tags: [2]string{"key=legacy", ""}, single: true},
|
||||
{name: "legacy-duplicates", tags: [2]string{"key=legacy", "key=legacy"}},
|
||||
} {
|
||||
t.Run(test.name, func(t *testing.T) {
|
||||
object := test.name
|
||||
var original ObjectInfo
|
||||
for pool := range 2 {
|
||||
if test.single && pool != test.winner {
|
||||
continue
|
||||
}
|
||||
metadata := map[string]string{
|
||||
xhttp.AmzObjectTagging: test.tags[pool],
|
||||
"copy-local": fmt.Sprint(pool),
|
||||
}
|
||||
if test.stamps[pool] != "" {
|
||||
metadata[timestamp] = test.stamps[pool]
|
||||
}
|
||||
original = putConsistencyObject(t, z, bucket, object, pool, "data", ObjectOptions{
|
||||
Versioned: true, VersionID: original.VersionID, MTime: original.ModTime, UserDefined: metadata,
|
||||
})
|
||||
got, err := z.serverPools[pool].GetObjectInfo(t.Context(), bucket, object, ObjectOptions{VersionID: original.VersionID})
|
||||
if err != nil || got.UserTags != test.tags[pool] || got.UserDefined[timestamp] != test.stamps[pool] {
|
||||
t.Fatalf("pool %d fixture: tags=%q timestamp=%q err=%v", pool, got.UserTags, got.UserDefined[timestamp], err)
|
||||
}
|
||||
if _, exists := got.UserDefined[xhttp.AmzObjectTagging]; exists {
|
||||
t.Fatal("fixture must use the cleaned ObjectInfo representation")
|
||||
}
|
||||
}
|
||||
wantTags, wantStamp := test.tags[test.winner], test.stamps[test.winner]
|
||||
called := 0
|
||||
got, err := z.PutObjectMetadata(t.Context(), bucket, object, ObjectOptions{
|
||||
VersionID: original.VersionID, MTime: original.ModTime,
|
||||
EvalMetadataFn: func(current *ObjectInfo, _ error) (ReplicateDecision, error) {
|
||||
called++
|
||||
if current.UserTags != wantTags || current.UserDefined[timestamp] != wantStamp {
|
||||
t.Errorf("callback tags=%q timestamp=%q; want %q %q", current.UserTags, current.UserDefined[timestamp], wantTags, wantStamp)
|
||||
}
|
||||
current.UserDefined["unrelated-update"] = "preserved"
|
||||
return ReplicateDecision{}, nil
|
||||
},
|
||||
})
|
||||
if err != nil || called != 1 {
|
||||
t.Fatalf("metadata update: %v, callbacks=%d", err, called)
|
||||
}
|
||||
if got.UserTags != wantTags || got.UserDefined[timestamp] != wantStamp {
|
||||
t.Errorf("response tags=%q timestamp=%q; want %q %q", got.UserTags, got.UserDefined[timestamp], wantTags, wantStamp)
|
||||
}
|
||||
for pool := range 2 {
|
||||
got, err := z.serverPools[pool].GetObjectInfo(t.Context(), bucket, object, ObjectOptions{VersionID: original.VersionID})
|
||||
if test.single && pool != test.winner {
|
||||
if !isErrVersionNotFound(err) {
|
||||
t.Errorf("metadata update created another copy: %v", err)
|
||||
}
|
||||
continue
|
||||
}
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
t.Logf("pool %d persisted tags=%q timestamp=%q", pool, got.UserTags, got.UserDefined[timestamp])
|
||||
if got.UserTags != wantTags || got.UserDefined[timestamp] != wantStamp {
|
||||
t.Errorf("pool %d persisted tags=%q timestamp=%q; want %q %q", pool, got.UserTags, got.UserDefined[timestamp], wantTags, wantStamp)
|
||||
}
|
||||
if got.UserDefined["unrelated-update"] != "preserved" || got.UserDefined["copy-local"] != fmt.Sprint(pool) {
|
||||
t.Errorf("pool %d lost unrelated metadata: %v", pool, got.UserDefined)
|
||||
}
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestReplicaWritesPreserveTagOrdering(t *testing.T) {
|
||||
z, bucket := consistencyPools(t)
|
||||
// Pool allocation checks the host's used-space percentage. Present only
|
||||
// its free space as fixture capacity; all reads and writes still use the
|
||||
// real disks. This keeps unrelated host disk usage out of the tag test.
|
||||
for _, pool := range z.serverPools {
|
||||
for _, set := range pool.sets {
|
||||
getDisks := set.getDisks
|
||||
disks := append([]StorageAPI(nil), getDisks()...)
|
||||
for i := range disks {
|
||||
disks[i] = tagTestCapacityDisk{StorageAPI: disks[i]}
|
||||
}
|
||||
set.getDisks = func() []StorageAPI { return disks }
|
||||
t.Cleanup(func() { set.getDisks = getDisks })
|
||||
}
|
||||
}
|
||||
const (
|
||||
old = "2026-09-09T09:00:00Z"
|
||||
recent = "2026-09-09T10:00:00Z"
|
||||
timestamp = ReservedMetadataPrefixLower + TaggingTimestamp
|
||||
)
|
||||
for _, path := range []struct {
|
||||
name, operation string
|
||||
direct bool
|
||||
winner int
|
||||
}{
|
||||
{name: "set-put", operation: "put", direct: true},
|
||||
{name: "set-multipart", operation: "multipart", direct: true},
|
||||
{name: "set-copy", operation: "copy", direct: true},
|
||||
{name: "pools-put", operation: "put", winner: 1},
|
||||
{name: "pools-multipart", operation: "multipart", winner: 1},
|
||||
{name: "pools-copy-primary", operation: "copy"},
|
||||
{name: "pools-copy-secondary", operation: "copy", winner: 1},
|
||||
} {
|
||||
for _, test := range []struct {
|
||||
name, storedTags, incomingTags, storedStamp, incomingStamp, wantTags string
|
||||
}{
|
||||
{"stored-newer", "key=stored", "key=incoming", recent, old, "key=stored"},
|
||||
{"stored-deleted", "", "key=incoming", recent, old, ""},
|
||||
{"incoming-newer", "key=stored", "key=incoming", old, recent, "key=incoming"},
|
||||
{"incoming-deleted", "key=stored", "", old, recent, ""},
|
||||
} {
|
||||
t.Run(path.name+"/"+test.name, func(t *testing.T) {
|
||||
object := path.name + "-" + test.name
|
||||
var original ObjectInfo
|
||||
for pool := range 2 {
|
||||
if path.direct && pool != 0 {
|
||||
continue
|
||||
}
|
||||
metadata := map[string]string{
|
||||
xhttp.AmzObjectTagging: "key=older-copy",
|
||||
timestamp: "2026-09-09T08:00:00Z",
|
||||
}
|
||||
if pool == path.winner {
|
||||
metadata[xhttp.AmzObjectTagging] = test.storedTags
|
||||
metadata[timestamp] = test.storedStamp
|
||||
}
|
||||
original = putConsistencyObject(t, z, bucket, object, pool, "data", ObjectOptions{
|
||||
Versioned: true, VersionID: original.VersionID, MTime: original.ModTime, UserDefined: metadata,
|
||||
})
|
||||
}
|
||||
opts := ObjectOptions{
|
||||
Versioned: true, VersionID: original.VersionID, MTime: original.ModTime, ReplicaLockReconcile: true,
|
||||
UserDefined: map[string]string{
|
||||
xhttp.AmzObjectTagging: test.incomingTags,
|
||||
timestamp: test.incomingStamp,
|
||||
},
|
||||
}
|
||||
var got ObjectInfo
|
||||
var err error
|
||||
switch path.operation {
|
||||
case "put":
|
||||
put := z.PutObject
|
||||
if path.direct {
|
||||
put = z.serverPools[0].PutObject
|
||||
}
|
||||
got, err = put(t.Context(), bucket, object, mustGetPutObjReader(t, strings.NewReader("data"), 4, "", ""), opts)
|
||||
case "multipart":
|
||||
// Persist the incoming tags with the upload, before completion
|
||||
// reconciles the destination version through its real resolver.
|
||||
mp, err := z.serverPools[0].NewMultipartUpload(t.Context(), bucket, object, opts)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
part, err := z.serverPools[0].PutObjectPart(t.Context(), bucket, object, mp.UploadID, 1,
|
||||
mustGetPutObjReader(t, strings.NewReader("data"), 4, "", ""), ObjectOptions{})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
complete := z.CompleteMultipartUpload
|
||||
if path.direct {
|
||||
complete = z.serverPools[0].CompleteMultipartUpload
|
||||
}
|
||||
got, err = complete(t.Context(), bucket, object, mp.UploadID, []CompletePart{{PartNumber: 1, ETag: part.ETag}}, ObjectOptions{
|
||||
Versioned: true, MTime: original.ModTime, ReplicaLockReconcile: true,
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
case "copy":
|
||||
src := original
|
||||
src.metadataOnly = true
|
||||
src.UserDefined = maps.Clone(opts.UserDefined)
|
||||
copyObject := z.CopyObject
|
||||
if path.direct {
|
||||
copyObject = z.serverPools[0].CopyObject
|
||||
}
|
||||
got, err = copyObject(t.Context(), bucket, object, bucket, object, src, ObjectOptions{VersionID: original.VersionID}, opts)
|
||||
}
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if got.UserTags != test.wantTags || got.UserDefined[timestamp] != recent {
|
||||
t.Errorf("response tags=%q timestamp=%q; want %q %q", got.UserTags, got.UserDefined[timestamp], test.wantTags, recent)
|
||||
}
|
||||
copies := 0
|
||||
for pool := range 2 {
|
||||
got, err := z.serverPools[pool].GetObjectInfo(t.Context(), bucket, object, ObjectOptions{VersionID: original.VersionID})
|
||||
if isErrVersionNotFound(err) {
|
||||
continue
|
||||
}
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
copies++
|
||||
if got.UserTags != test.wantTags || got.UserDefined[timestamp] != recent {
|
||||
t.Errorf("pool %d persisted tags=%q timestamp=%q; want %q %q", pool, got.UserTags, got.UserDefined[timestamp], test.wantTags, recent)
|
||||
}
|
||||
}
|
||||
if copies != 1 {
|
||||
t.Errorf("replacement left %d copies; want 1", copies)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
type tagTestCapacityDisk struct{ StorageAPI }
|
||||
|
||||
func (d tagTestCapacityDisk) DiskInfo(ctx context.Context, opts DiskInfoOptions) (DiskInfo, error) {
|
||||
info, err := d.StorageAPI.DiskInfo(ctx, opts)
|
||||
info.Total, info.Used = info.Free, 0
|
||||
return info, err
|
||||
}
|
||||
|
||||
func TestMergedPoolObjectInfoTagOrdering(t *testing.T) {
|
||||
const (
|
||||
old = "2026-09-09T09:00:00Z"
|
||||
recent = "2026-09-09T10:00:00Z"
|
||||
timestamp = ReservedMetadataPrefixLower + TaggingTimestamp
|
||||
)
|
||||
for _, test := range []struct {
|
||||
name, firstStamp, secondStamp, secondTags string
|
||||
winner int
|
||||
}{
|
||||
{"newer", old, recent, "key=second", 1},
|
||||
{"newer-removal", old, recent, "", 1},
|
||||
{"equal", recent, recent, "key=second", 0},
|
||||
{"unordered", "", "", "key=second", 0},
|
||||
{"missing-first", "", recent, "key=second", 1},
|
||||
{"missing-second", recent, "", "key=second", 0},
|
||||
{"invalid-first", "invalid", recent, "key=second", 1},
|
||||
{"invalid-second", recent, "invalid", "key=second", 0},
|
||||
} {
|
||||
t.Run(test.name, func(t *testing.T) {
|
||||
tagValues := []string{"key=first", test.secondTags}
|
||||
stamps := []string{test.firstStamp, test.secondStamp}
|
||||
copies := make([]PoolObjInfo, 2)
|
||||
before := make([]map[string]string, 2)
|
||||
for i := range copies {
|
||||
fi := FileInfo{Metadata: map[string]string{xhttp.AmzObjectTagging: tagValues[i]}}
|
||||
if stamps[i] != "" {
|
||||
fi.Metadata[timestamp] = stamps[i]
|
||||
}
|
||||
copies[i] = PoolObjInfo{Index: i, ObjInfo: fi.ToObjectInfo("bucket", "object", true)}
|
||||
before[i] = maps.Clone(copies[i].ObjInfo.UserDefined)
|
||||
}
|
||||
got := mergedPoolObjectInfo(copies)
|
||||
if got.UserTags != tagValues[test.winner] || got.UserDefined[timestamp] != stamps[test.winner] {
|
||||
t.Errorf("merged tags=%q timestamp=%q; want %q %q", got.UserTags, got.UserDefined[timestamp], tagValues[test.winner], stamps[test.winner])
|
||||
}
|
||||
if _, exists := got.UserDefined[xhttp.AmzObjectTagging]; exists {
|
||||
t.Error("merged ObjectInfo leaked the raw tagging key into UserDefined")
|
||||
}
|
||||
for i := range copies {
|
||||
if !maps.Equal(copies[i].ObjInfo.UserDefined, before[i]) || copies[i].ObjInfo.UserTags != tagValues[i] {
|
||||
t.Errorf("merge mutated input copy %d", i)
|
||||
}
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestPoolsMetadataCallbackReplacesTags(t *testing.T) {
|
||||
z, bucket := consistencyPools(t)
|
||||
const timestamp = ReservedMetadataPrefixLower + TaggingTimestamp
|
||||
for _, test := range []struct{ name, tags string }{
|
||||
{"replace", "key=callback"},
|
||||
{"remove", ""},
|
||||
} {
|
||||
t.Run(test.name, func(t *testing.T) {
|
||||
var original ObjectInfo
|
||||
for pool := range 2 {
|
||||
original = putConsistencyObject(t, z, bucket, test.name, pool, "data", ObjectOptions{
|
||||
Versioned: true, VersionID: original.VersionID, MTime: original.ModTime,
|
||||
UserDefined: map[string]string{
|
||||
xhttp.AmzObjectTagging: []string{"key=old", "key=new"}[pool],
|
||||
timestamp: []string{"2026-09-09T09:00:00Z", "2026-09-09T10:00:00Z"}[pool],
|
||||
},
|
||||
})
|
||||
}
|
||||
const updatedStamp = "2026-09-09T11:00:00Z"
|
||||
got, err := z.PutObjectMetadata(t.Context(), bucket, test.name, ObjectOptions{
|
||||
VersionID: original.VersionID, MTime: original.ModTime,
|
||||
EvalMetadataFn: func(current *ObjectInfo, _ error) (ReplicateDecision, error) {
|
||||
if current.UserTags != "key=new" {
|
||||
t.Errorf("callback read tags=%q; want key=new", current.UserTags)
|
||||
}
|
||||
if _, exists := current.UserDefined[xhttp.AmzObjectTagging]; exists {
|
||||
t.Error("callback received the raw tagging key")
|
||||
}
|
||||
current.UserDefined[xhttp.AmzObjectTagging] = test.tags
|
||||
current.UserDefined[timestamp] = updatedStamp
|
||||
return ReplicateDecision{}, nil
|
||||
},
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if got.UserTags != test.tags || got.UserDefined[timestamp] != updatedStamp {
|
||||
t.Errorf("callback update response tags=%q timestamp=%q", got.UserTags, got.UserDefined[timestamp])
|
||||
}
|
||||
for pool := range 2 {
|
||||
got, err := z.serverPools[pool].GetObjectInfo(t.Context(), bucket, test.name, ObjectOptions{VersionID: original.VersionID})
|
||||
if err != nil || got.UserTags != test.tags || got.UserDefined[timestamp] != updatedStamp {
|
||||
t.Errorf("pool %d did not persist callback tags: tags=%q timestamp=%q err=%v", pool, got.UserTags, got.UserDefined[timestamp], err)
|
||||
}
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestReconcileStoredObjectTagOrdering(t *testing.T) {
|
||||
const (
|
||||
old = "2026-09-09T09:00:00Z"
|
||||
recent = "2026-09-09T10:00:00Z"
|
||||
timestamp = ReservedMetadataPrefixLower + TaggingTimestamp
|
||||
)
|
||||
for _, test := range []struct {
|
||||
name, storedStamp, incomingStamp, storedTags string
|
||||
wantStored bool
|
||||
}{
|
||||
{"stored-newer", recent, old, "key=stored", true},
|
||||
{"incoming-newer", old, recent, "key=stored", false},
|
||||
{"equal", recent, recent, "key=stored", true},
|
||||
{"equal-removal", recent, recent, "", true},
|
||||
{"missing-stored", "", recent, "key=stored", false},
|
||||
{"missing-incoming", recent, "", "key=stored", true},
|
||||
{"invalid-stored", "invalid", recent, "key=stored", false},
|
||||
{"invalid-incoming", recent, "invalid", "key=stored", true},
|
||||
} {
|
||||
t.Run(test.name, func(t *testing.T) {
|
||||
metadata := map[string]string{
|
||||
xhttp.AmzObjectTagging: "key=incoming",
|
||||
timestamp: test.incomingStamp,
|
||||
"unrelated": "preserved",
|
||||
}
|
||||
reconcileStoredObjectTags(metadata, test.storedTags, test.storedStamp)
|
||||
wantTags, wantStamp := "key=incoming", test.incomingStamp
|
||||
if test.wantStored {
|
||||
wantTags, wantStamp = test.storedTags, test.storedStamp
|
||||
}
|
||||
if metadata[xhttp.AmzObjectTagging] != wantTags || metadata[timestamp] != wantStamp || metadata["unrelated"] != "preserved" {
|
||||
t.Errorf("reconciled metadata=%v; want tags=%q timestamp=%q and unrelated field preserved", metadata, wantTags, wantStamp)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestPoolsMetadataUpdatePreservesAbsentTags(t *testing.T) {
|
||||
z, bucket := consistencyPools(t)
|
||||
const (
|
||||
old = "2026-09-09T09:00:00Z"
|
||||
recent = "2026-09-09T10:00:00Z"
|
||||
timestamp = ReservedMetadataPrefixLower + TaggingTimestamp
|
||||
)
|
||||
for _, removal := range []bool{false, true} {
|
||||
t.Run(fmt.Sprintf("timestamp-only-removal=%t", removal), func(t *testing.T) {
|
||||
object := fmt.Sprintf("absent-tags-%t", removal)
|
||||
var original ObjectInfo
|
||||
checkStored := func(pool int, wantKey bool, wantTags, wantStamp string) {
|
||||
t.Helper()
|
||||
infos, errs := readAllFileInfo(t.Context(), z.serverPools[pool].getHashedSet(object).getDisks(), "", bucket, object, original.VersionID, false, false)
|
||||
for disk, info := range infos {
|
||||
if errs[disk] != nil {
|
||||
t.Fatal(errs[disk])
|
||||
}
|
||||
tags, exists := info.Metadata[xhttp.AmzObjectTagging]
|
||||
if exists != wantKey || tags != wantTags || info.Metadata[timestamp] != wantStamp {
|
||||
t.Errorf("pool %d disk %d raw tagging key=%t value=%q stamp=%q; want %t %q %q", pool, disk, exists, tags, info.Metadata[timestamp], wantKey, wantTags, wantStamp)
|
||||
}
|
||||
}
|
||||
}
|
||||
for pool := range 2 {
|
||||
metadata := map[string]string{}
|
||||
if removal {
|
||||
metadata[timestamp] = recent
|
||||
if pool == 1 {
|
||||
metadata[xhttp.AmzObjectTagging] = "key=old"
|
||||
metadata[timestamp] = old
|
||||
}
|
||||
}
|
||||
original = putConsistencyObject(t, z, bucket, object, pool, "data", ObjectOptions{
|
||||
Versioned: true, VersionID: original.VersionID, MTime: original.ModTime, UserDefined: metadata,
|
||||
})
|
||||
checkStored(pool, removal && pool == 1, metadata[xhttp.AmzObjectTagging], metadata[timestamp])
|
||||
}
|
||||
_, err := z.PutObjectMetadata(t.Context(), bucket, object, ObjectOptions{
|
||||
VersionID: original.VersionID, MTime: original.ModTime,
|
||||
EvalMetadataFn: func(current *ObjectInfo, _ error) (ReplicateDecision, error) {
|
||||
current.UserDefined["unrelated-update"] = "preserved"
|
||||
return ReplicateDecision{}, nil
|
||||
},
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
wantStamp := ""
|
||||
if removal {
|
||||
wantStamp = recent
|
||||
}
|
||||
for pool := range 2 {
|
||||
// A previously non-empty key needs an explicit empty value to
|
||||
// propagate deletion; an absent key should remain absent.
|
||||
checkStored(pool, removal && pool == 1, "", wantStamp)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
@@ -1447,7 +1447,7 @@ func (z *erasureServerPools) CopyObject(ctx context.Context, srcBucket, srcObjec
|
||||
}
|
||||
stored := mergedPoolObjectInfo(copies)
|
||||
reconcileStoredObjectLock(srcInfo.UserDefined, storedObjectLockState(stored.UserDefined))
|
||||
reconcileStoredObjectTags(srcInfo.UserDefined, stored.UserDefined)
|
||||
reconcileStoredObjectTags(srcInfo.UserDefined, stored.UserTags, stored.UserDefined[ReservedMetadataPrefixLower+TaggingTimestamp])
|
||||
idx := copies[0].Index
|
||||
oi, err := z.serverPools[idx].CopyObject(ctx, srcBucket, srcObject, dstBucket, dstObject, srcInfo, srcOpts, dstOpts)
|
||||
if err == nil {
|
||||
@@ -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
|
||||
|
||||
@@ -0,0 +1,133 @@
|
||||
// Copyright (c) 2026 PGSTY
|
||||
// SPDX-License-Identifier: AGPL-3.0-or-later
|
||||
|
||||
package cmd
|
||||
|
||||
import (
|
||||
"context"
|
||||
"os"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/minio/madmin-go/v3"
|
||||
"github.com/minio/minio/internal/auth"
|
||||
"github.com/minio/minio/internal/grid"
|
||||
"github.com/pgsty/silo-pkg/v3/policy"
|
||||
)
|
||||
|
||||
// Two independent IAM caches share the same real object backend, as sibling
|
||||
// nodes do. Deliver the actual peer handler only after the source committed.
|
||||
func TestIAMPeerDeleteNotificationReloadsCommittedState(t *testing.T) {
|
||||
for _, name := range []string{"deleted", "recreated", "recreated_without_grant"} {
|
||||
recreate := name != "deleted"
|
||||
t.Run(name, func(t *testing.T) {
|
||||
resetTestGlobals()
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
defer cancel()
|
||||
obj, disk, err := prepareFS(ctx)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
defer os.RemoveAll(disk)
|
||||
defer obj.Shutdown(ctx)
|
||||
defer resetTestGlobals()
|
||||
globalObjLayerMutex.Lock()
|
||||
globalObjectAPI = obj
|
||||
globalObjLayerMutex.Unlock()
|
||||
must := func(err error) {
|
||||
t.Helper()
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
source := globalIAMSys
|
||||
const user = "peer-reload-user"
|
||||
req := madmin.AddOrUpdateUserReq{SecretKey: "original-test-password", Status: madmin.AccountEnabled}
|
||||
_, err = source.CreateUser(ctx, user, req)
|
||||
must(err)
|
||||
_, err = source.PolicyDBSet(ctx, user, "readwrite", regUser, false)
|
||||
must(err)
|
||||
_, err = source.AddUsersToGroup(ctx, "peer-reload-group", []string{user})
|
||||
must(err)
|
||||
_, err = source.PolicyDBSet(ctx, "peer-reload-group", "readwrite", regUser, true)
|
||||
must(err)
|
||||
svc, _, err := source.NewServiceAccount(ctx, user, nil, newServiceAccountOpts{accessKey: "peer-reload-service", secretKey: "service-test-password"})
|
||||
must(err)
|
||||
signingKey, err := getTokenSigningKey()
|
||||
must(err)
|
||||
sts, err := auth.GetNewCredentialsWithMetadata(map[string]any{"exp": UTCNow().Add(time.Hour).Unix(), parentClaim: user}, signingKey)
|
||||
must(err)
|
||||
sts.ParentUser = user
|
||||
_, err = source.SetTempUser(ctx, sts.AccessKey, sts, "")
|
||||
must(err)
|
||||
|
||||
siblingStore := &IAMStoreSys{IAMStorageAPI: newIAMObjectStore(obj, MinIOUsersSysType)}
|
||||
must(siblingStore.LoadIAMCache(ctx, true))
|
||||
must(siblingStore.UserNotificationHandler(ctx, sts.AccessKey, stsUser))
|
||||
for _, key := range []string{user, svc.AccessKey, sts.AccessKey} {
|
||||
if _, ok := siblingStore.GetUser(key); !ok {
|
||||
t.Fatalf("fixture did not load %s", key)
|
||||
}
|
||||
}
|
||||
must(source.DeleteUser(ctx, user, false))
|
||||
if recreate {
|
||||
req.SecretKey = "recreated-test-password"
|
||||
_, err = source.CreateUser(ctx, user, req)
|
||||
must(err)
|
||||
if name == "recreated" {
|
||||
_, err = source.PolicyDBSet(ctx, user, "readonly", regUser, false)
|
||||
must(err)
|
||||
}
|
||||
}
|
||||
|
||||
sibling := &IAMSys{store: siblingStore, usersSysType: MinIOUsersSysType}
|
||||
globalIAMSys = sibling
|
||||
defer func() { globalIAMSys = source }()
|
||||
server := &peerRESTServer{}
|
||||
for range 2 {
|
||||
_, remoteErr := server.DeleteUserHandler(grid.NewMSSWith(map[string]string{peerRESTUser: user}))
|
||||
if remoteErr != nil {
|
||||
t.Fatal(remoteErr)
|
||||
}
|
||||
}
|
||||
if recreate {
|
||||
u, ok := siblingStore.GetUser(user)
|
||||
if !ok || u.Credentials.SecretKey != req.SecretKey {
|
||||
t.Fatal("delayed deletion notification removed the recreated user")
|
||||
}
|
||||
loaded := make(map[string]UserIdentity)
|
||||
must(source.store.loadUser(ctx, user, regUser, loaded))
|
||||
if loaded[user].Credentials.SecretKey != req.SecretKey {
|
||||
t.Fatal("notification changed the persisted recreated identity")
|
||||
}
|
||||
if allowed := sibling.IsAllowed(policy.Args{AccountName: user, Action: policy.GetObjectAction, BucketName: "bucket", ObjectName: "object"}); allowed != (name == "recreated") {
|
||||
t.Fatal("notification did not load the recreated user's current grant")
|
||||
}
|
||||
}
|
||||
if sibling.IsAllowed(policy.Args{AccountName: user, Action: policy.PutObjectAction, BucketName: "bucket", ObjectName: "object"}) {
|
||||
t.Fatal("notification retained an old direct or group grant")
|
||||
}
|
||||
for _, key := range []string{svc.AccessKey, sts.AccessKey} {
|
||||
if _, ok := siblingStore.GetUser(key); ok {
|
||||
t.Fatalf("notification retained a revoked child: %s", key)
|
||||
}
|
||||
}
|
||||
if !recreate {
|
||||
for _, key := range []string{user} {
|
||||
if _, ok := siblingStore.GetUser(key); ok {
|
||||
t.Fatalf("notification retained a revoked cached identity: %s", key)
|
||||
}
|
||||
}
|
||||
if sibling.IsAllowed(policy.Args{AccountName: user, Action: policy.GetObjectAction, BucketName: "bucket", ObjectName: "object"}) {
|
||||
t.Fatal("notification retained the user's old grant")
|
||||
}
|
||||
cache := siblingStore.rlock()
|
||||
member := cache.iamUserGroupMemberships[user].Contains("peer-reload-group")
|
||||
siblingStore.runlock()
|
||||
if member {
|
||||
t.Fatal("notification retained the deleted user's group membership")
|
||||
}
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
@@ -1814,6 +1814,52 @@ func (store *IAMStoreSys) PolicyMappingNotificationHandler(ctx context.Context,
|
||||
return err
|
||||
}
|
||||
|
||||
// UserDeletionNotificationHandler refreshes all cached state affected by a
|
||||
// parent deletion. Reloading only a recreated parent would retain the older
|
||||
// children, policy, and group memberships in this node's cache.
|
||||
func (store *IAMStoreSys) UserDeletionNotificationHandler(ctx context.Context, accessKey string) error {
|
||||
if accessKey == "" {
|
||||
return errInvalidArgument
|
||||
}
|
||||
cache := store.rlock()
|
||||
groups := cache.iamUserGroupMemberships[accessKey].ToSlice()
|
||||
children := make(map[string]IAMUserType)
|
||||
for key, u := range cache.iamUsersMap {
|
||||
if u.Credentials.ParentUser == accessKey && u.Credentials.IsServiceAccount() {
|
||||
children[key] = svcUser
|
||||
}
|
||||
}
|
||||
store.runlock()
|
||||
|
||||
if err := store.UserNotificationHandler(ctx, accessKey, regUser); err != nil {
|
||||
return err
|
||||
}
|
||||
if err := store.PolicyMappingNotificationHandler(ctx, accessKey, false, regUser); err != nil {
|
||||
return err
|
||||
}
|
||||
// Temporary credentials are loaded on demand. Invalidate their cached
|
||||
// copies; a later request must revalidate them against persisted state.
|
||||
cache = store.lock()
|
||||
for key, u := range cache.iamSTSAccountsMap {
|
||||
if u.Credentials.ParentUser == accessKey {
|
||||
delete(cache.iamSTSAccountsMap, key)
|
||||
}
|
||||
}
|
||||
cache.updatedAt = time.Now()
|
||||
store.unlock()
|
||||
for key, userType := range children {
|
||||
if err := store.UserNotificationHandler(ctx, key, userType); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
for _, group := range groups {
|
||||
if err := store.GroupNotificationHandler(ctx, group); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// UserNotificationHandler - handles updating a user/STS account/service account
|
||||
// from storage.
|
||||
func (store *IAMStoreSys) UserNotificationHandler(ctx context.Context, accessKey string, userType IAMUserType) error {
|
||||
|
||||
@@ -162,6 +162,15 @@ func (sys *IAMSys) LoadUser(ctx context.Context, objAPI ObjectLayer, accessKey s
|
||||
return sys.store.UserNotificationHandler(ctx, accessKey, userType)
|
||||
}
|
||||
|
||||
// LoadUserAfterDelete reloads a parent's identity and cached dependents after a
|
||||
// sibling committed a deletion. Each record may already have been recreated.
|
||||
func (sys *IAMSys) LoadUserAfterDelete(ctx context.Context, accessKey string) error {
|
||||
if !sys.Initialized() {
|
||||
return errServerNotInitialized
|
||||
}
|
||||
return sys.store.UserDeletionNotificationHandler(ctx, accessKey)
|
||||
}
|
||||
|
||||
// LoadServiceAccount - reloads a specific service account from backend disks or etcd.
|
||||
func (sys *IAMSys) LoadServiceAccount(ctx context.Context, accessKey string) error {
|
||||
if !sys.Initialized() {
|
||||
|
||||
@@ -230,7 +230,8 @@ func (s *peerRESTServer) LoadServiceAccountHandler(mss *grid.MSS) (np grid.NoPay
|
||||
return np, nerr
|
||||
}
|
||||
|
||||
// DeleteUserHandler - deletes a user on the server.
|
||||
// DeleteUserHandler reloads the state committed by another node. A delayed
|
||||
// notification must not delete an identity recreated since that commit.
|
||||
func (s *peerRESTServer) DeleteUserHandler(mss *grid.MSS) (np grid.NoPayload, nerr *grid.RemoteErr) {
|
||||
objAPI := newObjectLayerFn()
|
||||
if objAPI == nil {
|
||||
@@ -242,7 +243,9 @@ func (s *peerRESTServer) DeleteUserHandler(mss *grid.MSS) (np grid.NoPayload, ne
|
||||
return np, grid.NewRemoteErr(errors.New("username is missing"))
|
||||
}
|
||||
|
||||
if err := globalIAMSys.DeleteUser(context.Background(), accessKey, false); err != nil {
|
||||
ctx, cancel := context.WithTimeout(GlobalContext, defaultContextTimeout)
|
||||
defer cancel()
|
||||
if err := globalIAMSys.LoadUserAfterDelete(ctx, accessKey); err != nil {
|
||||
return np, grid.NewRemoteErr(err)
|
||||
}
|
||||
|
||||
|
||||
@@ -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.
|
||||
Reference in New Issue
Block a user