mirror of
https://github.com/pgsty/minio.git
synced 2026-10-15 22:26:00 +03:00
Compare commits
26 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| 2a4db6a4f9 | |||
| e90cb2067d | |||
| 00a375c4e2 | |||
| ab4c97e607 | |||
| 721dbfd1b7 | |||
| 2926474688 | |||
| a7370aa9ac | |||
| a592d69f45 | |||
| bfc7906064 | |||
| eeff112491 | |||
| a70156e45b | |||
| 99245e3ef8 | |||
| 7a3ca0f821 | |||
| d94b3b3c92 | |||
| 86c5a92eef | |||
| cdc3b8cb10 | |||
| 0d3ba5d4cb | |||
| 2d57b975d1 | |||
| cb85461c42 | |||
| ba6e8c8d69 | |||
| a54d0c4354 | |||
| 00d4a206f7 | |||
| f480086b6b | |||
| 968c30238c | |||
| 8c784f6133 | |||
| 258b167b1c |
@@ -96,6 +96,9 @@ jobs:
|
||||
- name: Run multipart listing and cancellation tests under race detector
|
||||
run: go test -race ./cmd -run '^Test(MultipartListing|MultipartAbort|PaginateMultipartUploads|ListMultipartUploads)' -count=1 -timeout=5m
|
||||
|
||||
- name: Run listing metadata recovery, pagination and cancellation tests under race detector
|
||||
run: go test -race ./cmd -run '^Test(ListFallback|ListingCache|ListingMerge|MergeXLV2LatestNullCrossParity|CountXLV2MixedECGroups|ListCrossParity)' -count=1 -timeout=5m
|
||||
|
||||
- name: Run CPU metrics tests under race detector
|
||||
run: go test -race ./cmd -run '^TestLoadCPUMetrics' -count=1 -timeout=5m
|
||||
|
||||
|
||||
@@ -70,6 +70,12 @@ jobs:
|
||||
buildscripts/verify-rebrand.sh
|
||||
dockerscripts/docker-entrypoint_test.sh
|
||||
|
||||
# Keep this gate aligned with the Object Lock race regression in go.yml.
|
||||
- name: Gate release on Object Lock authorization and overwrite tests under race detector
|
||||
env:
|
||||
MINIO_API_REQUESTS_MAX: "10000"
|
||||
run: go test -race -tags kqueue,dev ./cmd -run '^Test(APIPutObjectRetention|APIVersionAddressedWrite|PutObjectRetention|DirectoryObjectLock|ObjectLockOverwrite|APICopyObjectPreservesVersionAfterSuspension|MultipartNullObjectLock|PoolsObjectLockOverwrite|MultipartVersioningTransitions|PlanReview|RecoveredMultipartUpload)' -count=1 -timeout=5m
|
||||
|
||||
- name: Compute release variables
|
||||
env:
|
||||
# Passed through the environment, never interpolated into the script
|
||||
|
||||
@@ -0,0 +1,45 @@
|
||||
// Copyright (c) 2026 PGSTY
|
||||
//
|
||||
// This file is part of SILO Object Storage.
|
||||
// SPDX-License-Identifier: AGPL-3.0-or-later
|
||||
//
|
||||
// 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 (
|
||||
"fmt"
|
||||
"net/http"
|
||||
"testing"
|
||||
|
||||
"github.com/minio/minio/internal/hash"
|
||||
)
|
||||
|
||||
func TestChecksumMismatchAPIMessage(t *testing.T) {
|
||||
for _, algorithm := range []string{"CRC32", "CRC32C", "SHA1", "SHA256", "CRC64NVME", ""} {
|
||||
t.Run(algorithm, func(t *testing.T) {
|
||||
mismatch := hash.ChecksumMismatch{Algorithm: algorithm, Want: "expected", Got: "actual"}
|
||||
want := errorCodes.ToAPIErr(ErrContentChecksumMismatch).Description
|
||||
if algorithm != "" {
|
||||
want = fmt.Sprintf("The %s you specified did not match the calculated checksum.", algorithm)
|
||||
}
|
||||
for _, err := range []error{mismatch, fmt.Errorf("upload: %w", mismatch)} {
|
||||
got := toAPIError(t.Context(), err)
|
||||
if got.HTTPStatusCode != http.StatusBadRequest || got.Code != "XAmzContentChecksumMismatch" || got.Description != want {
|
||||
t.Fatalf("toAPIError(%v) = %+v, want 400 XAmzContentChecksumMismatch: %s", err, got, want)
|
||||
}
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
@@ -2505,6 +2505,14 @@ func toAPIError(ctx context.Context, err error) APIError {
|
||||
case errors.Is(err, errMissingPartChecksum):
|
||||
apiErr.Description = strings.TrimPrefix(err.Error(), errMissingPartChecksum.Error()+": ")
|
||||
}
|
||||
// PutObject/UploadPart checksum mismatches name the algorithm in the
|
||||
// message, matching S3 (e.g. "The CRC32 you specified did not match the
|
||||
// calculated checksum."). Falls through to the generic table description
|
||||
// when the algorithm is unknown.
|
||||
var checksumMismatch hash.ChecksumMismatch
|
||||
if errors.As(err, &checksumMismatch) && checksumMismatch.Algorithm != "" {
|
||||
apiErr.Description = fmt.Sprintf("The %s you specified did not match the calculated checksum.", checksumMismatch.Algorithm)
|
||||
}
|
||||
switch apiErr.Code {
|
||||
case "NotImplemented":
|
||||
apiErr = APIError{
|
||||
|
||||
+5
-4
@@ -744,10 +744,10 @@ func (f *folderScanner) scanFolder(ctx context.Context, folder cachedFolder, int
|
||||
}
|
||||
},
|
||||
// Some disks have data for this.
|
||||
partial: func(entries metaCacheEntries, errs []error) {
|
||||
partial: func(entries metaCacheEntries, errs []error) error {
|
||||
if !f.shouldHeal() {
|
||||
cancel()
|
||||
return
|
||||
return nil
|
||||
}
|
||||
entry, ok := entries.resolve(&resolver)
|
||||
if !ok {
|
||||
@@ -772,7 +772,7 @@ func (f *folderScanner) scanFolder(ctx context.Context, folder cachedFolder, int
|
||||
}
|
||||
|
||||
if entry.isDir() {
|
||||
return
|
||||
return nil
|
||||
}
|
||||
|
||||
// We got an entry which we should be able to heal.
|
||||
@@ -787,7 +787,7 @@ func (f *folderScanner) scanFolder(ctx context.Context, folder cachedFolder, int
|
||||
scannerLogIf(ctx, err)
|
||||
}
|
||||
foundObjs = foundObjs || err == nil
|
||||
return
|
||||
return nil
|
||||
}
|
||||
|
||||
custom["versions"] = fmt.Sprint(len(fiv.Versions))
|
||||
@@ -814,6 +814,7 @@ func (f *folderScanner) scanFolder(ctx context.Context, folder cachedFolder, int
|
||||
}
|
||||
custom["success_versions"] = fmt.Sprint(successVersions)
|
||||
custom["failed_versions"] = fmt.Sprint(failVersions)
|
||||
return nil
|
||||
},
|
||||
// Too many disks failed.
|
||||
finished: func(errs []error) {
|
||||
|
||||
@@ -86,7 +86,7 @@ func (er erasureObjects) listAndHeal(ctx context.Context, bucket, prefix string,
|
||||
cancel()
|
||||
}
|
||||
},
|
||||
partial: func(entries metaCacheEntries, _ []error) {
|
||||
partial: func(entries metaCacheEntries, _ []error) error {
|
||||
entry, ok := entries.resolve(&resolver)
|
||||
if !ok {
|
||||
// check if we can get one entry at least
|
||||
@@ -94,12 +94,12 @@ func (er erasureObjects) listAndHeal(ctx context.Context, bucket, prefix string,
|
||||
entry, _ = entries.firstFound()
|
||||
}
|
||||
if !recursive && prefix != entry.name {
|
||||
return
|
||||
return nil
|
||||
}
|
||||
if err := healEntry(bucket, *entry, scanMode); err != nil {
|
||||
cancel()
|
||||
return
|
||||
}
|
||||
return nil
|
||||
},
|
||||
finished: nil,
|
||||
}
|
||||
|
||||
@@ -20,6 +20,7 @@ package cmd
|
||||
import (
|
||||
"context"
|
||||
"encoding/hex"
|
||||
"errors"
|
||||
"fmt"
|
||||
"math/rand"
|
||||
"reflect"
|
||||
@@ -110,6 +111,96 @@ func TestReduceErrs(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
// TestReduceWriteQuorumErrsCommitMetaMissing verifies that an in-quorum
|
||||
// errCommitMetaMissing reduces to the sentinel, which renameData() converts
|
||||
// to errErasureWriteQuorum, while below-quorum reduces directly to
|
||||
// errErasureWriteQuorum.
|
||||
func TestReduceWriteQuorumErrsCommitMetaMissing(t *testing.T) {
|
||||
errs := make([]error, 16)
|
||||
for i := range 9 {
|
||||
errs[i] = errCommitMetaMissing
|
||||
}
|
||||
if err := reduceWriteQuorumErrs(t.Context(), errs, objectOpIgnoredErrs, 9); !errors.Is(err, errCommitMetaMissing) {
|
||||
t.Errorf("expected %s, got %s", errCommitMetaMissing, err)
|
||||
}
|
||||
// The sentinel must be distinguishable from errFileNotFound so that
|
||||
// commit failures are never mapped to ObjectNotFound (404).
|
||||
if errors.Is(errCommitMetaMissing, errFileNotFound) {
|
||||
t.Error("errCommitMetaMissing must not match errFileNotFound")
|
||||
}
|
||||
|
||||
// Below quorum the reduction must yield a write-quorum failure (503);
|
||||
// eight sentinels against eight committed drives is not a quorum.
|
||||
belowQuorum := make([]error, 16)
|
||||
for i := range 8 {
|
||||
belowQuorum[i] = errCommitMetaMissing
|
||||
}
|
||||
if err := reduceWriteQuorumErrs(t.Context(), belowQuorum, objectOpIgnoredErrs, 9); !errors.Is(err, errErasureWriteQuorum) {
|
||||
t.Errorf("expected %s, got %s", errErasureWriteQuorum, err)
|
||||
}
|
||||
}
|
||||
|
||||
// renameDataFailingDisk is a StorageAPI stub returning a fixed error
|
||||
// from RenameData, enough to drive renameData() in unit tests.
|
||||
type renameDataFailingDisk struct {
|
||||
StorageAPI
|
||||
err error
|
||||
undoWrite bool
|
||||
}
|
||||
|
||||
func (d *renameDataFailingDisk) RenameData(context.Context, string, string, FileInfo, string, string, RenameOptions) (RenameDataResp, error) {
|
||||
return RenameDataResp{}, d.err
|
||||
}
|
||||
|
||||
func (d *renameDataFailingDisk) DeleteVersion(_ context.Context, _, _ string, _ FileInfo, _ bool, opts DeleteOptions) error {
|
||||
d.undoWrite = opts.UndoWrite
|
||||
return nil
|
||||
}
|
||||
|
||||
// TestRenameDataCommitMetaMissing verifies that renameData() converts an
|
||||
// in-quorum errCommitMetaMissing into errErasureWriteQuorum (HTTP 503),
|
||||
// while an in-quorum errFileNotFound keeps its meaning so that
|
||||
// PutObject() still reports IncompleteBody for client disconnects.
|
||||
func TestRenameDataCommitMetaMissing(t *testing.T) {
|
||||
for _, tc := range []struct {
|
||||
name string
|
||||
failures int
|
||||
diskErr error
|
||||
wantErr error
|
||||
}{
|
||||
{"all_commit_metadata_missing", 16, errCommitMetaMissing, errErasureWriteQuorum},
|
||||
{"quorum_commit_metadata_missing", 9, errCommitMetaMissing, errErasureWriteQuorum},
|
||||
{"neither_side_has_quorum", 8, errCommitMetaMissing, errErasureWriteQuorum},
|
||||
{"successful_write_quorum", 7, errCommitMetaMissing, nil},
|
||||
{"staging_failure", 16, errFileNotFound, errFileNotFound},
|
||||
} {
|
||||
t.Run(tc.name, func(t *testing.T) {
|
||||
metadata := make([]FileInfo, 16)
|
||||
disks := make([]StorageAPI, 16)
|
||||
for i := range disks {
|
||||
metadata[i] = FileInfo{Deleted: true} // valid without erasure info
|
||||
disk := &renameDataFailingDisk{}
|
||||
if i < tc.failures {
|
||||
disk.err = tc.diskErr
|
||||
}
|
||||
disks[i] = disk
|
||||
}
|
||||
_, _, _, err := renameData(t.Context(), disks, minioMetaTmpBucket, "tmp", metadata, "bucket", "object", 9)
|
||||
if !errors.Is(err, tc.wantErr) {
|
||||
t.Fatalf("expected %v, got %v", tc.wantErr, err)
|
||||
}
|
||||
for i, disk := range disks {
|
||||
// Failed writes undo only the drives that committed. A
|
||||
// successful quorum must never be rolled back.
|
||||
wantUndo := tc.wantErr != nil && i >= tc.failures
|
||||
if got := disk.(*renameDataFailingDisk).undoWrite; got != wantUndo {
|
||||
t.Errorf("disk %d: UndoWrite = %v, want %v", i, got, wantUndo)
|
||||
}
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
// TestHashOrder - test order of ints in array
|
||||
func TestHashOrder(t *testing.T) {
|
||||
testCases := []struct {
|
||||
|
||||
@@ -0,0 +1,177 @@
|
||||
// Copyright (c) 2026 clown15
|
||||
//
|
||||
// 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"
|
||||
"encoding/base64"
|
||||
"fmt"
|
||||
"net/url"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"testing"
|
||||
"time"
|
||||
)
|
||||
|
||||
func TestExtractModTimeFromUploadIDDir(t *testing.T) {
|
||||
now := time.Unix(1792056747, 123456789).UTC()
|
||||
nano := now.UnixNano()
|
||||
uuid := "d16ad42a-6fe5-4a3b-8092-69983142238f"
|
||||
|
||||
testCases := []struct {
|
||||
name string
|
||||
dirName string
|
||||
expectedMod time.Time
|
||||
wantZero bool
|
||||
}{
|
||||
{
|
||||
name: "Modern Plaintext Format (<UUID>x<UnixNano>)",
|
||||
dirName: fmt.Sprintf("%sx%d", uuid, nano),
|
||||
expectedMod: time.Unix(0, nano),
|
||||
wantZero: false,
|
||||
},
|
||||
{
|
||||
// This valid UUID decodes to bytes containing 'x'. Checking
|
||||
// for that byte before parsing plaintext loses the timestamp.
|
||||
name: "Plaintext Also Decodes To Base64 Containing x",
|
||||
dirName: fmt.Sprintf("4eacd477-e733-4851-b821-6fd42785ff87x%d", nano),
|
||||
expectedMod: now,
|
||||
},
|
||||
{
|
||||
name: "Legacy Format Without Timestamp (<UUID>)",
|
||||
dirName: uuid,
|
||||
expectedMod: time.Time{},
|
||||
wantZero: true,
|
||||
},
|
||||
{
|
||||
name: "Base64 Encoded Format",
|
||||
dirName: base64.RawURLEncoding.EncodeToString([]byte(fmt.Sprintf("%sx%d", uuid, nano))),
|
||||
expectedMod: time.Unix(0, nano),
|
||||
wantZero: false,
|
||||
},
|
||||
{
|
||||
name: "Encoded Format With Deployment Prefix",
|
||||
dirName: base64.RawURLEncoding.EncodeToString([]byte(fmt.Sprintf("%s.%sx%d", uuid, uuid, nano))),
|
||||
expectedMod: now,
|
||||
},
|
||||
{
|
||||
name: "Malformed Suffix (<UUID>xinvalid)",
|
||||
dirName: fmt.Sprintf("%sxnotanumber", uuid),
|
||||
expectedMod: time.Time{},
|
||||
wantZero: true,
|
||||
},
|
||||
{
|
||||
name: "Timestamp Overflow",
|
||||
dirName: uuid + "x9223372036854775808",
|
||||
wantZero: true,
|
||||
},
|
||||
{
|
||||
name: "Malformed Plaintext UUID",
|
||||
dirName: fmt.Sprintf("not-a-uuidx%d", nano),
|
||||
wantZero: true,
|
||||
},
|
||||
{
|
||||
name: "Empty Directory Name",
|
||||
dirName: "",
|
||||
expectedMod: time.Time{},
|
||||
wantZero: true,
|
||||
},
|
||||
}
|
||||
|
||||
for _, tc := range testCases {
|
||||
t.Run(tc.name, func(t *testing.T) {
|
||||
got := extractModTimeFromUploadIDDir(tc.dirName)
|
||||
if tc.wantZero {
|
||||
if !got.IsZero() {
|
||||
t.Fatalf("expected zero time for %q, got %v", tc.dirName, got)
|
||||
}
|
||||
return
|
||||
}
|
||||
if !got.Equal(tc.expectedMod) {
|
||||
t.Fatalf("extractModTimeFromUploadIDDir(%q) = %v; want %v", tc.dirName, got, tc.expectedMod)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
type multipartCleanupDisk struct {
|
||||
StorageAPI
|
||||
root string
|
||||
modTime time.Time
|
||||
readErr error
|
||||
reads int
|
||||
}
|
||||
|
||||
func (d *multipartCleanupDisk) Endpoint() Endpoint {
|
||||
return Endpoint{URL: &url.URL{Path: d.root}}
|
||||
}
|
||||
|
||||
func (d *multipartCleanupDisk) ReadVersion(context.Context, string, string, string, string, ReadOptions) (FileInfo, error) {
|
||||
d.reads++
|
||||
return FileInfo{ModTime: d.modTime}, d.readErr
|
||||
}
|
||||
|
||||
func (d *multipartCleanupDisk) StatVol(context.Context, string) (VolInfo, error) {
|
||||
return VolInfo{Created: time.Now()}, nil
|
||||
}
|
||||
|
||||
func TestCleanupStaleUploadsOnDisk(t *testing.T) {
|
||||
const uuid = "d16ad42a-6fe5-4a3b-8092-69983142238f"
|
||||
expiry := globalAPIConfig.getStaleUploadsExpiry()
|
||||
for _, tc := range []struct {
|
||||
name, prefix string
|
||||
age time.Duration
|
||||
legacy bool
|
||||
readErr error
|
||||
wantReads int
|
||||
wantDeleted bool
|
||||
}{
|
||||
{name: "recent", prefix: uuid, age: expiry / 2, readErr: errFileNotFound},
|
||||
{name: "plaintext_base64_collision", prefix: "4eacd477-e733-4851-b821-6fd42785ff87", age: expiry / 2, readErr: errFileNotFound},
|
||||
{name: "expired", prefix: uuid, age: 2 * expiry, readErr: errFileNotFound, wantDeleted: true},
|
||||
{name: "legacy_recent", prefix: uuid, age: expiry / 2, legacy: true, wantReads: 1},
|
||||
{name: "legacy_expired", prefix: uuid, age: 2 * expiry, legacy: true, wantReads: 1, wantDeleted: true},
|
||||
{name: "legacy_metadata_missing", prefix: uuid, age: 2 * expiry, legacy: true, readErr: errFileNotFound, wantReads: 1},
|
||||
{name: "malformed_metadata_missing", prefix: "not-a-uuid", age: 2 * expiry, readErr: errFileNotFound, wantReads: 1},
|
||||
} {
|
||||
t.Run(tc.name, func(t *testing.T) {
|
||||
modTime := time.Now().Add(-tc.age)
|
||||
id := tc.prefix
|
||||
if !tc.legacy {
|
||||
id = fmt.Sprintf("%sx%d", tc.prefix, modTime.UnixNano())
|
||||
}
|
||||
disk := &multipartCleanupDisk{root: t.TempDir(), modTime: modTime, readErr: tc.readErr}
|
||||
uploadDir := filepath.Join(disk.root, minioMetaMultipartBucket, "sha", id)
|
||||
if err := os.MkdirAll(uploadDir, 0o755); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
erasureObjects{}.cleanupStaleUploadsOnDisk(t.Context(), disk)
|
||||
if disk.reads != tc.wantReads {
|
||||
t.Errorf("ReadVersion calls = %d, want %d", disk.reads, tc.wantReads)
|
||||
}
|
||||
_, err := os.Stat(uploadDir)
|
||||
if tc.wantDeleted {
|
||||
if !os.IsNotExist(err) {
|
||||
t.Errorf("expired upload must be removed: %v", err)
|
||||
}
|
||||
} else if err != nil {
|
||||
t.Errorf("upload must be preserved: %v", err)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
+19
-10
@@ -174,6 +174,24 @@ func (er erasureObjects) deleteAll(ctx context.Context, bucket, prefix string) {
|
||||
wg.Wait()
|
||||
}
|
||||
|
||||
// extractModTimeFromUploadIDDir extracts creation time from the upload ID directory on disk.
|
||||
// On-disk directory names are of the form <UUID>x<UnixNano>.
|
||||
// Parse the native format first: plaintext UUIDs can also be valid Base64.
|
||||
// Preserve the previous encoded-format parser for compatibility.
|
||||
func extractModTimeFromUploadIDDir(uploadIDDir string) time.Time {
|
||||
if modTime, ok := multipartUploadTime(uploadIDDir); ok {
|
||||
return modTime
|
||||
}
|
||||
if b64, err := base64.RawURLEncoding.DecodeString(uploadIDDir); err == nil {
|
||||
if split := strings.Split(string(b64), "x"); len(split) == 2 {
|
||||
if t, err := strconv.ParseInt(split[1], 10, 64); err == nil {
|
||||
return time.Unix(0, t)
|
||||
}
|
||||
}
|
||||
}
|
||||
return time.Time{}
|
||||
}
|
||||
|
||||
// Remove the old multipart uploads on the given disk.
|
||||
func (er erasureObjects) cleanupStaleUploadsOnDisk(ctx context.Context, disk StorageAPI) {
|
||||
drivePath := disk.Endpoint().Path
|
||||
@@ -181,16 +199,7 @@ func (er erasureObjects) cleanupStaleUploadsOnDisk(ctx context.Context, disk Sto
|
||||
readDirFn(pathJoin(drivePath, minioMetaMultipartBucket), func(shaDir string, typ os.FileMode) error {
|
||||
readDirFn(pathJoin(drivePath, minioMetaMultipartBucket, shaDir), func(uploadIDDir string, typ os.FileMode) error {
|
||||
uploadIDPath := pathJoin(shaDir, uploadIDDir)
|
||||
var modTime time.Time
|
||||
// Upload IDs are of the form base64_url(<UUID>x<UnixNano>), we can extract the time from the UUID.
|
||||
if b64, err := base64.RawURLEncoding.DecodeString(uploadIDDir); err == nil {
|
||||
if split := strings.Split(string(b64), "x"); len(split) == 2 {
|
||||
t, err := strconv.ParseInt(split[1], 10, 64)
|
||||
if err == nil {
|
||||
modTime = time.Unix(0, t)
|
||||
}
|
||||
}
|
||||
}
|
||||
modTime := extractModTimeFromUploadIDDir(uploadIDDir)
|
||||
// Fallback for older uploads without time in the ID.
|
||||
if modTime.IsZero() {
|
||||
wait := deleteMultipartCleanupSleeper.Timer(ctx)
|
||||
|
||||
@@ -1083,6 +1083,12 @@ func renameData(ctx context.Context, disks []StorageAPI, srcBucket, srcEntry str
|
||||
errs := g.Wait()
|
||||
|
||||
err := reduceWriteQuorumErrs(ctx, errs, objectOpIgnoredErrs, writeQuorum)
|
||||
if errors.Is(err, errCommitMetaMissing) {
|
||||
// Commit metadata went missing on a quorum of drives; the write
|
||||
// did not commit. Return a 503 class error so clients retry
|
||||
// instead of treating the object as absent (404).
|
||||
err = errErasureWriteQuorum
|
||||
}
|
||||
if err != nil {
|
||||
dg := errgroup.WithNErrs(len(disks))
|
||||
for index, nerr := range errs {
|
||||
|
||||
@@ -730,11 +730,12 @@ func (set *erasureObjects) listObjectsToDecommission(ctx context.Context, bi dec
|
||||
minDisks: listingQuorum,
|
||||
reportNotFound: false,
|
||||
agreed: fn,
|
||||
partial: func(entries metaCacheEntries, _ []error) {
|
||||
partial: func(entries metaCacheEntries, _ []error) error {
|
||||
entry, ok := entries.resolve(&resolver)
|
||||
if ok {
|
||||
fn(*entry)
|
||||
}
|
||||
return nil
|
||||
},
|
||||
finished: nil,
|
||||
})
|
||||
|
||||
@@ -525,11 +525,12 @@ func (set *erasureObjects) listObjectsToRebalance(ctx context.Context, bucketNam
|
||||
minDisks: listingQuorum,
|
||||
reportNotFound: false,
|
||||
agreed: fn,
|
||||
partial: func(entries metaCacheEntries, _ []error) {
|
||||
partial: func(entries metaCacheEntries, _ []error) error {
|
||||
entry, ok := entries.resolve(&resolver)
|
||||
if ok {
|
||||
fn(*entry)
|
||||
}
|
||||
return nil
|
||||
},
|
||||
finished: nil,
|
||||
})
|
||||
|
||||
@@ -2651,11 +2651,15 @@ func (z *erasureServerPools) Walk(ctx context.Context, bucket, prefix string, re
|
||||
minDisks: listingQuorum,
|
||||
reportNotFound: false,
|
||||
agreed: send,
|
||||
partial: func(entries metaCacheEntries, _ []error) {
|
||||
entry, ok := entries.resolve(&resolver)
|
||||
if ok {
|
||||
partial: func(entries metaCacheEntries, errs []error) error {
|
||||
entry, err := set.resolveListEntry(ctx, bucket, entries, errs, &resolver, z.listingNSLock(set))
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if entry != nil {
|
||||
send(*entry)
|
||||
}
|
||||
return nil
|
||||
},
|
||||
finished: nil,
|
||||
}
|
||||
|
||||
+2
-1
@@ -529,7 +529,7 @@ func (er *erasureObjects) healErasureSet(ctx context.Context, buckets []string,
|
||||
jt.Take()
|
||||
go healEntry(bucket, entry)
|
||||
},
|
||||
partial: func(entries metaCacheEntries, _ []error) {
|
||||
partial: func(entries metaCacheEntries, _ []error) error {
|
||||
entry, ok := entries.resolve(&resolver)
|
||||
if !ok {
|
||||
// check if we can get one entry at least
|
||||
@@ -538,6 +538,7 @@ func (er *erasureObjects) healErasureSet(ctx context.Context, buckets []string,
|
||||
}
|
||||
jt.Take()
|
||||
go healEntry(bucket, *entry)
|
||||
return nil
|
||||
},
|
||||
finished: func(errs []error) {
|
||||
success := countErrs(errs, nil)
|
||||
|
||||
@@ -670,6 +670,12 @@ func (m *metaCacheEntriesSorted) forwardPast(s string) {
|
||||
// If the context is canceled the function will return the error,
|
||||
// otherwise the function will return nil.
|
||||
func mergeEntryChannels(ctx context.Context, in []chan metaCacheEntry, out chan<- metaCacheEntry, readQuorum int) error {
|
||||
return mergeEntryChannelsWithErrors(ctx, in, out, readQuorum, nil)
|
||||
}
|
||||
|
||||
// terminalErr is inspected after each input closes and before the merger
|
||||
// advances beyond it. Callers must publish the error before closing the input.
|
||||
func mergeEntryChannelsWithErrors(ctx context.Context, in []chan metaCacheEntry, out chan<- metaCacheEntry, readQuorum int, terminalErr func(int) error) error {
|
||||
defer xioutil.SafeClose(out)
|
||||
top := make([]*metaCacheEntry, len(in))
|
||||
nDone := 0
|
||||
@@ -683,6 +689,9 @@ func mergeEntryChannels(ctx context.Context, in []chan metaCacheEntry, out chan<
|
||||
return ctx.Err()
|
||||
case v, ok := <-in[0]:
|
||||
if !ok {
|
||||
if terminalErr != nil {
|
||||
return terminalErr(0)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
select {
|
||||
@@ -700,6 +709,11 @@ func mergeEntryChannels(ctx context.Context, in []chan metaCacheEntry, out chan<
|
||||
return ctx.Err()
|
||||
case entry, ok := <-in[idx]:
|
||||
if !ok {
|
||||
if terminalErr != nil {
|
||||
if err := terminalErr(idx); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
top[idx] = nil
|
||||
nDone++
|
||||
} else {
|
||||
|
||||
@@ -0,0 +1,326 @@
|
||||
// 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"
|
||||
"context"
|
||||
"encoding/xml"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"net/url"
|
||||
"slices"
|
||||
"sync"
|
||||
"sync/atomic"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/minio/minio/internal/grid"
|
||||
)
|
||||
|
||||
type listingReadFailDisk struct {
|
||||
StorageAPI
|
||||
object string
|
||||
reads *atomic.Int64
|
||||
}
|
||||
|
||||
func (d *listingReadFailDisk) ReadXL(ctx context.Context, bucket, object string, readData bool) (RawFileInfo, error) {
|
||||
if object == d.object {
|
||||
d.reads.Add(1)
|
||||
return RawFileInfo{}, errDiskNotFound
|
||||
}
|
||||
return d.StorageAPI.ReadXL(ctx, bucket, object, readData)
|
||||
}
|
||||
|
||||
// Install before issuing requests: cache writers can outlive their HTTP
|
||||
// response, so disk faults must be switched without replacing getDisks.
|
||||
// Each published slice must remain immutable.
|
||||
func installListingDiskSwitch(set *erasureObjects) ([]StorageAPI, func([]StorageAPI)) {
|
||||
disks := set.getDisks()
|
||||
var current atomic.Value
|
||||
current.Store(disks)
|
||||
set.getDisks = func() []StorageAPI { return current.Load().([]StorageAPI) }
|
||||
return disks, func(disks []StorageAPI) { current.Store(disks) }
|
||||
}
|
||||
|
||||
func TestListingCachePreservesFallbackError(t *testing.T) {
|
||||
for _, continuation := range []bool{false, true} {
|
||||
name := "cold"
|
||||
if continuation {
|
||||
name = "continuation"
|
||||
}
|
||||
t.Run(name, func(t *testing.T) {
|
||||
testListingCacheFallbackError(t, continuation, false)
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestListingCacheCleanupKeepsOriginalID(t *testing.T) {
|
||||
testListingCacheFallbackError(t, true, true)
|
||||
}
|
||||
|
||||
// Use actual peer RPCs to verify the abandoned cache is updated after the
|
||||
// caller clears its own listing ID. The local-only path cannot observe this.
|
||||
func recordListingPeerCleanup(t *testing.T) func() {
|
||||
t.Helper()
|
||||
tg, err := grid.SetupTestGrid(2)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
previous := globalNotificationSys
|
||||
t.Cleanup(func() {
|
||||
tg.Cleanup()
|
||||
globalNotificationSys = previous
|
||||
})
|
||||
getIDs := make(chan string, 8)
|
||||
type cleanupResult struct {
|
||||
id string
|
||||
err *grid.RemoteErr
|
||||
}
|
||||
updated := make(chan cleanupResult, 1)
|
||||
server := &peerRESTServer{}
|
||||
if err := getMetacacheListingRPC.Register(tg.Managers[1], func(o *listPathOptions) (*metacache, *grid.RemoteErr) {
|
||||
getIDs <- o.ID
|
||||
return server.GetMetacacheListingHandler(o)
|
||||
}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := updateMetacacheListingRPC.Register(tg.Managers[1], func(m *metacache) (*metacache, *grid.RemoteErr) {
|
||||
result, err := server.UpdateMetacacheListingHandler(m)
|
||||
if m.error == "no longer used" {
|
||||
updated <- cleanupResult{id: m.id, err: err}
|
||||
}
|
||||
return result, err
|
||||
}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
client := &peerRESTClient{gridConn: func() *grid.Connection { return tg.Managers[0].Connection(tg.Hosts[1]) }}
|
||||
globalNotificationSys = &NotificationSys{peerClients: []*peerRESTClient{client}, allPeerClients: []*peerRESTClient{client}}
|
||||
return func() {
|
||||
ctx, cancel := context.WithTimeout(t.Context(), 5*time.Second)
|
||||
defer cancel()
|
||||
var ids []string
|
||||
for range 2 {
|
||||
select {
|
||||
case id := <-getIDs:
|
||||
ids = append(ids, id)
|
||||
case <-ctx.Done():
|
||||
t.Fatal("peer did not receive the failed cache cleanup lookup")
|
||||
}
|
||||
}
|
||||
if ids[0] == "" || ids[1] != ids[0] {
|
||||
t.Fatalf("cleanup lookup lost the original cache ID: %q", ids)
|
||||
}
|
||||
select {
|
||||
case result := <-updated:
|
||||
if result.err != nil || result.id != ids[0] {
|
||||
t.Fatalf("cleanup update: id=%q err=%v, want id=%q", result.id, result.err, ids[0])
|
||||
}
|
||||
t.Logf("peer cleanup Get/Update retained cache ID %q", result.id)
|
||||
case <-ctx.Done():
|
||||
t.Fatal("peer did not finish the failed cache cleanup update")
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func testListingCacheFallbackError(t *testing.T, continuation, peer bool) {
|
||||
t.Helper()
|
||||
z, bucket, router := nullQuorumBackend(t)
|
||||
var waitPeerCleanup func()
|
||||
if peer {
|
||||
waitPeerCleanup = recordListingPeerCleanup(t)
|
||||
}
|
||||
set := z.serverPools[0].sets[0]
|
||||
disks, switchDisks := installListingDiskSwitch(set)
|
||||
body := bytes.Repeat([]byte{'a'}, 8192)
|
||||
mtime := time.Now().UTC().Add(-time.Hour)
|
||||
for _, object := range []string{"a", "b", "z"} {
|
||||
nullQuorumPut(t, z, bucket, object, body, ObjectOptions{MTime: mtime})
|
||||
}
|
||||
old := make([][]byte, 3)
|
||||
for i := range old {
|
||||
old[i] = mustReadNullQuorumMeta(t, disks[i], bucket, "z")
|
||||
}
|
||||
nullQuorumPut(t, z, bucket, "z", body, ObjectOptions{MTime: mtime.Add(time.Minute)})
|
||||
var token string
|
||||
if continuation {
|
||||
rec := nullQuorumRequest(t, router, http.MethodGet, getListObjectsV2URL("", bucket, "", "1", "", "", ""))
|
||||
var list ListObjectsV2Response
|
||||
if rec.Code != http.StatusOK || xml.Unmarshal(rec.Body.Bytes(), &list) != nil || !list.IsTruncated || list.NextContinuationToken == "" {
|
||||
t.Fatalf("first page: %d %s", rec.Code, rec.Body.String())
|
||||
}
|
||||
token = list.NextContinuationToken
|
||||
}
|
||||
for i := range old {
|
||||
if err := disks[i].WriteAll(t.Context(), bucket, "z/"+xlStorageFormatFile, old[i]); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
var reads atomic.Int64
|
||||
walkers := slices.Clone(disks)
|
||||
for i := range walkers {
|
||||
walkers[i] = &listingReadFailDisk{StorageAPI: disks[i], object: "z", reads: &reads}
|
||||
if i >= 8 {
|
||||
walkers[i] = &failedMetadataWalkDisk{StorageAPI: walkers[i]}
|
||||
}
|
||||
}
|
||||
switchDisks(walkers)
|
||||
target := getListObjectsV2URL("", bucket, "", "1000", "", "", "")
|
||||
if token != "" {
|
||||
target += "&continuation-token=" + url.QueryEscape(token)
|
||||
}
|
||||
rec := nullQuorumRequest(t, router, http.MethodGet, target)
|
||||
t.Logf("LIST=%d failed metadata reads=%d body=%s", rec.Code, reads.Load(), rec.Body.String())
|
||||
if reads.Load() == 0 {
|
||||
t.Fatal("fallback not exercised")
|
||||
}
|
||||
var apiErr APIErrorResponse
|
||||
if rec.Code != http.StatusServiceUnavailable || xml.Unmarshal(rec.Body.Bytes(), &apiErr) != nil || apiErr.Code != "SlowDownRead" {
|
||||
t.Fatalf("fallback failure lost its read-quorum error: %d %s", rec.Code, rec.Body.String())
|
||||
}
|
||||
if waitPeerCleanup != nil {
|
||||
waitPeerCleanup()
|
||||
}
|
||||
if continuation {
|
||||
caches, _ := localMetacacheMgr.getBucket(t.Context(), bucket).cloneCaches()
|
||||
if len(caches) == 0 {
|
||||
t.Fatal("continuation did not create a cache")
|
||||
}
|
||||
for _, cache := range caches {
|
||||
if cache.status != scanStateError {
|
||||
t.Fatalf("incomplete cache was not marked failed: %+v", cache)
|
||||
}
|
||||
}
|
||||
}
|
||||
switchDisks(disks)
|
||||
retry := nullQuorumRequest(t, router, http.MethodGet, target)
|
||||
var recovered ListObjectsV2Response
|
||||
if retry.Code != http.StatusOK || xml.Unmarshal(retry.Body.Bytes(), &recovered) != nil {
|
||||
t.Fatalf("retry after metadata recovers: %d %s", retry.Code, retry.Body.String())
|
||||
}
|
||||
var keys []string
|
||||
for _, entry := range recovered.Contents {
|
||||
keys = append(keys, entry.Key)
|
||||
}
|
||||
want := []string{"a", "b", "z"}
|
||||
if continuation {
|
||||
want = []string{"b", "z"}
|
||||
}
|
||||
if !slices.Equal(keys, want) || recovered.IsTruncated {
|
||||
t.Fatalf("retry returned incomplete result: %+v", recovered)
|
||||
}
|
||||
}
|
||||
|
||||
func TestListingCachePagination(t *testing.T) {
|
||||
z, bucket, router := nullQuorumBackend(t)
|
||||
for _, key := range []string{"a", "b", "c", "d"} {
|
||||
nullQuorumPut(t, z, bucket, key, []byte(key), ObjectOptions{})
|
||||
}
|
||||
var keys []string
|
||||
var token string
|
||||
var finished bool
|
||||
for page := 0; page < 8; page++ {
|
||||
target := getListObjectsV2URL("", bucket, "", "1", "", "", "")
|
||||
if token != "" {
|
||||
target += "&continuation-token=" + url.QueryEscape(token)
|
||||
}
|
||||
rec := nullQuorumRequest(t, router, http.MethodGet, target)
|
||||
var list ListObjectsV2Response
|
||||
if rec.Code != http.StatusOK || xml.Unmarshal(rec.Body.Bytes(), &list) != nil {
|
||||
t.Fatalf("page %d: %d %s", page, rec.Code, rec.Body.String())
|
||||
}
|
||||
for _, entry := range list.Contents {
|
||||
keys = append(keys, entry.Key)
|
||||
}
|
||||
if !list.IsTruncated {
|
||||
finished = true
|
||||
break
|
||||
}
|
||||
if list.NextContinuationToken == "" || list.NextContinuationToken == token {
|
||||
t.Fatal("pagination did not advance")
|
||||
}
|
||||
token = list.NextContinuationToken
|
||||
}
|
||||
if !finished || !slices.Equal(keys, []string{"a", "b", "c", "d"}) {
|
||||
t.Fatalf("pagination returned %v, finished=%v", keys, finished)
|
||||
}
|
||||
}
|
||||
|
||||
// A cached full page must remain streamable even if a later key is slow. The
|
||||
// completion barrier belongs at EOF, not before every page response.
|
||||
func TestListingCacheFullPageDoesNotWaitForTail(t *testing.T) {
|
||||
z, bucket, router := nullQuorumBackend(t)
|
||||
disks, switchDisks := installListingDiskSwitch(z.serverPools[0].sets[0])
|
||||
for _, key := range []string{"a", "b", "c", "d", "e", "zz"} {
|
||||
nullQuorumPut(t, z, bucket, key, []byte(key), ObjectOptions{})
|
||||
}
|
||||
first := nullQuorumRequest(t, router, http.MethodGet, getListObjectsV2URL("", bucket, "", "1", "", "", ""))
|
||||
var page ListObjectsV2Response
|
||||
if first.Code != http.StatusOK || xml.Unmarshal(first.Body.Bytes(), &page) != nil || !page.IsTruncated {
|
||||
t.Fatalf("first page: %d %s", first.Code, first.Body.String())
|
||||
}
|
||||
resume := make(chan struct{})
|
||||
release := sync.OnceFunc(func() { close(resume) })
|
||||
defer release()
|
||||
var blocked, active atomic.Int64
|
||||
walkers := slices.Clone(disks)
|
||||
for i, disk := range walkers {
|
||||
walkers[i] = &listingPausedTailDisk{StorageAPI: disk, bucket: bucket, tail: "zz", resume: resume, blocked: &blocked, active: &active}
|
||||
}
|
||||
switchDisks(walkers)
|
||||
ctx, cancel := context.WithTimeout(t.Context(), 10*time.Second)
|
||||
defer cancel()
|
||||
target := getListObjectsV2URL("", bucket, "", "1", "", "", "") + "&continuation-token=" + url.QueryEscape(page.NextContinuationToken)
|
||||
req, err := newTestSignedRequestV4(http.MethodGet, target, 0, nil, globalActiveCred.AccessKey, globalActiveCred.SecretKey, nil)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
response := make(chan *httptest.ResponseRecorder, 1)
|
||||
go func() {
|
||||
rec := httptest.NewRecorder()
|
||||
router.ServeHTTP(rec, req.WithContext(ctx))
|
||||
response <- rec
|
||||
}()
|
||||
select {
|
||||
case rec := <-response:
|
||||
page = ListObjectsV2Response{}
|
||||
if rec.Code != http.StatusOK || xml.Unmarshal(rec.Body.Bytes(), &page) != nil || !page.IsTruncated || len(page.Contents) != 1 || page.Contents[0].Key != "b" || page.NextContinuationToken == "" {
|
||||
t.Fatalf("full page before tail: %d %s", rec.Code, rec.Body.String())
|
||||
}
|
||||
if blocked.Load() == 0 || active.Load() == 0 {
|
||||
t.Fatal("full page did not overlap the paused tail")
|
||||
}
|
||||
case <-ctx.Done():
|
||||
t.Fatal("full page waited for the paused tail")
|
||||
}
|
||||
release()
|
||||
waitForListingCondition(t, "disk walkers did not finish", func() bool { return active.Load() == 0 })
|
||||
target = getListObjectsV2URL("", bucket, "", "1000", "", "", "") + "&continuation-token=" + url.QueryEscape(page.NextContinuationToken)
|
||||
rec := nullQuorumRequest(t, router, http.MethodGet, target)
|
||||
page = ListObjectsV2Response{}
|
||||
if rec.Code != http.StatusOK || xml.Unmarshal(rec.Body.Bytes(), &page) != nil || page.IsTruncated {
|
||||
t.Fatalf("page after tail resumes: %d %s", rec.Code, rec.Body.String())
|
||||
}
|
||||
var keys []string
|
||||
for _, entry := range page.Contents {
|
||||
keys = append(keys, entry.Key)
|
||||
}
|
||||
if !slices.Equal(keys, []string{"c", "d", "e", "zz"}) {
|
||||
t.Fatalf("cached remainder lost keys: %v", keys)
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,133 @@
|
||||
// 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"
|
||||
"fmt"
|
||||
"net/http"
|
||||
"slices"
|
||||
"testing"
|
||||
"time"
|
||||
)
|
||||
|
||||
func TestListFallbackPreservesPoolDeleteMarker(t *testing.T) {
|
||||
for _, markerPool := range []int{0, 1} {
|
||||
t.Run(fmt.Sprint(markerPool), func(t *testing.T) {
|
||||
z, bucket, router := listingPoolsBackend(t)
|
||||
ctx := t.Context()
|
||||
live, deleted := z.serverPools[1-markerPool].sets[0], z.serverPools[markerPool].sets[0]
|
||||
disks, switchDisks := installListingDiskSwitch(deleted)
|
||||
const object = "gone"
|
||||
body := bytes.Repeat([]byte{'a'}, 8192)
|
||||
oldTime := time.Now().UTC().Add(-time.Hour)
|
||||
for _, set := range []*erasureObjects{live, deleted} {
|
||||
_, err := set.PutObject(ctx, bucket, object, mustGetPutObjReader(t, bytes.NewReader(body), int64(len(body)), "", ""), ObjectOptions{MTime: oldTime, VersionSuspended: true})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
old := make([][]byte, 3)
|
||||
for i := range old {
|
||||
old[i] = mustReadNullQuorumMeta(t, disks[i], bucket, object)
|
||||
}
|
||||
_, err := deleted.DeleteObject(ctx, bucket, object, ObjectOptions{VersionSuspended: true, DeleteMarker: true, MTime: oldTime.Add(time.Minute)})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
get := nullQuorumRequest(t, router, http.MethodGet, getGetObjectURL("", bucket, object))
|
||||
if get.Code != http.StatusNotFound {
|
||||
t.Fatalf("expected deleted GET, got %d %s", get.Code, get.Body.String())
|
||||
}
|
||||
first := nullQuorumRequest(t, router, http.MethodGet, getListObjectsV2URL("", bucket, "", "1000", "", "", ""))
|
||||
var page ListObjectsV2Response
|
||||
if first.Code != http.StatusOK || xml.Unmarshal(first.Body.Bytes(), &page) != nil || len(page.Contents) != 0 {
|
||||
t.Fatalf("normal LIST marker: %d %s", first.Code, first.Body.String())
|
||||
}
|
||||
for i := range old {
|
||||
if err := disks[i].WriteAll(ctx, bucket, object+"/"+xlStorageFormatFile, old[i]); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
walkers := slices.Clone(disks)
|
||||
for i := 8; i < len(walkers); i++ {
|
||||
walkers[i] = &failedMetadataWalkDisk{StorageAPI: disks[i]}
|
||||
}
|
||||
switchDisks(walkers)
|
||||
fi, _, _, err := deleted.getObjectFileInfo(ctx, bucket, object, ObjectOptions{}, false)
|
||||
if err != nil || !fi.Deleted {
|
||||
t.Fatalf("read fallback marker: %+v %v", fi, err)
|
||||
}
|
||||
rec := nullQuorumRequest(t, router, http.MethodGet, getListObjectsV2URL("", bucket, "", "1000", "", "", ""))
|
||||
page = ListObjectsV2Response{}
|
||||
if rec.Code != http.StatusOK || xml.Unmarshal(rec.Body.Bytes(), &page) != nil || len(page.Contents) != 0 {
|
||||
t.Fatalf("fallback resurrected old pool copy: %d %s", rec.Code, rec.Body.String())
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
// The regular metadata reader can translate a persistent generation split to
|
||||
// ObjectNotFound even though every drive still has valid metadata. LIST must
|
||||
// preserve that uncertainty instead of treating the key as deleted.
|
||||
func TestListFallbackRejectsSplitGenerations(t *testing.T) {
|
||||
z, bucket, _ := nullQuorumBackend(t)
|
||||
set := z.serverPools[0].sets[0]
|
||||
disks := set.getDisks()
|
||||
const object = "object"
|
||||
mtime := time.Now().UTC().Add(-time.Hour)
|
||||
var generations [3][][]byte
|
||||
for generation := range generations {
|
||||
body := bytes.Repeat([]byte{byte('a' + generation)}, 8192)
|
||||
nullQuorumPut(t, z, bucket, object, body, ObjectOptions{MTime: mtime.Add(time.Duration(generation) * time.Minute)})
|
||||
generations[generation] = make([][]byte, len(disks))
|
||||
for i, disk := range disks {
|
||||
generations[generation][i] = mustReadNullQuorumMeta(t, disk, bucket, object)
|
||||
}
|
||||
}
|
||||
// All sixteen drives are online, but the 7/5/4 split leaves no generation
|
||||
// with the eight matching snapshots required by the listing resolver.
|
||||
for i, disk := range disks {
|
||||
generation := 0
|
||||
if i >= 7 {
|
||||
generation = 1
|
||||
}
|
||||
if i >= 12 {
|
||||
generation = 2
|
||||
}
|
||||
if err := disk.WriteAll(t.Context(), bucket, object+"/"+xlStorageFormatFile, generations[generation][i]); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
if _, _, _, err := set.getObjectFileInfo(t.Context(), bucket, object, ObjectOptions{}, false); !isErrObjectNotFound(err) {
|
||||
t.Fatalf("split metadata did not exercise ambiguous not-found: %v", err)
|
||||
}
|
||||
entries := make(metaCacheEntries, len(disks))
|
||||
for i, disk := range disks {
|
||||
entries[i] = metaCacheEntry{name: object, metadata: mustReadNullQuorumMeta(t, disk, bucket, object)}
|
||||
if _, err := entries[i].xlmeta(); err != nil {
|
||||
t.Fatalf("drive %d no longer has valid metadata: %v", i, err)
|
||||
}
|
||||
}
|
||||
resolver := metadataResolutionParams{dirQuorum: 8, objQuorum: 8, bucket: bucket, requestedVersions: 1}
|
||||
entry, err := set.resolveListEntry(t.Context(), bucket, entries, nil, &resolver, z.listingNSLock(set))
|
||||
if entry != nil || !isErrReadQuorum(err) {
|
||||
t.Fatalf("split live metadata became absence: entry=%v err=%v", entry, err)
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,101 @@
|
||||
// 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"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"slices"
|
||||
"testing"
|
||||
"time"
|
||||
)
|
||||
|
||||
// Exercise the actual LIST pipeline with the disputed entry in pool 1.
|
||||
func TestListFallbackMultiPoolHTTPReadLock(t *testing.T) {
|
||||
for _, object := range []string{"object", "dir/"} {
|
||||
t.Run(object, func(t *testing.T) {
|
||||
z, bucket, router := listingPoolsBackend(t)
|
||||
ctx := t.Context()
|
||||
set := z.serverPools[1].sets[0]
|
||||
disks := set.getDisks()
|
||||
encoded := encodeDirObject(object)
|
||||
body := bytes.Repeat([]byte{'a'}, 8192)
|
||||
if object == "dir/" {
|
||||
body = nil
|
||||
}
|
||||
mtime := time.Now().UTC().Add(-time.Hour)
|
||||
put := func(tm time.Time) {
|
||||
_, err := set.PutObject(ctx, bucket, encoded, mustGetPutObjReader(t, bytes.NewReader(body), int64(len(body)), "", ""), ObjectOptions{MTime: tm})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
put(mtime)
|
||||
old := make([][]byte, 3)
|
||||
for i := range old {
|
||||
old[i] = mustReadNullQuorumMeta(t, disks[i], bucket, encoded)
|
||||
}
|
||||
put(mtime.Add(time.Minute))
|
||||
for i := range old {
|
||||
if err := disks[i].WriteAll(ctx, bucket, encoded+"/"+xlStorageFormatFile, old[i]); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
walkers := slices.Clone(disks)
|
||||
for i := 8; i < len(walkers); i++ {
|
||||
walkers[i] = &failedMetadataWalkDisk{StorageAPI: disks[i]}
|
||||
}
|
||||
set.getDisks = func() []StorageAPI { return walkers }
|
||||
lock := z.NewNSLock(bucket, encoded)
|
||||
lkctx, err := lock.GetLock(ctx, globalOperationTimeout)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
locked := true
|
||||
defer func() {
|
||||
if locked {
|
||||
lock.Unlock(lkctx)
|
||||
}
|
||||
}()
|
||||
target := getListObjectsV2URL("", bucket, "", "1000", "", "", "")
|
||||
req, err := newTestSignedRequestV4(http.MethodGet, target, 0, nil, globalActiveCred.AccessKey, globalActiveCred.SecretKey, nil)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
rec := httptest.NewRecorder()
|
||||
done := make(chan struct{})
|
||||
go func() {
|
||||
defer close(done)
|
||||
router.ServeHTTP(rec, req.WithContext(ctx))
|
||||
}()
|
||||
waitForListingReadLock(t, z.serverPools[0].sets[0].nsMutex, bucket, encoded, done)
|
||||
lock.Unlock(lkctx)
|
||||
locked = false
|
||||
<-done
|
||||
var list ListObjectsV2Response
|
||||
if rec.Code != http.StatusOK || xml.Unmarshal(rec.Body.Bytes(), &list) != nil {
|
||||
t.Fatalf("LIST HTTP=%d body=%s", rec.Code, rec.Body.String())
|
||||
}
|
||||
if len(list.Contents) != 1 || list.Contents[0].Key != object {
|
||||
t.Fatalf("LIST lost key: %+v", list)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,84 @@
|
||||
// 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"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/minio/minio/internal/dsync"
|
||||
)
|
||||
|
||||
// Use the real distributed-lock algorithm with in-process lock servers.
|
||||
// This is a protocol-level check, not a network or rolling-restart test.
|
||||
func TestListFallbackDistributedLockRoute(t *testing.T) {
|
||||
for _, tc := range []struct {
|
||||
name string
|
||||
multiPool, sharedLockers bool
|
||||
}{
|
||||
{"multi-pool-shared-lockers", true, true},
|
||||
{"multi-pool-disjoint-lockers", true, false},
|
||||
{"single-pool-target-set-lockers", false, false},
|
||||
} {
|
||||
t.Run(tc.name, func(t *testing.T) {
|
||||
a := []dsync.NetLocker{newLocker(), newLocker()}
|
||||
b := []dsync.NetLocker{newLocker(), newLocker()}
|
||||
if tc.sharedLockers {
|
||||
b = a
|
||||
}
|
||||
set0 := &erasureObjects{nsMutex: newNSLock(true), getLockers: func() ([]dsync.NetLocker, string) { return a, "listing-test" }}
|
||||
target := &erasureObjects{nsMutex: newNSLock(true), getLockers: func() ([]dsync.NetLocker, string) { return b, "listing-test" }}
|
||||
z := &erasureServerPools{serverPools: []*erasureSets{{sets: []*erasureObjects{set0, target}}}}
|
||||
if tc.multiPool {
|
||||
z.serverPools = []*erasureSets{{sets: []*erasureObjects{set0}}, {sets: []*erasureObjects{target}}}
|
||||
}
|
||||
writer := target.NewNSLock("bucket", "object")
|
||||
if tc.multiPool {
|
||||
writer = z.NewNSLock("bucket", "object")
|
||||
}
|
||||
wctx, err := writer.GetLock(t.Context(), globalOperationTimeout)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
locked := true
|
||||
defer func() {
|
||||
if locked {
|
||||
writer.Unlock(wctx)
|
||||
}
|
||||
}()
|
||||
reader := z.listingNSLock(target)("bucket", "object")
|
||||
ctx, cancel := context.WithTimeout(t.Context(), 80*time.Millisecond)
|
||||
defer cancel()
|
||||
rctx, err := reader.GetRLock(ctx, globalOperationTimeout)
|
||||
if err == nil {
|
||||
reader.RUnlock(rctx)
|
||||
t.Fatal("fallback read lock acquired while ordinary writer lock is held")
|
||||
}
|
||||
writer.Unlock(wctx)
|
||||
locked = false
|
||||
ctx, cancel = context.WithTimeout(t.Context(), 5*time.Second)
|
||||
defer cancel()
|
||||
rctx, err = reader.GetRLock(ctx, globalOperationTimeout)
|
||||
if err != nil {
|
||||
t.Fatalf("fallback read lock failed after writer unlock: %v", err)
|
||||
}
|
||||
reader.RUnlock(rctx)
|
||||
})
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,137 @@
|
||||
// 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"
|
||||
"context"
|
||||
"net/http"
|
||||
"testing"
|
||||
"time"
|
||||
)
|
||||
|
||||
// The fallback on pool 1 must contend with the
|
||||
// pools-layer lock used by ordinary multi-pool writers (anchored at pool 0).
|
||||
func TestListFallbackMultiPoolReadLock(t *testing.T) {
|
||||
z, bucket, _ := listingPoolsBackend(t)
|
||||
ctx := t.Context()
|
||||
set := z.serverPools[1].sets[0]
|
||||
disks := set.getDisks()
|
||||
const object = "object"
|
||||
body := bytes.Repeat([]byte{'a'}, 8192)
|
||||
mtime := time.Now().UTC().Add(-time.Hour)
|
||||
put := func(tm time.Time) {
|
||||
_, err := set.PutObject(ctx, bucket, object, mustGetPutObjReader(t, bytes.NewReader(body), int64(len(body)), "", ""), ObjectOptions{MTime: tm})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
put(mtime)
|
||||
old := make([][]byte, len(disks))
|
||||
for i := range old {
|
||||
old[i] = mustReadNullQuorumMeta(t, disks[i], bucket, object)
|
||||
}
|
||||
put(mtime.Add(time.Minute))
|
||||
entries := make(metaCacheEntries, len(disks))
|
||||
errs := make([]error, len(disks))
|
||||
for i := range 3 {
|
||||
entries[i] = metaCacheEntry{name: object, metadata: old[i]}
|
||||
}
|
||||
for i := 3; i < 8; i++ {
|
||||
entries[i] = metaCacheEntry{name: object, metadata: mustReadNullQuorumMeta(t, disks[i], bucket, object)}
|
||||
}
|
||||
for i := 8; i < len(disks); i++ {
|
||||
errs[i] = errDiskNotFound
|
||||
}
|
||||
resolver := metadataResolutionParams{dirQuorum: 8, objQuorum: 8, bucket: bucket, requestedVersions: 1}
|
||||
lock := z.NewNSLock(bucket, object)
|
||||
lkctx, err := lock.GetLock(ctx, globalOperationTimeout)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
locked := true
|
||||
defer func() {
|
||||
if locked {
|
||||
lock.Unlock(lkctx)
|
||||
}
|
||||
}()
|
||||
done := make(chan struct{})
|
||||
var entry *metaCacheEntry
|
||||
var readErr error
|
||||
go func() {
|
||||
defer close(done)
|
||||
entry, readErr = set.resolveListEntry(ctx, bucket, entries, errs, &resolver, z.listingNSLock(set))
|
||||
}()
|
||||
waitForListingReadLock(t, z.serverPools[0].sets[0].nsMutex, bucket, object, done)
|
||||
lock.Unlock(lkctx)
|
||||
locked = false
|
||||
<-done
|
||||
if readErr != nil || entry == nil || entry.name != object {
|
||||
t.Fatalf("fallback after writer unlock: entry=%v err=%v", entry, readErr)
|
||||
}
|
||||
}
|
||||
|
||||
// Observe the reader registered on the writer's actual namespace lock before
|
||||
// releasing the writer. This tests mutual exclusion without a timed sleep.
|
||||
func waitForListingReadLock(t *testing.T, mutex *nsLockMap, bucket, object string, done <-chan struct{}) {
|
||||
t.Helper()
|
||||
timer := time.NewTimer(5 * time.Second)
|
||||
defer timer.Stop()
|
||||
tick := time.NewTicker(time.Millisecond)
|
||||
defer tick.Stop()
|
||||
for {
|
||||
mutex.lockMapMutex.Lock()
|
||||
lock := mutex.lockMap[pathJoin(bucket, object)]
|
||||
waiting := lock != nil && lock.ref > 1
|
||||
mutex.lockMapMutex.Unlock()
|
||||
if waiting {
|
||||
return
|
||||
}
|
||||
select {
|
||||
case <-done:
|
||||
t.Fatal("listing completed while the writer held its namespace lock")
|
||||
case <-timer.C:
|
||||
t.Fatal("listing did not wait on the writer's namespace lock")
|
||||
case <-tick.C:
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func listingPoolsBackend(t *testing.T) (*erasureServerPools, string, http.Handler) {
|
||||
t.Helper()
|
||||
ctx, cancel := context.WithCancel(t.Context())
|
||||
obj, dirs, err := prepareErasurePoolsWithContext(ctx)
|
||||
if err != nil {
|
||||
cancel()
|
||||
t.Fatal(err)
|
||||
}
|
||||
z := obj.(*erasureServerPools)
|
||||
previous := newObjectLayerFn()
|
||||
setObjectLayer(z)
|
||||
t.Cleanup(func() {
|
||||
cancel()
|
||||
z.Shutdown(context.Background())
|
||||
removeRoots(dirs)
|
||||
setObjectLayer(previous)
|
||||
})
|
||||
bucket, router, err := initAPIHandlerTest(ctx, z, nil, MakeBucketOptions{})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
return z, bucket, router
|
||||
}
|
||||
@@ -0,0 +1,244 @@
|
||||
// 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"
|
||||
"context"
|
||||
"encoding/xml"
|
||||
"errors"
|
||||
"io"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"net/url"
|
||||
"slices"
|
||||
"sync/atomic"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/klauspost/compress/s2"
|
||||
"github.com/tinylib/msgp/msgp"
|
||||
)
|
||||
|
||||
// Replay real disk-walk metadata in its existing wire format, but pause before
|
||||
// the final key. This keeps the healthy pool's listing alive after the other
|
||||
// pool has already failed, without forging entries or listing errors.
|
||||
type listingPausedTailDisk struct {
|
||||
StorageAPI
|
||||
bucket string
|
||||
tail string
|
||||
resume <-chan struct{}
|
||||
blocked *atomic.Int64
|
||||
active *atomic.Int64
|
||||
}
|
||||
|
||||
func (d *listingPausedTailDisk) WalkDir(ctx context.Context, opts WalkDirOptions, out io.Writer) (result error) {
|
||||
if opts.Bucket != d.bucket {
|
||||
return d.StorageAPI.WalkDir(ctx, opts, out)
|
||||
}
|
||||
if d.active != nil {
|
||||
d.active.Add(1)
|
||||
defer d.active.Add(-1)
|
||||
}
|
||||
var source bytes.Buffer
|
||||
if err := d.StorageAPI.WalkDir(ctx, opts, &source); err != nil {
|
||||
return err
|
||||
}
|
||||
r := newMetacacheReader(bytes.NewReader(source.Bytes()))
|
||||
defer r.Close()
|
||||
compressed := s2.NewWriter(out)
|
||||
defer func() {
|
||||
if err := compressed.Close(); result == nil {
|
||||
result = err
|
||||
}
|
||||
}()
|
||||
w := msgp.NewWriter(compressed)
|
||||
if err := w.WriteByte(metacacheStreamVersion); err != nil {
|
||||
return err
|
||||
}
|
||||
for {
|
||||
entry, err := r.next()
|
||||
if err == io.EOF {
|
||||
if err = w.WriteBool(false); err != nil {
|
||||
return err
|
||||
}
|
||||
if err = w.Flush(); err != nil {
|
||||
return err
|
||||
}
|
||||
return nil
|
||||
}
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if entry.name == d.tail {
|
||||
if err = w.Flush(); err != nil {
|
||||
return err
|
||||
}
|
||||
if err = compressed.Flush(); err != nil {
|
||||
return err
|
||||
}
|
||||
d.blocked.Add(1)
|
||||
select {
|
||||
case <-d.resume:
|
||||
case <-ctx.Done():
|
||||
return ctx.Err()
|
||||
}
|
||||
}
|
||||
if err = w.WriteBool(true); err != nil {
|
||||
return err
|
||||
}
|
||||
if err = w.WriteString(entry.name); err != nil {
|
||||
return err
|
||||
}
|
||||
if err = w.WriteBytes(entry.metadata); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestListingMergeDoesNotSkipFailedPool(t *testing.T) {
|
||||
ctx, cancel := context.WithTimeout(t.Context(), 20*time.Second)
|
||||
defer cancel()
|
||||
z, bucket, router := listingPoolsBackend(t)
|
||||
broken, healthy := z.serverPools[0].sets[0], z.serverPools[1].sets[0]
|
||||
brokenDisks, switchBrokenDisks := installListingDiskSwitch(broken)
|
||||
healthyDisks, switchHealthyDisks := installListingDiskSwitch(healthy)
|
||||
body := bytes.Repeat([]byte{'a'}, 8192)
|
||||
mtime := time.Now().UTC().Add(-time.Hour)
|
||||
put := func(set *erasureObjects, name string, tm time.Time) {
|
||||
_, err := set.PutObject(ctx, bucket, name, mustGetPutObjReader(t, bytes.NewReader(body), int64(len(body)), "", ""), ObjectOptions{MTime: tm})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
put(broken, "a", mtime)
|
||||
put(broken, "e", mtime)
|
||||
for _, name := range []string{"f", "g", "h", "i", "j", "k", "zz"} {
|
||||
put(healthy, name, mtime)
|
||||
}
|
||||
first := nullQuorumRequest(t, router, http.MethodGet, getListObjectsV2URL("", bucket, "", "1", "", "", ""))
|
||||
var page ListObjectsV2Response
|
||||
if first.Code != http.StatusOK || xml.Unmarshal(first.Body.Bytes(), &page) != nil || !page.IsTruncated || page.NextContinuationToken == "" || len(page.Contents) != 1 || page.Contents[0].Key != "a" {
|
||||
t.Fatalf("first page HTTP=%d body=%s", first.Code, first.Body.String())
|
||||
}
|
||||
firstToken := page.NextContinuationToken
|
||||
old := make([][]byte, 3)
|
||||
for i := range old {
|
||||
old[i] = mustReadNullQuorumMeta(t, brokenDisks[i], bucket, "e")
|
||||
}
|
||||
put(broken, "e", mtime.Add(time.Minute))
|
||||
for i := range old {
|
||||
if err := brokenDisks[i].WriteAll(ctx, bucket, "e/"+xlStorageFormatFile, old[i]); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
var reads, blocked, active atomic.Int64
|
||||
brokenWalkers := slices.Clone(brokenDisks)
|
||||
for i := range brokenWalkers {
|
||||
brokenWalkers[i] = &listingReadFailDisk{StorageAPI: brokenDisks[i], object: "e", reads: &reads}
|
||||
if i >= 8 {
|
||||
brokenWalkers[i] = &failedMetadataWalkDisk{StorageAPI: brokenWalkers[i]}
|
||||
}
|
||||
}
|
||||
switchBrokenDisks(brokenWalkers)
|
||||
resume := make(chan struct{})
|
||||
defer close(resume)
|
||||
healthyWalkers := slices.Clone(healthyDisks)
|
||||
for i, disk := range healthyWalkers {
|
||||
healthyWalkers[i] = &listingPausedTailDisk{StorageAPI: disk, bucket: bucket, tail: "zz", resume: resume, blocked: &blocked, active: &active}
|
||||
}
|
||||
switchHealthyDisks(healthyWalkers)
|
||||
|
||||
target := getListObjectsV2URL("", bucket, "", "2", "", "", "") + "&continuation-token=" + url.QueryEscape(firstToken)
|
||||
req, err := newTestSignedRequestV4(http.MethodGet, target, 0, nil, globalActiveCred.AccessKey, globalActiveCred.SecretKey, nil)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
response := make(chan *httptest.ResponseRecorder, 1)
|
||||
go func() {
|
||||
rec := httptest.NewRecorder()
|
||||
router.ServeHTTP(rec, req.WithContext(ctx))
|
||||
response <- rec
|
||||
}()
|
||||
select {
|
||||
case rec := <-response:
|
||||
t.Logf("continuation LIST=%d failed fallback ReadXL=%d paused healthy walkers=%d body=%s", rec.Code, reads.Load(), blocked.Load(), rec.Body.String())
|
||||
if reads.Load() == 0 || blocked.Load() == 0 {
|
||||
t.Fatal("schedule did not exercise fallback failure and unfinished healthy pool")
|
||||
}
|
||||
var apiErr APIErrorResponse
|
||||
if rec.Code != http.StatusServiceUnavailable || xml.Unmarshal(rec.Body.Bytes(), &apiErr) != nil || apiErr.Code != "SlowDownRead" {
|
||||
t.Fatalf("failed pool advanced the page instead of reporting read quorum: %d %s", rec.Code, rec.Body.String())
|
||||
}
|
||||
waitForListingCondition(t, "canceled disk walkers did not exit", func() bool { return active.Load() == 0 })
|
||||
case <-ctx.Done():
|
||||
t.Fatalf("no early response: %v reads=%d blocked=%d", ctx.Err(), reads.Load(), blocked.Load())
|
||||
}
|
||||
}
|
||||
|
||||
func TestListingMergeCancellation(t *testing.T) {
|
||||
z, bucket, _ := nullQuorumBackend(t)
|
||||
for _, key := range []string{"a", "b", "zz"} {
|
||||
nullQuorumPut(t, z, bucket, key, []byte(key), ObjectOptions{})
|
||||
}
|
||||
set := z.serverPools[0].sets[0]
|
||||
var blocked, active atomic.Int64
|
||||
resume := make(chan struct{})
|
||||
defer close(resume)
|
||||
walkers := slices.Clone(set.getDisks())
|
||||
for i, disk := range walkers {
|
||||
walkers[i] = &listingPausedTailDisk{StorageAPI: disk, bucket: bucket, tail: "zz", resume: resume, blocked: &blocked, active: &active}
|
||||
}
|
||||
set.getDisks = func() []StorageAPI { return walkers }
|
||||
ctx, cancel := context.WithCancel(t.Context())
|
||||
defer cancel()
|
||||
entries := make(chan metaCacheEntry, 10)
|
||||
done := make(chan error, 1)
|
||||
go func() {
|
||||
done <- z.listMerged(ctx, listPathOptions{Bucket: bucket, Recursive: true, Limit: 10}, entries)
|
||||
}()
|
||||
waitForListingCondition(t, "disk walks did not reach the paused tail", func() bool { return blocked.Load() > 0 })
|
||||
cancel()
|
||||
select {
|
||||
case err := <-done:
|
||||
if !errors.Is(err, context.Canceled) {
|
||||
t.Fatalf("canceled merge returned %v", err)
|
||||
}
|
||||
case <-time.After(5 * time.Second):
|
||||
t.Fatal("canceled merge did not finish")
|
||||
}
|
||||
// listMerged must close its output and release every paused disk walk.
|
||||
for range entries {
|
||||
}
|
||||
waitForListingCondition(t, "canceled disk walkers did not exit", func() bool { return active.Load() == 0 })
|
||||
}
|
||||
|
||||
func waitForListingCondition(t *testing.T, failure string, ready func() bool) {
|
||||
t.Helper()
|
||||
timer := time.NewTimer(5 * time.Second)
|
||||
defer timer.Stop()
|
||||
tick := time.NewTicker(time.Millisecond)
|
||||
defer tick.Stop()
|
||||
for !ready() {
|
||||
select {
|
||||
case <-timer.C:
|
||||
t.Fatal(failure)
|
||||
case <-tick.C:
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -52,6 +52,37 @@ type nullQuorumWalkDisk struct {
|
||||
index int
|
||||
}
|
||||
|
||||
type failedMetadataWalkDisk struct {
|
||||
StorageAPI
|
||||
}
|
||||
|
||||
func (d *failedMetadataWalkDisk) WalkDir(context.Context, WalkDirOptions, io.Writer) error {
|
||||
return errDiskNotFound
|
||||
}
|
||||
|
||||
// offlineNullQuorumDisk models a drive of a node that is away.
|
||||
type offlineNullQuorumDisk struct {
|
||||
StorageAPI
|
||||
}
|
||||
|
||||
func (d *offlineNullQuorumDisk) IsOnline() bool { return false }
|
||||
|
||||
func (d *offlineNullQuorumDisk) DiskInfo(context.Context, DiskInfoOptions) (DiskInfo, error) {
|
||||
return DiskInfo{}, errDiskNotFound
|
||||
}
|
||||
|
||||
func (d *offlineNullQuorumDisk) WalkDir(context.Context, WalkDirOptions, io.Writer) error {
|
||||
return errDiskNotFound
|
||||
}
|
||||
|
||||
func (d *offlineNullQuorumDisk) ReadXL(context.Context, string, string, bool) (RawFileInfo, error) {
|
||||
return RawFileInfo{}, errDiskNotFound
|
||||
}
|
||||
|
||||
func (d *offlineNullQuorumDisk) ReadVersion(context.Context, string, string, string, string, ReadOptions) (FileInfo, error) {
|
||||
return FileInfo{}, errDiskNotFound
|
||||
}
|
||||
|
||||
func (d *nullQuorumWalkDisk) WalkDir(ctx context.Context, opts WalkDirOptions, out io.Writer) error {
|
||||
if opts.Bucket != d.schedule.bucket {
|
||||
return d.StorageAPI.WalkDir(ctx, opts, out)
|
||||
@@ -119,6 +150,30 @@ func nullQuorumRequest(t *testing.T, router http.Handler, method, target string)
|
||||
return rec
|
||||
}
|
||||
|
||||
func nullQuorumPut(t *testing.T, z *erasureServerPools, bucket, object string, body []byte, opts ObjectOptions) {
|
||||
t.Helper()
|
||||
if _, err := z.PutObject(t.Context(), bucket, object, mustGetPutObjReader(t, bytes.NewReader(body), int64(len(body)), "", ""), opts); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
|
||||
func nullQuorumListV2Keys(t *testing.T, router http.Handler, bucket string) (int, []string) {
|
||||
t.Helper()
|
||||
rec := nullQuorumRequest(t, router, http.MethodGet, getListObjectsV2URL("", bucket, "", "1000", "", "", ""))
|
||||
if rec.Code != http.StatusOK {
|
||||
return rec.Code, nil
|
||||
}
|
||||
var list ListObjectsV2Response
|
||||
if err := xml.Unmarshal(rec.Body.Bytes(), &list); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
keys := make([]string, 0, len(list.Contents))
|
||||
for _, object := range list.Contents {
|
||||
keys = append(keys, object.Key)
|
||||
}
|
||||
return rec.Code, keys
|
||||
}
|
||||
|
||||
func TestListObjectsSingleNullQuorumHTTP(t *testing.T) {
|
||||
z, bucket, router := nullQuorumBackend(t)
|
||||
oldBody := bytes.Repeat([]byte{'a'}, 8192)
|
||||
@@ -227,6 +282,357 @@ func TestListObjectsSingleNullQuorumHTTP(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
func TestListObjectsFallsBackToReadableMetadata(t *testing.T) {
|
||||
z, bucket, router := nullQuorumBackend(t)
|
||||
set := z.serverPools[0].sets[0]
|
||||
disks := set.getDisks()
|
||||
const object = "object"
|
||||
oldBody := bytes.Repeat([]byte{'a'}, 8192)
|
||||
newBody := bytes.Repeat([]byte{'b'}, 8192)
|
||||
oldTime := time.Now().UTC().Add(-time.Hour)
|
||||
|
||||
if _, err := z.PutObject(t.Context(), bucket, object, mustGetPutObjReader(t, bytes.NewReader(oldBody), int64(len(oldBody)), "", ""), ObjectOptions{MTime: oldTime}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
oldMeta := make([][]byte, len(disks))
|
||||
for i, disk := range disks {
|
||||
oldMeta[i] = mustReadNullQuorumMeta(t, disk, bucket, object)
|
||||
}
|
||||
if _, err := z.PutObject(t.Context(), bucket, object, mustGetPutObjReader(t, bytes.NewReader(newBody), int64(len(newBody)), "", ""), ObjectOptions{MTime: oldTime.Add(time.Minute)}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
// The eight readable walk streams are split 3/5 across two generations;
|
||||
// the other eight streams fail. Neither generation is proven readable by
|
||||
// the listing inputs, so returning HTTP 200 would silently omit the key.
|
||||
for i := range 3 {
|
||||
if err := disks[i].WriteAll(t.Context(), bucket, object+"/"+xlStorageFormatFile, oldMeta[i]); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
wrapped := make([]StorageAPI, len(disks))
|
||||
for i, disk := range disks {
|
||||
wrapped[i] = disk
|
||||
if i >= 8 {
|
||||
wrapped[i] = &failedMetadataWalkDisk{StorageAPI: disk}
|
||||
}
|
||||
}
|
||||
set.getDisks = func() []StorageAPI { return wrapped }
|
||||
|
||||
rec := nullQuorumRequest(t, router, http.MethodGet, getListObjectsV2URL("", bucket, "", "1000", "", "", ""))
|
||||
var list ListObjectsV2Response
|
||||
if rec.Code != http.StatusOK || xml.Unmarshal(rec.Body.Bytes(), &list) != nil {
|
||||
t.Fatalf("LIST failed to resolve readable metadata: %d %s", rec.Code, rec.Body.String())
|
||||
}
|
||||
if list.KeyCount != 1 || len(list.Contents) != 1 || list.Contents[0].Key != object {
|
||||
t.Fatalf("LIST omitted the readable key: %+v", list)
|
||||
}
|
||||
|
||||
// The object itself still has a read quorum outside the failed walk
|
||||
// streams, demonstrating why the LIST must report uncertainty, not absence.
|
||||
get := nullQuorumRequest(t, router, http.MethodGet, getGetObjectURL("", bucket, object))
|
||||
if get.Code != http.StatusOK || !bytes.Equal(get.Body.Bytes(), newBody) {
|
||||
t.Fatalf("GET lost the readable generation: %d %q", get.Code, get.Body.String())
|
||||
}
|
||||
}
|
||||
|
||||
// The fallback must read under the object read lock. A lock-free read of an
|
||||
// object being overwritten stably reports a live object as absent, which is
|
||||
// the omission this repair exists to remove. Hold the write lock and assert
|
||||
// the listing waits for it rather than resolving straight through.
|
||||
func TestListFallbackTakesObjectReadLock(t *testing.T) {
|
||||
z, bucket, router := nullQuorumBackend(t)
|
||||
set := z.serverPools[0].sets[0]
|
||||
disks := set.getDisks()
|
||||
const object = "object"
|
||||
oldBody, newBody := bytes.Repeat([]byte{'a'}, 8192), bytes.Repeat([]byte{'b'}, 8192)
|
||||
oldTime := time.Now().UTC().Add(-time.Hour)
|
||||
|
||||
if _, err := z.PutObject(t.Context(), bucket, object, mustGetPutObjReader(t, bytes.NewReader(oldBody), int64(len(oldBody)), "", ""), ObjectOptions{MTime: oldTime}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
oldMeta := make([][]byte, len(disks))
|
||||
for i, disk := range disks {
|
||||
oldMeta[i] = mustReadNullQuorumMeta(t, disk, bucket, object)
|
||||
}
|
||||
if _, err := z.PutObject(t.Context(), bucket, object, mustGetPutObjReader(t, bytes.NewReader(newBody), int64(len(newBody)), "", ""), ObjectOptions{MTime: oldTime.Add(time.Minute)}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
// Split the readable walk streams so the entry cannot be resolved from
|
||||
// the snapshots and the fallback has to run.
|
||||
for i := range 3 {
|
||||
if err := disks[i].WriteAll(t.Context(), bucket, object+"/"+xlStorageFormatFile, oldMeta[i]); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
wrapped := make([]StorageAPI, len(disks))
|
||||
for i, disk := range disks {
|
||||
wrapped[i] = disk
|
||||
if i >= 8 {
|
||||
wrapped[i] = &failedMetadataWalkDisk{StorageAPI: disk}
|
||||
}
|
||||
}
|
||||
set.getDisks = func() []StorageAPI { return wrapped }
|
||||
|
||||
lock := set.NewNSLock(bucket, object)
|
||||
lkctx, err := lock.GetLock(t.Context(), globalOperationTimeout)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
locked := true
|
||||
defer func() {
|
||||
if locked {
|
||||
lock.Unlock(lkctx)
|
||||
}
|
||||
}()
|
||||
target := getListObjectsV2URL("", bucket, "", "1000", "", "", "")
|
||||
req, err := newTestSignedRequestV4(http.MethodGet, target, 0, nil, globalActiveCred.AccessKey, globalActiveCred.SecretKey, nil)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
rec := httptest.NewRecorder()
|
||||
done := make(chan struct{})
|
||||
go func() {
|
||||
defer close(done)
|
||||
router.ServeHTTP(rec, req)
|
||||
}()
|
||||
waitForListingReadLock(t, set.nsMutex, bucket, object, done)
|
||||
lock.Unlock(lkctx)
|
||||
locked = false
|
||||
<-done
|
||||
|
||||
var list ListObjectsV2Response
|
||||
if rec.Code != http.StatusOK || xml.Unmarshal(rec.Body.Bytes(), &list) != nil {
|
||||
t.Fatalf("LIST failed after the lock was released: %d %s", rec.Code, rec.Body.String())
|
||||
}
|
||||
if list.KeyCount != 1 || len(list.Contents) != 1 || list.Contents[0].Key != object {
|
||||
t.Fatalf("LIST omitted the readable key: %+v", list)
|
||||
}
|
||||
}
|
||||
|
||||
// A truncated xl.meta is what a concurrent overwrite looks like on one drive,
|
||||
// and the walker skips that entry. Skipping is safe and deliberate: the key is
|
||||
// only omitted when no generation reaches quorum, which listPath now resolves
|
||||
// against the read path. Pin that, so the walker is not "hardened" into
|
||||
// failing a whole listing over a minority of unreadable drives.
|
||||
func TestListObjectsSurvivesTruncatedMetadataMinority(t *testing.T) {
|
||||
z, bucket, router := nullQuorumBackend(t)
|
||||
set := z.serverPools[0].sets[0]
|
||||
disks := set.getDisks()
|
||||
const object = "object"
|
||||
body := bytes.Repeat([]byte{'a'}, 8192)
|
||||
|
||||
if _, err := z.PutObject(t.Context(), bucket, object, mustGetPutObjReader(t, bytes.NewReader(body), int64(len(body)), "", ""), ObjectOptions{}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
// Truncate xl.meta on a minority of drives, as an overwrite in flight does.
|
||||
for i := range 4 {
|
||||
meta := mustReadNullQuorumMeta(t, disks[i], bucket, object)
|
||||
if err := disks[i].WriteAll(t.Context(), bucket, object+"/"+xlStorageFormatFile, meta[:len(meta)/2]); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
|
||||
rec := nullQuorumRequest(t, router, http.MethodGet, getListObjectsV2URL("", bucket, "", "1000", "", "", ""))
|
||||
var list ListObjectsV2Response
|
||||
if rec.Code != http.StatusOK || xml.Unmarshal(rec.Body.Bytes(), &list) != nil {
|
||||
t.Fatalf("a truncated minority failed the listing: %d %s", rec.Code, rec.Body.String())
|
||||
}
|
||||
if list.KeyCount != 1 || len(list.Contents) != 1 || list.Contents[0].Key != object {
|
||||
t.Fatalf("a truncated minority omitted the key: %+v", list)
|
||||
}
|
||||
}
|
||||
|
||||
func TestListStaleMinorityIgnored(t *testing.T) {
|
||||
z, bucket, router := nullQuorumBackend(t)
|
||||
disks := z.serverPools[0].sets[0].getDisks()
|
||||
body := bytes.Repeat([]byte{'a'}, 8192)
|
||||
nullQuorumPut(t, z, bucket, "gone", body, ObjectOptions{})
|
||||
nullQuorumPut(t, z, bucket, "keep", body, ObjectOptions{})
|
||||
for _, disk := range disks[4:] {
|
||||
if err := disk.Delete(t.Context(), bucket, "gone", DeleteOptions{Recursive: true, Immediate: true}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
|
||||
if code, keys := nullQuorumListV2Keys(t, router, bucket); code != http.StatusOK || !slices.Equal(keys, []string{"keep"}) {
|
||||
t.Errorf("ListObjectsV2: got %d %v; want 200 [keep]", code, keys)
|
||||
}
|
||||
results := make(chan itemOrErr[ObjectInfo], 16)
|
||||
if err := z.Walk(t.Context(), bucket, "", results, WalkOptions{}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
var names []string
|
||||
for result := range results {
|
||||
if result.Err != nil {
|
||||
t.Fatal(result.Err)
|
||||
}
|
||||
names = append(names, result.Item.Name)
|
||||
}
|
||||
if !slices.Equal(names, []string{"keep"}) {
|
||||
t.Errorf("Walk: got %v; want [keep]", names)
|
||||
}
|
||||
|
||||
rec := nullQuorumRequest(t, router, http.MethodGet, getListObjectVersionsURL("", bucket, "", "1000", ""))
|
||||
var list struct {
|
||||
Versions []struct {
|
||||
Key string `xml:"Key"`
|
||||
} `xml:"Version"`
|
||||
}
|
||||
if rec.Code != http.StatusOK {
|
||||
t.Errorf("ListObjectVersions: got %d; want 200", rec.Code)
|
||||
} else if err := xml.Unmarshal(rec.Body.Bytes(), &list); err != nil {
|
||||
t.Fatal(err)
|
||||
} else if len(list.Versions) != 1 || list.Versions[0].Key != "keep" {
|
||||
t.Errorf("ListObjectVersions: got %+v; want [keep]", list.Versions)
|
||||
}
|
||||
|
||||
if _, err := z.DeleteObject(t.Context(), bucket, "keep", ObjectOptions{}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := z.DeleteBucket(t.Context(), bucket, DeleteBucketOptions{}); err != nil {
|
||||
t.Errorf("DeleteBucket with only a stale minority: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestIAMListingIgnoresStaleMinority(t *testing.T) {
|
||||
z, _, _ := nullQuorumBackend(t)
|
||||
body := []byte(`{"version":1}`)
|
||||
alive := iamConfigUsersPrefix + "alive/identity.json"
|
||||
ghost := iamConfigUsersPrefix + "ghost/identity.json"
|
||||
nullQuorumPut(t, z, minioMetaBucket, alive, body, ObjectOptions{})
|
||||
nullQuorumPut(t, z, minioMetaBucket, ghost, body, ObjectOptions{})
|
||||
for _, disk := range z.serverPools[0].getHashedSet(ghost).getDisks()[4:] {
|
||||
if err := disk.Delete(t.Context(), minioMetaBucket, ghost, DeleteOptions{Recursive: true, Immediate: true}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
|
||||
items, err := newIAMObjectStore(z, MinIOUsersSysType).listAllIAMConfigItems(t.Context())
|
||||
if err != nil {
|
||||
t.Fatalf("IAM listing failed on a stale minority (retriable=%v): %v", configRetriableErrors(err), err)
|
||||
}
|
||||
if got := items["users/"]; !slices.Equal(got, []string{"alive/identity.json"}) {
|
||||
t.Errorf("IAM users: got %v; want [alive/identity.json]", got)
|
||||
}
|
||||
}
|
||||
|
||||
func TestListSplitDirectoryObject(t *testing.T) {
|
||||
z, bucket, router := nullQuorumBackend(t)
|
||||
set := z.serverPools[0].sets[0]
|
||||
disks := set.getDisks()
|
||||
const object = "dir/"
|
||||
encoded := encodeDirObject(object)
|
||||
mtime := time.Now().UTC().Add(-time.Hour)
|
||||
|
||||
nullQuorumPut(t, z, bucket, object, nil, ObjectOptions{MTime: mtime})
|
||||
old := make([][]byte, 3)
|
||||
for i := range old {
|
||||
old[i] = mustReadNullQuorumMeta(t, disks[i], bucket, encoded)
|
||||
}
|
||||
nullQuorumPut(t, z, bucket, object, nil, ObjectOptions{MTime: mtime.Add(time.Minute)})
|
||||
for i := range old {
|
||||
if err := disks[i].WriteAll(t.Context(), bucket, encoded+"/"+xlStorageFormatFile, old[i]); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
|
||||
walkers := slices.Clone(disks)
|
||||
for i := 8; i < len(walkers); i++ {
|
||||
walkers[i] = &failedMetadataWalkDisk{StorageAPI: disks[i]}
|
||||
}
|
||||
set.getDisks = func() []StorageAPI { return walkers }
|
||||
if code, keys := nullQuorumListV2Keys(t, router, bucket); code != http.StatusOK || !slices.Equal(keys, []string{object}) {
|
||||
t.Errorf("ListObjectsV2: got %d %v; want 200 [%s]", code, keys, object)
|
||||
}
|
||||
}
|
||||
|
||||
func TestListFallbackDoesNotTrustNotFound(t *testing.T) {
|
||||
z, bucket, _ := nullQuorumBackend(t)
|
||||
set := z.serverPools[0].sets[0]
|
||||
disks := set.getDisks()
|
||||
const object = "object"
|
||||
body := bytes.Repeat([]byte{'a'}, 8192)
|
||||
mtime := time.Now().UTC().Add(-time.Hour)
|
||||
|
||||
nullQuorumPut(t, z, bucket, object, body, ObjectOptions{MTime: mtime})
|
||||
old := make([][]byte, len(disks))
|
||||
for i := range disks {
|
||||
old[i] = mustReadNullQuorumMeta(t, disks[i], bucket, object)
|
||||
}
|
||||
nullQuorumPut(t, z, bucket, object, body, ObjectOptions{MTime: mtime.Add(time.Minute)})
|
||||
newer := make([][]byte, len(disks))
|
||||
for i := range disks {
|
||||
newer[i] = mustReadNullQuorumMeta(t, disks[i], bucket, object)
|
||||
if err := disks[i].Delete(t.Context(), bucket, object, DeleteOptions{Recursive: true, Immediate: true}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
|
||||
entries := make(metaCacheEntries, len(disks))
|
||||
errs := make([]error, len(disks))
|
||||
for i := range 3 {
|
||||
entries[i] = metaCacheEntry{name: object, metadata: old[i]}
|
||||
}
|
||||
for i := 3; i < 8; i++ {
|
||||
entries[i] = metaCacheEntry{name: object, metadata: newer[i]}
|
||||
}
|
||||
for i := 8; i < len(errs); i++ {
|
||||
errs[i] = errDiskNotFound
|
||||
}
|
||||
resolver := metadataResolutionParams{dirQuorum: 8, objQuorum: 8, bucket: bucket, requestedVersions: 1}
|
||||
entry, err := set.resolveListEntry(t.Context(), bucket, entries, errs, &resolver, z.listingNSLock(set))
|
||||
if entry != nil || !isErrReadQuorum(err) {
|
||||
t.Fatalf("ambiguous not-found: got entry=%v err=%v; want read quorum error", entry, err)
|
||||
}
|
||||
}
|
||||
|
||||
// G0 was written 8+8 while a node was away. G1 was later written 12+4 but
|
||||
// missed drive 0, which still holds G0. Then a node holding four G1 drives
|
||||
// goes away: eleven G1 and one G0 remain online. G1 exists but lacks one of
|
||||
// its twelve data shards until the node returns, so HEAD must not report 404
|
||||
// and LIST must still show the key.
|
||||
func TestListCrossParityMinorityStaysVisible(t *testing.T) {
|
||||
z, bucket, router := nullQuorumBackend(t)
|
||||
set := z.serverPools[0].sets[0]
|
||||
disks, switchDisks := installListingDiskSwitch(set)
|
||||
const object = "object"
|
||||
mtime := time.Now().UTC().Add(-time.Hour)
|
||||
|
||||
nullQuorumPut(t, z, bucket, object, bytes.Repeat([]byte{'a'}, 256<<10), ObjectOptions{MTime: mtime, MaxParity: true})
|
||||
g0 := mustReadNullQuorumMeta(t, disks[0], bucket, object)
|
||||
var x xlMetaV2
|
||||
if err := x.LoadOrConvert(g0); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if h := x.versions[0].header; h.EcM != 8 || h.EcN != 8 {
|
||||
t.Fatalf("G0 is %d+%d, want 8+8", h.EcM, h.EcN)
|
||||
}
|
||||
g1Body := bytes.Repeat([]byte{'b'}, 256<<10)
|
||||
nullQuorumPut(t, z, bucket, object, g1Body, ObjectOptions{MTime: mtime.Add(time.Minute)})
|
||||
if err := disks[0].WriteAll(t.Context(), bucket, object+"/"+xlStorageFormatFile, g0); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
online := slices.Clone(disks)
|
||||
for i := 12; i < len(online); i++ {
|
||||
online[i] = &offlineNullQuorumDisk{StorageAPI: disks[i]}
|
||||
}
|
||||
switchDisks(online)
|
||||
if code := nullQuorumRequest(t, router, http.MethodHead, getHeadObjectURL("", bucket, object)).Code; code == http.StatusNotFound {
|
||||
t.Errorf("HEAD with a node away: 404 for an existing object")
|
||||
}
|
||||
if code, keys := nullQuorumListV2Keys(t, router, bucket); code != http.StatusOK || !slices.Equal(keys, []string{object}) {
|
||||
t.Errorf("ListObjectsV2 with a node away: got %d %v; want 200 [%s]", code, keys, object)
|
||||
}
|
||||
|
||||
switchDisks(disks)
|
||||
if get := nullQuorumRequest(t, router, http.MethodGet, getGetObjectURL("", bucket, object)); get.Code != http.StatusOK || !bytes.Equal(get.Body.Bytes(), g1Body) {
|
||||
t.Fatalf("GET after the node returns: %d", get.Code)
|
||||
}
|
||||
}
|
||||
|
||||
func TestSingleNullQuorumReadAndConsumers(t *testing.T) {
|
||||
z, bucket, router := nullQuorumBackend(t)
|
||||
set := z.serverPools[0].sets[0]
|
||||
|
||||
@@ -191,19 +191,19 @@ func (z *erasureServerPools) listPath(ctx context.Context, o *listPathOptions) (
|
||||
return entries, err
|
||||
}
|
||||
entries.truncate(0)
|
||||
go func() {
|
||||
rpc := globalNotificationSys.restClientFromHash(pathJoin(o.Bucket, o.Prefix))
|
||||
rpc := globalNotificationSys.restClientFromHash(pathJoin(o.Bucket, o.Prefix))
|
||||
go func(o listPathOptions) {
|
||||
if rpc != nil {
|
||||
ctx, cancel := context.WithTimeout(GlobalContext, 5*time.Second)
|
||||
defer cancel()
|
||||
c, err := rpc.GetMetacacheListing(ctx, *o)
|
||||
c, err := rpc.GetMetacacheListing(ctx, o)
|
||||
if err == nil {
|
||||
c.error = "no longer used"
|
||||
c.status = scanStateError
|
||||
rpc.UpdateMetacacheListing(ctx, *c)
|
||||
}
|
||||
}
|
||||
}()
|
||||
}(*o)
|
||||
o.ID = ""
|
||||
}
|
||||
|
||||
@@ -264,6 +264,16 @@ func (z *erasureServerPools) listPathShortcutEOF(ctx context.Context, bucket str
|
||||
return io.EOF
|
||||
}
|
||||
|
||||
// listingNSLock chooses the namespace used by regular object writes.
|
||||
// Multiple pools serialize at the pools layer; single-pool PUT and GET use
|
||||
// the destination set. This keeps the existing single-pool lock routing.
|
||||
func (z *erasureServerPools) listingNSLock(er *erasureObjects) func(string, ...string) RWLocker {
|
||||
if z.SinglePool() {
|
||||
return er.NewNSLock
|
||||
}
|
||||
return z.NewNSLock
|
||||
}
|
||||
|
||||
// listMerged will list across all sets and return a merged results stream.
|
||||
// The result channel is closed when no more results are expected.
|
||||
func (z *erasureServerPools) listMerged(ctx context.Context, o listPathOptions, results chan<- metaCacheEntry) error {
|
||||
@@ -283,13 +293,16 @@ func (z *erasureServerPools) listMerged(ctx context.Context, o listPathOptions,
|
||||
inputs = append(inputs, innerResults)
|
||||
go func(i int, set *erasureObjects) {
|
||||
defer wg.Done()
|
||||
err := set.listPath(listCtx, o, innerResults)
|
||||
err := set.listPath(listCtx, o, innerResults, z.listingNSLock(set))
|
||||
mu.Lock()
|
||||
defer mu.Unlock()
|
||||
if err == nil {
|
||||
allAtEOF = false
|
||||
}
|
||||
errs[i] = err
|
||||
mu.Unlock()
|
||||
// The merger must observe the terminal error before treating
|
||||
// this stream as exhausted and advancing to another pool.
|
||||
xioutil.SafeClose(innerResults)
|
||||
}(len(errs), set)
|
||||
errs = append(errs, nil)
|
||||
}
|
||||
@@ -298,7 +311,16 @@ func (z *erasureServerPools) listMerged(ctx context.Context, o listPathOptions,
|
||||
|
||||
// Gather results to a single channel.
|
||||
// Quorum is one since we are merging across sets.
|
||||
err := mergeEntryChannels(ctx, inputs, results, 1)
|
||||
err := mergeEntryChannelsWithErrors(ctx, inputs, results, 1, func(i int) error {
|
||||
mu.Lock()
|
||||
err := errs[i]
|
||||
mu.Unlock()
|
||||
// Keep the existing aggregate handling of absent paths and buckets.
|
||||
if errors.Is(err, io.EOF) || isAllNotFound([]error{err}) {
|
||||
return nil
|
||||
}
|
||||
return err
|
||||
})
|
||||
|
||||
cancelList()
|
||||
wg.Wait()
|
||||
@@ -415,13 +437,18 @@ func (z *erasureServerPools) listAndSave(ctx context.Context, o *listPathOptions
|
||||
cancel()
|
||||
}()
|
||||
|
||||
// Do listing...
|
||||
// Publish the typed terminal result before either consumer observes EOF.
|
||||
// mergeEntryChannels closes inCh before listMerged has inspected set errors.
|
||||
listingDone := make(chan struct{})
|
||||
var listingErr error
|
||||
go func(o listPathOptions) {
|
||||
err := z.listMerged(listCtx, o, inCh)
|
||||
if err != nil {
|
||||
meta.setErr(err.Error())
|
||||
listingErr = z.listMerged(listCtx, o, inCh)
|
||||
if listingErr != nil && !errors.Is(listingErr, io.EOF) {
|
||||
meta.setErr(listingErr.Error())
|
||||
cancel()
|
||||
}
|
||||
o.debugln("listAndSave: listing", o.ID, "finished with ", err)
|
||||
close(listingDone)
|
||||
o.debugln("listAndSave: listing", o.ID, "finished with ", listingErr)
|
||||
}(*o)
|
||||
|
||||
// Keep track of when we return since we no longer have to send entries to output.
|
||||
@@ -451,6 +478,9 @@ func (z *erasureServerPools) listAndSave(ctx context.Context, o *listPathOptions
|
||||
entry.reusable = returned
|
||||
saveCh <- entry
|
||||
}
|
||||
// Do not publish an end-of-stream cache block or a final response
|
||||
// before the scan has recorded its terminal error.
|
||||
<-listingDone
|
||||
if !returned {
|
||||
xioutil.SafeClose(outCh)
|
||||
}
|
||||
@@ -458,9 +488,19 @@ func (z *erasureServerPools) listAndSave(ctx context.Context, o *listPathOptions
|
||||
}()
|
||||
|
||||
entries, err = filteredResults()
|
||||
if err == nil {
|
||||
// Check if listing recorded an error.
|
||||
err = meta.getErr()
|
||||
if err == nil || errors.Is(err, io.EOF) {
|
||||
// Full pages may return before the scan ends. A final page has
|
||||
// crossed the completion barrier above and must preserve scan errors.
|
||||
select {
|
||||
case <-listingDone:
|
||||
if listingErr != nil && !errors.Is(listingErr, io.EOF) {
|
||||
return entries, listingErr
|
||||
}
|
||||
default:
|
||||
}
|
||||
if cacheErr := meta.getErr(); cacheErr != nil {
|
||||
return entries, cacheErr
|
||||
}
|
||||
}
|
||||
return entries, err
|
||||
}
|
||||
|
||||
+92
-8
@@ -39,7 +39,6 @@ import (
|
||||
"github.com/minio/minio/internal/bucket/versioning"
|
||||
"github.com/minio/minio/internal/color"
|
||||
"github.com/minio/minio/internal/hash"
|
||||
xioutil "github.com/minio/minio/internal/ioutil"
|
||||
"github.com/pgsty/silo-pkg/v3/console"
|
||||
)
|
||||
|
||||
@@ -700,9 +699,88 @@ func getQuorumDisks(disks []StorageAPI, infos []DiskInfo, readQuorum int) (newDi
|
||||
return newDisks
|
||||
}
|
||||
|
||||
// resolveListEntry resolves walk snapshots and, for latest-only listings,
|
||||
// falls back to the regular object read path when those snapshots disagree.
|
||||
//
|
||||
// Only unresolved entries pay for the additional namespace read lock and
|
||||
// metadata read. The caller supplies the lock domain used by object writes.
|
||||
func (er *erasureObjects) resolveListEntry(ctx context.Context, bucket string, entries metaCacheEntries, errs []error, resolver *metadataResolutionParams, newNSLock func(string, ...string) RWLocker) (*metaCacheEntry, error) {
|
||||
entry, ok := entries.resolve(resolver)
|
||||
if ok {
|
||||
return entry, nil
|
||||
}
|
||||
if resolver.requestedVersions != 1 {
|
||||
return nil, nil
|
||||
}
|
||||
valid := 0
|
||||
for i := range entries {
|
||||
if !entries[i].isObject() {
|
||||
continue
|
||||
}
|
||||
if _, err := entries[i].xlmeta(); err != nil {
|
||||
continue
|
||||
}
|
||||
valid++
|
||||
if entry == nil {
|
||||
entry = &entries[i]
|
||||
}
|
||||
}
|
||||
for _, err := range errs {
|
||||
if err != nil {
|
||||
valid++
|
||||
}
|
||||
}
|
||||
if entry == nil || valid < resolver.objQuorum {
|
||||
return nil, nil
|
||||
}
|
||||
readQuorumErr := func() error {
|
||||
return InsufficientReadQuorum{
|
||||
Bucket: bucket,
|
||||
Object: entry.name,
|
||||
Err: errErasureReadQuorum,
|
||||
Type: RQInconsistentMeta,
|
||||
}
|
||||
}
|
||||
// Read under the object read lock, exactly as the GET path does. A
|
||||
// lock-free read of an object being overwritten stably reports a live
|
||||
// object as absent, so an unlocked fallback only trades one silent
|
||||
// omission for another.
|
||||
object := encodeDirObject(entry.name)
|
||||
lock := newNSLock(bucket, object)
|
||||
lockCtx, cancel := context.WithTimeout(ctx, 5*time.Second)
|
||||
defer cancel()
|
||||
lkctx, err := lock.GetRLock(lockCtx, globalOperationTimeout)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
defer lock.RUnlock(lkctx)
|
||||
|
||||
fi, _, _, err := er.getObjectFileInfo(lkctx.Context(), bucket, object, ObjectOptions{}, false)
|
||||
if err != nil {
|
||||
// getObjectFileInfo maps inconsistent metadata to not-found, so the
|
||||
// walk evidence means not-found is still undecidable here.
|
||||
if isErrObjectNotFound(err) || isErrVersionNotFound(err) {
|
||||
return nil, readQuorumErr()
|
||||
}
|
||||
return nil, err
|
||||
}
|
||||
// Preserve delete markers until the cross-pool merge has discarded any
|
||||
// older live copy. The normal listing filter removes the merged marker.
|
||||
xl := &xlMetaV2{}
|
||||
if err = xl.AddVersion(fi); err != nil {
|
||||
return nil, readQuorumErr()
|
||||
}
|
||||
resolved := &metaCacheEntry{name: entry.name, cached: xl, reusable: true}
|
||||
resolved.metadata, err = xl.AppendTo(metaDataPoolGet())
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return resolved, nil
|
||||
}
|
||||
|
||||
// Will return io.EOF if continuing would not yield more results.
|
||||
func (er *erasureObjects) listPath(ctx context.Context, o listPathOptions, results chan<- metaCacheEntry) (err error) {
|
||||
defer xioutil.SafeClose(results)
|
||||
func (er *erasureObjects) listPath(ctx context.Context, o listPathOptions, results chan<- metaCacheEntry, newNSLock func(string, ...string) RWLocker) (err error) {
|
||||
// The caller records the terminal error before closing results.
|
||||
o.debugf(color.Green("listPath:")+" with options: %#v", o)
|
||||
|
||||
// get prioritized non-healing disks for listing
|
||||
@@ -776,15 +854,19 @@ func (er *erasureObjects) listPath(ctx context.Context, o listPathOptions, resul
|
||||
case results <- entry:
|
||||
}
|
||||
},
|
||||
partial: func(entries metaCacheEntries, errs []error) {
|
||||
partial: func(entries metaCacheEntries, errs []error) error {
|
||||
// Results Disagree :-(
|
||||
entry, ok := entries.resolve(&resolver)
|
||||
if ok {
|
||||
entry, err := er.resolveListEntry(ctx, o.Bucket, entries, errs, &resolver, newNSLock)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if entry != nil {
|
||||
select {
|
||||
case <-ctxDone:
|
||||
case results <- *entry:
|
||||
}
|
||||
}
|
||||
return nil
|
||||
},
|
||||
})
|
||||
}
|
||||
@@ -980,7 +1062,7 @@ type listPathRawOptions struct {
|
||||
// partial will be called when there is disagreement between disks.
|
||||
// if disk did not return any result, but also haven't errored
|
||||
// the entry will be empty and errs will
|
||||
partial func(entries metaCacheEntries, errs []error)
|
||||
partial func(entries metaCacheEntries, errs []error) error
|
||||
|
||||
// finished will be called when all streams have finished and
|
||||
// more than one disk returned an error.
|
||||
@@ -1191,7 +1273,9 @@ func listPathRaw(ctx context.Context, opts listPathRawOptions) (err error) {
|
||||
continue
|
||||
}
|
||||
if opts.partial != nil {
|
||||
opts.partial(topEntries, errs)
|
||||
if err := opts.partial(topEntries, errs); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
// Skip the inputs we used.
|
||||
for i, r := range readers {
|
||||
|
||||
@@ -162,7 +162,7 @@ func toObjectErr(oerr error, params ...string) error {
|
||||
apiErr.Type = v.Type
|
||||
}
|
||||
return apiErr
|
||||
case errErasureWriteQuorum.Error():
|
||||
case errErasureWriteQuorum.Error(), errCommitMetaMissing.Error():
|
||||
apiErr := InsufficientWriteQuorum{}
|
||||
if len(params) >= 1 {
|
||||
apiErr.Bucket = params[0]
|
||||
|
||||
@@ -62,12 +62,13 @@ func testAPIPutObjectChunkedChecksum(obj ObjectLayer, instanceType, bucketName s
|
||||
// A well-formed 4-byte value that does not match the content.
|
||||
wrongCRC := base64.StdEncoding.EncodeToString([]byte{0x01, 0x02, 0x03, 0x04})
|
||||
|
||||
apiCode := func(rec *httptest.ResponseRecorder) string {
|
||||
apiErrOf := func(rec *httptest.ResponseRecorder) APIErrorResponse {
|
||||
var apiErr APIErrorResponse
|
||||
b, _ := io.ReadAll(rec.Body)
|
||||
_ = xml.Unmarshal(b, &apiErr)
|
||||
return apiErr.Code
|
||||
return apiErr
|
||||
}
|
||||
apiCode := func(rec *httptest.ResponseRecorder) string { return apiErrOf(rec).Code }
|
||||
|
||||
// newChunkedJavaForm builds a non-trailer signed chunked PutObject request that
|
||||
// mirrors the AWS Java SDK v2 wire form: the checksum value sits in the header
|
||||
@@ -131,8 +132,12 @@ func testAPIPutObjectChunkedChecksum(obj ObjectLayer, instanceType, bucketName s
|
||||
if rec.Code != http.StatusBadRequest {
|
||||
t.Fatalf("%s: chunked+wrong CRC32: expected 400, got %d", instanceType, rec.Code)
|
||||
}
|
||||
if code := apiCode(rec); code != "XAmzContentChecksumMismatch" {
|
||||
t.Fatalf("%s: chunked+wrong CRC32: want XAmzContentChecksumMismatch, got %q", instanceType, code)
|
||||
gotErr := apiErrOf(rec)
|
||||
if gotErr.Code != "XAmzContentChecksumMismatch" {
|
||||
t.Fatalf("%s: chunked+wrong CRC32: want XAmzContentChecksumMismatch, got %q", instanceType, gotErr.Code)
|
||||
}
|
||||
if gotErr.Message != "The CRC32 you specified did not match the calculated checksum." {
|
||||
t.Fatalf("%s: chunked+wrong CRC32: message = %q, want S3 algorithm-specific wording", instanceType, gotErr.Message)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -70,6 +70,11 @@ var errDiskAccessDenied = StorageErr("drive access denied")
|
||||
// errFileNotFound - cannot find the file.
|
||||
var errFileNotFound = StorageErr("file not found")
|
||||
|
||||
// errCommitMetaMissing - the staged xl.meta disappeared between the
|
||||
// data rename and the meta-file commit; a write-quorum failure, not
|
||||
// a missing object.
|
||||
var errCommitMetaMissing = StorageErr("commit metadata is missing")
|
||||
|
||||
// errFileNotFound - cannot find requested file version.
|
||||
var errFileVersionNotFound = StorageErr("file version not found")
|
||||
|
||||
|
||||
@@ -106,6 +106,8 @@ func toStorageErr(err error) error {
|
||||
return errVolumeExists
|
||||
case errFileNotFound.Error():
|
||||
return errFileNotFound
|
||||
case errCommitMetaMissing.Error():
|
||||
return errCommitMetaMissing
|
||||
case errFileVersionNotFound.Error():
|
||||
return errFileVersionNotFound
|
||||
case errFileNameTooLong.Error():
|
||||
|
||||
@@ -290,6 +290,9 @@ func (s *storageRESTServer) NSScannerHandler(ctx context.Context, params *nsScan
|
||||
for update := range updates {
|
||||
resp := storageNSScannerRPC.NewResponse()
|
||||
resp.Update = &update
|
||||
// Pooled responses may carry fields from a previous
|
||||
// scan, always clear the fields not being set.
|
||||
resp.Final = nil
|
||||
out <- resp
|
||||
}
|
||||
}()
|
||||
@@ -301,6 +304,7 @@ func (s *storageRESTServer) NSScannerHandler(ctx context.Context, params *nsScan
|
||||
// Send final response.
|
||||
resp := storageNSScannerRPC.NewResponse()
|
||||
resp.Final = &ui
|
||||
resp.Update = nil
|
||||
out <- resp
|
||||
return nil
|
||||
}
|
||||
|
||||
@@ -0,0 +1,94 @@
|
||||
// Copyright (c) 2026 Julien Laurenceau and Feng Ruohang
|
||||
//
|
||||
// This file is part of Silo Object Storage.
|
||||
//
|
||||
// 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"
|
||||
"testing"
|
||||
|
||||
"github.com/minio/madmin-go/v3"
|
||||
"github.com/minio/minio/internal/grid"
|
||||
)
|
||||
|
||||
type scannerResponseTestDisk struct {
|
||||
StorageAPI
|
||||
}
|
||||
|
||||
func (scannerResponseTestDisk) NSScanner(_ context.Context, cache dataUsageCache, updates chan<- dataUsageEntry, _ madmin.HealScanMode, _ func() bool) (dataUsageCache, error) {
|
||||
defer close(updates)
|
||||
updates <- dataUsageEntry{Size: int64(cache.Info.NextCycle)}
|
||||
cache.Info.NextCycle++
|
||||
return cache, nil
|
||||
}
|
||||
|
||||
func TestNSScannerHandlerClearsStaleResponses(t *testing.T) {
|
||||
endpoint, err := NewEndpoint(t.TempDir())
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
globalLocalDrivesMu.Lock()
|
||||
oldDrives, oldSets := globalLocalDrivesMap, globalLocalSetDrives
|
||||
globalLocalDrivesMap = map[string]StorageAPI{endpoint.String(): scannerResponseTestDisk{}}
|
||||
globalLocalSetDrives = nil
|
||||
globalLocalDrivesMu.Unlock()
|
||||
t.Cleanup(func() {
|
||||
globalLocalDrivesMu.Lock()
|
||||
globalLocalDrivesMap, globalLocalSetDrives = oldDrives, oldSets
|
||||
globalLocalDrivesMu.Unlock()
|
||||
})
|
||||
|
||||
oldRPC := storageNSScannerRPC
|
||||
t.Cleanup(func() { storageNSScannerRPC = oldRPC })
|
||||
// Every new allocation is dirty too: the check must not depend on
|
||||
// sync.Pool returning a previously used object.
|
||||
storageNSScannerRPC = grid.NewStream[*nsScannerOptions, grid.NoPayload, *nsScannerResp](grid.HandlerNSScanner,
|
||||
func() *nsScannerOptions { return &nsScannerOptions{} }, nil,
|
||||
func() *nsScannerResp {
|
||||
return &nsScannerResp{Update: &dataUsageEntry{Size: -1}, Final: &dataUsageCache{}}
|
||||
})
|
||||
server := storageRESTServer{endpoint: endpoint}
|
||||
for cycle := uint32(1); cycle <= 3; cycle++ {
|
||||
out := make(chan *nsScannerResp, 2)
|
||||
params := &nsScannerOptions{Cache: &dataUsageCache{Info: dataUsageCacheInfo{Name: "bucket", NextCycle: cycle}}}
|
||||
if err := server.NSScannerHandler(t.Context(), params, out); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if len(out) != 2 {
|
||||
t.Fatalf("cycle %d: got %d responses, want an update and a final cache", cycle, len(out))
|
||||
}
|
||||
for i := 0; i < 2; i++ {
|
||||
resp := <-out
|
||||
wire, err := resp.MarshalMsg(nil)
|
||||
storageNSScannerRPC.PutResponse(resp)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
var decoded nsScannerResp
|
||||
if _, err := decoded.UnmarshalMsg(wire); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if i == 0 {
|
||||
if decoded.Update == nil || decoded.Final != nil || decoded.Update.Size != int64(cycle) {
|
||||
t.Errorf("cycle %d: invalid update response: %+v", cycle, decoded)
|
||||
}
|
||||
} else if decoded.Update != nil || decoded.Final == nil || decoded.Final.Info.NextCycle != cycle+1 {
|
||||
t.Errorf("cycle %d: invalid final response: %+v", cycle, decoded)
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -18,10 +18,16 @@
|
||||
package cmd
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"net/http"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"runtime"
|
||||
"sync"
|
||||
"syscall"
|
||||
"testing"
|
||||
"time"
|
||||
)
|
||||
|
||||
func TestSysErrors(t *testing.T) {
|
||||
@@ -56,3 +62,120 @@ func TestSysErrors(t *testing.T) {
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// TestCommitMetaMissingMapping verifies the RPC and object-layer mapping
|
||||
// of errCommitMetaMissing, returned when the staged xl.meta disappears
|
||||
// between the data rename and the meta-file commit in RenameData().
|
||||
func TestCommitMetaMissingMapping(t *testing.T) {
|
||||
// The sentinel must survive a storage REST/grid round-trip; errors
|
||||
// cross the wire as strings and are remapped by toStorageErr().
|
||||
if got := toStorageErr(errors.New(errCommitMetaMissing.Error())); !errors.Is(got, errCommitMetaMissing) {
|
||||
t.Fatalf("expected %s, got %s", errCommitMetaMissing, got)
|
||||
}
|
||||
if got := toStorageErr(errCommitMetaMissing); got != errCommitMetaMissing {
|
||||
t.Fatalf("expected %s, got %s", errCommitMetaMissing, got)
|
||||
}
|
||||
|
||||
// toObjectErr() must map the sentinel to InsufficientWriteQuorum
|
||||
// (503), never to ObjectNotFound (404).
|
||||
err := toObjectErr(errCommitMetaMissing, "bucket", "object")
|
||||
qerr, ok := err.(InsufficientWriteQuorum)
|
||||
if !ok {
|
||||
t.Fatalf("expected InsufficientWriteQuorum, got %s (%T)", err, err)
|
||||
}
|
||||
if qerr.Bucket != "bucket" || qerr.Object != "object" {
|
||||
t.Fatalf("unexpected bucket/object %s/%s", qerr.Bucket, qerr.Object)
|
||||
}
|
||||
apiErr := toAPIError(t.Context(), err)
|
||||
if apiErr.Code != "SlowDownWrite" || apiErr.HTTPStatusCode != http.StatusServiceUnavailable {
|
||||
t.Fatalf("expected SlowDownWrite/503, got %+v", apiErr)
|
||||
}
|
||||
}
|
||||
|
||||
// missingCommitMetaContext removes staged metadata at an existing cancellation
|
||||
// check, after WriteAll and (for non-inline objects) the data rename succeeded.
|
||||
// This deterministically models concurrent cleanup without changing production
|
||||
// code or relying on a filesystem race. The context itself remains live.
|
||||
type missingCommitMetaContext struct {
|
||||
context.Context
|
||||
mu sync.Mutex
|
||||
meta, dataDest string
|
||||
removed bool
|
||||
removeErr error
|
||||
}
|
||||
|
||||
func (c *missingCommitMetaContext) Done() <-chan struct{} {
|
||||
c.mu.Lock()
|
||||
defer c.mu.Unlock()
|
||||
if !c.removed {
|
||||
if _, err := os.Stat(c.meta); err == nil {
|
||||
dataMoved := c.dataDest == ""
|
||||
if !dataMoved {
|
||||
_, err = os.Stat(c.dataDest)
|
||||
dataMoved = err == nil
|
||||
}
|
||||
if dataMoved {
|
||||
c.removeErr = os.Remove(c.meta)
|
||||
c.removed = c.removeErr == nil
|
||||
}
|
||||
}
|
||||
}
|
||||
return c.Context.Done()
|
||||
}
|
||||
|
||||
func TestXLStorageRenameDataCommitMetaMissing(t *testing.T) {
|
||||
for _, tc := range []struct {
|
||||
name string
|
||||
dataFile bool
|
||||
}{
|
||||
{"empty_object", false},
|
||||
{"data_directory", true},
|
||||
} {
|
||||
t.Run(tc.name, func(t *testing.T) {
|
||||
disk, root, err := newXLStorageTestSetup(t)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
defer disk.Close()
|
||||
s := disk.storage
|
||||
if err := s.MakeVol(t.Context(), "bucket"); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
fi := newFileInfo("object", 2, 2)
|
||||
fi.DataDir, fi.VersionID = mustGetUUID(), mustGetUUID()
|
||||
fi.Erasure.Index = 1
|
||||
fi.ModTime = time.Now().UTC()
|
||||
fi.Fresh = true
|
||||
ctx := &missingCommitMetaContext{
|
||||
Context: t.Context(),
|
||||
meta: filepath.Join(root, minioMetaTmpBucket, "staged", xlStorageFormatFile),
|
||||
}
|
||||
if tc.dataFile {
|
||||
fi.Size = 1
|
||||
fi.AddObjectPart(1, "", 1, 1, fi.ModTime, nil, nil)
|
||||
part := filepath.Join(root, minioMetaTmpBucket, "staged", fi.DataDir, "part.1")
|
||||
if err := os.MkdirAll(filepath.Dir(part), 0o755); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := os.WriteFile(part, []byte("a"), 0o644); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
ctx.dataDest = filepath.Join(root, "bucket", "object", fi.DataDir)
|
||||
}
|
||||
_, err = s.RenameData(ctx, minioMetaTmpBucket, "staged", fi, "bucket", "object", RenameOptions{})
|
||||
if !ctx.removed {
|
||||
t.Fatalf("metadata removal did not execute: removal error = %v, RenameData error = %v", ctx.removeErr, err)
|
||||
}
|
||||
if !errors.Is(err, errCommitMetaMissing) {
|
||||
t.Fatalf("expected commit metadata missing, got %v", err)
|
||||
}
|
||||
apiErr := toAPIError(t.Context(), toObjectErr(err, "bucket", "object"))
|
||||
if apiErr.Code != "SlowDownWrite" || apiErr.HTTPStatusCode != http.StatusServiceUnavailable {
|
||||
t.Fatalf("expected SlowDownWrite/503, got %+v", apiErr)
|
||||
}
|
||||
if _, err := os.Stat(filepath.Join(root, "bucket", "object", xlStorageFormatFile)); !os.IsNotExist(err) {
|
||||
t.Fatalf("failed commit must not produce destination metadata: %v", err)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
@@ -0,0 +1,222 @@
|
||||
// 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"
|
||||
"fmt"
|
||||
"net/http"
|
||||
"slices"
|
||||
"testing"
|
||||
"time"
|
||||
)
|
||||
|
||||
func crossParityNullVersion(t *testing.T, generation int64, data, parity int) xlMetaV2ShallowVersion {
|
||||
t.Helper()
|
||||
fi := newFileInfo("bucket/object", data, parity)
|
||||
fi.Erasure.Index = 1
|
||||
fi.DataDir = "11111111-1111-1111-1111-111111111111"
|
||||
fi.ModTime = time.Unix(generation, 0).UTC()
|
||||
fi.Size = 1024
|
||||
fi.Parts = []ObjectPartInfo{{Number: 1, Size: fi.Size, ActualSize: fi.Size}}
|
||||
fi.Metadata = map[string]string{"etag": fmt.Sprintf("%032d", generation)}
|
||||
var xl xlMetaV2
|
||||
if err := xl.AddVersion(fi); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
return xl.versions[0]
|
||||
}
|
||||
|
||||
// A restarted node can retain four 12+4 shards while a successful degraded
|
||||
// write leaves twelve newer 8+8 shards. Neither disk order nor the minority's
|
||||
// EC must hide the independently sufficient newer group.
|
||||
func TestMergeXLV2LatestNullCrossParity(t *testing.T) {
|
||||
versions := []xlMetaV2ShallowVersion{
|
||||
crossParityNullVersion(t, 100, 12, 4),
|
||||
crossParityNullVersion(t, 200, 8, 8),
|
||||
crossParityNullVersion(t, 300, 10, 6),
|
||||
}
|
||||
for _, tt := range []struct {
|
||||
name string
|
||||
counts [3]int
|
||||
empty int
|
||||
quorum int
|
||||
want int
|
||||
}{
|
||||
{"new-twelve-old-four", [3]int{4, 12}, 0, 8, 1},
|
||||
{"old-twelve-new-four", [3]int{12, 4}, 0, 8, 0},
|
||||
{"new-nine-old-seven", [3]int{7, 9}, 0, 8, 1},
|
||||
{"eight-meets-eight", [3]int{4, 8}, 4, 8, 1},
|
||||
{"eight-does-not-meet-nine", [3]int{8, 8}, 0, 9, -1},
|
||||
{"three-subquorums", [3]int{7, 5, 4}, 0, 8, -1},
|
||||
} {
|
||||
t.Run(tt.name, func(t *testing.T) {
|
||||
var input [][]xlMetaV2ShallowVersion
|
||||
for group, count := range tt.counts {
|
||||
for range count {
|
||||
input = append(input, []xlMetaV2ShallowVersion{versions[group]})
|
||||
}
|
||||
}
|
||||
input = append(input, make([][]xlMetaV2ShallowVersion, tt.empty)...)
|
||||
for order, streams := range quorumVersionOrders(input) {
|
||||
got := mergeXLV2Versions(tt.quorum, false, 1, streams...)
|
||||
if tt.want < 0 {
|
||||
if len(got) != 0 {
|
||||
t.Fatalf("order %d: subquorum groups were combined: %v", order, got)
|
||||
}
|
||||
continue
|
||||
}
|
||||
if len(got) != 1 || got[0].header != versions[tt.want].header {
|
||||
t.Fatalf("order %d: got %v, want group %d", order, got, tt.want)
|
||||
}
|
||||
xl := xlMetaV2{versions: got}
|
||||
fi, err := xl.ToFileInfo("bucket", "object", "", false, true)
|
||||
if err != nil || !fi.IsValid() || fi.Erasure.DataBlocks != int(versions[tt.want].header.EcM) {
|
||||
t.Fatalf("order %d: selected metadata is invalid: %+v, %v", order, fi, err)
|
||||
}
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestMergeXLV2LatestNullCrossParityScope(t *testing.T) {
|
||||
old := crossParityNullVersion(t, 100, 12, 4)
|
||||
newer := crossParityNullVersion(t, 200, 8, 8)
|
||||
for _, kind := range []string{"strict", "all-versions", "uuid", "delete", "free", "history", "legacy-modern-ec", "partial-ec"} {
|
||||
t.Run(kind, func(t *testing.T) {
|
||||
a, b := old, newer
|
||||
strict, requested := false, 1
|
||||
switch kind {
|
||||
case "strict":
|
||||
strict = true
|
||||
case "all-versions":
|
||||
requested = 0
|
||||
case "uuid":
|
||||
a.header.VersionID, b.header.VersionID = [16]byte{1}, [16]byte{1}
|
||||
case "delete":
|
||||
b.header.Type = DeleteType
|
||||
case "free":
|
||||
b.header.Flags |= xlFlagFreeVersion
|
||||
case "legacy-modern-ec":
|
||||
b.header.EcN, b.header.EcM = 0, 0
|
||||
case "partial-ec":
|
||||
b.header.EcN = 0
|
||||
}
|
||||
// These previously unresolved inputs remain outside the repair.
|
||||
// Strict and legacy-header cases use an older majority followed by
|
||||
// a newer minority; the other cases use the opposite grouping.
|
||||
input := make([][]xlMetaV2ShallowVersion, 16)
|
||||
for i := range input {
|
||||
input[i] = []xlMetaV2ShallowVersion{b}
|
||||
if i >= 12 {
|
||||
input[i] = []xlMetaV2ShallowVersion{a}
|
||||
}
|
||||
}
|
||||
if kind == "strict" || kind == "legacy-modern-ec" || kind == "partial-ec" {
|
||||
for i := range input {
|
||||
input[i] = []xlMetaV2ShallowVersion{a}
|
||||
if i >= 12 {
|
||||
input[i] = []xlMetaV2ShallowVersion{b}
|
||||
}
|
||||
}
|
||||
}
|
||||
if kind == "history" {
|
||||
history := old
|
||||
history.header.VersionID = [16]byte{1}
|
||||
input[0] = append(input[0], history)
|
||||
}
|
||||
if got := mergeXLV2Versions(8, strict, requested, input...); len(got) != 0 {
|
||||
t.Fatalf("excluded input changed: %v", got)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestCountXLV2MixedECGroups(t *testing.T) {
|
||||
a := crossParityNullVersion(t, 100, 12, 4)
|
||||
b := crossParityNullVersion(t, 100, 8, 8)
|
||||
c := crossParityNullVersion(t, 200, 10, 6)
|
||||
// Same-time groups remain separate, and equal counts have a deterministic
|
||||
// final EC tiebreak even though sortsBefore does not compare EC itself.
|
||||
input := [][]xlMetaV2ShallowVersion{{a}, {a}, {a}, {b}, {b}, {b}, {c}}
|
||||
for order, streams := range quorumVersionOrders(input) {
|
||||
tops := make([]xlMetaV2ShallowVersion, len(streams))
|
||||
for i := range streams {
|
||||
tops[i] = streams[i][0]
|
||||
}
|
||||
for range 20 {
|
||||
got, count := countXLV2Versions(tops, c.header, c, false, true)
|
||||
if count != 3 || got.header != a.header {
|
||||
t.Fatalf("order %d: got %v, count %d; want distinct group of three", order, got.header, count)
|
||||
}
|
||||
}
|
||||
}
|
||||
differentSig := a
|
||||
differentSig.header.Signature[0]++
|
||||
if _, count := countXLV2Versions([]xlMetaV2ShallowVersion{a, a, differentSig}, a.header, a, false, true); count != 3 {
|
||||
t.Fatalf("non-strict signature groups did not combine: count %d", count)
|
||||
}
|
||||
// Same EC with different time or flags must not accumulate votes either.
|
||||
flags := a
|
||||
flags.header.Flags ^= xlFlagInlineData
|
||||
for _, other := range []xlMetaV2ShallowVersion{crossParityNullVersion(t, 200, 12, 4), flags} {
|
||||
got, count := countXLV2Versions([]xlMetaV2ShallowVersion{a, a, other, other}, a.header, a, false, true)
|
||||
if count != 2 {
|
||||
t.Fatalf("incompatible header groups combined: %v, count %d", got.header, count)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Exercise the regular full-metadata read and HTTP LIST with actual on-disk
|
||||
// object metadata, not merely the metadata selection helper.
|
||||
func TestListCrossParityReadableMajority(t *testing.T) {
|
||||
for _, size := range []int{1024, 256 << 10} {
|
||||
t.Run(fmt.Sprintf("bytes-%d", size), func(t *testing.T) {
|
||||
z, bucket, router := nullQuorumBackend(t)
|
||||
set := z.serverPools[0].sets[0]
|
||||
disks := set.getDisks()
|
||||
const object = "object"
|
||||
mtime := time.Now().UTC().Add(-time.Hour)
|
||||
oldBody := bytes.Repeat([]byte{'a'}, size)
|
||||
newBody := bytes.Repeat([]byte{'b'}, size)
|
||||
nullQuorumPut(t, z, bucket, object, oldBody, ObjectOptions{MTime: mtime})
|
||||
old := make([][]byte, len(disks))
|
||||
for i, disk := range disks {
|
||||
old[i] = mustReadNullQuorumMeta(t, disk, bucket, object)
|
||||
}
|
||||
nullQuorumPut(t, z, bucket, object, newBody, ObjectOptions{MTime: mtime.Add(time.Minute), MaxParity: true})
|
||||
for i := 12; i < len(disks); i++ {
|
||||
if err := disks[i].WriteAll(t.Context(), bucket, object+"/"+xlStorageFormatFile, old[i]); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
for _, readData := range []bool{false, true} {
|
||||
fi, _, _, err := set.getObjectFileInfo(t.Context(), bucket, object, ObjectOptions{}, readData)
|
||||
if err != nil || !fi.ModTime.Equal(mtime.Add(time.Minute)) || fi.Erasure.DataBlocks != 8 {
|
||||
t.Fatalf("readData=%v: full metadata read lost readable majority: %+v, %v", readData, fi, err)
|
||||
}
|
||||
}
|
||||
if get := nullQuorumRequest(t, router, http.MethodGet, getGetObjectURL("", bucket, object)); get.Code != http.StatusOK || !bytes.Equal(get.Body.Bytes(), newBody) {
|
||||
t.Fatalf("GET lost readable majority: status %d", get.Code)
|
||||
}
|
||||
if code, keys := nullQuorumListV2Keys(t, router, bucket); code != http.StatusOK || !slices.Equal(keys, []string{object}) {
|
||||
t.Fatalf("LIST lost readable majority: %d %v", code, keys)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
@@ -40,6 +40,10 @@ func (x *xlMetaV2VersionHeader) unmarshalV(v uint8, bts []byte) (o []byte, err e
|
||||
|
||||
// unmarshalV1 decodes version 1, never released.
|
||||
func (x *xlMetaV2VersionHeader) unmarshalV1(bts []byte) (o []byte, err error) {
|
||||
// Fields absent from the V1 wire format must be reset since the
|
||||
// header may be reused across decodes.
|
||||
x.Signature = [4]byte{}
|
||||
x.EcN, x.EcM = 0, 0
|
||||
var zb0001 uint32
|
||||
zb0001, bts, err = msgp.ReadArrayHeaderBytes(bts)
|
||||
if err != nil {
|
||||
@@ -134,7 +138,7 @@ type xlMetaV2VersionHeaderV2 struct {
|
||||
|
||||
// UnmarshalMsg implements msgp.Unmarshaler
|
||||
func (z *xlMetaV2VersionHeaderV2) UnmarshalMsg(bts []byte) (o []byte, err error) {
|
||||
z.EcN, z.EcN = 0, 0
|
||||
z.EcN, z.EcM = 0, 0
|
||||
var zb0001 uint32
|
||||
zb0001, bts, err = msgp.ReadArrayHeaderBytes(bts)
|
||||
if err != nil {
|
||||
@@ -184,7 +188,7 @@ func (z *xlMetaV2VersionHeaderV2) UnmarshalMsg(bts []byte) (o []byte, err error)
|
||||
|
||||
// DecodeMsg implements msgp.Decodable
|
||||
func (z *xlMetaV2VersionHeaderV2) DecodeMsg(dc *msgp.Reader) (err error) {
|
||||
z.EcN, z.EcN = 0, 0
|
||||
z.EcN, z.EcM = 0, 0
|
||||
var zb0001 uint32
|
||||
zb0001, err = dc.ReadArrayHeader()
|
||||
if err != nil {
|
||||
|
||||
+35
-14
@@ -1944,7 +1944,7 @@ func mergeXLV2Versions(quorum int, strict bool, requestedVersions int, versions
|
||||
// Keep the original stream shapes: pruning must not make a versioned object
|
||||
// eligible for the single null-version recount below.
|
||||
originalVersions := versions
|
||||
var checkedSingleNull, singleNull bool
|
||||
var checkedSingleNull, singleNull, mixedECNull bool
|
||||
// Shallow copy input
|
||||
versions = append(make([][]xlMetaV2ShallowVersion, 0, len(versions)), versions...)
|
||||
|
||||
@@ -2019,19 +2019,24 @@ func mergeXLV2Versions(quorum int, strict bool, requestedVersions int, versions
|
||||
// Version IDs match, but otherwise unable to resolve.
|
||||
// We are either strict, or don't have enough information to match.
|
||||
// Switch to a pure counting algo.
|
||||
latest, latestCount = countXLV2Versions(tops, ver.header, latest, strict)
|
||||
latest, latestCount = countXLV2Versions(tops, ver.header, latest, strict, false)
|
||||
break
|
||||
}
|
||||
}
|
||||
if latestCount < quorum {
|
||||
if !checkedSingleNull {
|
||||
singleNull = singleNullVersionStreams(originalVersions)
|
||||
singleNull = singleNullVersionStreams(originalVersions, false)
|
||||
// Availability-optimized writes can change EC while a drive
|
||||
// is offline. Only latest-only non-strict reads may recount
|
||||
// these ordinary null versions across separate modern EC groups.
|
||||
mixedECNull = !singleNull && !strict && requestedVersions == 1 &&
|
||||
singleNullVersionStreams(originalVersions, true)
|
||||
checkedSingleNull = true
|
||||
}
|
||||
if singleNull {
|
||||
if singleNull || mixedECNull {
|
||||
// A newer minority at the end can hide an older quorum from
|
||||
// the selection loop. Recount before discarding the null ID.
|
||||
if candidate, count := countXLV2Versions(tops, latest.header, latest, strict); count >= quorum {
|
||||
if candidate, count := countXLV2Versions(tops, latest.header, latest, strict, mixedECNull); count >= quorum {
|
||||
latest, latestCount = candidate, count
|
||||
}
|
||||
}
|
||||
@@ -2098,7 +2103,8 @@ func mergeXLV2Versions(quorum int, strict bool, requestedVersions int, versions
|
||||
|
||||
// singleNullVersionStreams excludes histories and other version types from the
|
||||
// additional recount. Check the original inputs, before any stream is pruned.
|
||||
func singleNullVersionStreams(versions [][]xlMetaV2ShallowVersion) bool {
|
||||
// Mixed EC groups require known modern EC headers on every non-empty stream.
|
||||
func singleNullVersionStreams(versions [][]xlMetaV2ShallowVersion, allowMixedEC bool) bool {
|
||||
var ec xlMetaV2VersionHeader
|
||||
var haveEC bool
|
||||
for _, stream := range versions {
|
||||
@@ -2112,7 +2118,10 @@ func singleNullVersionStreams(versions [][]xlMetaV2ShallowVersion) bool {
|
||||
if h.VersionID != [16]byte{} || h.Type != ObjectType || h.FreeVersion() {
|
||||
return false
|
||||
}
|
||||
if haveEC && (h.EcN != ec.EcN || h.EcM != ec.EcM) {
|
||||
if allowMixedEC && !h.hasEC() {
|
||||
return false
|
||||
}
|
||||
if !allowMixedEC && haveEC && (h.EcN != ec.EcN || h.EcM != ec.EcM) {
|
||||
return false
|
||||
}
|
||||
ec, haveEC = h, true
|
||||
@@ -2121,16 +2130,18 @@ func singleNullVersionStreams(versions [][]xlMetaV2ShallowVersion) bool {
|
||||
}
|
||||
|
||||
// countXLV2Versions selects the most frequent compatible header for reference's
|
||||
// VersionID, retaining the existing sort tiebreak and last matching entry.
|
||||
func countXLV2Versions(tops []xlMetaV2ShallowVersion, reference xlMetaV2VersionHeader, latest xlMetaV2ShallowVersion, strict bool) (xlMetaV2ShallowVersion, int) {
|
||||
// VersionID and the last matching entry. The mixed-EC null-version recount adds
|
||||
// a stable EC tiebreak without changing the ordinary counting path.
|
||||
func countXLV2Versions(tops []xlMetaV2ShallowVersion, reference xlMetaV2VersionHeader, latest xlMetaV2ShallowVersion, strict, allowMixedEC bool) (xlMetaV2ShallowVersion, int) {
|
||||
x := make(map[xlMetaV2VersionHeader]int, len(tops))
|
||||
for _, a := range tops {
|
||||
if a.header.VersionID != reference.VersionID {
|
||||
continue
|
||||
}
|
||||
if !strict {
|
||||
// we must match EC, when we are not strict.
|
||||
if !a.header.matchesEC(reference) {
|
||||
// The narrow null-version recount can consider other EC groups,
|
||||
// but EcN/EcM stay in the map key: their votes never combine.
|
||||
if !allowMixedEC && !a.header.matchesEC(reference) {
|
||||
continue
|
||||
}
|
||||
a.header.Signature = [4]byte{}
|
||||
@@ -2142,9 +2153,19 @@ func countXLV2Versions(tops []xlMetaV2ShallowVersion, reference xlMetaV2VersionH
|
||||
if v < latestCount {
|
||||
continue
|
||||
}
|
||||
if v == latestCount && latest.header.sortsBefore(k) {
|
||||
// Tiebreak, use sort.
|
||||
continue
|
||||
if v == latestCount {
|
||||
if allowMixedEC {
|
||||
current := latest.header
|
||||
current.Signature = [4]byte{}
|
||||
// Signatures are not part of these groups. Use EC counts as
|
||||
// the final tiebreak when the existing sort fields are equal.
|
||||
if current.sortsBefore(k) || !k.sortsBefore(current) &&
|
||||
(current.EcM > k.EcM || current.EcM == k.EcM && current.EcN > k.EcN) {
|
||||
continue
|
||||
}
|
||||
} else if latest.header.sortsBefore(k) {
|
||||
continue
|
||||
}
|
||||
}
|
||||
for _, a := range tops {
|
||||
hdr := a.header
|
||||
|
||||
@@ -38,6 +38,7 @@ import (
|
||||
"github.com/klauspost/compress/zstd"
|
||||
"github.com/minio/minio/internal/bucket/lifecycle"
|
||||
xhttp "github.com/minio/minio/internal/http"
|
||||
"github.com/tinylib/msgp/msgp"
|
||||
)
|
||||
|
||||
func TestReadXLMetaNoData(t *testing.T) {
|
||||
@@ -1190,3 +1191,92 @@ func BenchmarkToFileInfoNoParts(b *testing.B) {
|
||||
func BenchmarkToFileInfoWithParts(b *testing.B) {
|
||||
benchmarkManyPartsOptionally(b, true)
|
||||
}
|
||||
|
||||
// TestXLMetaV2HeaderLegacyUnmarshalResetsEc checks that decoding legacy
|
||||
// version headers, which carry no erasure metadata fields, cannot leak
|
||||
// stale EcN/EcM values from a previously decoded object into the same
|
||||
// structure.
|
||||
func TestXLMetaV2HeaderLegacyUnmarshalResetsEc(t *testing.T) {
|
||||
base := xlMetaV2VersionHeader{
|
||||
VersionID: [16]byte{1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16},
|
||||
ModTime: time.Now().Unix(),
|
||||
Signature: [4]byte{1, 2, 3, 4},
|
||||
Type: ObjectType,
|
||||
Flags: xlFlagInlineData,
|
||||
}
|
||||
|
||||
// Legacy V2 header wire format: 5 fields, no EcN/EcM.
|
||||
var legacy []byte
|
||||
legacy = msgp.AppendArrayHeader(legacy, 5)
|
||||
legacy = msgp.AppendBytes(legacy, base.VersionID[:])
|
||||
legacy = msgp.AppendInt64(legacy, base.ModTime)
|
||||
legacy = msgp.AppendBytes(legacy, base.Signature[:])
|
||||
legacy = msgp.AppendUint8(legacy, uint8(base.Type))
|
||||
legacy = msgp.AppendUint8(legacy, uint8(base.Flags))
|
||||
|
||||
for _, decoder := range []string{"UnmarshalMsg", "DecodeMsg", "loadIndexed"} {
|
||||
t.Run(decoder, func(t *testing.T) {
|
||||
header := base
|
||||
header.EcN, header.EcM = 4, 8
|
||||
x2 := xlMetaV2VersionHeaderV2{xlMetaV2VersionHeader: &header}
|
||||
var err error
|
||||
switch decoder {
|
||||
case "UnmarshalMsg":
|
||||
_, err = x2.UnmarshalMsg(legacy)
|
||||
case "DecodeMsg":
|
||||
err = x2.DecodeMsg(msgp.NewReader(bytes.NewReader(legacy)))
|
||||
case "loadIndexed":
|
||||
indexed := msgp.AppendUint(nil, 2)
|
||||
indexed = msgp.AppendUint(indexed, xlMetaVersion)
|
||||
indexed = msgp.AppendInt(indexed, 1)
|
||||
indexed = msgp.AppendBytes(indexed, legacy)
|
||||
indexed = msgp.AppendBytes(indexed, nil)
|
||||
xl := xlMetaV2{versions: []xlMetaV2ShallowVersion{{header: header}}}
|
||||
err = xl.loadIndexed(indexed, nil)
|
||||
header = xl.versions[0].header
|
||||
}
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if header != base {
|
||||
t.Fatalf("legacy header: got %+v, want %+v", header, base)
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
// V1 headers (4 fields, never released) must zero the fields too.
|
||||
var v1 []byte
|
||||
v1 = msgp.AppendArrayHeader(v1, 4)
|
||||
v1 = msgp.AppendBytes(v1, base.VersionID[:])
|
||||
v1 = msgp.AppendInt64(v1, base.ModTime)
|
||||
v1 = msgp.AppendUint8(v1, uint8(base.Type))
|
||||
v1 = msgp.AppendUint8(v1, uint8(base.Flags))
|
||||
|
||||
header := base
|
||||
header.EcN, header.EcM = 4, 8
|
||||
header.Signature = [4]byte{9, 9, 9, 9}
|
||||
if _, err := header.unmarshalV1(v1); err != nil {
|
||||
t.Fatalf("legacy V1 header unmarshal failed: %v", err)
|
||||
}
|
||||
if header.EcN != 0 || header.EcM != 0 {
|
||||
t.Fatalf("stale erasure metadata leaked on V1 decode: EcN %d, EcM %d", header.EcN, header.EcM)
|
||||
}
|
||||
if header.Signature != ([4]byte{}) {
|
||||
t.Fatalf("stale signature leaked on V1 decode: %v", header.Signature)
|
||||
}
|
||||
|
||||
// Current headers must keep their erasure metadata.
|
||||
current := base
|
||||
current.EcN, current.EcM = 4, 8
|
||||
wire, err := current.MarshalMsg(nil)
|
||||
if err != nil {
|
||||
t.Fatalf("header marshal failed: %v", err)
|
||||
}
|
||||
decoded := base
|
||||
if _, err := decoded.UnmarshalMsg(wire); err != nil {
|
||||
t.Fatalf("header unmarshal failed: %v", err)
|
||||
}
|
||||
if decoded.EcN != 4 || decoded.EcM != 8 {
|
||||
t.Fatalf("erasure metadata lost on current decode: EcN %d, EcM %d", decoded.EcN, decoded.EcM)
|
||||
}
|
||||
}
|
||||
|
||||
+9
-1
@@ -2925,7 +2925,15 @@ func (s *xlStorage) RenameData(ctx context.Context, srcVolume, srcPath string, f
|
||||
// if its a partial rename() do not attempt to delete recursively.
|
||||
// this can be healed since all parts are available.
|
||||
s.deleteFile(dstVolumeDir, dstDataPath, false, false)
|
||||
return res, osErrToFileErr(err)
|
||||
err = osErrToFileErr(err)
|
||||
if errors.Is(err, errFileNotFound) {
|
||||
// The staged xl.meta disappeared between staging and commit.
|
||||
// This is a write-quorum failure, not a missing object; use a
|
||||
// dedicated error so the object layer returns 503 instead of
|
||||
// mapping it to ObjectNotFound (404).
|
||||
err = errCommitMetaMissing
|
||||
}
|
||||
return res, err
|
||||
}
|
||||
|
||||
if srcVolume != minioMetaMultipartBucket {
|
||||
|
||||
@@ -4,7 +4,7 @@ go 1.27.2
|
||||
|
||||
// Console and MC retain their historical module paths for best-effort upstream
|
||||
// compatibility. Pin the maintained PGSTY implementations used by SILO.
|
||||
replace github.com/minio/console => github.com/pgsty/silo-console v0.0.0-20260916075814-1360e26d976d
|
||||
replace github.com/minio/console => github.com/pgsty/silo-console v0.0.0-20261015132217-e5b165e4c0ee
|
||||
|
||||
replace github.com/minio/mc => github.com/pgsty/mc v0.0.0-20260916070421-e952aa78f10a
|
||||
|
||||
|
||||
@@ -547,8 +547,8 @@ github.com/pborman/getopt v0.0.0-20170112200414-7148bc3a4c30/go.mod h1:85jBQOZwp
|
||||
github.com/pelletier/go-toml v1.2.0/go.mod h1:5z9KED0ma1S8pY6P1sdut58dfprrGBbd/94hg7ilaic=
|
||||
github.com/pgsty/mc v0.0.0-20260916070421-e952aa78f10a h1:FY1vcTQT67HGBe23ozlhOiQnWCzUCGmeSN5x6DT/kFs=
|
||||
github.com/pgsty/mc v0.0.0-20260916070421-e952aa78f10a/go.mod h1:6z9lX8kOvaWz2vLlP7BjjF3JWHnIH86XDA2HwcmpZH8=
|
||||
github.com/pgsty/silo-console v0.0.0-20260916075814-1360e26d976d h1:A9ou1GkUckGZd3BOgEXeZ393nbYST967SKcI4Lgr/Vs=
|
||||
github.com/pgsty/silo-console v0.0.0-20260916075814-1360e26d976d/go.mod h1:JS165CF2yoTBIx1X8k1IQ6vunJcZFG3pRZlWHJ8UT8o=
|
||||
github.com/pgsty/silo-console v0.0.0-20261015132217-e5b165e4c0ee h1:e51AnrGe0Zq+nHTYaGfv/GkehQMO9Mekuw/ew6zYwOM=
|
||||
github.com/pgsty/silo-console v0.0.0-20261015132217-e5b165e4c0ee/go.mod h1:jpt39bx4wjhXoFB+BTOa/bx/pmU3pXzp2cT7PnT21Lo=
|
||||
github.com/pgsty/silo-pkg/v3 v3.14.1 h1:8MONT3Hky9EOnsuPk29590pEBE6MILaOnJtaejPPi7M=
|
||||
github.com/pgsty/silo-pkg/v3 v3.14.1/go.mod h1:jxKWxi52jSsDaDhXXPqjOyPDYdxx2iqdBXyLfM/xSIw=
|
||||
github.com/philhofer/fwd v1.2.0 h1:e6DnBTl7vGY+Gz322/ASL4Gyp1FspeMvx1RNDoToZuM=
|
||||
|
||||
@@ -0,0 +1,73 @@
|
||||
// Copyright (c) 2026 PGSTY
|
||||
//
|
||||
// This file is part of SILO Object Storage.
|
||||
// SPDX-License-Identifier: AGPL-3.0-or-later
|
||||
//
|
||||
// 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 hash
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"errors"
|
||||
"io"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"testing"
|
||||
)
|
||||
|
||||
func TestChecksumMismatchAlgorithm(t *testing.T) {
|
||||
content := []byte("actual content")
|
||||
for _, typ := range []ChecksumType{ChecksumCRC32, ChecksumCRC32C, ChecksumSHA1, ChecksumSHA256, ChecksumCRC64NVME} {
|
||||
t.Run(typ.String(), func(t *testing.T) {
|
||||
wrong := NewChecksumFromData(typ, []byte("wrong content"))
|
||||
check := func(t *testing.T, err error) {
|
||||
t.Helper()
|
||||
var mismatch ChecksumMismatch
|
||||
if !errors.As(err, &mismatch) || mismatch.Algorithm != typ.String() {
|
||||
t.Fatalf("error=%+v, want %s checksum mismatch", err, typ)
|
||||
}
|
||||
}
|
||||
check(t, wrong.Matches(content, 0))
|
||||
parts := NewChecksumFromData(typ, content)
|
||||
parts.WantParts = 2
|
||||
check(t, parts.Matches(content, 1))
|
||||
for _, mode := range []string{"header", "trailer", "malformed-trailer"} {
|
||||
t.Run(mode, func(t *testing.T) {
|
||||
reader, err := NewReader(t.Context(), bytes.NewReader(content), int64(len(content)), "", "", int64(len(content)))
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
req := httptest.NewRequest(http.MethodPut, "http://localhost/bucket/object", nil)
|
||||
if mode == "header" {
|
||||
req.Header.Set(typ.Key(), wrong.Encoded)
|
||||
} else {
|
||||
req.Header.Set("x-amz-trailer", typ.Key())
|
||||
req.Trailer = make(http.Header)
|
||||
value := wrong.Encoded
|
||||
if mode == "malformed-trailer" {
|
||||
value = "!invalid-base64!"
|
||||
}
|
||||
req.Trailer.Set(typ.Key(), value)
|
||||
}
|
||||
if err = reader.AddChecksum(req, false); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
_, err = io.ReadAll(reader)
|
||||
check(t, err)
|
||||
})
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
@@ -584,15 +584,17 @@ func (c Checksum) Matches(content []byte, parts int) error {
|
||||
sum := hasher.Sum(nil)
|
||||
if c.WantParts > 0 && c.WantParts != parts {
|
||||
return ChecksumMismatch{
|
||||
Want: fmt.Sprintf("%s-%d", c.Encoded, c.WantParts),
|
||||
Got: fmt.Sprintf("%s-%d", base64.StdEncoding.EncodeToString(sum), parts),
|
||||
Algorithm: c.Type.String(),
|
||||
Want: fmt.Sprintf("%s-%d", c.Encoded, c.WantParts),
|
||||
Got: fmt.Sprintf("%s-%d", base64.StdEncoding.EncodeToString(sum), parts),
|
||||
}
|
||||
}
|
||||
|
||||
if !bytes.Equal(sum, c.Raw) {
|
||||
return ChecksumMismatch{
|
||||
Want: c.Encoded,
|
||||
Got: base64.StdEncoding.EncodeToString(sum),
|
||||
Algorithm: c.Type.String(),
|
||||
Want: c.Encoded,
|
||||
Got: base64.StdEncoding.EncodeToString(sum),
|
||||
}
|
||||
}
|
||||
return nil
|
||||
|
||||
@@ -74,8 +74,11 @@ func (e SizeMismatch) Error() string {
|
||||
|
||||
// ChecksumMismatch - when content checksum does not match with what was sent from client.
|
||||
type ChecksumMismatch struct {
|
||||
Want string
|
||||
Got string
|
||||
// Algorithm carries the requested checksum type (e.g. "CRC32") so API
|
||||
// error rendering can name it, matching S3's wording.
|
||||
Algorithm string
|
||||
Want string
|
||||
Got string
|
||||
}
|
||||
|
||||
func (e ChecksumMismatch) Error() string {
|
||||
|
||||
@@ -302,13 +302,14 @@ func (r *Reader) Read(p []byte) (int, error) {
|
||||
r.contentHash.Encoded = r.trailer.Get(r.contentHash.Type.Key())
|
||||
r.contentHash.Raw, err = base64.StdEncoding.DecodeString(r.contentHash.Encoded)
|
||||
if err != nil || len(r.contentHash.Raw) == 0 {
|
||||
return 0, ChecksumMismatch{Got: r.contentHash.Encoded}
|
||||
return 0, ChecksumMismatch{Algorithm: r.contentHash.Type.String(), Got: r.contentHash.Encoded}
|
||||
}
|
||||
}
|
||||
if sum := r.contentHasher.Sum(nil); !bytes.Equal(r.contentHash.Raw, sum) {
|
||||
err := ChecksumMismatch{
|
||||
Want: r.contentHash.Encoded,
|
||||
Got: base64.StdEncoding.EncodeToString(sum),
|
||||
Algorithm: r.contentHash.Type.String(),
|
||||
Want: r.contentHash.Encoded,
|
||||
Got: base64.StdEncoding.EncodeToString(sum),
|
||||
}
|
||||
return n, err
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user