fix(decom): repeat the walk of a bucket while it removes source versions

Signed-off-by: Denis Babchuk <babchukdenis@icloud.com>

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) {

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

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

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,

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")
 			}

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 {