Compare commits

..

4 Commits

Author SHA1 Message Date
Feng Ruohang bf053386a4 docs: link compression and federation to maintained design records
Signed-off-by: Feng Ruohang <rh@vonng.com>
2026-09-16 18:26:03 +08:00
Feng Ruohang f99ed829b5 Merge pull request #213 from pgsty/codex/pr198-release-compat
fix: preserve multipart defaults and rollback compatibility
2026-09-16 16:42:56 +08:00
Feng Ruohang 956a1a8e33 Merge pull request #212 from pgsty/codex/docs-migration-cleanup
docs: consolidate repository entry points and retire migrated work records
2026-09-16 16:39:40 +08:00
Feng Ruohang 82f0a9828e fix: preserve multipart defaults and rollback compatibility
Signed-off-by: Feng Ruohang <rh@vonng.com>
2026-09-16 16:30:43 +08:00
12 changed files with 524 additions and 63 deletions
+18 -9
View File
@@ -56,19 +56,28 @@ and [complete commit range](https://github.com/pgsty/silo/compare/RELEASE.2026-0
quorum-written `xl.meta`; completion removes those upload-only fields. Native quorum-written `xl.meta`; completion removes those upload-only fields. Native
markers remain usable after their upload is completed or canceled. Strict markers remain usable after their upload is completed or canceled. Strict
listing returns a diagnostic 503 for legacy uploads or uncertain coverage; listing returns a diagnostic 503 for legacy uploads or uncertain coverage;
`api multipart_listing=legacy` is an explicit temporary migration mode. the default remains the released exact-key/cache-based `legacy` behavior.
Upgrade every writer, drain old uploads and check the read-only admin Opt into strict mode only through `MINIO_API_MULTIPART_LISTING=strict`, after
`multipart-preflight` report before relying on strict listing. Per-process upgrading every writer, draining old uploads, checking the read-only admin
`multipart-preflight` report and validating scan capacity. The process-only
setting is not persisted into shared API configuration. Historical
`multipart_listing` keys are ignored and can be removed with a targeted
`mcli admin config reset ALIAS api multipart_listing` before rollback. Per-process
admission, directory-entry, worker and time budgets bound scan scheduling; admission, directory-entry, worker and time budgets bound scan scheduling;
each page still scans durable state. See [issue #79](https://github.com/pgsty/silo/issues/79) each page still scans durable state. See [issue #79](https://github.com/pgsty/silo/issues/79)
and its [design record](https://silo.pgsty.com/blog/design/list-multipart-uploads/). and its [design record](https://silo.pgsty.com/blog/design/list-multipart-uploads/).
Thanks to mr javad seydi (@mrjavadseydi) for the original implementation. Thanks to mr javad seydi (@mrjavadseydi) for the original implementation.
- Confirm multipart cancellation on a strict majority of each relevant set, - Retain released read-quorum and best-effort multipart cancellation in default
and allow retries after partial deletion. Uncertain pools or insufficient legacy mode. Strict mode requires majority deletion acknowledgements and
confirmations return 503 rather than acknowledging a cancellation whose permits retries below read quorum. When most drives were already empty,
static remnants can later become readable. **Known boundary:** creation failed deletion of an observed remnant now returns 503 instead of being
writes that finish after a storage timeout can still restore an upload after masked by empty-drive successes. Wrong-key or wrong-bucket cancellation
successful cancellation; this change does not add a durable creation fence. preserves the valid upload's cache entry; successful cancellation notifies
peers with the request context after the distributed lock is released.
The HTTP response for an absent
upload remains 204; this is not proof of physical cleanup. **Known boundary:**
delayed creation writes can still restore an upload after cancellation;
this change does not add a durable creation fence.
- Preserve object tags during multi-pool metadata reconciliation by reading the - Preserve object tags during multi-pool metadata reconciliation by reading the
resolved tag field together with its revision (#189). Previously, reconciliation resolved tag field together with its revision (#189). Previously, reconciliation
could replace existing tags with an empty value. could replace existing tags with an empty value.
+23
View File
@@ -31,6 +31,7 @@ func TestMultipartListingAbortBetweenHTTPPages(t *testing.T) {
if err != nil { if err != nil {
t.Fatal(err) t.Fatal(err)
} }
setMultipartListingTestMode(t, false)
var firstID, secondID string var firstID, secondID string
for attempt := 0; attempt < 32; attempt++ { for attempt := 0; attempt < 32; attempt++ {
one, err := z.NewMultipartUpload(t.Context(), bucket, "a", ObjectOptions{}) one, err := z.NewMultipartUpload(t.Context(), bucket, "a", ObjectOptions{})
@@ -147,6 +148,22 @@ func TestMultipartListingLegacyPreflight(t *testing.T) {
t.Fatal(err) t.Fatal(err)
} }
z.mpCache.Clear() z.mpCache.Clear()
t.Run("legacy-upgrade", func(t *testing.T) {
setMultipartListingTestMode(t, true)
exact, err := z.ListMultipartUploads(t.Context(), other, "old", "", "", "", 10)
if err != nil {
t.Fatal(err)
}
requireMultipartUploadKeys(t, exact, "old")
const empty = "multipart-legacy-empty"
if err := z.MakeBucket(t.Context(), empty, MakeBucketOptions{}); err != nil {
t.Fatal(err)
}
got, err := z.ListMultipartUploads(t.Context(), empty, "", "", "", "", 10)
if err != nil || len(got.Uploads) != 0 {
t.Fatalf("old upload in another bucket broke legacy listing: %+v %v", got, err)
}
})
_, err = z.ListMultipartUploads(t.Context(), bucket, "", "", "", "", 10) _, err = z.ListMultipartUploads(t.Context(), bucket, "", "", "", "", 10)
if !errors.Is(err, errMultipartListingLegacy) { if !errors.Is(err, errMultipartListingLegacy) {
t.Fatalf("old upload in another bucket: %v", err) t.Fatalf("old upload in another bucket: %v", err)
@@ -214,6 +231,7 @@ func TestMultipartListingIdentityFallback(t *testing.T) {
func TestMultipartAbortPoolsAndRetry(t *testing.T) { func TestMultipartAbortPoolsAndRetry(t *testing.T) {
z, bucket := consistencyPools(t) z, bucket := consistencyPools(t)
setMultipartListingTestMode(t, false)
mp, err := z.serverPools[1].NewMultipartUpload(t.Context(), bucket, "a", ObjectOptions{}) mp, err := z.serverPools[1].NewMultipartUpload(t.Context(), bucket, "a", ObjectOptions{})
if err != nil { if err != nil {
t.Fatal(err) t.Fatal(err)
@@ -286,6 +304,7 @@ func TestMultipartListingMarkerHTTP(t *testing.T) {
if err != nil { if err != nil {
t.Fatal(err) t.Fatal(err)
} }
setMultipartListingTestMode(t, false)
for _, tc := range []struct { for _, tc := range []struct {
key, marker string key, marker string
status int status int
@@ -369,6 +388,7 @@ func TestMultipartAbortLateCreateBoundary(t *testing.T) {
t.Fatal(err) t.Fatal(err)
} }
z := obj.(*erasureServerPools) z := obj.(*erasureServerPools)
setMultipartListingTestMode(t, false)
t.Cleanup(func() { z.Shutdown(context.Background()); removeRoots(dirs) }) t.Cleanup(func() { z.Shutdown(context.Background()); removeRoots(dirs) })
saved := globalStorageClass saved := globalStorageClass
globalStorageClass.Update(storageclass.Config{Standard: storageclass.StorageClass{Parity: 8}}) globalStorageClass.Update(storageclass.Config{Standard: storageclass.StorageClass{Parity: 8}})
@@ -454,6 +474,7 @@ func TestMultipartListingScanCosts(t *testing.T) {
t.Fatal(err) t.Fatal(err)
} }
z := obj.(*erasureServerPools) z := obj.(*erasureServerPools)
setMultipartListingTestMode(t, false)
t.Cleanup(func() { z.Shutdown(context.Background()); removeRoots(dirs) }) t.Cleanup(func() { z.Shutdown(context.Background()); removeRoots(dirs) })
const bucket, otherBucket = "r9-scan-target", "r9-scan-unrelated" const bucket, otherBucket = "r9-scan-target", "r9-scan-unrelated"
for _, name := range []string{bucket, otherBucket} { for _, name := range []string{bucket, otherBucket} {
@@ -523,6 +544,7 @@ func multipartListingFixture(t *testing.T) (*erasureServerPools, *erasureObjects
if err := z.MakeBucket(t.Context(), bucket, MakeBucketOptions{}); err != nil { if err := z.MakeBucket(t.Context(), bucket, MakeBucketOptions{}); err != nil {
t.Fatal(err) t.Fatal(err)
} }
setMultipartListingTestMode(t, false)
return z, z.serverPools[0].getHashedSet("a"), bucket return z, z.serverPools[0].getHashedSet("a"), bucket
} }
@@ -673,6 +695,7 @@ func TestMultipartListingAdmissionHTTP(t *testing.T) {
if err != nil { if err != nil {
t.Fatal(err) t.Fatal(err)
} }
setMultipartListingTestMode(t, false)
for range cap(multipartScanSlots) { for range cap(multipartScanSlots) {
scan, err := startMultipartScan(t.Context(), false) scan, err := startMultipartScan(t.Context(), false)
if err != nil { if err != nil {
+331
View File
@@ -0,0 +1,331 @@
// Copyright (c) 2026 Ruohang Feng
// SPDX-License-Identifier: AGPL-3.0-or-later
package cmd
import (
"context"
"encoding/base64"
"errors"
"fmt"
"strings"
"sync/atomic"
"testing"
"github.com/minio/minio/internal/config/api"
"github.com/minio/minio/internal/config/storageclass"
"github.com/minio/minio/internal/dsync"
"github.com/minio/minio/internal/grid"
xnet "github.com/pgsty/silo-pkg/v3/net"
)
func setMultipartListingTestMode(t *testing.T, legacy bool) {
t.Helper()
globalAPIConfig.mu.Lock()
previous := globalAPIConfig.multipartListingStrict
globalAPIConfig.multipartListingStrict = !legacy
globalAPIConfig.mu.Unlock()
t.Cleanup(func() {
globalAPIConfig.mu.Lock()
globalAPIConfig.multipartListingStrict = previous
globalAPIConfig.mu.Unlock()
})
}
func TestMultipartListingDefaultMode(t *testing.T) {
var uninitialized apiConfig
if !uninitialized.getMultipartListingLegacy() {
t.Fatal("runtime zero value enabled strict mode before config initialization")
}
for _, mode := range []string{"", "legacy", "strict"} {
var local apiConfig
local.init(api.Config{RequestsMax: 1, MultipartListing: mode}, []int{4}, false)
if local.getMultipartListingLegacy() != (mode != "strict") {
t.Fatalf("unexpected mode for %q", mode)
}
}
}
func TestMultipartAbortLegacyAvailability(t *testing.T) {
for _, drives := range []int{4, 16} {
for _, offline := range []int{0, 1, drives / 2} {
t.Run(fmt.Sprintf("drives=%d/offline=%d", drives, offline), func(t *testing.T) {
obj, dirs, err := prepareErasure(t.Context(), drives)
if err != nil {
t.Fatal(err)
}
z := obj.(*erasureServerPools)
t.Cleanup(func() { z.Shutdown(context.Background()); removeRoots(dirs) })
setMultipartListingTestMode(t, true)
previous := globalStorageClass
globalStorageClass.Update(storageclass.Config{Standard: storageclass.StorageClass{Parity: drives / 2}})
t.Cleanup(func() { globalStorageClass.Update(previous) })
const bucket, key = "multipart-legacy-availability", "keep-available"
if err := z.MakeBucket(t.Context(), bucket, MakeBucketOptions{}); err != nil {
t.Fatal(err)
}
mp, err := z.NewMultipartUpload(t.Context(), bucket, key, ObjectOptions{})
if err != nil {
t.Fatal(err)
}
set := z.serverPools[0].getHashedSet(key)
original := set.getDisks
disks := original()
set.getDisks = func() []StorageAPI {
visible := append([]StorageAPI(nil), disks...)
clear(visible[:offline])
return visible
}
t.Cleanup(func() { set.getDisks = original })
if err := z.AbortMultipartUpload(t.Context(), bucket, key, mp.UploadID, ObjectOptions{}); err != nil {
t.Fatalf("released read-quorum availability regressed: %v", err)
}
for _, disk := range disks[offline:] {
_, err := disk.ReadVersion(t.Context(), bucket, minioMetaMultipartBucket, set.getUploadIDDir(bucket, key, mp.UploadID), "", ReadOptions{})
if !errors.Is(err, errFileNotFound) {
t.Fatalf("online replica not cleaned: %v", err)
}
}
})
}
}
}
func TestMultipartAbortLegacyPoolOrder(t *testing.T) {
for _, owner := range []int{0, 1} {
t.Run(fmt.Sprintf("owner=%d", owner), func(t *testing.T) {
z, bucket := consistencyPools(t)
setMultipartListingTestMode(t, true)
mp, err := z.serverPools[owner].NewMultipartUpload(t.Context(), bucket, "a", ObjectOptions{})
if err != nil {
t.Fatal(err)
}
emptySet := z.serverPools[1-owner].getHashedSet("a")
original := emptySet.getDisks
t.Cleanup(func() { emptySet.getDisks = original })
emptySet.getDisks = func() []StorageAPI { return make([]StorageAPI, emptySet.setDriveCount) }
err = z.AbortMultipartUpload(t.Context(), bucket, "a", mp.UploadID, ObjectOptions{})
if owner == 0 && err != nil {
t.Fatalf("unrelated later pool blocked legacy cancellation: %v", err)
}
if owner == 1 {
if err == nil || toAPIError(t.Context(), err).HTTPStatusCode != 503 {
t.Fatalf("unknown earlier pool must retain released error behavior: %v", err)
}
if _, err := z.serverPools[owner].GetMultipartInfo(t.Context(), bucket, "a", mp.UploadID, ObjectOptions{}); err != nil {
t.Fatalf("later pool visited despite earlier error: %v", err)
}
emptySet.getDisks = original
if err := z.AbortMultipartUpload(t.Context(), bucket, "a", mp.UploadID, ObjectOptions{}); err != nil {
t.Fatal(err)
}
}
})
}
}
func TestMultipartAbortMinorityLegacyCleanup(t *testing.T) {
for _, failDelete := range []bool{false, true} {
t.Run(fmt.Sprintf("delete-fails=%v", failDelete), func(t *testing.T) {
z, set, bucket := multipartListingFixture(t)
mp, err := z.NewMultipartUpload(t.Context(), bucket, "old", ObjectOptions{})
if err != nil {
t.Fatal(err)
}
path := set.getUploadIDDir(bucket, "old", mp.UploadID)
original := set.getDisks
disks := original()
fi, err := disks[0].ReadVersion(t.Context(), bucket, minioMetaMultipartBucket, path, "", ReadOptions{})
if err != nil {
t.Fatal(err)
}
delete(fi.Metadata, multipartMetaBucket)
delete(fi.Metadata, multipartMetaObject)
if err := disks[0].WriteMetadata(t.Context(), bucket, minioMetaMultipartBucket, path, fi); err != nil {
t.Fatal(err)
}
for _, d := range disks[1:] {
if err := d.Delete(t.Context(), minioMetaMultipartBucket, path, DeleteOptions{Recursive: true}); err != nil {
t.Fatal(err)
}
}
report, err := z.multipartPreflight(t.Context())
if err != nil || report.Ready || report.LegacyUploads != 1 {
t.Fatalf("invalid legacy fixture: %+v %v", report, err)
}
if failDelete {
set.getDisks = func() []StorageAPI {
wrapped := append([]StorageAPI(nil), disks...)
wrapped[0] = multipartListingFaultDisk{StorageAPI: disks[0], delete: func(context.Context, string, string, DeleteOptions) error { return errFaultyDisk }}
return wrapped
}
t.Cleanup(func() { set.getDisks = original })
err := z.AbortMultipartUpload(t.Context(), bucket, "old", mp.UploadID, ObjectOptions{})
if err == nil || toAPIError(t.Context(), err).HTTPStatusCode != 503 {
t.Fatalf("empty drives masked failed deletion of the observed replica: %v", err)
}
if _, present := z.mpCache.Load(mp.UploadID); !present {
t.Fatal("failed cancellation evicted upload from cache")
}
set.getDisks = original
}
if err := z.AbortMultipartUpload(t.Context(), bucket, "old", mp.UploadID, ObjectOptions{}); err != nil {
t.Fatal(err)
}
report, err = z.multipartPreflight(t.Context())
if err != nil || !report.Ready || report.LegacyUploads != 0 {
t.Fatalf("acknowledged cleanup left online legacy remnants: %+v %v", report, err)
}
if _, present := z.mpCache.Load(mp.UploadID); present {
t.Fatal("successful cancellation retained cache entry")
}
})
}
}
func TestMultipartAbortWrongTargetKeepsCache(t *testing.T) {
for _, legacy := range []bool{true, false} {
t.Run(fmt.Sprintf("legacy=%v", legacy), func(t *testing.T) {
z, _, bucket := multipartListingFixture(t)
setMultipartListingTestMode(t, legacy)
const other = "multipart-other-bucket"
if err := z.MakeBucket(t.Context(), other, MakeBucketOptions{}); err != nil {
t.Fatal(err)
}
mp, err := z.NewMultipartUpload(t.Context(), bucket, "valid-key", ObjectOptions{})
if err != nil {
t.Fatal(err)
}
for _, target := range [][2]string{{bucket, "wrong-key"}, {other, "valid-key"}} {
err := z.AbortMultipartUpload(t.Context(), target[0], target[1], mp.UploadID, ObjectOptions{})
var invalid InvalidUploadID
if !errors.As(err, &invalid) {
t.Fatalf("wrong target result: %v", err)
}
if _, present := z.mpCache.Load(mp.UploadID); !present {
t.Fatal("wrong-target abort evicted another upload")
}
if _, err := z.GetMultipartInfo(t.Context(), bucket, "valid-key", mp.UploadID, ObjectOptions{}); err != nil {
t.Fatalf("wrong-target abort harmed valid upload: %v", err)
}
}
if err := z.AbortMultipartUpload(t.Context(), bucket, "valid-key", mp.UploadID, ObjectOptions{}); err != nil {
t.Fatal(err)
}
var invalid InvalidUploadID
if err := z.AbortMultipartUpload(t.Context(), bucket, "valid-key", mp.UploadID, ObjectOptions{}); !errors.As(err, &invalid) {
t.Fatalf("repeat abort: %v", err)
}
})
}
}
func TestMultipartAbortRejectsUnsafeID(t *testing.T) {
for _, legacy := range []bool{true, false} {
t.Run(fmt.Sprintf("legacy=%v", legacy), func(t *testing.T) {
z, _, bucket := multipartListingFixture(t)
setMultipartListingTestMode(t, legacy)
mp, err := z.NewMultipartUpload(t.Context(), bucket, "keep", ObjectOptions{})
if err != nil {
t.Fatal(err)
}
for _, suffix := range []string{".", "..", "../other", "a/b", "a\\b", ""} {
id := base64.RawURLEncoding.EncodeToString([]byte("deployment." + suffix))
err := z.AbortMultipartUpload(t.Context(), bucket, "keep", id, ObjectOptions{})
var invalid InvalidUploadID
if !errors.As(err, &invalid) {
t.Fatalf("unsafe ID accepted: suffix=%q err=%v", suffix, err)
}
if _, err := z.GetMultipartInfo(t.Context(), bucket, "keep", mp.UploadID, ObjectOptions{}); err != nil {
t.Fatalf("valid upload harmed by rejected ID: %v", err)
}
}
})
}
}
func TestMultipartAbortPeerNotification(t *testing.T) {
z, set, bucket := multipartListingFixture(t)
tg, err := grid.SetupTestGrid(2)
if err != nil {
t.Fatal(err)
}
t.Cleanup(tg.Cleanup)
var notifications atomic.Int32
if err := cleanupUploadIDCacheMetaRPC.Register(tg.Managers[1], func(*grid.MSS) (grid.NoPayload, *grid.RemoteErr) {
notifications.Add(1)
return grid.NoPayload{}, nil
}); err != nil {
t.Fatal(err)
}
host, err := xnet.ParseHost(strings.TrimPrefix(tg.Hosts[1], "http://"))
if err != nil {
t.Fatal(err)
}
previous := globalNotificationSys
globalNotificationSys = &NotificationSys{peerClients: []*peerRESTClient{{
host: host,
gridConn: func() *grid.Connection { return tg.Managers[0].Connection(tg.Hosts[1]) },
}}}
t.Cleanup(func() { globalNotificationSys = previous })
// Use the real distributed lock implementation: Unlock cancels its derived
// context. Local locks do not, and would hide a broken notification context.
oldMutex, oldLockers := set.nsMutex, set.getLockers
lockers := []dsync.NetLocker{newLocker(), newLocker()}
set.nsMutex = newNSLock(true)
set.getLockers = func() ([]dsync.NetLocker, string) { return lockers, "multipart-test" }
t.Cleanup(func() { set.nsMutex, set.getLockers = oldMutex, oldLockers })
for _, legacy := range []bool{true, false} {
t.Run(fmt.Sprintf("legacy=%v", legacy), func(t *testing.T) {
setMultipartListingTestMode(t, legacy)
for range 4 {
mp, err := z.NewMultipartUpload(t.Context(), bucket, "valid", ObjectOptions{})
if err != nil {
t.Fatal(err)
}
before := notifications.Load()
var invalid InvalidUploadID
if err := z.AbortMultipartUpload(t.Context(), bucket, "wrong", mp.UploadID, ObjectOptions{}); !errors.As(err, &invalid) {
t.Fatalf("wrong target: %v", err)
}
if notifications.Load() != before {
t.Fatal("wrong-target abort sent a destructive peer notification")
}
if err := z.AbortMultipartUpload(t.Context(), bucket, "valid", mp.UploadID, ObjectOptions{}); err != nil {
t.Fatal(err)
}
if notifications.Load() != before+1 {
t.Fatal("successful abort lost peer notification after unlocking")
}
}
})
}
}
func TestMultipartAbortStrictMajority(t *testing.T) {
z, set, bucket := multipartListingFixture(t)
mp, err := z.NewMultipartUpload(t.Context(), bucket, "a", ObjectOptions{})
if err != nil {
t.Fatal(err)
}
original := set.getDisks
disks := original()
set.getDisks = func() []StorageAPI {
wrapped := append([]StorageAPI(nil), disks...)
wrapped[0] = multipartListingFaultDisk{StorageAPI: disks[0], delete: func(context.Context, string, string, DeleteOptions) error { return errFaultyDisk }}
return wrapped
}
t.Cleanup(func() { set.getDisks = original })
if err := z.AbortMultipartUpload(t.Context(), bucket, "a", mp.UploadID, ObjectOptions{}); err != nil {
t.Fatalf("ordinary majority cancellation was tightened: %v", err)
}
// One failed deletion is tolerated in the ordinary majority path. Retrying
// after recovery must also clean the now-minority remnant.
if _, err := disks[0].ReadVersion(t.Context(), bucket, minioMetaMultipartBucket, set.getUploadIDDir(bucket, "a", mp.UploadID), "", ReadOptions{}); err != nil {
t.Fatalf("fault fixture did not leave a remnant: %v", err)
}
set.getDisks = original
if err := z.AbortMultipartUpload(t.Context(), bucket, "a", mp.UploadID, ObjectOptions{}); err != nil {
t.Fatal(err)
}
}
+37 -12
View File
@@ -1753,11 +1753,11 @@ func (er erasureObjects) CompleteMultipartUpload(ctx context.Context, bucket str
return fi.ToObjectInfo(bucket, object, opts.Versioned || opts.VersionSuspended), nil return fi.ToObjectInfo(bucket, object, opts.Versioned || opts.VersionSuspended), nil
} }
// abortMultipartUpload confirms absence on a strict majority of this set. // abortMultipartUpload retains read-quorum validation and best-effort cleanup
// Unlike an existence read followed by best-effort deletion, it also permits // in legacy mode. Strict mode requires majority deletion acknowledgements and
// retrying a partial deletion which no longer has a readable metadata quorum. // permits retrying remnants below read quorum. Neither mode fences creation
// This does not fence creation writes still executing after a storage timeout. // writes still executing after a storage timeout.
func (er erasureObjects) abortMultipartUpload(ctx context.Context, bucket, object, uploadID string, opts ObjectOptions) (bool, error) { func (er erasureObjects) abortMultipartUpload(ctx context.Context, bucket, object, uploadID string, opts ObjectOptions, legacy bool) (bool, error) {
if !opts.NoAuditLog { if !opts.NoAuditLog {
auditObjectErasureSet(ctx, "AbortMultipartUpload", object, &er) auditObjectErasureSet(ctx, "AbortMultipartUpload", object, &er)
} }
@@ -1769,21 +1769,34 @@ func (er erasureObjects) abortMultipartUpload(ctx context.Context, bucket, objec
if !ok || internalID == "" || internalID == "." || internalID == ".." || strings.ContainsAny(internalID, "/\\") { if !ok || internalID == "" || internalID == "." || internalID == ".." || strings.ContainsAny(internalID, "/\\") {
return false, InvalidUploadID{Bucket: bucket, Object: object, UploadID: uploadID} return false, InvalidUploadID{Bucket: bucket, Object: object, UploadID: uploadID}
} }
if legacy {
// Keep the released read-quorum and best-effort cleanup behavior.
// The upload ID safety check above applies to both modes.
defer er.deleteAll(ctx, minioMetaMultipartBucket, er.getUploadIDDir(bucket, object, uploadID))
_, _, err := er.checkUploadIDExists(ctx, bucket, object, uploadID, false)
err = toObjectErr(err, bucket, object, uploadID)
if _, absent := err.(InvalidUploadID); absent {
return false, nil
}
return err == nil, err
}
disks := er.getDisks() disks := er.getDisks()
uploadPath := er.getUploadIDDir(bucket, object, uploadID) uploadPath := er.getUploadIDDir(bucket, object, uploadID)
_, errs := readAllFileInfo(ctx, disks, bucket, minioMetaMultipartBucket, uploadPath, "", false, false) _, errs := readAllFileInfo(ctx, disks, bucket, minioMetaMultipartBucket, uploadPath, "", false, false)
quorum := er.setDriveCount/2 + 1 quorum := er.setDriveCount/2 + 1
found, absent := false, 0 found, absent := false, 0
for _, err := range errs { observed := make([]bool, len(disks))
for i, err := range errs {
switch { switch {
case err == nil, errors.Is(err, errFileCorrupt): case err == nil, errors.Is(err, errFileCorrupt):
found = true found = true
observed[i] = true
case errors.Is(err, errFileNotFound), errors.Is(err, errFileVersionNotFound): case errors.Is(err, errFileNotFound), errors.Is(err, errFileVersionNotFound):
absent++ absent++
} }
} }
if absent >= quorum { if absent >= quorum && !found {
return found, nil return false, nil
} }
if !found { if !found {
return false, toObjectErr(errErasureReadQuorum, bucket, object, uploadID) return false, toObjectErr(errErasureReadQuorum, bucket, object, uploadID)
@@ -1801,13 +1814,25 @@ func (er erasureObjects) abortMultipartUpload(ctx context.Context, bucket, objec
return err return err
}, i) }, i)
} }
return true, toObjectErr(reduceWriteQuorumErrs(ctx, g.Wait(), nil, quorum), bucket, object, uploadID) deleteErrs := g.Wait()
if err := reduceWriteQuorumErrs(ctx, deleteErrs, nil, quorum); err != nil {
return true, toObjectErr(err, bucket, object, uploadID)
}
if absent >= quorum {
// Missing disks must not mask a failed cleanup of known remnants.
for i, found := range observed {
if found && deleteErrs[i] != nil {
return true, toObjectErr(errErasureWriteQuorum, bucket, object, uploadID)
}
}
}
return true, nil
} }
// AbortMultipartUpload confirms logical cancellation. Offline part data may // AbortMultipartUpload cancels an upload using the configured mode. Offline
// still need stale-upload cleanup after its drives return. // part data may still need stale-upload cleanup after its drives return.
func (er erasureObjects) AbortMultipartUpload(ctx context.Context, bucket, object, uploadID string, opts ObjectOptions) (err error) { func (er erasureObjects) AbortMultipartUpload(ctx context.Context, bucket, object, uploadID string, opts ObjectOptions) (err error) {
found, err := er.abortMultipartUpload(ctx, bucket, object, uploadID, opts) found, err := er.abortMultipartUpload(ctx, bucket, object, uploadID, opts, globalAPIConfig.getMultipartListingLegacy())
if err != nil { if err != nil {
return err return err
} }
+11 -5
View File
@@ -2160,13 +2160,14 @@ func (z *erasureServerPools) AbortMultipartUpload(ctx context.Context, bucket, o
return toObjectErr(err, bucket) return toObjectErr(err, bucket)
} }
defer func() { // Unlock cancels the derived lock context before this notification runs.
_, absent := err.(InvalidUploadID) // Keep the request context so successful cancellation reaches peer caches.
if err == nil || absent { defer func(ctx context.Context) {
if err == nil {
z.mpCache.Delete(uploadID) z.mpCache.Delete(uploadID)
globalNotificationSys.DeleteUploadID(ctx, uploadID) globalNotificationSys.DeleteUploadID(ctx, uploadID)
} }
}() }(ctx)
lk := z.NewNSLock(bucket, pathJoin(object, uploadID)) lk := z.NewNSLock(bucket, pathJoin(object, uploadID))
lkctx, err := lk.GetLock(ctx, globalOperationTimeout) lkctx, err := lk.GetLock(ctx, globalOperationTimeout)
@@ -2176,13 +2177,18 @@ func (z *erasureServerPools) AbortMultipartUpload(ctx context.Context, bucket, o
ctx = lkctx.Context() ctx = lkctx.Context()
defer lk.Unlock(lkctx) defer lk.Unlock(lkctx)
legacy := globalAPIConfig.getMultipartListingLegacy()
found := false found := false
var firstErr error var firstErr error
for idx, pool := range z.serverPools { for idx, pool := range z.serverPools {
if z.IsSuspended(idx) { if z.IsSuspended(idx) {
continue continue
} }
poolFound, err := pool.getHashedSet(object).abortMultipartUpload(ctx, bucket, object, uploadID, opts) poolFound, err := pool.getHashedSet(object).abortMultipartUpload(ctx, bucket, object, uploadID, opts, legacy)
if legacy && (poolFound || err != nil) {
// Match the released first-matching-pool behavior.
return err
}
found = found || poolFound found = found || poolFound
if err != nil && firstErr == nil { if err != nil && firstErr == nil {
firstErr = err firstErr = err
+3 -3
View File
@@ -50,7 +50,7 @@ type apiConfig struct {
transitionWorkers int transitionWorkers int
staleUploadsExpiry time.Duration staleUploadsExpiry time.Duration
multipartListingLegacy bool multipartListingStrict bool
staleUploadsCleanupInterval time.Duration staleUploadsCleanupInterval time.Duration
deleteCleanupInterval time.Duration deleteCleanupInterval time.Duration
enableODirect bool enableODirect bool
@@ -182,7 +182,7 @@ func (t *apiConfig) init(cfg api.Config, setDriveCounts []int, legacy bool) {
t.transitionWorkers = cfg.TransitionWorkers t.transitionWorkers = cfg.TransitionWorkers
t.staleUploadsExpiry = cfg.StaleUploadsExpiry t.staleUploadsExpiry = cfg.StaleUploadsExpiry
t.multipartListingLegacy = cfg.MultipartListing == "legacy" t.multipartListingStrict = cfg.MultipartListing == "strict"
t.deleteCleanupInterval = cfg.DeleteCleanupInterval t.deleteCleanupInterval = cfg.DeleteCleanupInterval
t.enableODirect = cfg.EnableODirect t.enableODirect = cfg.EnableODirect
t.gzipObjects = cfg.GzipObjects t.gzipObjects = cfg.GzipObjects
@@ -211,7 +211,7 @@ func (t *apiConfig) odirectEnabled() bool {
func (t *apiConfig) getMultipartListingLegacy() bool { func (t *apiConfig) getMultipartListingLegacy() bool {
t.mu.RLock() t.mu.RLock()
defer t.mu.RUnlock() defer t.mu.RUnlock()
return t.multipartListingLegacy return !t.multipartListingStrict
} }
func (t *apiConfig) shouldGzipObjects() bool { func (t *apiConfig) shouldGzipObjects() bool {
+3 -9
View File
@@ -47,6 +47,7 @@ func TestListMultipartUploadsS3Compatibility(t *testing.T) {
t.Fatal(err) t.Fatal(err)
} }
z := obj.(*erasureServerPools) z := obj.(*erasureServerPools)
setMultipartListingTestMode(t, false)
t.Cleanup(func() { t.Cleanup(func() {
z.Shutdown(t.Context()) z.Shutdown(t.Context())
removeRoots(dirs) removeRoots(dirs)
@@ -201,15 +202,7 @@ func TestListMultipartUploadsS3Compatibility(t *testing.T) {
if !errors.Is(err, errMultipartListingLegacy) { if !errors.Is(err, errMultipartListingLegacy) {
t.Fatalf("legacy strict listing: %v", err) t.Fatalf("legacy strict listing: %v", err)
} }
globalAPIConfig.mu.Lock() setMultipartListingTestMode(t, true)
oldLegacy := globalAPIConfig.multipartListingLegacy
globalAPIConfig.multipartListingLegacy = true
globalAPIConfig.mu.Unlock()
t.Cleanup(func() {
globalAPIConfig.mu.Lock()
globalAPIConfig.multipartListingLegacy = oldLegacy
globalAPIConfig.mu.Unlock()
})
legacy, err := z.ListMultipartUploads(t.Context(), bucket, legacyObject, "", "", "", 100) legacy, err := z.ListMultipartUploads(t.Context(), bucket, legacyObject, "", "", "", 100)
if err != nil { if err != nil {
t.Fatal(err) t.Fatal(err)
@@ -267,6 +260,7 @@ func TestPaginateMultipartUploads(t *testing.T) {
func TestListMultipartUploadsGlobalPageAcrossPools(t *testing.T) { func TestListMultipartUploadsGlobalPageAcrossPools(t *testing.T) {
z, bucket := consistencyPools(t) z, bucket := consistencyPools(t)
setMultipartListingTestMode(t, false)
objects := []string{"a/one", "b/two", "c/three", "d/four"} objects := []string{"a/one", "b/two", "c/three", "d/four"}
for i, object := range objects { for i, object := range objects {
if _, err := z.serverPools[i%len(z.serverPools)].NewMultipartUpload(t.Context(), bucket, object, ObjectOptions{}); err != nil { if _, err := z.serverPools[i%len(z.serverPools)].NewMultipartUpload(t.Context(), bucket, object, ObjectOptions{}); err != nil {
+2
View File
@@ -1,5 +1,7 @@
# Compression Guide # Compression Guide
For SILO's current SSE-C compression and historical-object boundary, see the [maintained encryption guide](https://silo.pgsty.com/administration/server-side-encryption/server-side-encryption-sse-c/) and [SSE-C replica design](https://silo.pgsty.com/blog/design/ssec-replica-integrity/). Check the documented release boundary before applying main-branch behavior to an older binary.
Silo server allows streaming compression to ensure efficient disk space usage. Silo server allows streaming compression to ensure efficient disk space usage.
Compression happens inflight, i.e objects are compressed before being written to disk(s). Compression happens inflight, i.e objects are compressed before being written to disk(s).
Silo uses [`klauspost/compress/s2`](https://github.com/klauspost/compress/tree/master/s2) Silo uses [`klauspost/compress/s2`](https://github.com/klauspost/compress/tree/master/s2)
+2
View File
@@ -1,5 +1,7 @@
# Federation Quickstart Guide *Federation feature is deprecated and should be avoided for future deployments* # Federation Quickstart Guide *Federation feature is deprecated and should be avoided for future deployments*
The maintained [federated CopyObject design](https://silo.pgsty.com/blog/design/federated-copy-object/) records the destination encryption, checksum, Object Lock and committed-response contract, including its Server release boundary.
This document explains how to configure Silo with `Bucket lookup from DNS` style federation. This document explains how to configure Silo with `Bucket lookup from DNS` style federation.
## Cross-deployment copy behavior ## Cross-deployment copy behavior
+7 -5
View File
@@ -135,7 +135,6 @@ var (
Key: apiStaleUploadsExpiry, Key: apiStaleUploadsExpiry,
Value: "24h", Value: "24h",
}, },
config.KV{Key: apiMultipartListing, Value: "strict"},
config.KV{ config.KV{
Key: apiDeleteCleanupInterval, Key: apiDeleteCleanupInterval,
Value: "5m", Value: "5m",
@@ -210,6 +209,7 @@ func LookupConfig(kvs config.KVS) (cfg Config, err error) {
apiReplicationWorkers, apiReplicationWorkers,
apiReplicationFailedWorkers, apiReplicationFailedWorkers,
"expiry_workers", "expiry_workers",
apiMultipartListing, // Ignore the retired shared-config key during migration.
} }
disableODirect := env.Get(EnvAPIDisableODirect, kvs.Get(apiDisableODirect)) == config.EnableOn disableODirect := env.Get(EnvAPIDisableODirect, kvs.Get(apiDisableODirect)) == config.EnableOn
@@ -324,10 +324,6 @@ func LookupConfig(kvs config.KVS) (cfg Config, err error) {
return cfg, err return cfg, err
} }
cfg.StaleUploadsExpiry = staleUploadsExpiry cfg.StaleUploadsExpiry = staleUploadsExpiry
cfg.MultipartListing = env.Get(EnvAPIMultipartListing, kvs.GetWithDefault(apiMultipartListing, DefaultKVS))
if cfg.MultipartListing != "strict" && cfg.MultipartListing != "legacy" {
return cfg, fmt.Errorf("%s must be strict or legacy", apiMultipartListing)
}
cfg.SyncEvents = env.Get(EnvAPISyncEvents, kvs.Get(apiSyncEvents)) == config.EnableOn cfg.SyncEvents = env.Get(EnvAPISyncEvents, kvs.Get(apiSyncEvents)) == config.EnableOn
@@ -348,5 +344,11 @@ func LookupConfig(kvs config.KVS) (cfg Config, err error) {
cfg.ObjectMaxVersions = math.MaxInt64 cfg.ObjectMaxVersions = math.MaxInt64
} }
cfg.MultipartListing = env.Get(EnvAPIMultipartListing, "legacy")
if cfg.MultipartListing != "strict" && cfg.MultipartListing != "legacy" {
cfg.MultipartListing = "legacy"
return cfg, fmt.Errorf("%s must be strict or legacy; using legacy", EnvAPIMultipartListing)
}
return cfg, nil return cfg, nil
} }
+87 -14
View File
@@ -4,38 +4,111 @@
package api package api
import ( import (
"encoding/json"
"strings"
"testing" "testing"
"github.com/minio/minio/internal/config" "github.com/minio/minio/internal/config"
) )
func TestMultipartListingMigrationMode(t *testing.T) { func TestMultipartListingMigrationMode(t *testing.T) {
t.Setenv(EnvAPIRequestsMax, "")
t.Setenv(EnvAPISyncEvents, "")
t.Setenv(EnvAPIObjectMaxVersions, "")
for _, tc := range []struct { for _, tc := range []struct {
name, stored, override, want string name, stored, override, want string
invalid bool invalid bool
}{ }{
{name: "default", want: "strict"}, {name: "default", want: "legacy"},
{name: "explicit-migration", stored: "legacy", want: "legacy"}, {name: "explicit-legacy", override: "legacy", want: "legacy"},
{name: "environment-override", stored: "strict", override: "legacy", want: "legacy"}, {name: "explicit-strict", override: "strict", want: "strict"},
{name: "invalid-stored", stored: "automatic", invalid: true}, {name: "retired-stored-strict", stored: "strict", want: "legacy"},
{name: "invalid-environment", stored: "strict", override: "automatic", invalid: true}, {name: "retired-stored-legacy", stored: "legacy", want: "legacy"},
{name: "retired-stored-invalid", stored: "automatic", want: "legacy"},
{name: "environment-override", stored: "legacy", override: "strict", want: "strict"},
{name: "invalid-environment", stored: "strict", override: "automatic", want: "legacy", invalid: true},
} { } {
t.Run(tc.name, func(t *testing.T) { t.Run(tc.name, func(t *testing.T) {
t.Setenv(EnvAPIMultipartListing, tc.override) t.Setenv(EnvAPIMultipartListing, tc.override)
var kvs config.KVS kvs := DefaultKVS.Clone()
kvs.Set(apiRequestsMax, "17")
kvs.Set(apiSyncEvents, config.EnableOn)
kvs.Set(apiObjectMaxVersions, "71")
if tc.stored != "" { if tc.stored != "" {
kvs = config.KVS{{Key: apiMultipartListing, Value: tc.stored}} kvs.Set(apiMultipartListing, tc.stored)
} }
cfg, err := LookupConfig(kvs) cfg, err := LookupConfig(kvs)
if tc.invalid { if (err != nil) != tc.invalid || cfg.MultipartListing != tc.want {
if err == nil { t.Fatalf("mode=%q err=%v, want %q invalid=%v", cfg.MultipartListing, err, tc.want, tc.invalid)
t.Fatal("invalid migration mode silently accepted")
}
return
} }
if err != nil || cfg.MultipartListing != tc.want { if cfg.RequestsMax != 17 || !cfg.SyncEvents || cfg.ObjectMaxVersions != 71 {
t.Fatalf("mode=%q err=%v, want %q", cfg.MultipartListing, err, tc.want) t.Fatalf("migration setting discarded unrelated config: %+v", cfg)
} }
}) })
} }
} }
func TestMultipartListingConfigPersistence(t *testing.T) {
t.Setenv(EnvAPIRequestsMax, "")
t.Setenv(EnvAPISyncEvents, "")
t.Setenv(EnvAPIObjectMaxVersions, "")
t.Setenv(EnvAPIMultipartListing, "strict")
previous := config.DefaultKVS
config.DefaultKVS = map[string]config.KVS{config.APISubSys: DefaultKVS}
t.Cleanup(func() { config.DefaultKVS = previous })
for _, help := range Help {
if help.Key == apiMultipartListing {
t.Fatal("process-only setting advertised as shared config")
}
}
for _, input := range []string{
"",
"api requests_max=17 sync_events=on object_max_versions=71",
`api requests_max=17 comment="later disable multipart_listing=strict"`,
} {
t.Run(input, func(t *testing.T) {
c := config.New().Clone()
if _, err := c.ReadConfig(strings.NewReader(input)); err != nil {
t.Fatal(err)
}
// Exercise the JSON save/load and Merge paths used for shared config.
data, err := json.Marshal(c)
if err != nil {
t.Fatal(err)
}
var loaded config.Config
if err = json.Unmarshal(data, &loaded); err != nil {
t.Fatal(err)
}
kvs := loaded.Merge()[config.APISubSys][config.Default]
if _, present := kvs.Lookup(apiMultipartListing); present {
t.Fatal("process-only setting persisted in shared config")
}
cfg, err := LookupConfig(kvs)
if err != nil || cfg.MultipartListing != "strict" {
t.Fatalf("environment mode lost: %+v %v", cfg, err)
}
if input != "" && cfg.RequestsMax != 17 {
t.Fatalf("unrelated setting lost: %+v", cfg)
}
if strings.Contains(input, "comment=") && kvs.Get(config.Comment) != "later disable multipart_listing=strict" {
t.Fatalf("literal comment changed: %q", kvs.Get(config.Comment))
}
})
}
// A key persisted by a previous development build is tolerated, retained for
// explicit cleanup, and removable without resetting other API settings.
c := config.New().Clone()
kvs := c[config.APISubSys][config.Default]
kvs.Set(apiMultipartListing, "strict")
kvs.Set(apiRequestsMax, "17")
c[config.APISubSys][config.Default] = kvs
c = c.Merge()
if err := c.DelKVS("api multipart_listing"); err != nil {
t.Fatal(err)
}
kvs = c.Merge()[config.APISubSys][config.Default]
if _, present := kvs.Lookup(apiMultipartListing); present || kvs.Get(apiRequestsMax) != "17" {
t.Fatalf("targeted reset changed unrelated config or restored retired key: %v", kvs)
}
}
-6
View File
@@ -26,12 +26,6 @@ var (
// Help holds configuration keys and their default values for api subsystem. // Help holds configuration keys and their default values for api subsystem.
Help = config.HelpKVS{ Help = config.HelpKVS{
config.HelpKV{
Key: apiMultipartListing,
Description: "multipart listing mode: strict, or temporary legacy mode during coordinated upgrade" + defaultHelpPostfix(apiMultipartListing),
Type: "string",
Optional: true,
},
config.HelpKV{ config.HelpKV{
Key: apiRequestsMax, Key: apiRequestsMax,
Description: `set the maximum number of concurrent requests (default: auto)`, Description: `set the maximum number of concurrent requests (default: auto)`,