diff --git a/cmd/erasure-server-pool-consistency.go b/cmd/erasure-server-pool-consistency.go index 2e8550503..71f99e3ee 100644 --- a/cmd/erasure-server-pool-consistency.go +++ b/cmd/erasure-server-pool-consistency.go @@ -273,15 +273,36 @@ func (z *erasureServerPools) applyMigrationExpiry(ctx context.Context, bucket, o // lifecycle rule. A migration cleanup must never turn an asynchronous ILM // decision into an unconditional physical purge. It returns how many source // versions were removed, including versions deleted by ILM before cleanup. -func (z *erasureServerPools) removeMovedVersions(ctx context.Context, bucket, object string, source int, versions []FileInfo, expired []migrationExpiredVersion) (removed int, err error) { +func (z *erasureServerPools) removeMovedVersions(ctx context.Context, bucket, object string, source int, versions []FileInfo, expired []migrationExpiredVersion) (int, error) { + err := z.retireMovedVersions(ctx, bucket, object, source, versions, expired) + removed, countErr := z.countAbsentMigrationVersions(ctx, bucket, object, source, versions) + if ctx.Err() != nil { + return 0, ctx.Err() + } + if err != nil { + return removed, err + } + return removed, countErr +} + +// retireMovedVersions does not infer physical progress from a pool-level read +// or delete result. Decommission verifies the entire walk snapshot once after +// this cleanup and tier GC, including versions concurrently deleted by clients. +func (z *erasureServerPools) retireMovedVersions(ctx context.Context, bucket, object string, source int, versions []FileInfo, expired []migrationExpiredVersion) error { + if err := ctx.Err(); err != nil { + return err + } if err := z.applyMigrationExpiry(ctx, bucket, object, source, expired); err != nil { - return 0, err + return err + } + if len(versions) == 0 { + return nil } object = encodeDirObject(object) lk := z.NewNSLock(bucket, object) lkctx, err := lk.GetLock(ctx, globalOperationTimeout) if err != nil { - return 0, err + return err } ctx = lkctx.Context() defer lk.Unlock(lkctx) @@ -293,15 +314,10 @@ func (z *erasureServerPools) removeMovedVersions(ctx context.Context, bucket, ob expiredVersion := isExpiredVersion(expired, version) copies, err := z.objectPoolInfos(ctx, bucket, object, ObjectOptions{VersionID: vid, NoAuditLog: true}) if isErrObjectNotFound(err) || isErrVersionNotFound(err) { - // ILM can remove the parent key before cleanup. Count that as - // progress so the next walk discovers keys below it. - if expiredVersion { - removed++ - } continue } if err != nil { - return removed, err + return err } var src *ObjectInfo var retained []PoolObjInfo @@ -313,13 +329,10 @@ func (z *erasureServerPools) removeMovedVersions(ctx context.Context, bucket, ob } } if src == nil { - if expiredVersion { - removed++ - } continue } if !src.ModTime.Equal(version.ModTime) || src.DeleteMarker != version.Deleted { - return removed, PreConditionFailed{} + return PreConditionFailed{} } if expiredVersion { // The logical expiry path owns this version. Keeping the source is @@ -329,20 +342,17 @@ 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 removed, PreConditionFailed{} + return 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 removed, err - } - if err == nil { - removed++ + return err } } - return removed, nil + return 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 4f1614380..6d4155002 100644 --- a/cmd/erasure-server-pool-decom.go +++ b/cmd/erasure-server-pool-decom.go @@ -28,6 +28,7 @@ import ( "slices" "sort" "strings" + "sync" "sync/atomic" "time" @@ -762,16 +763,25 @@ func (set *erasureObjects) listObjectsToDecommission(ctx context.Context, bi dec // no new writes, so the repeats end. func (z *erasureServerPools) decommissionPool(ctx context.Context, idx int, pool *erasureSets, bi decomBucketInfo) error { for { + if err := ctx.Err(); err != nil { + return err + } removed, err := z.decommissionPoolWalk(ctx, idx, pool, bi) - if err != nil || removed == 0 || contextCanceled(ctx) { + if ctx.Err() != nil { + return ctx.Err() + } + if err != nil || removed == 0 { return err } } } // decommissionPoolWalk walks bucket bi of pool idx once and returns the number -// of keys whose source versions it removed. +// of snapshot versions verified absent from every source drive afterwards. func (z *erasureServerPools) decommissionPoolWalk(ctx context.Context, idx int, pool *erasureSets, bi decomBucketInfo) (int64, error) { + if err := ctx.Err(); err != nil { + return 0, err + } ctx = logger.SetReqInfo(ctx, &logger.ReqInfo{}) const envDecomWorkers = "_MINIO_DECOMMISSION_WORKERS" @@ -786,6 +796,8 @@ func (z *erasureServerPools) decommissionPoolWalk(ctx context.Context, idx int, workerSize += len(pool.sets) var removed atomic.Int64 + var verifyErr error + var verifyOnce sync.Once wk, err := workers.New(workerSize) if err != nil { @@ -876,7 +888,7 @@ func (z *erasureServerPools) decommissionPoolWalk(ctx context.Context, idx int, // to create the appropriate stack. versionsSorter(fivs.Versions).reverse() - var decommissioned int + var candidates []FileInfo var expired []migrationExpiredVersion // The key's history in all pools, read once for its versions. var history migrationHistory @@ -884,8 +896,8 @@ func (z *erasureServerPools) decommissionPoolWalk(ctx context.Context, idx int, stopFn := globalDecommissionMetrics.log(decomMetricDecommissionObject, idx, bi.Name, version.Name, version.VersionID) // Apply lifecycle rules on the objects that are expired. if action := filterLifecycle(bi.Name, version.Name, version, &history); action.Delete() { - decommissioned++ if ilmDeletesVersionOnly(action) || (action.DeleteAll() && !version.IsRemote()) { + candidates = append(candidates, version) // Run the logical expiry before retiring the source. // A full-object expiry is safe here only for hot-tier // metadata; transitioned objects retain their tier cleanup @@ -933,7 +945,6 @@ func (z *erasureServerPools) decommissionPoolWalk(ctx context.Context, idx int, // This can happen when rebalance stop races with ongoing rebalance workers. // These rebalance failures can be ignored. if isErrObjectNotFound(err) || isErrVersionNotFound(err) || isDataMovementOverWriteErr(err) { - ignore = true stopFn(0, nil) continue } @@ -947,8 +958,7 @@ func (z *erasureServerPools) decommissionPoolWalk(ctx context.Context, idx int, z.poolMeta.CountItem(idx, 0, failure) z.poolMetaMutex.Unlock() if !failure { - // Success keep a count. - decommissioned++ + candidates = append(candidates, version) } auditLogDecom(ctx, "DecomCopyDeleteMarker", bi.Name, version.Name, versionID, err) continue @@ -1034,19 +1044,32 @@ func (z *erasureServerPools) decommissionPoolWalk(ctx context.Context, idx int, if failure { break // break out on first error } - decommissioned++ + candidates = append(candidates, version) } - // if all versions were decommissioned, then we can delete the object versions. - if decommissioned == len(fivs.Versions) { + // Retire only successful moves and eligible expiries. A concurrently + // deleted or failed sibling must not strand the successful copies. + // Skipped versions are checked for absence, never purged here. + { stopFn := globalDecommissionMetrics.log(decomMetricDecommissionRemoveObject, idx, bi.Name, entry.name) - n, err := z.removeMovedVersions(ctx, bi.Name, entry.name, idx, fivs.Versions, expired) - if n > 0 { - removed.Add(1) - } + err := z.retireMovedVersions(ctx, bi.Name, entry.name, idx, candidates, expired) if err == nil { err = z.removeFreeVersions(ctx, bi.Name, entry.name, idx, fivs.FreeVersions) } + // One metadata-only read of each source drive per entry, regardless + // of its version count. No read error can become progress or trigger + // another walk; leave the bucket queued for an explicit retry. + snapshot := append(fivs.Versions, fivs.FreeVersions...) + n, countErr := z.countAbsentMigrationVersions(ctx, bi.Name, entry.name, idx, snapshot) + removed.Add(int64(n)) + // The disposable usage cache is also excluded by the final + // verification. Its unreadability still never counts as progress. + if countErr != nil && !(bi.Name == minioMetaBucket && strings.Contains(entry.name, dataUsageCacheName)) { + verifyOnce.Do(func() { verifyErr = countErr }) + if err == nil { + err = countErr + } + } stopFn(0, err) auditLogDecom(ctx, "DecomDeleteObject", bi.Name, entry.name, "", err) if err != nil { @@ -1081,12 +1104,19 @@ func (z *erasureServerPools) decommissionPoolWalk(ctx context.Context, idx int, setN := humanize.Ordinal(setIdx + 1) retryDur := time.Duration(rand.Float64() * float64(5*time.Second)) decomLogOnceIf(ctx, fmt.Errorf("listing objects from %s set failed with %v, retrying in %v", setN, err, retryDur), "decom-listing-failed"+setN) - time.Sleep(retryDur) + select { + case <-ctx.Done(): + return + case <-time.After(retryDur): + } } }(setIdx) } wk.Wait() - return removed.Load(), nil + if err := ctx.Err(); err != nil { + return removed.Load(), err + } + return removed.Load(), verifyErr } //msgp:ignore decomMetrics @@ -1137,6 +1167,9 @@ func (m *decomMetrics) log(d decomMetric, poolIdx int, paths ...string) func(z i // decommission fails and a retry resumes with what is left. A canceled // context returns at once. func (z *erasureServerPools) decommissionInBackground(ctx context.Context, idx int) error { + if err := ctx.Err(); err != nil { + return err + } pool := z.serverPools[idx] z.poolMetaMutex.RLock() pending := z.poolMeta.PendingBuckets(idx) @@ -1175,7 +1208,7 @@ func (z *erasureServerPools) decommissionInBackground(ctx context.Context, idx i if err := z.decommissionPool(ctx, idx, pool, bucket); err != nil { stopFn(0, err) if contextCanceled(ctx) { - return err + return ctx.Err() } failed(bucket, err) continue @@ -1188,7 +1221,7 @@ func (z *erasureServerPools) decommissionInBackground(ctx context.Context, idx i if err := z.checkAfterDecomBucket(ctx, idx, bucket); err != nil { stopFn(0, err) if contextCanceled(ctx) { - return err + return ctx.Err() } failed(bucket, err) continue @@ -1201,6 +1234,9 @@ func (z *erasureServerPools) decommissionInBackground(ctx context.Context, idx i } z.poolMetaMutex.Unlock() } + if err := ctx.Err(); err != nil { + return err + } return firstErr } @@ -1241,28 +1277,54 @@ func (z *erasureServerPools) removeFreeVersions(ctx context.Context, bucket, obj // until the expiry path has performed its logical delete, replication, and // tier cleanup. func (z *erasureServerPools) checkAfterDecomBucket(ctx context.Context, idx int, bi decomBucketInfo) error { - versionsFound := 0 + if err := ctx.Err(); err != nil { + return err + } for _, set := range z.serverPools[idx].sets { - if err := set.listObjectsToDecommission(ctx, bi, func(entry metaCacheEntry) { - if !entry.isObject() { - return + disks := set.getDisks() + if len(disks) == 0 { + return toObjectErr(errErasureReadQuorum, bi.Name) + } + for _, disk := range disks { + if disk == nil { + return toObjectErr(errDiskNotFound, bi.Name) } - if bi.Name == minioMetaBucket && strings.Contains(entry.name, dataUsageCacheName) { - return + // Quorum listings omit minority versions, and WalkDir can skip + // unreadable metadata. Inspect physical directory entries instead: + // even a corrupt or truncated xl.meta must keep this pool queued. + pending := []string{bi.Prefix} + for len(pending) > 0 { + if err := ctx.Err(); err != nil { + return err + } + dir := pending[len(pending)-1] + pending = pending[:len(pending)-1] + entries, err := disk.ListDir(ctx, "", bi.Name, dir, -1) + if ctx.Err() != nil { + return ctx.Err() + } + if errors.Is(err, errFileNotFound) || errors.Is(err, errVolumeNotFound) { + continue // A concurrent deletion can remove the directory. + } + if err != nil { + return err + } + for _, entry := range entries { + name := pathJoin(dir, entry) + if bi.Name == minioMetaBucket && strings.Contains(name, dataUsageCacheName) { + continue + } + if entry == xlStorageFormatFile || entry == xlStorageFormatFileV1 { + return fmt.Errorf("object metadata %s remains in bucket %s after decommissioning", name, bi.Name) + } + if strings.HasSuffix(entry, SlashSeparator) { + pending = append(pending, name) + } + } } - fivs, err := entry.fileInfoVersions(bi.Name) - if err != nil { - return - } - versionsFound += len(fivs.Versions) + len(fivs.FreeVersions) - }); err != nil { - return err } } - if versionsFound > 0 { - return fmt.Errorf("at least %d object(s)/version(s) were found in bucket `%s` after decommissioning", versionsFound, bi.Name) - } - return nil + return ctx.Err() } func (z *erasureServerPools) checkAfterDecom(ctx context.Context, idx int) error { diff --git a/cmd/erasure-server-pool-migration-progress_test.go b/cmd/erasure-server-pool-migration-progress_test.go new file mode 100644 index 000000000..d3972b010 --- /dev/null +++ b/cmd/erasure-server-pool-migration-progress_test.go @@ -0,0 +1,528 @@ +// Copyright (c) 2026 Feng Ruohang +// SPDX-License-Identifier: AGPL-3.0-or-later +// +// 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 . + +package cmd + +import ( + "bytes" + "context" + "errors" + "io" + "slices" + "sync" + "sync/atomic" + "testing" + "time" + + "github.com/minio/minio/internal/bucket/lifecycle" +) + +// The walk has already captured the version when its data read reaches this +// hook. Metadata-only checks made by a concurrent DELETE do not reenter it. +type migrationProgressReadDisk struct { + StorageAPI + bucket, object, version string + before func() error +} + +func (d migrationProgressReadDisk) ReadVersion(ctx context.Context, orig, bucket, object, version string, opts ReadOptions) (FileInfo, error) { + if opts.ReadData && bucket == d.bucket && object == d.object && version == d.version { + if err := d.before(); err != nil { + return FileInfo{}, err + } + } + return d.StorageAPI.ReadVersion(ctx, orig, bucket, object, version, opts) +} + +// Buffer every drive's first walk before publishing any entry. Thus deleting +// the parent cannot accidentally let this same walk discover the child. +type migrationSnapshotWalk struct { + bucket string + drives int32 + calls atomic.Int32 + ready chan struct{} +} + +type migrationSnapshotWalkDisk struct { + StorageAPI + snapshot *migrationSnapshotWalk +} + +func (d migrationSnapshotWalkDisk) WalkDir(ctx context.Context, opts WalkDirOptions, out io.Writer) error { + if opts.Bucket != d.snapshot.bucket { + return d.StorageAPI.WalkDir(ctx, opts, out) + } + var buf bytes.Buffer + if err := d.StorageAPI.WalkDir(ctx, opts, &buf); err != nil { + return err + } + if d.snapshot.calls.Add(1) == d.snapshot.drives { + close(d.snapshot.ready) + } + select { + case <-d.snapshot.ready: + case <-ctx.Done(): + return ctx.Err() + } + _, err := io.Copy(out, &buf) + return err +} + +func migrationProgressSnapshot(t *testing.T, z *erasureServerPools, bucket, key string) *migrationSnapshotWalk { + t.Helper() + s := &migrationSnapshotWalk{bucket: bucket, drives: int32(len(z.serverPools[0].getHashedSet(key).getDisks())), ready: make(chan struct{})} + t.Cleanup(conditionalPutSwapDisks(z.serverPools[0], key, func(d StorageAPI) StorageAPI { + return migrationSnapshotWalkDisk{StorageAPI: d, snapshot: s} + })) + return s +} + +func migrationProgressPrepare(t *testing.T, z *erasureServerPools, kind string) { + t.Helper() + migrationTestMover(t, z, 0, kind) + z.poolMetaMutex.Lock() + z.poolMeta.Pools[0].LastUpdate = UTCNow() + z.poolMetaMutex.Unlock() +} + +func TestDecommissionSplitMetadataDoesNotAdvance(t *testing.T) { + m := newMigrationILM(t, "", false, migrationILMExpireCurrent()) + const key = "obj-01" + m.put(key, "old", ObjectOptions{MTime: UTCNow().Add(-96 * time.Hour)}) + disks := m.z.serverPools[0].getHashedSet(key).getDisks() + old := make([][]byte, len(disks)/2) + for i := range old { + var err error + old[i], err = disks[i].ReadAll(t.Context(), m.bucket, pathJoin(key, xlStorageFormatFile)) + mustIAM(t, err) + } + m.put(key, "new-and-longer", ObjectOptions{MTime: UTCNow().Add(-72 * time.Hour)}) + for i := range old { + mustIAM(t, disks[i].WriteAll(t.Context(), m.bucket, pathJoin(key, xlStorageFormatFile), old[i])) + } + if _, err := m.z.serverPools[0].GetObjectInfo(t.Context(), m.bucket, key, ObjectOptions{VersionID: nullVersionID}); !isErrVersionNotFound(err) { + t.Fatalf("fixture must expose the ambiguous logical NotFound, got %v", err) + } + migrationProgressPrepare(t, m.z, "decommission") + walk := migrationProgressSnapshot(t, m.z, m.bucket, key) + bi := decomBucketInfo{Name: m.bucket} + // Check a single pass first, so the unfixed implementation fails rather + // than looping until the test timeout. + removed, err := m.z.decommissionPoolWalk(t.Context(), 0, m.z.serverPools[0], bi) + if err != nil || removed != 0 { + t.Fatalf("split metadata reported progress: removed=%d err=%v", removed, err) + } + mustIAM(t, m.z.decommissionPool(t.Context(), 0, m.z.serverPools[0], bi)) + if got := walk.calls.Load(); got != 2*walk.drives { + t.Fatalf("unchanged metadata repeated the walk: calls=%d", got) + } + if err := m.z.checkAfterDecomBucket(t.Context(), 0, bi); err == nil { + t.Fatal("final verification accepted split metadata still on disk") + } + for _, disk := range disks { + if _, err := disk.ReadVersion(t.Context(), "", m.bucket, key, nullVersionID, ReadOptions{}); err != nil { + t.Fatalf("migration removed an ambiguous source revision: %v", err) + } + } +} + +func TestPoolsMigrationConcurrentDeleteCleanup(t *testing.T) { + for _, kind := range []string{"decommission", "rebalance"} { + t.Run(kind, func(t *testing.T) { + z, bucket := migrationTestPools(t) + v1 := putConsistencyObject(t, z, bucket, "k", 0, "older", ObjectOptions{Versioned: true, MTime: UTCNow().Add(-time.Hour)}) + v2 := putConsistencyObject(t, z, bucket, "k", 0, "newer", ObjectOptions{Versioned: true}) + var once sync.Once + var deleteErr error + before := func() error { + once.Do(func() { + _, deleteErr = z.DeleteObject(t.Context(), bucket, "k", ObjectOptions{Versioned: true, VersionID: v2.VersionID}) + }) + return deleteErr + } + restore := conditionalPutSwapDisks(z.serverPools[0], "k", func(d StorageAPI) StorageAPI { + return migrationProgressReadDisk{StorageAPI: d, bucket: bucket, object: "k", version: v2.VersionID, before: before} + }) + defer restore() + if failed := migrationTestWalk(t, z, bucket, 0, kind); failed != 0 { + t.Fatalf("failed moves: %d", failed) + } + restore() + for _, v := range []ObjectInfo{v1, v2} { + for _, disk := range z.serverPools[0].getHashedSet("k").getDisks() { + _, err := disk.ReadVersion(t.Context(), "", bucket, "k", v.VersionID, ReadOptions{}) + if !errors.Is(err, errFileVersionNotFound) && !errors.Is(err, errFileNotFound) { + t.Fatalf("source version %s remained: %v", v.VersionID, err) + } + } + } + if body := migrationTestBody(t, z.GetObjectNInfo, bucket, "k", v1.VersionID); body != "older" { + t.Fatalf("moved version reads %q", body) + } + if _, err := z.GetObjectInfo(t.Context(), bucket, "k", ObjectOptions{VersionID: v2.VersionID}); !isErrVersionNotFound(err) { + t.Fatalf("concurrently deleted version resurrected: %v", err) + } + }) + } +} + +func TestPoolsMigrationSiblingFailureCleanup(t *testing.T) { + for _, kind := range []string{"decommission", "rebalance"} { + for _, loseDestination := range []bool{false, true} { + name := kind + "/destination-retained" + if loseDestination { + name = kind + "/destination-removed" + } + t.Run(name, func(t *testing.T) { + z, bucket := migrationTestPools(t) + v1 := putConsistencyObject(t, z, bucket, "k", 0, "older", ObjectOptions{Versioned: true, MTime: UTCNow().Add(-time.Hour)}) + v2 := putConsistencyObject(t, z, bucket, "k", 0, "newer", ObjectOptions{Versioned: true}) + var once sync.Once + var deleteErr error + before := func() error { + once.Do(func() { + if loseDestination { + _, deleteErr = z.serverPools[1].DeleteObject(t.Context(), bucket, "k", ObjectOptions{VersionID: v1.VersionID}) + } + }) + return errDiskNotFound // only the sibling's data read fails + } + restore := conditionalPutSwapDisks(z.serverPools[0], "k", func(d StorageAPI) StorageAPI { + return migrationProgressReadDisk{StorageAPI: d, bucket: bucket, object: "k", version: v2.VersionID, before: before} + }) + defer restore() + migrationProgressPrepare(t, z, kind) + if kind == "rebalance" { + mustIAM(t, z.rebalanceBucket(t.Context(), bucket, 0)) + } else { + _, err := z.decommissionPoolWalk(t.Context(), 0, z.serverPools[0], decomBucketInfo{Name: bucket}) + mustIAM(t, err) + if err := z.checkAfterDecomBucket(t.Context(), 0, decomBucketInfo{Name: bucket}); err == nil { + t.Fatal("failed sibling must block final decommission verification") + } + } + restore() + mustIAM(t, deleteErr) + if body := migrationTestBody(t, z.GetObjectNInfo, bucket, "k", v2.VersionID); body != "newer" { + t.Fatalf("failed sibling reads %q", body) + } + if body := migrationTestBody(t, z.GetObjectNInfo, bucket, "k", v1.VersionID); body != "older" { + t.Fatalf("successful candidate lost: %q", body) + } + _, err := z.serverPools[0].GetObjectInfo(t.Context(), bucket, "k", ObjectOptions{VersionID: v1.VersionID}) + if loseDestination && err != nil || !loseDestination && !isErrVersionNotFound(err) { + t.Fatalf("source cleanup ignored destination state: %v", err) + } + }) + } + } +} + +func TestDecommissionConcurrentParentDeleteRescansChildren(t *testing.T) { + z, bucket := migrationTestPools(t) + putConsistencyObject(t, z, bucket, "k", 0, "parent", ObjectOptions{}) + putConsistencyObject(t, z, bucket, "k/b", 0, "child", ObjectOptions{}) + migrationProgressSnapshot(t, z, bucket, "k") + var once sync.Once + var deleteErr error + t.Cleanup(conditionalPutSwapDisks(z.serverPools[0], "k", func(d StorageAPI) StorageAPI { + return migrationProgressReadDisk{StorageAPI: d, bucket: bucket, object: "k", version: nullVersionID, before: func() error { + once.Do(func() { + _, deleteErr = z.DeleteObject(t.Context(), bucket, "k", ObjectOptions{VersionID: nullVersionID}) + }) + return deleteErr + }} + })) + if failed := migrationTestWalk(t, z, bucket, 0, "decommission"); failed != 0 { + t.Fatalf("failed moves: %d", failed) + } + mustIAM(t, z.checkAfterDecomBucket(t.Context(), 0, decomBucketInfo{Name: bucket})) + if body := migrationTestBody(t, z.GetObjectNInfo, bucket, "k/b", ""); body != "child" { + t.Fatalf("child reads %q", body) + } +} + +func TestFreeVersionDecommissionRescansChildren(t *testing.T) { + f := newFreeVersionTest(t, "") + f.bucket = "free-parent" + mustIAM(t, f.z.MakeBucket(t.Context(), f.bucket, MakeBucketOptions{VersioningEnabled: true})) + f.deleteVersion(f.tiered("k", "parent", UTCNow().Add(-time.Hour))) + f.put("k/b", "child", UTCNow()) + migrationProgressSnapshot(t, f.z, f.bucket, "k") + migrationProgressPrepare(t, f.z, "decommission") + bi := decomBucketInfo{Name: f.bucket} + removed, err := f.z.decommissionPoolWalk(t.Context(), 0, f.z.serverPools[0], bi) + if err != nil || removed != 1 { + t.Fatalf("free-only parent removal: removed=%d err=%v", removed, err) + } + if err := f.z.checkAfterDecomBucket(t.Context(), 0, bi); err == nil { + t.Fatal("the newly exposed child must still block completion") + } + mustIAM(t, f.z.decommissionPool(t.Context(), 0, f.z.serverPools[0], bi)) + mustIAM(t, f.z.checkAfterDecomBucket(t.Context(), 0, bi)) + if body := migrationTestBody(t, f.z.GetObjectNInfo, f.bucket, "k/b", ""); body != "k/b child" { + t.Fatalf("child reads %q", body) + } + if !slices.Contains(f.tier.removed, f.remote("k", "parent")) { + t.Fatal("free version cleanup did not delete its remote object") + } +} + +func TestPoolsMigrationCancellation(t *testing.T) { + for _, kind := range []string{"decommission", "rebalance"} { + for _, phase := range []string{"before-walk", "during-read", "deadline"} { + name := kind + "/" + phase + t.Run(name, func(t *testing.T) { + z, bucket := migrationTestPools(t) + oi := putConsistencyObject(t, z, bucket, "k", 0, "source", ObjectOptions{Versioned: true}) + migrationProgressPrepare(t, z, kind) + ctx, cancel := context.WithCancel(t.Context()) + wantErr := context.Canceled + if phase == "deadline" { + cancel() + ctx, cancel = context.WithDeadline(t.Context(), time.Time{}) + wantErr = context.DeadlineExceeded + } + defer cancel() + if phase == "during-read" { + t.Cleanup(conditionalPutSwapDisks(z.serverPools[0], "k", func(d StorageAPI) StorageAPI { + return migrationProgressReadDisk{StorageAPI: d, bucket: bucket, object: "k", version: oi.VersionID, before: func() error { cancel(); return ctx.Err() }} + })) + } else { + cancel() + } + var err error + if kind == "decommission" { + err = z.decommissionPool(ctx, 0, z.serverPools[0], decomBucketInfo{Name: bucket}) + } else { + err = z.rebalanceBucket(ctx, bucket, 0) + } + if !errors.Is(err, wantErr) { + t.Fatalf("cancellation returned %v", err) + } + if err := z.decommissionInBackground(ctx, 0); !errors.Is(err, wantErr) { + t.Fatalf("canceled background migration returned %v", err) + } + if err := z.rebalanceBuckets(ctx, 0); !errors.Is(err, wantErr) { + t.Fatalf("canceled background rebalance returned %v", err) + } + if n, err := z.countAbsentMigrationVersions(ctx, bucket, "k", 0, []FileInfo{{VersionID: oi.VersionID}}); n != 0 || !errors.Is(err, wantErr) { + t.Fatalf("canceled progress verification: removed=%d err=%v", n, err) + } + if _, err := z.serverPools[0].GetObjectInfo(t.Context(), bucket, "k", ObjectOptions{VersionID: oi.VersionID}); err != nil { + t.Fatalf("canceled migration lost source: %v", err) + } + if err := z.checkAfterDecomBucket(ctx, 0, decomBucketInfo{Name: bucket}); !errors.Is(err, wantErr) { + t.Fatalf("canceled verification returned %v", err) + } + }) + } + } +} + +type migrationProgressRawDisk struct { + StorageAPI + buf []byte + err error + reads *atomic.Int32 +} + +func (d migrationProgressRawDisk) ReadXL(ctx context.Context, bucket, object string, data bool) (RawFileInfo, error) { + d.reads.Add(1) + if data { + return RawFileInfo{}, errors.New("progress check requested object data") + } + return RawFileInfo{Buf: d.buf}, d.err +} + +func (d migrationProgressRawDisk) ListDir(context.Context, string, string, string, int) ([]string, error) { + return nil, errFileNotFound +} + +func TestPoolsMigrationPhysicalProgress(t *testing.T) { + f := newFreeVersionTest(t, "") + buf := f.xlMeta(0, "obj-01") + versions, err := getFileInfoVersions(buf, f.bucket, "obj-01", true) + mustIAM(t, err) + if len(versions.Versions) != 3 || !versions.Versions[0].TierFreeVersion() { + t.Fatal("fixture must contain two live versions and one free version") + } + missing := FileInfo{VersionID: mustGetUUID()} + snapshot := append(slices.Clone(versions.Versions), missing) + for _, tc := range []struct { + name string + buf []byte + err error + want int + wantErr bool + }{ + {name: "all-absent", err: errFileNotFound, want: 4}, + {name: "missing-volume", err: errVolumeNotFound, want: 4}, + {name: "minority-live-and-free", buf: buf, want: 1}, + {name: "unavailable", err: errDiskNotFound, wantErr: true}, + {name: "read-quorum", err: errErasureReadQuorum, wantErr: true}, + {name: "corrupt-read", err: errFileCorrupt, wantErr: true}, + {name: "corrupt-metadata", buf: []byte("broken xl.meta"), wantErr: true}, + } { + t.Run(tc.name, func(t *testing.T) { + var reads atomic.Int32 + disks := make([]StorageAPI, 4) + for i := range disks { + disks[i] = migrationProgressRawDisk{err: errFileNotFound, reads: &reads} + } + disks[3] = migrationProgressRawDisk{buf: tc.buf, err: tc.err, reads: &reads} + er := erasureObjects{getDisks: func() []StorageAPI { return disks }} + n, err := er.countAbsentMigrationVersions(t.Context(), f.bucket, "obj-01", snapshot) + if n != tc.want || (err != nil) != tc.wantErr { + t.Fatalf("removed=%d err=%v, want %d / error=%t", n, err, tc.want, tc.wantErr) + } + if reads.Load() != int32(len(disks)) { + t.Fatalf("read count %d must not depend on version count", reads.Load()) + } + }) + } + putConsistencyObject(t, f.z, f.bucket, "null", 0, "null-data", ObjectOptions{}) + nullBuf := f.xlMeta(0, "null") + var reads atomic.Int32 + er := erasureObjects{getDisks: func() []StorageAPI { + return []StorageAPI{migrationProgressRawDisk{buf: nullBuf, reads: &reads}} + }} + if n, err := er.countAbsentMigrationVersions(t.Context(), f.bucket, "null", []FileInfo{{VersionID: nullVersionID}, {VersionID: ""}}); n != 0 || err != nil { + t.Fatalf("null version normalization: removed=%d err=%v", n, err) + } +} + +// A successful quorum deletion can leave one source drive behind. It must not +// count as physical removal or pass the final check, nor cause a retry loop. +func TestDecommissionQuorumDeletionIsNotPhysicalProgress(t *testing.T) { + z, bucket := migrationTestPools(t) + oi := putConsistencyObject(t, z, bucket, "k", 0, "payload", ObjectOptions{Versioned: true}) + move := migrationTestMover(t, z, 0, "decommission") + mustIAM(t, move(t.Context(), 0, bucket, migrationTestReader(t, z, 0, oi))) + first := true + restore := conditionalPutSwapDisks(z.serverPools[0], "k", func(d StorageAPI) StorageAPI { + if first { + first = false + return consistencyDeleteFaultDisk{StorageAPI: d, bucket: bucket, object: "k", version: oi.VersionID} + } + return d + }) + defer restore() + snapshot := []FileInfo{{VersionID: oi.VersionID, ModTime: oi.ModTime}} + for range 2 { + n, err := z.removeMovedVersions(t.Context(), bucket, "k", 0, snapshot, nil) + if err != nil || n != 0 { + t.Fatalf("partial deletion reported progress: removed=%d err=%v", n, err) + } + } + if err := z.checkAfterDecomBucket(t.Context(), 0, decomBucketInfo{Name: bucket}); err == nil { + t.Fatal("final verification accepted a minority-drive source copy") + } + if body := migrationTestBody(t, z.GetObjectNInfo, bucket, "k", oi.VersionID); body != "payload" { + t.Fatalf("destination copy lost: %q", body) + } +} + +func TestDecommissionProgressReadFailureStopsWalk(t *testing.T) { + z, bucket := migrationTestPools(t) + putConsistencyObject(t, z, bucket, "k", 0, "payload", ObjectOptions{Versioned: true}) + migrationProgressPrepare(t, z, "decommission") + walk := migrationProgressSnapshot(t, z, bucket, "k") + restore := conditionalPutSwapDisks(z.serverPools[0], "k", func(d StorageAPI) StorageAPI { + return tierReadFaultDisk{StorageAPI: d, bucket: bucket, object: "k", err: errDiskNotFound} + }) + defer restore() + if err := z.decommissionPool(t.Context(), 0, z.serverPools[0], decomBucketInfo{Name: bucket}); err == nil { + t.Fatal("unreadable source progress check reported success") + } + if got := walk.calls.Load(); got != walk.drives { + t.Fatalf("verification failure repeated the walk: calls=%d", got) + } + restore() + if body := migrationTestBody(t, z.GetObjectNInfo, bucket, "k", ""); body != "payload" { + t.Fatalf("quorum move lost destination: %q", body) + } +} + +type migrationProgressListFaultDisk struct { + StorageAPI +} + +func (d migrationProgressListFaultDisk) ListDir(context.Context, string, string, string, int) ([]string, error) { + return nil, errDiskAccessDenied +} + +func TestDecommissionFinalVerificationFailsClosed(t *testing.T) { + for _, state := range []string{"offline-drive", "directory-error", "corrupt-metadata", "truncated-metadata", "legacy-metadata"} { + t.Run(state, func(t *testing.T) { + z, bucket := migrationTestPools(t) + switch state { + case "offline-drive", "directory-error": + t.Cleanup(conditionalPutSwapDisks(z.serverPools[0], "k", func(d StorageAPI) StorageAPI { + if state == "directory-error" { + return migrationProgressListFaultDisk{StorageAPI: d} + } + return nil + })) + default: + // Real metadata on only one drive, without a readable quorum. + // This also covers errors that WalkDir deliberately skips. + disk := z.serverPools[0].getHashedSet("k").getDisks()[0] + file, data := xlStorageFormatFile, []byte("broken xl.meta") + if state == "truncated-metadata" { + data = nil + } + if state == "legacy-metadata" { + file = xlStorageFormatFileV1 + } + mustIAM(t, disk.WriteAll(t.Context(), bucket, pathJoin("k", file), data)) + } + if err := z.checkAfterDecomBucket(t.Context(), 0, decomBucketInfo{Name: bucket}); err == nil { + t.Fatal("unreadable storage was accepted as physically empty") + } + if n, err := z.countAbsentMigrationVersions(t.Context(), bucket, "k", 0, []FileInfo{{VersionID: mustGetUUID()}}); n != 0 || err == nil { + t.Fatalf("unreadable storage was counted as progress: removed=%d err=%v", n, err) + } + }) + } +} + +func TestFreeVersionDecommissionPreservesLiveRemoteOwner(t *testing.T) { + f := newFreeVersionTest(t, "") + const key = "obj-02" + fv, tuple := f.freeRecord(0, key) + owner := putConsistencyObject(t, f.z, f.bucket, key, 1, "live-owner", ObjectOptions{Versioned: true}) + infos, errs := readAllFileInfo(t.Context(), f.z.serverPools[1].getHashedSet(key).getDisks(), "", f.bucket, key, owner.VersionID, false, false) + mustIAM(t, errs[0]) + fi := infos[0] + fi.TransitionStatus, fi.TransitionTier = lifecycle.TransitionComplete, tuple.Tier + fi.TransitionedObjName, fi.TransitionVersionID = tuple.Object, tuple.VersionID + mustIAM(t, f.z.serverPools[1].DecomTieredObject(t.Context(), f.bucket, key, fi, ObjectOptions{})) + if err := f.z.removeFreeVersions(t.Context(), f.bucket, key, 0, []FileInfo{fv}); !errors.Is(err, errTierStillReferenced) { + t.Fatalf("cleanup with a live remote owner: %v", err) + } + if n, err := f.z.countAbsentMigrationVersions(t.Context(), f.bucket, key, 0, []FileInfo{fv}); err != nil || n != 0 { + t.Fatalf("retained free record reported progress: removed=%d err=%v", n, err) + } + if slices.Contains(f.tier.removed, tuple.Object) { + t.Fatal("cleanup deleted a live remote object") + } + if err := f.z.checkAfterDecomBucket(t.Context(), 0, decomBucketInfo{Name: f.bucket}); err == nil { + t.Fatal("retained free record must block final decommission verification") + } +} diff --git a/cmd/erasure-server-pool-migration.go b/cmd/erasure-server-pool-migration.go index 8cbe91fa5..9ddc91501 100644 --- a/cmd/erasure-server-pool-migration.go +++ b/cmd/erasure-server-pool-migration.go @@ -17,7 +17,117 @@ package cmd -import xhttp "github.com/minio/minio/internal/http" +import ( + "context" + "errors" + + xhttp "github.com/minio/minio/internal/http" + "github.com/pgsty/silo-pkg/v3/sync/errgroup" +) + +func migrationVersionKey(versionID string) string { + if versionID == "" || versionID == nullVersionID { + return nullVersionID + } + return versionID +} + +// countAbsentMigrationVersions counts snapshot versions that no source drive +// holds anymore, including free versions. A pool-level NotFound is insufficient: +// the ordinary read path can return it for inconsistent metadata. Only readable +// per-drive metadata (or physical absence) proves progress for another walk. +// The caller holds the object's writer lock while checking the source. +func (er erasureObjects) countAbsentMigrationVersions(ctx context.Context, bucket, object string, versions []FileInfo) (int, error) { + if err := ctx.Err(); err != nil { + return 0, err + } + if len(versions) == 0 { + return 0, nil + } + absent := make(map[string]struct{}, len(versions)) + for _, version := range versions { + absent[migrationVersionKey(version.VersionID)] = struct{}{} + } + disks := er.getDisks() + if len(disks) == 0 { + return 0, toObjectErr(errErasureReadQuorum, bucket, object) + } + stored := make([][]FileInfo, len(disks)) + g := errgroup.WithNErrs(len(disks)) + for i := range disks { + g.Go(func() error { + if disks[i] == nil { + return errDiskNotFound + } + raw, err := disks[i].ReadXL(ctx, bucket, object, false) + if errors.Is(err, errFileNotFound) { + // ReadXL also returns NotFound for an empty xl.meta, and may + // ignore legacy metadata. Confirm physical absence before using + // that result as progress. This does not descend into child keys. + // Both reads run in parallel across drives, once per object. + entries, err := disks[i].ListDir(ctx, "", bucket, object, -1) + if err != nil && !errors.Is(err, errFileNotFound) && !errors.Is(err, errVolumeNotFound) { + return err + } + for _, entry := range entries { + if entry == xlStorageFormatFile || entry == xlStorageFormatFileV1 { + return errFileCorrupt + } + } + return nil + } + if errors.Is(err, errVolumeNotFound) { + return nil + } + if err != nil { + return err + } + // allParts=false omits part details, not free versions. Decode every + // version before concluding absence, so corrupt metadata fails closed. + fivs, err := getAllFileInfoVersions(raw.Buf, bucket, object, false) + stored[i] = fivs.Versions + return err + }, i) + } + errs := g.Wait() + if err := ctx.Err(); err != nil { + return 0, err + } + for i, err := range errs { + if err != nil { + return 0, toObjectErr(err, bucket, object) + } + for _, version := range stored[i] { + delete(absent, migrationVersionKey(version.VersionID)) + } + } + if err := ctx.Err(); err != nil { + return 0, err + } + return len(absent), nil +} + +// countAbsentMigrationVersions checks ignored walk entries without authorizing +// their deletion. A concurrent client delete may have uncovered child keys. +func (z *erasureServerPools) countAbsentMigrationVersions(ctx context.Context, bucket, object string, source int, versions []FileInfo) (int, error) { + if err := ctx.Err(); err != nil { + return 0, err + } + if len(versions) == 0 { + return 0, nil + } + object = encodeDirObject(object) + lk := z.NewNSLock(bucket, object) + lkctx, err := lk.GetLock(ctx, globalOperationTimeout) + if err != nil { + if ctx.Err() != nil { + return 0, ctx.Err() + } + return 0, err + } + defer lk.Unlock(lkctx) + return z.serverPools[source].getHashedSet(object).countAbsentMigrationVersions(lkctx.Context(), bucket, object, versions) +} // migrationObjectMetadata restores tags removed from the read representation. // Keep the original revision, including ordered empty states, for the existing diff --git a/cmd/erasure-server-pool-rebalance.go b/cmd/erasure-server-pool-rebalance.go index fb9eb996d..35e6d29de 100644 --- a/cmd/erasure-server-pool-rebalance.go +++ b/cmd/erasure-server-pool-rebalance.go @@ -376,6 +376,9 @@ func (z *erasureServerPools) IsPoolRebalancing(poolIndex int) bool { } func (z *erasureServerPools) rebalanceBuckets(ctx context.Context, poolIdx int) (err error) { + if err := ctx.Err(); err != nil { + return err + } doneCh := make(chan error, 1) defer xioutil.SafeClose(doneCh) @@ -450,7 +453,7 @@ func (z *erasureServerPools) rebalanceBuckets(ctx context.Context, poolIdx int) select { case <-ctx.Done(): doneCh <- ctx.Err() - return err + return ctx.Err() default: } @@ -474,6 +477,10 @@ func (z *erasureServerPools) rebalanceBuckets(ctx context.Context, poolIdx int) z.bucketRebalanceDone(bucket, poolIdx) } + if err := ctx.Err(); err != nil { + doneCh <- err + return err + } rebalanceLogEvent(GlobalContext, "Pool %d rebalancing is done", poolIdx+1) return err @@ -538,6 +545,9 @@ func (set *erasureObjects) listObjectsToRebalance(ctx context.Context, bucketNam // rebalanceBucket rebalances objects under bucket in poolIdx pool func (z *erasureServerPools) rebalanceBucket(ctx context.Context, bucket string, poolIdx int) (err error) { + if err := ctx.Err(); err != nil { + return err + } ctx = logger.SetReqInfo(ctx, &logger.ReqInfo{}) var vc *versioning.Versioning @@ -643,7 +653,7 @@ func (z *erasureServerPools) rebalanceBucket(ctx context.Context, bucket string, // to create the appropriate stack. versionsSorter(fivs.Versions).reverse() - var rebalanced int + var candidates []FileInfo var expired []migrationExpiredVersion // The key's history in all pools, read once for its versions. var history migrationHistory @@ -658,12 +668,8 @@ func (z *erasureServerPools) rebalanceBucket(ctx context.Context, bucket string, // Apply lifecycle rules on the objects that are expired. if action := filterLifecycle(bucket, version.Name, version, &history); action.Delete() { - // Count the version, or the source copies of the moved - // ones stay behind as duplicates. Its own source copy - // goes with them when ILM deletes it alone; any other - // version stays until the expiry has run. - rebalanced++ if ilmDeletesVersionOnly(action) || action.DeleteAll() { + candidates = append(candidates, version) expired = append(expired, migrationExpiredVersion{ version: version, event: lifecycle.Event{Action: action}, @@ -706,7 +712,6 @@ func (z *erasureServerPools) rebalanceBucket(ctx context.Context, bucket string, // This can happen when rebalance stop races with ongoing rebalance workers. // These rebalance failures can be ignored. if isErrObjectNotFound(err) || isErrVersionNotFound(err) || isDataMovementOverWriteErr(err) { - ignore = true stopFn(0, nil) continue } @@ -716,7 +721,7 @@ func (z *erasureServerPools) rebalanceBucket(ctx context.Context, bucket string, failure = err != nil if !failure { z.updatePoolStats(poolIdx, bucket, version) - rebalanced++ + candidates = append(candidates, version) } auditLogRebalance(ctx, "Rebalance:DeleteMarker", bucket, version.Name, versionID, err) continue @@ -773,13 +778,14 @@ func (z *erasureServerPools) rebalanceBucket(ctx context.Context, bucket string, break // break out on first error } z.updatePoolStats(poolIdx, bucket, version) - rebalanced++ + candidates = append(candidates, version) } - // if all versions were rebalanced, we can delete the object versions. - if rebalanced == len(fivs.Versions) { + // A skipped or failed sibling does not prevent retiring successful + // moves. Cleanup still rechecks each candidate and its destination. + if len(candidates) > 0 { 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, candidates, expired) stopFn(0, err) auditLogRebalance(ctx, "Rebalance:DeleteObject", bucket, entry.name, "", err) if err != nil { @@ -806,7 +812,7 @@ func (z *erasureServerPools) rebalanceBucket(ctx context.Context, bucket string, } wk.Wait() - return nil + return ctx.Err() } type rebalSaveOpts uint8