From b524fc4fb73436f1c6081fecfe20782ff9525324 Mon Sep 17 00:00:00 2001 From: Feng Ruohang Date: Sat, 7 Nov 2026 01:10:01 +0800 Subject: [PATCH] fix(ilm): queue explicit version heals after corrupt history refusal Signed-off-by: Feng Ruohang --- ...ure-server-pool-delete-all-reasons_test.go | 23 ++++ cmd/erasure-server-pool-delete-all.go | 7 ++ cmd/erasure-server-pool-delete-all_test.go | 10 ++ cmd/review-completion_test.go | 107 ++++++++++++++++++ 4 files changed, 147 insertions(+) create mode 100644 cmd/review-completion_test.go diff --git a/cmd/erasure-server-pool-delete-all-reasons_test.go b/cmd/erasure-server-pool-delete-all-reasons_test.go index cb1502e57..815ef6868 100644 --- a/cmd/erasure-server-pool-delete-all-reasons_test.go +++ b/cmd/erasure-server-pool-delete-all-reasons_test.go @@ -22,6 +22,7 @@ import ( "errors" "slices" "strings" + "sync" "testing" madmin "github.com/minio/madmin-go/v3" @@ -174,6 +175,17 @@ func TestPurgeAllVersionsRefusesCorruptDrive(t *testing.T) { t.Fatalf("CRC classification after storage RPC: %v", got) } + var healMu sync.Mutex + queued := make(map[string]bool) + hook := func(op PartialOperation) { + if op.Bucket == p.bucket && op.Object == key { + healMu.Lock() + queued[op.VersionID] = true + healMu.Unlock() + } + } + previousHook := mrfPartialOpHook.Swap(&hook) + t.Cleanup(func() { mrfPartialOpHook.Store(previousHook) }) delta := incompleteDelta(func() { p.run(ops) }) if delta[ilmIncompleteHistoryUnavailable] != 1 { t.Errorf("counted %v, want one %s", delta, ilmIncompleteHistoryUnavailable) @@ -191,6 +203,17 @@ func TestPurgeAllVersionsRefusesCorruptDrive(t *testing.T) { if err != nil { t.Fatal(err) } + mrfPartialOpHook.Store(previousHook) + healMu.Lock() + for _, fi := range versions.Versions { + if !queued[deleteAllVersionID(fi.VersionID)] { + t.Errorf("no MRF heal for %q: %v", fi.VersionID, queued) + } + } + if queued[""] { + t.Error("MRF used latest instead of explicit null version") + } + healMu.Unlock() for _, mode := range []madmin.HealScanMode{madmin.HealNormalScan, madmin.HealDeepScan} { if err := bad.WriteAll(ctx, p.bucket, metaPath, buf); err != nil { t.Fatal(err) diff --git a/cmd/erasure-server-pool-delete-all.go b/cmd/erasure-server-pool-delete-all.go index 670c98d0d..f20f8f05c 100644 --- a/cmd/erasure-server-pool-delete-all.go +++ b/cmd/erasure-server-pool-delete-all.go @@ -435,6 +435,13 @@ func (er erasureObjects) deleteAllDrives(ctx context.Context, bucket, object str } var xl xlMetaV2 if err := xl.LoadOrConvert(raws[i].Buf); err != nil { + if errors.Is(err, errFileCorrupt) { + // The purge remains refused. Heal each approved version explicitly, + // including a noncurrent null version, before a later attempt retries. + for id := range ids { + er.addPartial(bucket, object, id) + } + } return nil, errDeleteAllHistoryUnavailable } versions, err := xl.ListVersions(bucket, object, false) diff --git a/cmd/erasure-server-pool-delete-all_test.go b/cmd/erasure-server-pool-delete-all_test.go index 31787b5f9..1e7c24e27 100644 --- a/cmd/erasure-server-pool-delete-all_test.go +++ b/cmd/erasure-server-pool-delete-all_test.go @@ -370,6 +370,16 @@ func (p *purgeTest) matched(name string) expireObjInfo { // end, and returns the counts of the job. func (p *purgeTest) batch(walked ...expireObjInfo) *batchJobInfo { p.t.Helper() + // Direct object-layer fixtures do not start IAM. A real batch worker + // requires its initial cache load even when the submitter is root. + previousIAM := globalIAMSys + sys := NewIAMSys() + sys.initStore(p.z, nil) + if err := sys.Load(p.t.Context(), true); err != nil { + p.t.Fatal(err) + } + globalIAMSys = sys + defer func() { globalIAMSys = previousIAM }() globalBatchJobsMetrics.Lock() if globalBatchJobsMetrics.metrics == nil { globalBatchJobsMetrics.metrics = make(map[string]*batchJobInfo) diff --git a/cmd/review-completion_test.go b/cmd/review-completion_test.go new file mode 100644 index 000000000..455fa4892 --- /dev/null +++ b/cmd/review-completion_test.go @@ -0,0 +1,107 @@ +// Copyright (c) 2026 PGSTY +// SPDX-License-Identifier: AGPL-3.0-or-later +package cmd + +import ( + "bytes" + "context" + "testing" + + madmin "github.com/minio/madmin-go/v3" +) + +func TestHealCRCWithoutQuorumPreservesMetadata(t *testing.T) { + z := purgeSinglePool(t, 16) + p := newPurgeTest(t, z, "heal-crc-quorum", MakeBucketOptions{}, "") + const key = "crc-quorum" + p.put(0, key, "payload", daysAgo(2), ObjectOptions{}) + disks := z.serverPools[0].getHashedSet(key).getDisks() + before := make([][]byte, len(disks)) + for i, disk := range disks { + raw, err := disk.ReadAll(t.Context(), p.bucket, pathJoin(key, xlStorageFormatFile)) + if err != nil { + t.Fatal(err) + } + if i <= len(disks)/2 { + meta, _, err := isIndexedMetaV2(raw) + if err != nil { + t.Fatal(err) + } + meta[len(meta)-1] ^= 1 + if err := disk.WriteAll(t.Context(), p.bucket, pathJoin(key, xlStorageFormatFile), raw); err != nil { + t.Fatal(err) + } + } + before[i] = raw + } + for _, dry := range []bool{true, false} { + if _, err := z.HealObject(t.Context(), p.bucket, key, nullVersionID, madmin.HealOpts{DryRun: dry, Remove: true, ScanMode: madmin.HealDeepScan}); err == nil { + t.Fatal("healed without a metadata quorum") + } + for i, disk := range disks { + after, err := disk.ReadAll(t.Context(), p.bucket, pathJoin(key, xlStorageFormatFile)) + if err != nil || !bytes.Equal(after, before[i]) { + t.Fatalf("heal without quorum changed drive %d: %v", i, err) + } + } + } +} + +func TestHealListedVersionDeletedBeforeRepair(t *testing.T) { + z := purgeSinglePool(t, 16) + p := newPurgeTest(t, z, "heal-delete-race", MakeBucketOptions{VersioningEnabled: true}, "") + const key = "deleted-after-listing" + listed := p.version(0, key, "old", 3) + p.version(0, key, "latest", 2) + // A healer has already listed this exact version. Its deletion commits + // before the heal acquires the object's lock and re-reads the metadata. + if _, err := z.DeleteObject(t.Context(), p.bucket, key, ObjectOptions{VersionID: listed.VersionID, Versioned: true}); err != nil { + t.Fatal(err) + } + if _, err := z.HealObject(t.Context(), p.bucket, key, listed.VersionID, madmin.HealOpts{ScanMode: madmin.HealDeepScan}); !isErrVersionNotFound(err) { + t.Fatalf("heal of deleted listing: %v", err) + } + if got := p.get(key); got != "latest" { + t.Fatalf("latest changed: %s", got) + } + for _, disk := range z.serverPools[0].getHashedSet(key).getDisks() { + if _, err := disk.ReadVersion(t.Context(), "", p.bucket, key, listed.VersionID, ReadOptions{}); err == nil { + t.Fatal("heal resurrected a deleted version") + } + } +} + +func TestDecommissionCancellationSurvivesReload(t *testing.T) { + z, _ := migrationTestPools(t) + z.poolMeta.dontSave = false + if err := z.StartDecommission(t.Context(), 0); err != nil { + t.Fatal(err) + } + // Install the worker's cancel handle without racing an actual migration. + _, cancel := context.WithCancel(t.Context()) + defer cancel() + z.decommissionCancelers[0] = cancel + if err := z.DecommissionCancel(t.Context(), 0); err != nil { + t.Fatal(err) + } + // Late worker completion/failure must not overwrite the durable cancel. + if err := z.DecommissionFailed(t.Context(), 0); err != nil { + t.Fatal(err) + } + if err := z.CompleteDecommission(t.Context(), 0); err != nil { + t.Fatal(err) + } + for _, pool := range z.serverPools { + var loaded poolMeta + if err := loaded.load(t.Context(), pool, z.serverPools); err != nil { + t.Fatal(err) + } + state := loaded.Pools[0].Decommission + if state == nil || !state.Canceled || state.Complete || state.Failed { + t.Fatalf("persisted cancellation changed: %+v", state) + } + if got := loaded.returnResumablePools(); len(got) != 0 { + t.Fatalf("canceled pool resumes after reload: %v", got) + } + } +}