From b404f787643b51813e6f9acb0fa66bde3c67890d Mon Sep 17 00:00:00 2001 From: Denis Babchuk Date: Mon, 9 Nov 2026 16:20:15 +0300 Subject: [PATCH 1/2] test(decom): cover keys below a key that the walk does not see Signed-off-by: Denis Babchuk --- ...asure-server-pool-decom-keys-below_test.go | 66 +++++++++++++++++++ 1 file changed, 66 insertions(+) create mode 100644 cmd/erasure-server-pool-decom-keys-below_test.go diff --git a/cmd/erasure-server-pool-decom-keys-below_test.go b/cmd/erasure-server-pool-decom-keys-below_test.go new file mode 100644 index 000000000..7435bd6a4 --- /dev/null +++ b/cmd/erasure-server-pool-decom-keys-below_test.go @@ -0,0 +1,66 @@ +// Copyright (c) 2026 Denis Babchuk +// +// This file is part of MinIO Object Storage stack +// +// This program is free software: you can redistribute it and/or modify +// it under the terms of the GNU Affero General Public License as published by +// the Free Software Foundation, either version 3 of the License, or +// (at your option) any later version. +// +// This program is distributed in the hope that it will be useful +// but WITHOUT ANY WARRANTY; without even the implied warranty of +// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +// GNU Affero General Public License for more details. +// +// You should have received a copy of the GNU Affero General Public License +// along with this program. If not, see . + +package cmd + +import "testing" + +// A key below another key ("k/b" under "k") is a directory inside the +// directory of that key. The listing treats the directory of an object as a +// leaf, so a decommission walk does not see "k/b" while "k" is still in the +// pool. The walk has to be repeated once the keys above are gone, or the +// check after the bucket finds them and the decommission fails. + +func keysBelowDecommission(t *testing.T, z *erasureServerPools) PoolDecommissionInfo { + t.Helper() + if err := z.StartDecommission(t.Context(), 0); err != nil { + t.Fatal(err) + } + z.doDecommissionInRoutine(t.Context(), 0) + z.poolMetaMutex.RLock() + defer z.poolMetaMutex.RUnlock() + return *z.poolMeta.Pools[0].Decommission +} + +func keysBelowPools(t *testing.T) (*erasureServerPools, string, []string) { + t.Helper() + z, _ := migrationTestPools(t) + const bucket = "keys-below" + if err := z.MakeBucket(t.Context(), bucket, MakeBucketOptions{VersioningEnabled: true}); err != nil { + t.Fatal(err) + } + keys := []string{"k", "k/b", "k/b/c"} + for _, key := range keys { + putConsistencyObject(t, z, bucket, key, 0, key, ObjectOptions{Versioned: true}) + } + return z, bucket, keys +} + +func TestDecommissionMovesKeysBelowAKey(t *testing.T) { + z, bucket, keys := keysBelowPools(t) + if info := keysBelowDecommission(t, z); !info.Complete || info.Failed { + t.Errorf("decommission: complete %v, failed %v; want complete", info.Complete, info.Failed) + } + for _, key := range keys { + if _, err := z.serverPools[0].GetObjectInfo(t.Context(), bucket, key, ObjectOptions{}); !isErrObjectNotFound(err) { + t.Errorf("%s still in pool 0: %v", key, err) + } + if body := migrationTestBody(t, z.GetObjectNInfo, bucket, key, ""); body != key { + t.Errorf("%s reads %q", key, body) + } + } +} From ce5d19107707c44973c75f7999fbf2fa8606d142 Mon Sep 17 00:00:00 2001 From: Denis Babchuk Date: Mon, 9 Nov 2026 16:55:38 +0300 Subject: [PATCH 2/2] fix(decom): repeat the walk of a bucket while it removes source versions Signed-off-by: Denis Babchuk --- cmd/batch-authz-review6_test.go | 2 +- cmd/erasure-server-pool-consistency.go | 22 ++++++----- cmd/erasure-server-pool-decom.go | 38 +++++++++++++++---- ...re-server-pool-lifecycle-authority_test.go | 2 +- ...asure-server-pool-migration-delete_test.go | 2 +- cmd/erasure-server-pool-rebalance.go | 2 +- 6 files changed, 48 insertions(+), 20 deletions(-) diff --git a/cmd/batch-authz-review6_test.go b/cmd/batch-authz-review6_test.go index 7fefe6f74..c11a84c4d 100644 --- a/cmd/batch-authz-review6_test.go +++ b/cmd/batch-authz-review6_test.go @@ -330,7 +330,7 @@ func TestBatchReviewRotationKeepsMigrationCleanup(t *testing.T) { t.Errorf("rotation changed a draining source after its version was copied (rotation error: %v)", rotationErr) } snapshot := []FileInfo{{VersionID: before.VersionID, ModTime: beforeInfo.ModTime}} - if err := z.removeMovedVersions(t.Context(), bucket, object, 0, snapshot, nil); err != nil { + if _, err := z.removeMovedVersions(t.Context(), bucket, object, 0, snapshot, nil); err != nil { t.Fatalf("rotation prevented migration cleanup: %v", err) } if _, err := z.serverPools[0].GetObjectInfo(t.Context(), bucket, object, ObjectOptions{VersionID: before.VersionID}); !isErrObjectNotFound(err) && !isErrVersionNotFound(err) { diff --git a/cmd/erasure-server-pool-consistency.go b/cmd/erasure-server-pool-consistency.go index f44e2ce1f..79ba0f01c 100644 --- a/cmd/erasure-server-pool-consistency.go +++ b/cmd/erasure-server-pool-consistency.go @@ -271,16 +271,17 @@ func (z *erasureServerPools) applyMigrationExpiry(ctx context.Context, bucket, o // in the source until the ordinary expiry path has deleted them: that path // owns replication state and tier cleanup, and it re-evaluates the current // lifecycle rule. A migration cleanup must never turn an asynchronous ILM -// decision into an unconditional physical purge. -func (z *erasureServerPools) removeMovedVersions(ctx context.Context, bucket, object string, source int, versions []FileInfo, expired []migrationExpiredVersion) error { +// decision into an unconditional physical purge. It returns how many source +// versions it deleted. +func (z *erasureServerPools) removeMovedVersions(ctx context.Context, bucket, object string, source int, versions []FileInfo, expired []migrationExpiredVersion) (removed int, err error) { if err := z.applyMigrationExpiry(ctx, bucket, object, source, expired); err != nil { - return err + return 0, err } object = encodeDirObject(object) lk := z.NewNSLock(bucket, object) lkctx, err := lk.GetLock(ctx, globalOperationTimeout) if err != nil { - return err + return 0, err } ctx = lkctx.Context() defer lk.Unlock(lkctx) @@ -294,7 +295,7 @@ func (z *erasureServerPools) removeMovedVersions(ctx context.Context, bucket, ob continue } if err != nil { - return err + return removed, err } var src *ObjectInfo var retained []PoolObjInfo @@ -309,7 +310,7 @@ func (z *erasureServerPools) removeMovedVersions(ctx context.Context, bucket, ob continue } if !src.ModTime.Equal(version.ModTime) || src.DeleteMarker != version.Deleted { - return PreConditionFailed{} + return removed, PreConditionFailed{} } if isExpiredVersion(expired, version) { // The logical expiry path owns this version. Keeping the source is @@ -319,17 +320,20 @@ func (z *erasureServerPools) removeMovedVersions(ctx context.Context, bucket, ob // Successful movement is not enough if the destination changed again. // Keep the source unless another copy still preserves its read result. if len(retained) == 0 || retained[0].ObjInfo.ModTime.Before(src.ModTime) { - return PreConditionFailed{} + return removed, PreConditionFailed{} } _, err = z.serverPools[source].DeleteObject(ctx, bucket, object, ObjectOptions{ VersionID: vid, NoLock: true, NoAuditLog: true, SkipFreeVersion: true, // preserve the old prefix cleanup's tier-GC behavior }) if err != nil && !isErrObjectNotFound(err) && !isErrVersionNotFound(err) { - return err + return removed, err + } + if err == nil { + removed++ } } - return nil + return removed, nil } // ilmDeletesVersionOnly reports whether the ILM action queued for a version diff --git a/cmd/erasure-server-pool-decom.go b/cmd/erasure-server-pool-decom.go index 16ea90b4d..4f1614380 100644 --- a/cmd/erasure-server-pool-decom.go +++ b/cmd/erasure-server-pool-decom.go @@ -28,6 +28,7 @@ import ( "slices" "sort" "strings" + "sync/atomic" "time" "github.com/dustin/go-humanize" @@ -752,7 +753,25 @@ func (set *erasureObjects) listObjectsToDecommission(ctx context.Context, bi dec return err } +// decommissionPool moves bucket bi out of pool idx. A key below another key +// ("k/b" under "k") is a directory inside the directory of that key, and the +// listing does not descend into the directory of an object: a walk sees "k/b" +// only once "k" has left the pool. The walk is repeated while it removes +// source versions, so each pass moves the keys that the previous one +// uncovered. Each repeat needs a version removed from the pool, which takes +// no new writes, so the repeats end. func (z *erasureServerPools) decommissionPool(ctx context.Context, idx int, pool *erasureSets, bi decomBucketInfo) error { + for { + removed, err := z.decommissionPoolWalk(ctx, idx, pool, bi) + if err != nil || removed == 0 || contextCanceled(ctx) { + return err + } + } +} + +// decommissionPoolWalk walks bucket bi of pool idx once and returns the number +// of keys whose source versions it removed. +func (z *erasureServerPools) decommissionPoolWalk(ctx context.Context, idx int, pool *erasureSets, bi decomBucketInfo) (int64, error) { ctx = logger.SetReqInfo(ctx, &logger.ReqInfo{}) const envDecomWorkers = "_MINIO_DECOMMISSION_WORKERS" @@ -766,9 +785,11 @@ func (z *erasureServerPools) decommissionPool(ctx context.Context, idx int, pool // add that many extra workers. workerSize += len(pool.sets) + var removed atomic.Int64 + wk, err := workers.New(workerSize) if err != nil { - return err + return 0, err } var vc *versioning.Versioning @@ -779,24 +800,24 @@ func (z *erasureServerPools) decommissionPool(ctx context.Context, idx int, pool if bi.Name != minioMetaBucket { vc, err = globalBucketVersioningSys.Get(bi.Name) if err != nil { - return err + return 0, err } // Check if the current bucket has a configured lifecycle policy lc, lifecycleUpdatedAt, err = globalBucketMetadataSys.GetLifecycleConfig(bi.Name) if err != nil && !errors.Is(err, BucketLifecycleNotFound{Bucket: bi.Name}) { - return err + return 0, err } // Check if bucket is object locked. lr, err = globalBucketObjectLockSys.Get(bi.Name) if err != nil { - return err + return 0, err } rcfg, err = getReplicationConfig(ctx, bi.Name) if err != nil { - return err + return 0, err } } @@ -1019,7 +1040,10 @@ func (z *erasureServerPools) decommissionPool(ctx context.Context, idx int, pool // if all versions were decommissioned, then we can delete the object versions. if decommissioned == len(fivs.Versions) { stopFn := globalDecommissionMetrics.log(decomMetricDecommissionRemoveObject, idx, bi.Name, entry.name) - err := z.removeMovedVersions(ctx, bi.Name, entry.name, idx, fivs.Versions, expired) + n, err := z.removeMovedVersions(ctx, bi.Name, entry.name, idx, fivs.Versions, expired) + if n > 0 { + removed.Add(1) + } if err == nil { err = z.removeFreeVersions(ctx, bi.Name, entry.name, idx, fivs.FreeVersions) } @@ -1062,7 +1086,7 @@ func (z *erasureServerPools) decommissionPool(ctx context.Context, idx int, pool }(setIdx) } wk.Wait() - return nil + return removed.Load(), nil } //msgp:ignore decomMetrics diff --git a/cmd/erasure-server-pool-lifecycle-authority_test.go b/cmd/erasure-server-pool-lifecycle-authority_test.go index e3bd7aaa5..82a13dde2 100644 --- a/cmd/erasure-server-pool-lifecycle-authority_test.go +++ b/cmd/erasure-server-pool-lifecycle-authority_test.go @@ -1666,7 +1666,7 @@ func TestPoolsMigrationLifecycleRefusedExpiryKeepsSource(t *testing.T) { t.Fatal(err) } version := FileInfo{Name: key, VersionID: id, ModTime: source.ModTime} - err = s.z.removeMovedVersions(ctx, s.bucket, key, 0, []FileInfo{version}, []migrationExpiredVersion{{ + _, err = s.z.removeMovedVersions(ctx, s.bucket, key, 0, []FileInfo{version}, []migrationExpiredVersion{{ version: version, event: lifecycle.Event{Action: lifecycle.DeleteAction}, lifecycleUpdatedAt: updatedAt, diff --git a/cmd/erasure-server-pool-migration-delete_test.go b/cmd/erasure-server-pool-migration-delete_test.go index 55bab37fc..557cc2efa 100644 --- a/cmd/erasure-server-pool-migration-delete_test.go +++ b/cmd/erasure-server-pool-migration-delete_test.go @@ -132,7 +132,7 @@ func TestPoolsMigrationCleanupRechecksSource(t *testing.T) { _, err := z.serverPools[1].DeleteObject(t.Context(), bucket, oi.Name, ObjectOptions{VersionID: nullVersionID}) mustIAM(t, err) } - err := z.removeMovedVersions(t.Context(), bucket, oi.Name, 0, snapshot, nil) + _, err := z.removeMovedVersions(t.Context(), bucket, oi.Name, 0, snapshot, nil) if change != "delete-uuid" && err == nil { t.Fatal("cleanup accepted changed or uncertain state") } diff --git a/cmd/erasure-server-pool-rebalance.go b/cmd/erasure-server-pool-rebalance.go index 81160fb0e..fb9eb996d 100644 --- a/cmd/erasure-server-pool-rebalance.go +++ b/cmd/erasure-server-pool-rebalance.go @@ -779,7 +779,7 @@ func (z *erasureServerPools) rebalanceBucket(ctx context.Context, bucket string, // if all versions were rebalanced, we can delete the object versions. if rebalanced == len(fivs.Versions) { stopFn := globalRebalanceMetrics.log(rebalanceMetricRebalanceRemoveObject, poolIdx, bucket, entry.name) - err := z.removeMovedVersions(ctx, bucket, entry.name, poolIdx, fivs.Versions, expired) + _, err := z.removeMovedVersions(ctx, bucket, entry.name, poolIdx, fivs.Versions, expired) stopFn(0, err) auditLogRebalance(ctx, "Rebalance:DeleteObject", bucket, entry.name, "", err) if err != nil {