Files
minio/cmd/erasure-server-pool-delete-all_test.go
T
Feng Ruohang bd39bab339 test(ilm): cover the history majority, approval check and tier ownership gates
Three safeguards of the lifecycle and tier cleanup paths had no test
that failed when they were removed:

- authoritativeHistory accepts a revision only on more than half of the
  drives. TestPoolsLifecycleHalfDriveMarker holds a delete marker on
  exactly half of sixteen drives, which is the read quorum of a marker,
  and checks that the history is refused, that the expiry of the
  version under it and of the key are refused and every drive keeps what
  it has, and that one more drive decides. With the guard removed the
  strict merge at the same majority still refuses the history; with the
  merge relaxed as well the test fails, so it pins the refusal, not one
  of its two layers.
- deleteAllApproved refuses a purge whose placement read differs from
  the versions it approved. deleteAllBeforePlacement is a test-only
  injection point between the two reads, nil in production.
  TestPurgeAllVersionsRefusesChangedApproval writes a new version and
  rewrites the latest one there, as a writer that bypasses the object
  lock would, and checks the write quorum refusal, the
  history_unavailable count, that nothing is deleted and that the next
  purge removes the key. With the check disabled the purge deletes the
  versions it never approved.
- readTierHistory refuses the ownership of a remote object while a
  drive of any pool cannot be read, and xlMetaV2.DeleteVersion deletes
  a free record only with the remote tuple it names.
  TestTierOwnershipReadFailsClosed fails one drive of the pool that
  holds nothing of the key as offline, not found, faulty and
  unformatted, and checks that the expiry worker's and the
  decommission's cleanup are refused with a read quorum error and
  succeed once the drive reads again.
  TestFreeRecordDeleteNeedsMatchingTuple deletes a free record with
  another tier, remote object and remote version, in memory and on an
  xlStorage drive, and checks the record stays until the tuple matches.
  With either check removed the remote object or the record is deleted.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
Signed-off-by: Feng Ruohang <rh@vonng.com>
2026-11-07 17:04:02 +08:00

1157 lines
42 KiB
Go

// Copyright (c) 2026 Feng Ruohang
//
// 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 <http://www.gnu.org/licenses/>.
package cmd
import (
"bytes"
"context"
"errors"
"fmt"
"io"
"path"
"slices"
"strings"
"testing"
"time"
"github.com/minio/madmin-go/v3"
"github.com/minio/minio/internal/amztime"
"github.com/minio/minio/internal/bucket/lifecycle"
objectlock "github.com/minio/minio/internal/bucket/object/lock"
"github.com/minio/minio/internal/bucket/versioning"
"github.com/minio/minio/internal/event"
xhttp "github.com/minio/minio/internal/http"
"github.com/minio/minio/internal/pubsub"
"github.com/pgsty/silo-pkg/v3/workers"
)
// An expiry of all versions of a key must remove the versions of that key
// and nothing else. On the drives, the keys under K/ live inside the
// directory of K, in the same erasure set or another one, and in other
// pools the same name is a directory too.
const purgeRules = "<Rule><ID>eav</ID><Status>Enabled</Status><Filter><Prefix>ea/</Prefix></Filter>" +
"<Expiration><Days>1</Days><ExpiredObjectAllVersions>true</ExpiredObjectAllVersions></Expiration></Rule>" +
"<Rule><ID>dme</ID><Status>Enabled</Status><Filter><Prefix>dm/</Prefix></Filter>" +
"<DelMarkerExpiration><Days>1</Days></DelMarkerExpiration></Rule>"
// purgeTest is a bucket with purgeRules and a remote tier WARM.
type purgeTest struct {
t *testing.T
z *erasureServerPools
bucket string
tier *tierExpiryTestBackend
bodies map[string]string // body -> label
}
// purgeSinglePool returns one pool of drives drives, two erasure sets for 32.
func purgeSinglePool(t *testing.T, drives int) *erasureServerPools {
t.Helper()
ctx, cancel := context.WithCancel(t.Context())
obj, dirs, err := prepareErasure(ctx, drives)
if err != nil {
cancel()
t.Fatal(err)
}
z := obj.(*erasureServerPools)
previous := newObjectLayerFn()
setObjectLayer(z)
t.Cleanup(func() { cancel(); z.Shutdown(context.Background()); removeRoots(dirs); setObjectLayer(previous) })
return z
}
func newPurgeTest(t *testing.T, z *erasureServerPools, bucket string, opts MakeBucketOptions, rules string) *purgeTest {
t.Helper()
ctx := t.Context()
if err := z.MakeBucket(ctx, bucket, opts); err != nil {
t.Fatal(err)
}
if rules != "" {
meta, err := globalBucketMetadataSys.Get(bucket)
if err != nil {
t.Fatal(err)
}
meta.LifecycleConfigXML = []byte("<LifecycleConfiguration>" + rules + "</LifecycleConfiguration>")
if err := meta.Save(ctx, z); err != nil {
t.Fatal(err)
}
globalBucketMetadataSys.Set(bucket, meta)
}
p := &purgeTest{t: t, z: z, bucket: bucket, tier: &tierExpiryTestBackend{objects: map[string][]byte{}}, bodies: map[string]string{}}
prevTier := globalTierConfigMgr
globalTierConfigMgr = NewTierConfigMgr()
globalTierConfigMgr.drivercache["WARM"] = p.tier
t.Cleanup(func() { globalTierConfigMgr = prevTier })
return p
}
func daysAgo(days float64) time.Time {
return UTCNow().Add(-time.Duration(days * float64(24*time.Hour)))
}
// put writes a version of name to pool with opts, at mtime.
func (p *purgeTest) put(pool int, name, label string, mtime time.Time, opts ObjectOptions) ObjectInfo {
p.t.Helper()
body := name + " " + label
if HasSuffix(name, SlashSeparator) {
body = ""
}
opts.MTime = mtime
oi, err := p.z.serverPools[pool].PutObject(p.t.Context(), p.bucket, encodeDirObject(name),
mustGetPutObjReader(p.t, bytes.NewBufferString(body), int64(len(body)), "", ""), opts)
if err != nil {
p.t.Fatalf("put %s %s: %v", name, label, err)
}
p.bodies[body] = label
return oi
}
func (p *purgeTest) version(pool int, name, label string, days float64) ObjectInfo {
p.t.Helper()
return p.put(pool, name, label, daysAgo(days), ObjectOptions{Versioned: true})
}
func (p *purgeTest) marker(pool int, name string, days float64) {
p.t.Helper()
if _, err := p.z.serverPools[pool].DeleteObject(p.t.Context(), p.bucket, encodeDirObject(name),
ObjectOptions{Versioned: true, VersionID: mustGetUUID(), DeleteMarker: true, MTime: daysAgo(days)}); err != nil {
p.t.Fatal(err)
}
}
// tiered writes a version of name to pool whose data is in the remote tier.
func (p *purgeTest) tiered(pool int, name, label string, days float64) ObjectInfo {
p.t.Helper()
ctx := p.t.Context()
oi := p.version(pool, name, label, days)
infos, errs := readAllFileInfo(ctx, p.z.serverPools[pool].getHashedSet(name).getDisks(), "", p.bucket, name, oi.VersionID, false, false)
if errs[0] != nil {
p.t.Fatal(errs[0])
}
fi := infos[0]
remote := "remote/" + name + "/" + label
fi.TransitionStatus, fi.TransitionTier, fi.TransitionedObjName = lifecycle.TransitionComplete, "WARM", remote
p.tier.mu.Lock()
p.tier.objects[remote] = []byte(name)
p.tier.mu.Unlock()
if err := p.z.serverPools[pool].DecomTieredObject(ctx, p.bucket, name, fi, ObjectOptions{}); err != nil {
p.t.Fatal(err)
}
return oi
}
func (p *purgeTest) remotes() int {
p.tier.mu.Lock()
defer p.tier.mu.Unlock()
return len(p.tier.objects)
}
// held returns what every drive of the set of name in pool holds, newest
// first, free versions included, as they agree, or nil without xl.meta.
func (p *purgeTest) held(pool int, name string) []string {
p.t.Helper()
enc := encodeDirObject(name)
var first []string
for i, disk := range p.z.serverPools[pool].getHashedSet(enc).getDisks() {
var got []string
buf, err := disk.ReadAll(p.t.Context(), p.bucket, pathJoin(enc, xlStorageFormatFile))
if err == nil {
fivs, err := getFileInfoVersions(buf, p.bucket, enc, true)
if err != nil {
p.t.Fatal(err)
}
for _, fi := range fivs.Versions {
switch {
case fi.TierFreeVersion():
got = append(got, "free")
case fi.Deleted:
got = append(got, "DM")
case fi.TransitionStatus == lifecycle.TransitionComplete:
got = append(got, "tiered")
default:
got = append(got, "hot")
}
}
} else if !errors.Is(err, errFileNotFound) {
p.t.Fatalf("drive %d of %s: %v", i, name, err)
}
if i == 0 {
first = got
} else if !slices.Equal(got, first) {
p.t.Fatalf("%s in pool %d: drive %d holds %v, drive 0 %v", name, pool, i, got, first)
}
}
return first
}
// get returns the label of the version a GET of name reads, or "-".
func (p *purgeTest) get(name string) string {
p.t.Helper()
gr, err := p.z.GetObjectNInfo(p.t.Context(), p.bucket, name, nil, nil, ObjectOptions{})
if err != nil {
if isErrObjectNotFound(err) || isErrVersionNotFound(err) || isErrMethodNotAllowed(err) {
return "-"
}
return fmt.Sprintf("<%v>", err)
}
defer gr.Close()
data, err := io.ReadAll(gr)
if err != nil {
p.t.Fatal(err)
}
if l, ok := p.bodies[string(data)]; ok {
return l
}
return "?" + string(data)
}
// childIn returns a key under parent/ in the same or another erasure set of
// pool than parent.
func (p *purgeTest) childIn(pool int, parent string, same bool) string {
sets := p.z.serverPools[pool]
for i := 0; ; i++ {
name := fmt.Sprintf("%s/c%03d", parent, i)
if (sets.getHashedSetIndex(name) == sets.getHashedSetIndex(parent)) == same {
return name
}
}
}
// scanned returns the versions of name that the first drive of its set in
// pool holds, as the scanner of that drive reads them.
func (p *purgeTest) scanned(pool int, name string) []ObjectInfo {
p.t.Helper()
enc := encodeDirObject(name)
disk := p.z.serverPools[pool].getHashedSet(enc).getDisks()[0]
buf, err := disk.ReadAll(p.t.Context(), p.bucket, pathJoin(enc, xlStorageFormatFile))
if err != nil {
p.t.Fatalf("read %s: %v", name, err)
}
fivs, err := getFileInfoVersions(buf, p.bucket, enc, false)
if err != nil {
p.t.Fatal(err)
}
vc, _ := globalBucketVersioningSys.Get(p.bucket)
objInfos := make([]ObjectInfo, len(fivs.Versions))
for i, fi := range fivs.Versions {
objInfos[i] = fi.ToObjectInfo(p.bucket, name, vc != nil && vc.Versioned(name))
}
return objInfos
}
// scanQueue applies the lifecycle of the bucket to the versions that pool
// holds of name, as the scanner does, and returns what it queued.
func (p *purgeTest) scanQueue(pool int, name string) []expiryOp {
p.t.Helper()
ops, _ := p.scan(pool, name)
return ops
}
// scan is scanQueue that also returns the size the scanner accounts for
// each version, by version ID.
func (p *purgeTest) scan(pool int, name string) ([]expiryOp, map[string]int64) {
p.t.Helper()
ctx := p.t.Context()
enc := encodeDirObject(name)
objInfos := p.scanned(pool, name)
lc, err := globalLifecycleSys.Get(p.bucket)
if err != nil {
p.t.Fatal(err)
}
lr, err := globalBucketObjectLockSys.Get(p.bucket)
if err != nil {
p.t.Fatal(err)
}
queue := make(chan expiryOp, 100)
es := &expiryState{ctx: ctx, objAPI: p.z}
workers := []chan expiryOp{queue}
es.workers.Store(&workers)
prev := globalExpiryState
globalExpiryState = es
defer func() { globalExpiryState = prev }()
item := scannerItem{Path: enc, bucket: p.bucket, objectName: path.Base(enc), lifeCycle: lc}
sizes := map[string]int64{}
item.applyActions(ctx, p.z, objInfos, lr, &sizeSummary{}, func(oi ObjectInfo, sz, _ int64, _ *sizeSummary) {
sizes[deleteAllVersionID(oi.VersionID)] += sz
})
var ops []expiryOp
for len(queue) > 0 {
ops = append(ops, <-queue)
}
return ops, sizes
}
// deleteAllTask is the expiry of all versions of name that a scanner queues
// for the latest version that pool holds, as the scanner of a drive or a
// pool that holds no tiered version of the key queues it.
func (p *purgeTest) deleteAllTask(pool int, name string, action lifecycle.Action) expiryOp {
p.t.Helper()
return expiryTask{objInfo: p.scanned(pool, name)[0], event: lifecycle.Event{Action: action}, src: lcEventSrc_Scanner}
}
// queuesDeleteAll reports whether ops holds an expiry of all versions.
func queuesDeleteAll(ops []expiryOp) bool {
return slices.ContainsFunc(ops, func(op expiryOp) bool {
task, ok := op.(expiryTask)
return ok && task.event.Action.DeleteAll()
})
}
// expire runs what the scanner queues for name in pool with an expiry
// worker, and fails unless it queued an expiry of all versions.
func (p *purgeTest) expire(pool int, name string) {
p.t.Helper()
ops := p.scanQueue(pool, name)
if !queuesDeleteAll(ops) {
p.t.Fatalf("the scanner queued no expiry of all versions of %s: %v", name, ops)
}
p.run(ops)
}
func (p *purgeTest) run(ops []expiryOp) {
ch := make(chan expiryOp, len(ops))
for _, op := range ops {
ch <- op
}
close(ch)
es := &expiryState{ctx: p.t.Context(), objAPI: p.z}
es.Worker(ch)
}
// matched returns the latest version of name in all pools, with the number
// of its versions and delete markers, as the walk of a batch expiration
// matches it for a purge of all versions.
func (p *purgeTest) matched(name string) expireObjInfo {
p.t.Helper()
var latest FileInfo
ids := map[string]bool{}
var markers int64
for _, pool := range p.z.serverPools {
versions, err := pool.getHashedSet(name).objectVersions(p.t.Context(), p.bucket, encodeDirObject(name))
if err != nil {
p.t.Fatalf("versions of %s: %v", name, err)
}
for _, fi := range versions {
if fi.ModTime.After(latest.ModTime) {
latest = fi
}
if !ids[fi.VersionID] && fi.Deleted {
markers++
}
ids[fi.VersionID] = true
}
}
if len(ids) == 0 {
p.t.Fatalf("%s has no versions", name)
}
vc, _ := globalBucketVersioningSys.Get(p.bucket)
oi := latest.ToObjectInfo(p.bucket, name, vc != nil && vc.Versioned(name))
oi.IsLatest, oi.NumVersions = true, len(ids)
return expireObjInfo{ObjectInfo: oi, ExpireAll: true, DeleteMarkerCount: markers}
}
// batch runs the deletes of a batch expiration job of the root user for
// walked, the keys its walk matched, as one batch of a walk that reached its
// 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)
}
globalBatchJobsMetrics.Unlock()
sub, err := newBatchJobSubmitter(globalActiveCred)
if err != nil {
p.t.Fatal(err)
}
job := BatchJobRequest{ID: "purge-" + mustGetUUID(), Started: time.Now(), Submitter: sub, Expire: &BatchJobExpire{Bucket: p.bucket}}
ri := &batchJobInfo{JobID: job.ID}
wk, err := workers.New(1)
if err != nil {
p.t.Fatal(err)
}
ch := make(chan []expireObjInfo, 1)
ch <- walked
close(ch)
refuse := func(err error) { p.t.Errorf("batch job refused: %v", err) }
batchObjsForDelete(p.t.Context(), refuse, func() bool { return true }, job.Expire, ri, job, p.z, wk, ch)
wk.Wait()
return ri
}
// purgeObserver collects the removal events and ILM expiry traces.
type purgeObserver struct {
events chan event.Event
traces chan madmin.TraceInfo
}
func observePurges(t *testing.T) *purgeObserver {
t.Helper()
done := make(chan struct{})
t.Cleanup(func() { close(done) })
o := &purgeObserver{events: make(chan event.Event, 64), traces: make(chan madmin.TraceInfo, 64)}
var mask pubsub.Mask
mask.MergeMaskable(event.ObjectRemovedAll)
mask.MergeMaskable(event.ILMDelMarkerExpirationDelete)
if err := globalHTTPListen.Subscribe(mask, o.events, done, nil); err != nil {
t.Fatal(err)
}
if err := globalTrace.Subscribe(madmin.TraceILM, o.traces, done, nil); err != nil {
t.Fatal(err)
}
return o
}
// drain returns the removal events and the custom fields of the ILM expiry
// traces published so far. Both publishers deliver synchronously.
func (o *purgeObserver) drain() (events []string, traces []map[string]string) {
for {
select {
case ev := <-o.events:
events = append(events, ev.EventName.String()+" "+ev.S3.Object.Key)
case tr := <-o.traces:
if tr.FuncName == ILMExpiry {
traces = append(traces, tr.Custom)
}
default:
return events, traces
}
}
}
// purgeFaultDisk fails the version deletes of one key, or of one of its
// versions.
type purgeFaultDisk struct {
StorageAPI
bucket, object, versionID string
}
func (d purgeFaultDisk) fails(volume, path, versionID string) bool {
return volume == d.bucket && path == d.object && (d.versionID == "" || versionID == d.versionID)
}
func (d purgeFaultDisk) DeleteVersion(ctx context.Context, volume, path string, fi FileInfo, forceDelMarker bool, opts DeleteOptions) error {
if d.fails(volume, path, fi.VersionID) {
return errFaultyDisk
}
return d.StorageAPI.DeleteVersion(ctx, volume, path, fi, forceDelMarker, opts)
}
func (d purgeFaultDisk) DeleteVersions(ctx context.Context, volume string, versions []FileInfoVersions, opts DeleteOptions) []error {
errs := make([]error, len(versions))
for i, v := range versions {
for _, fi := range v.Versions {
if d.fails(volume, v.Name, fi.VersionID) {
errs[i] = errFaultyDisk
}
}
if errs[i] == nil {
errs[i] = d.StorageAPI.DeleteVersions(ctx, volume, []FileInfoVersions{v}, opts)[0]
}
}
return errs
}
// failVersionDeletes makes every drive of pool fail the version deletes of
// object, of versionID only when set, until the returned function runs.
func (p *purgeTest) failVersionDeletes(pool int, object, versionID string) func() {
p.t.Helper()
sets := p.z.serverPools[pool]
sets.erasureDisksMu.Lock()
prior := sets.erasureDisks
wrapped := make([][]StorageAPI, len(prior))
for i, disks := range prior {
wrapped[i] = append([]StorageAPI(nil), disks...)
for j := range disks {
wrapped[i][j] = purgeFaultDisk{StorageAPI: disks[j], bucket: p.bucket, object: object, versionID: versionID}
}
}
sets.erasureDisks = wrapped
sets.erasureDisksMu.Unlock()
restored := false
restore := func() {
if restored {
return
}
restored = true
sets.erasureDisksMu.Lock()
sets.erasureDisks = prior
sets.erasureDisksMu.Unlock()
}
p.t.Cleanup(restore)
return restore
}
// Children of the key in its own erasure set and in the other one keep all
// their versions, for both lifecycle actions that remove all versions.
func TestPurgeAllVersionsKeepsKeysUnderTheKey(t *testing.T) {
z := purgeSinglePool(t, 32)
p := newPurgeTest(t, z, "purge-children", MakeBucketOptions{VersioningEnabled: true}, purgeRules)
if len(z.serverPools[0].sets) < 2 {
t.Fatalf("fixture needs two erasure sets, has %d", len(z.serverPools[0].sets))
}
for _, parent := range []string{"ea/parent", "dm/parent"} {
p.version(0, parent, "v1", 5)
if strings.HasPrefix(parent, "dm/") {
p.marker(0, parent, 3)
} else {
p.version(0, parent, "v2", 4)
}
children := []string{p.childIn(0, parent, true), p.childIn(0, parent, false), parent + "/deep/grandchild"}
for _, child := range children {
p.version(0, child, "old", 5)
p.version(0, child, "new", 0)
}
p.expire(0, parent)
if got := p.held(0, parent); got != nil {
t.Errorf("%s keeps %v after the expiry of all its versions", parent, got)
}
for _, child := range children {
if got := p.held(0, child); !slices.Equal(got, []string{"hot", "hot"}) {
t.Errorf("%s in set %d (parent in set %d) holds %v, want both versions",
child, z.serverPools[0].getHashedSetIndex(child), z.serverPools[0].getHashedSetIndex(parent), got)
}
if got := p.get(child); got != "new" {
t.Errorf("%s reads %s, want new", child, got)
}
}
}
}
// A child in another pool, and copies of the key in both pools.
func TestPurgeAllVersionsKeepsKeysInOtherPools(t *testing.T) {
z, _ := consistencyPools(t)
p := newPurgeTest(t, z, "purge-pools", MakeBucketOptions{VersioningEnabled: true}, purgeRules)
const parent = "ea/parent"
p.version(0, parent, "v1", 6)
moved := p.version(1, parent, "v2", 5)
// A copy of v2 in pool 0 as an interrupted move leaves it.
p.put(0, parent, "v2", moved.ModTime, ObjectOptions{Versioned: true, VersionID: moved.VersionID})
p.version(1, parent+"/child", "new", 0)
p.version(0, parent+"/child0", "new", 0)
p.expire(1, parent)
for pool := range z.serverPools {
if got := p.held(pool, parent); got != nil {
t.Errorf("pool %d keeps %v of %s", pool, got, parent)
}
}
if got := p.held(1, parent+"/child"); !slices.Equal(got, []string{"hot"}) {
t.Errorf("child in the other pool holds %v", got)
}
if got := p.held(0, parent+"/child0"); !slices.Equal(got, []string{"hot"}) {
t.Errorf("child in the key's pool holds %v", got)
}
}
// Controls: a key without children loses all its versions, and neighbors
// and directory objects behave as before.
func TestPurgeAllVersionsControls(t *testing.T) {
z := purgeSinglePool(t, 32)
p := newPurgeTest(t, z, "purge-controls", MakeBucketOptions{VersioningEnabled: true}, purgeRules)
p.version(0, "ea/plain", "v1", 6)
p.marker(0, "ea/plain", 5)
p.version(0, "ea/plain", "v2", 4)
for _, n := range []string{"ea/plainx", "ea/plain-1"} {
p.version(0, n, "new", 0)
}
p.version(0, "ea/dir/", "", 5)
p.version(0, "ea/dir/kid", "new", 0)
p.expire(0, "ea/plain")
p.expire(0, "ea/dir/")
for _, n := range []string{"ea/plain", "ea/dir/"} {
if got := p.held(0, n); got != nil {
t.Errorf("%s keeps %v", n, got)
}
}
for _, n := range []string{"ea/plainx", "ea/plain-1", "ea/dir/kid"} {
if got := p.get(n); got != "new" {
t.Errorf("%s reads %s, want new", n, got)
}
}
}
// The directory of a directory object K/ can hold keys whose names contain
// its encoded name. They stay too.
func TestPurgeAllVersionsKeepsKeysUnderDirectoryObject(t *testing.T) {
z := purgeSinglePool(t, 16)
p := newPurgeTest(t, z, "purge-dir-encoded", MakeBucketOptions{VersioningEnabled: true}, purgeRules)
p.version(0, "ea/dir/", "", 5)
if _, err := z.PutObject(t.Context(), p.bucket, "ea/dir"+globalDirSuffix+"/odd",
mustGetPutObjReader(t, bytes.NewBufferString("odd"), 3, "", ""), ObjectOptions{Versioned: true}); err != nil {
t.Fatal(err)
}
p.expire(0, "ea/dir/")
if got := p.held(0, "ea/dir/"); got != nil {
t.Errorf("ea/dir/ keeps %v", got)
}
if _, err := z.GetObjectInfo(t.Context(), p.bucket, "ea/dir"+globalDirSuffix+"/odd", ObjectOptions{}); err != nil {
t.Errorf("the key inside the directory object's directory: %v", err)
}
}
// Force delete stays a prefix delete.
func TestPurgeAllVersionsForceDeleteControl(t *testing.T) {
z := purgeSinglePool(t, 16)
p := newPurgeTest(t, z, "purge-force", MakeBucketOptions{VersioningEnabled: true}, "")
for _, n := range []string{"fd/parent", "fd/parent/child", "fd/parentx"} {
p.version(0, n, "v1", 0)
}
if _, err := z.DeleteObject(t.Context(), p.bucket, "fd/parent", ObjectOptions{DeletePrefix: true}); err != nil {
t.Fatal(err)
}
for n, want := range map[string]string{"fd/parent": "-", "fd/parent/child": "-", "fd/parentx": "v1"} {
if got := p.get(n); got != want {
t.Errorf("%s reads %s after a force delete of fd/parent, want %s", n, got, want)
}
}
}
// A batch expiry that keeps no versions purges the key it matched and
// counts what it removed.
func TestPurgeAllVersionsBatchKeepsKeysUnderTheKey(t *testing.T) {
z := purgeSinglePool(t, 32)
p := newPurgeTest(t, z, "purge-batch", MakeBucketOptions{VersioningEnabled: true}, "")
const parent = "bx/parent"
p.version(0, parent, "v1", 5)
p.marker(0, parent, 4)
p.version(0, parent, "v2", 3)
others := []string{p.childIn(0, parent, true), p.childIn(0, parent, false), "bx/parentx"}
for _, n := range others {
p.version(0, n, "old", 5)
}
ri := p.batch(p.matched(parent))
if got := p.held(0, parent); got != nil {
t.Errorf("%s keeps %v", parent, got)
}
for _, n := range others {
if got := p.get(n); got != "old" {
t.Errorf("%s reads %s, want old", n, got)
}
}
if ri.Objects != 2 || ri.DeleteMarkers != 1 || ri.ObjectsFailed != 0 || ri.DeleteMarkersFailed != 0 {
t.Errorf("job counts objects=%d markers=%d failed=%d/%d, want 2, 1, 0/0", ri.Objects, ri.DeleteMarkers, ri.ObjectsFailed, ri.DeleteMarkersFailed)
}
}
// Decommission and rebalance run the expiry of all versions of a key whose
// child is new; the child stays. A walk does not list keys under an object
// key, so the child is left for a later pass, as without ILM.
func TestPurgeAllVersionsMigrationKeepsKeysUnderTheKey(t *testing.T) {
for _, kind := range []string{"decommission", "rebalance"} {
t.Run(kind, func(t *testing.T) {
m := newMigrationILM(t, versioning.Enabled, false,
migrationILMRule("<Expiration><Days>1</Days><ExpiredObjectAllVersions>true</ExpiredObjectAllVersions></Expiration>"))
m.aged(3, "obj-01")
m.put("obj-01/child", "v1", ObjectOptions{Versioned: true})
m.put("obj-01x", "v1", ObjectOptions{Versioned: true})
if kind == "rebalance" {
migrationTestWalk(t, m.z, m.bucket, 0, kind)
} else {
if err := m.z.StartDecommission(t.Context(), 0); err != nil {
t.Fatal(err)
}
m.z.doDecommissionInRoutine(t.Context(), 0)
}
for _, n := range []string{"obj-01/child", "obj-01x"} {
if _, err := m.z.GetObjectInfo(t.Context(), m.bucket, n, ObjectOptions{}); err != nil {
t.Errorf("%s after %s: %v", n, kind, err)
}
}
if _, err := m.z.GetObjectInfo(t.Context(), m.bucket, "obj-01", ObjectOptions{}); !isErrObjectNotFound(err) {
t.Errorf("obj-01 after %s: %v", kind, err)
}
for pool := range m.z.serverPools {
if v := m.view(pool)["obj-01"]; len(v) > 0 {
t.Errorf("pool %d keeps %v of obj-01", pool, v)
}
}
})
}
}
// A version written after the decision is kept: the expiry is decided again
// on the history under the lock, also on one pool, and a batch expiry only
// purges while the version it matched is the latest.
func TestPurgeAllVersionsRefusesStaleDecision(t *testing.T) {
z := purgeSinglePool(t, 16)
p := newPurgeTest(t, z, "purge-stale", MakeBucketOptions{VersioningEnabled: true}, purgeRules)
p.version(0, "ea/stale", "old", 5)
ops := p.scanQueue(0, "ea/stale")
p.version(0, "ea/stale", "fresh", 0)
p.run(ops)
if got := p.held(0, "ea/stale"); !slices.Equal(got, []string{"hot", "hot"}) {
t.Errorf("lifecycle: ea/stale holds %v, want both versions", got)
}
p.version(0, "bx/stale", "old", 5)
walked := p.matched("bx/stale")
p.version(0, "bx/stale", "fresh", 0)
ri := p.batch(walked)
if got := p.held(0, "bx/stale"); !slices.Equal(got, []string{"hot", "hot"}) {
t.Errorf("batch: bx/stale holds %v, want both versions", got)
}
if ri.Objects != 0 || ri.ObjectsFailed != 0 {
t.Errorf("batch counts a stale match: objects=%d failed=%d", ri.Objects, ri.ObjectsFailed)
}
}
// The purge reads the placement of the versions it approved and may only
// confirm them. A write that reaches the key between the approval and that
// read, as one that bypasses the object lock would, changes what the purge
// is about to delete: a version it did not approve, or another write of an
// approved one. The purge refuses the key with a write quorum error, which
// counts as an unavailable history, and deletes nothing; the next purge,
// which approves what it finds, removes the key.
func TestPurgeAllVersionsRefusesChangedApproval(t *testing.T) {
for _, kind := range []string{"new-version", "latest-rewritten"} {
t.Run(kind, func(t *testing.T) {
z := purgeSinglePool(t, 16)
p := newPurgeTest(t, z, "purge-approval", MakeBucketOptions{VersioningEnabled: true}, purgeRules)
ctx := t.Context()
const key = "ea/approval"
p.version(0, key, "v1", 5)
latest := p.version(0, key, "v2", 4)
ops := p.scanQueue(0, key)
// Each purge finds the history changed since its approval.
var writes int
deleteAllBeforePlacement = func(bucket, object string) {
if bucket != p.bucket || object != key {
return
}
writes++
label := fmt.Sprintf("v%d", 2+writes)
opts := ObjectOptions{Versioned: true, NoLock: true}
mtime := daysAgo(3)
if kind == "latest-rewritten" {
label = fmt.Sprintf("v2-rewrite-%d", writes)
opts.VersionID, mtime = latest.VersionID, latest.ModTime
}
p.put(0, key, label, mtime, opts)
}
t.Cleanup(func() { deleteAllBeforePlacement = nil })
_, err := z.DeleteObject(ctx, p.bucket, key, ObjectOptions{DeleteAllVersions: true, Expiration: ExpirationOptions{Expire: true}})
if !isErrWriteQuorum(err) {
t.Fatalf("purge of a changed history: %v, want a write quorum error", err)
}
delta := incompleteDelta(func() { p.run(ops) })
if delta[ilmIncompleteHistoryUnavailable] != 1 {
t.Errorf("counted %v, want one %s", delta, ilmIncompleteHistoryUnavailable)
}
if writes != 2 {
t.Fatalf("the history changed %d times, want 2", writes)
}
want, read := []string{"hot", "hot", "hot", "hot"}, "v4"
if kind == "latest-rewritten" {
want, read = []string{"hot", "hot"}, "v2-rewrite-2"
}
if got := p.held(0, key); !slices.Equal(got, want) {
t.Fatalf("the refused purges left %v, want %v", got, want)
}
if got := p.get(key); got != read {
t.Errorf("the key reads %s, want %s", got, read)
}
deleteAllBeforePlacement = nil
p.expire(0, key)
if got := p.held(0, key); got != nil {
t.Errorf("the purge of the unchanged history left %v", got)
}
})
}
}
// Control: on two pools the stale decision was already refused.
func TestPurgeAllVersionsStaleDecisionTwoPoolsControl(t *testing.T) {
z, _ := consistencyPools(t)
p := newPurgeTest(t, z, "purge-stale-pools", MakeBucketOptions{VersioningEnabled: true}, purgeRules)
p.version(0, "ea/stale", "old", 5)
ops := p.scanQueue(0, "ea/stale")
p.version(1, "ea/stale", "fresh", 0)
p.run(ops)
if got := p.get("ea/stale"); got != "fresh" {
t.Errorf("ea/stale reads %s, want fresh", got)
}
}
// When an older version cannot be deleted, the latest version stays and so
// the read result does not change. The worker audits the partial purge and
// sends no removal event; the next expiry completes it.
func TestPurgeAllVersionsDeletesLatestLast(t *testing.T) {
z, _ := consistencyPools(t)
p := newPurgeTest(t, z, "purge-barrier", MakeBucketOptions{VersioningEnabled: true}, purgeRules)
const key = "ea/barrier"
p.version(0, key, "v1", 6)
v2 := p.version(1, key, "v2", 5)
p.version(0, key, "v3", 4)
o := observePurges(t)
restore := p.failVersionDeletes(1, key, v2.VersionID)
p.expire(0, key)
if got := p.get(key); got != "v3" {
t.Errorf("reads %s after a failed purge, want v3", got)
}
if got := p.held(0, key); !slices.Equal(got, []string{"hot"}) {
t.Errorf("pool 0 holds %v, want only the latest version", got)
}
if got := p.held(1, key); !slices.Equal(got, []string{"hot"}) {
t.Errorf("pool 1 holds %v, want the version that failed", got)
}
events, traces := o.drain()
if len(events) != 0 {
t.Errorf("a partial purge sent removal events %v", events)
}
if len(traces) != 1 || traces[0]["ilm-partial"] != "true" || traces[0]["ilm-versions-removed"] != "1" {
t.Errorf("a partial purge traced %v, want one partial expiry that removed 1 version", traces)
}
restore()
p.expire(0, key)
for pool := range z.serverPools {
if got := p.held(pool, key); got != nil {
t.Errorf("pool %d keeps %v once the purge completed", pool, got)
}
}
events, traces = o.drain()
if !slices.Equal(events, []string{event.ObjectRemovedDeleteAllVersions.String() + " " + key}) {
t.Errorf("completed purge sent %v", events)
}
if len(traces) != 1 || traces[0]["ilm-partial"] != "" || traces[0]["ilm-versions-removed"] != "2" {
t.Errorf("completed purge traced %v, want one expiry that removed 2 versions", traces)
}
}
// A key with a transitioned version is refused, whether the latest version
// is hot or tiered; its versions and remote objects stay.
func TestPurgeAllVersionsRefusesTieredKeys(t *testing.T) {
z := purgeSinglePool(t, 16)
p := newPurgeTest(t, z, "purge-tiered", MakeBucketOptions{VersioningEnabled: true}, purgeRules)
p.tiered(0, "ea/tiered-older", "v1", 6)
p.version(0, "ea/tiered-older", "v2", 5)
p.version(0, "bx/tiered-older", "v1", 6)
p.tiered(0, "bx/tiered-older", "v2", 5)
p.version(0, "bx/tiered-older", "v3", 4)
p.version(0, "bx/tiered-latest", "v1", 6)
p.tiered(0, "bx/tiered-latest", "v2", 5)
remotes := p.remotes()
// The scanner of a pool that holds the tiered version does not queue
// the expiry; one that does not hold it queues it, and the purge refuses.
if ops := p.scanQueue(0, "ea/tiered-older"); queuesDeleteAll(ops) {
t.Errorf("the scanner queued an expiry of all versions that the purge refuses: %v", ops)
}
p.run([]expiryOp{p.deleteAllTask(0, "ea/tiered-older", lifecycle.DeleteAllVersionsAction)})
ri := p.batch(p.matched("bx/tiered-older"), p.matched("bx/tiered-latest"))
for name, want := range map[string][]string{
"ea/tiered-older": {"hot", "tiered"},
"bx/tiered-older": {"hot", "tiered", "hot"},
"bx/tiered-latest": {"tiered", "hot"},
} {
if got := p.held(0, name); !slices.Equal(got, want) {
t.Errorf("%s holds %v, want %v", name, got, want)
}
}
if got := p.remotes(); got != remotes {
t.Errorf("tier holds %d remote objects, want %d", got, remotes)
}
if ri.Objects != 0 || ri.ObjectsFailed != 0 || ri.KeysRefused != 2 {
t.Errorf("refused batch purges counted objects=%d failed=%d refused=%d, want 2 refused keys", ri.Objects, ri.ObjectsFailed, ri.KeysRefused)
}
}
// Control: an expiry of all versions of a transitioned latest version takes
// the transitioned path as before.
func TestPurgeAllVersionsTieredLatestControl(t *testing.T) {
z := purgeSinglePool(t, 16)
p := newPurgeTest(t, z, "purge-tiered-latest", MakeBucketOptions{VersioningEnabled: true}, purgeRules)
p.version(0, "ea/obj", "v1", 6)
p.tiered(0, "ea/obj", "v2", 5)
p.expire(0, "ea/obj")
if got := p.held(0, "ea/obj"); !slices.Equal(got, []string{"DM", "tiered", "hot"}) {
t.Errorf("holds %v, want a delete marker over both versions", got)
}
if got := p.remotes(); got != 1 {
t.Errorf("tier holds %d remote objects, want 1", got)
}
}
// A free version that an earlier delete of a tiered version left stays for
// the tier cleanup.
func TestPurgeAllVersionsKeepsFreeVersions(t *testing.T) {
z := purgeSinglePool(t, 16)
p := newPurgeTest(t, z, "purge-free", MakeBucketOptions{VersioningEnabled: true}, purgeRules)
tv := p.tiered(0, "ea/obj", "v1", 6)
if _, err := z.DeleteObject(t.Context(), p.bucket, "ea/obj", ObjectOptions{Versioned: true, VersionID: tv.VersionID}); err != nil {
t.Fatal(err)
}
p.version(0, "ea/obj", "v2", 5)
p.expire(0, "ea/obj")
if got := p.held(0, "ea/obj"); !slices.Equal(got, []string{"free"}) {
t.Errorf("holds %v, want only the free version", got)
}
if got := p.get("ea/obj"); got != "-" {
t.Errorf("reads %s", got)
}
}
// Retention and legal hold protect every write that the purge would delete.
func TestPurgeAllVersionsKeepsProtectedWrites(t *testing.T) {
z, _ := consistencyPools(t)
p := newPurgeTest(t, z, "purge-protected", MakeBucketOptions{VersioningEnabled: true, LockEnabled: true}, "")
retained := map[string]string{
strings.ToLower(xhttp.AmzObjectLockMode): string(objectlock.RetCompliance),
strings.ToLower(xhttp.AmzObjectLockRetainUntilDate): amztime.ISO8601Format(UTCNow().Add(24 * time.Hour)),
}
held := map[string]string{strings.ToLower(xhttp.AmzObjectLockLegalHold): string(objectlock.LegalHoldOn)}
p.put(0, "lk/retained", "v1", daysAgo(5), ObjectOptions{Versioned: true, UserDefined: retained})
p.version(0, "lk/retained", "v2", 4)
p.put(0, "lk/held", "v1", daysAgo(5), ObjectOptions{Versioned: true, UserDefined: held})
p.version(0, "lk/held", "v2", 4)
// An older null write in another pool shares the ID of the latest.
p.put(0, "lk/null", "old", daysAgo(5), ObjectOptions{UserDefined: retained})
p.put(1, "lk/null", "new", daysAgo(4), ObjectOptions{})
p.version(0, "lk/free", "v1", 5)
p.version(0, "lk/free", "v2", 4)
ri := p.batch(p.matched("lk/retained"), p.matched("lk/held"), p.matched("lk/null"), p.matched("lk/free"))
for name, want := range map[string]string{"lk/retained": "v2", "lk/held": "v2", "lk/null": "new", "lk/free": "-"} {
if got := p.get(name); got != want {
t.Errorf("%s reads %s, want %s", name, got, want)
}
}
for pool, want := range [][]string{{"hot"}, {"hot"}} {
if got := p.held(pool, "lk/null"); !slices.Equal(got, want) {
t.Errorf("lk/null: pool %d holds %v, want %v", pool, got, want)
}
}
for _, name := range []string{"lk/retained", "lk/held"} {
if got := p.held(0, name); !slices.Equal(got, []string{"hot", "hot"}) {
t.Errorf("%s holds %v, want both versions", name, got)
}
}
if ri.Objects != 2 || ri.KeysRefused != 3 || ri.ObjectsFailed != 0 {
t.Errorf("job counts objects=%d refused=%d failed=%d, want 2, 3 and 0", ri.Objects, ri.KeysRefused, ri.ObjectsFailed)
}
}
// Writes that share a version ID: the older null write in another pool goes
// first, so that a failure keeps what reads return, and the copies of one
// version count once.
func TestPurgeAllVersionsRetiresOlderWritesFirst(t *testing.T) {
z, _ := consistencyPools(t)
p := newPurgeTest(t, z, "purge-writes", MakeBucketOptions{VersioningEnabled: true}, purgeRules)
const null, dup = "ea/null", "ea/dup"
p.put(0, null, "old", daysAgo(6), ObjectOptions{})
p.put(1, null, "new", daysAgo(5), ObjectOptions{})
if got := p.get(null); got != "new" {
t.Fatalf("fixture reads %s", got)
}
v1 := p.version(0, dup, "v1", 5)
p.put(1, dup, "v1", v1.ModTime, ObjectOptions{Versioned: true, VersionID: v1.VersionID})
o := observePurges(t)
restore := p.failVersionDeletes(1, null, "")
p.expire(1, null)
if got := p.get(null); got != "new" {
t.Errorf("reads %s after the newer write failed, want new", got)
}
if got := p.held(0, null); got != nil {
t.Errorf("the older write is still in pool 0: %v", got)
}
// A copy is gone, though no version is: the purge is partial.
if events, traces := o.drain(); len(events) != 0 || len(traces) != 1 ||
traces[0]["ilm-partial"] != "true" || traces[0]["ilm-versions-removed"] != "0" {
t.Errorf("a purge that removed a copy sent %v and traced %v, want one partial expiry of 0 versions", events, traces)
}
restore()
p.expire(1, null)
if got := p.held(1, null); got != nil {
t.Errorf("pool 1 keeps %v", got)
}
o.drain()
p.expire(0, dup)
for pool := range z.serverPools {
if got := p.held(pool, dup); got != nil {
t.Errorf("pool %d keeps %v of %s", pool, got, dup)
}
}
if _, traces := o.drain(); len(traces) != 1 || traces[0]["ilm-versions-removed"] != "1" {
t.Errorf("copies of one version traced %v, want 1 version removed", traces)
}
}
// A minority of drives can hold another revision of a version than the
// accepted one. The purge checks every revision on every drive it deletes
// on: a transitioned or a protected one refuses the key.
func TestPurgeAllVersionsChecksEveryDrive(t *testing.T) {
z := purgeSinglePool(t, 16)
// The minority revision is on the last drive; the scanner of the first
// drive, which the test runs, does not see it and queues the expiry.
minority := func(name string) StorageAPI {
disks := z.serverPools[0].getHashedSet(name).getDisks()
return disks[len(disks)-1]
}
rewrite := func(p *purgeTest, name, versionID string, metadata map[string]string) {
t.Helper()
disk := minority(name)
if err := disk.UpdateMetadata(t.Context(), p.bucket, name, FileInfo{VersionID: versionID, Metadata: metadata}, UpdateMetadataOpts{}); err != nil {
t.Fatal(err)
}
}
tiered := newPurgeTest(t, z, "purge-drive-tier", MakeBucketOptions{VersioningEnabled: true}, purgeRules)
v := tiered.version(0, "ea/minority", "v1", 5)
tiered.version(0, "ea/minority", "v2", 4)
rewrite(tiered, "ea/minority", v.VersionID, map[string]string{
metaTierStatus: lifecycle.TransitionComplete,
metaTierName: "WARM",
metaTierObjName: "remote/ea/minority/v1",
})
tiered.expire(0, "ea/minority")
if got := tiered.get("ea/minority"); got != "v2" {
t.Errorf("a key with a transitioned revision on one drive reads %s, want v2", got)
}
locked := newPurgeTest(t, z, "purge-drive-hold", MakeBucketOptions{VersioningEnabled: true, LockEnabled: true}, "")
v = locked.version(0, "lk/minority", "v1", 5)
locked.version(0, "lk/minority", "v2", 4)
rewrite(locked, "lk/minority", v.VersionID, map[string]string{
strings.ToLower(xhttp.AmzObjectLockLegalHold): string(objectlock.LegalHoldOn),
})
locked.batch(locked.matched("lk/minority"))
if got := locked.get("lk/minority"); got != "v2" {
t.Errorf("a key with a legal hold on one drive reads %s, want v2", got)
}
disk := minority("lk/minority")
if fi, err := disk.ReadVersion(t.Context(), "", locked.bucket, "lk/minority", v.VersionID, ReadOptions{}); err != nil ||
fi.Metadata[strings.ToLower(xhttp.AmzObjectLockLegalHold)] != string(objectlock.LegalHoldOn) {
t.Errorf("the held revision: %v", err)
}
}
// A drive that cannot be read may hold a protected or transitioned revision,
// and heal would remove what the purge leaves there. It refuses the key.
func TestPurgeAllVersionsRefusesUnreadableDrive(t *testing.T) {
z := purgeSinglePool(t, 16)
failRead := func(bucket, name string) {
t.Helper()
sets := z.serverPools[0]
index := sets.getHashedSetIndex(name)
sets.erasureDisksMu.Lock()
prior := sets.erasureDisks[index][0]
sets.erasureDisks[index][0] = authorityDisk{StorageAPI: prior, bucket: bucket, key: name, failRead: true}
sets.erasureDisksMu.Unlock()
t.Cleanup(func() {
sets.erasureDisksMu.Lock()
sets.erasureDisks[index][0] = prior
sets.erasureDisksMu.Unlock()
})
}
locked := newPurgeTest(t, z, "purge-unreadable-hold", MakeBucketOptions{VersioningEnabled: true, LockEnabled: true}, "")
v := locked.version(0, "lk/hidden", "v1", 5)
locked.version(0, "lk/hidden", "v2", 4)
disk := z.serverPools[0].getHashedSet("lk/hidden").getDisks()[0]
if err := disk.UpdateMetadata(t.Context(), locked.bucket, "lk/hidden", FileInfo{VersionID: v.VersionID, Metadata: map[string]string{
strings.ToLower(xhttp.AmzObjectLockLegalHold): string(objectlock.LegalHoldOn),
}}, UpdateMetadataOpts{}); err != nil {
t.Fatal(err)
}
exp := locked.matched("lk/hidden")
failRead(locked.bucket, "lk/hidden")
ri := locked.batch(exp)
if fi, err := disk.ReadVersion(t.Context(), "", locked.bucket, "lk/hidden", v.VersionID, ReadOptions{}); err != nil ||
fi.Metadata[strings.ToLower(xhttp.AmzObjectLockLegalHold)] != string(objectlock.LegalHoldOn) {
t.Errorf("the held revision on the unreadable drive: %v", err)
}
if got := locked.get("lk/hidden"); got != "v2" || ri.Objects != 0 || ri.ObjectsFailed == 0 {
t.Errorf("reads %s; job counts objects=%d failed=%d", got, ri.Objects, ri.ObjectsFailed)
}
plain := newPurgeTest(t, z, "purge-unreadable-drive", MakeBucketOptions{VersioningEnabled: true}, purgeRules)
plain.version(0, "ea/unreadable-drive", "v1", 5)
ops := plain.scanQueue(0, "ea/unreadable-drive")
failRead(plain.bucket, "ea/unreadable-drive")
plain.run(ops)
if got := plain.get("ea/unreadable-drive"); got != "v1" {
t.Errorf("a key with one unreadable drive reads %s, want v1", got)
}
}
// A batch purge that fails on the latest delete marker counts that marker
// as failed.
func TestPurgeAllVersionsBatchCountsFailedMarker(t *testing.T) {
z := purgeSinglePool(t, 16)
p := newPurgeTest(t, z, "purge-batch-marker", MakeBucketOptions{VersioningEnabled: true}, "")
const key = "bx/marker"
p.version(0, key, "old", 5)
p.marker(0, key, 4)
exp := p.matched(key)
p.failVersionDeletes(0, key, exp.VersionID)
ri := p.batch(exp)
if got := p.held(0, key); !slices.Equal(got, []string{"DM"}) {
t.Errorf("holds %v, want the latest marker alone", got)
}
if ri.Objects != 1 || ri.DeleteMarkers != 0 || ri.DeleteMarkersFailed != 1 {
t.Errorf("job counts objects=%d markers=%d failed=%d/%d, want 1, 0, 0/1", ri.Objects, ri.DeleteMarkers, ri.ObjectsFailed, ri.DeleteMarkersFailed)
}
}
// Unreadable drives are never absence: an expiry of all versions on a history
// that the drives cannot tell deletes nothing, also on one pool.
func TestPurgeAllVersionsRefusesUnreadableHistory(t *testing.T) {
z := purgeSinglePool(t, 16)
p := newPurgeTest(t, z, "purge-unreadable", MakeBucketOptions{VersioningEnabled: true}, purgeRules)
const key = "ea/unreadable"
p.version(0, key, "v1", 5)
ops := p.scanQueue(0, key)
sets := z.serverPools[0]
sets.erasureDisksMu.Lock()
prior := sets.erasureDisks
wrapped := make([][]StorageAPI, len(prior))
for i, disks := range prior {
wrapped[i] = append([]StorageAPI(nil), disks...)
for j := 0; j < len(disks)/2; j++ {
wrapped[i][j] = authorityDisk{StorageAPI: disks[j], bucket: p.bucket, key: key, failRead: true}
}
}
sets.erasureDisks = wrapped
sets.erasureDisksMu.Unlock()
p.run(ops)
sets.erasureDisksMu.Lock()
sets.erasureDisks = prior
sets.erasureDisksMu.Unlock()
if got := p.held(0, key); !slices.Equal(got, []string{"hot"}) {
t.Errorf("holds %v after an expiry on an unreadable history", got)
}
}