mirror of
https://github.com/pgsty/minio.git
synced 2026-11-10 01:31:52 +03:00
fix(ilm): queue explicit version heals after corrupt history refusal
Signed-off-by: Feng Ruohang <rh@vonng.com>
This commit is contained in:
@@ -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)
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user