diff --git a/cmd/batch-expire.go b/cmd/batch-expire.go
index 6de32920a..17dcf556f 100644
--- a/cmd/batch-expire.go
+++ b/cmd/batch-expire.go
@@ -445,31 +445,13 @@ func batchObjsForDelete(ctx context.Context, r *BatchJobExpire, ri *batchJobInfo
oiCache.Add(od, &exp.ObjectInfo)
}
- // DeleteObject(deletePrefix: true) to expire all versions of an object
+ // Purge all versions of exactly each key while the version the
+ // walk matched is still its latest; see
+ // erasureServerPools.deleteAllVersions.
for _, exp := range toExpireAll {
- var success bool
- for attempts := 1; attempts <= retryAttempts; attempts++ {
- select {
- case <-ctx.Done():
- ri.trackMultipleObjectVersions(exp, success)
- return
- default:
- }
- stopFn := globalBatchJobsMetrics.trace(batchJobMetricExpire, ri.JobID, attempts)
- _, err := api.DeleteObject(ctx, exp.Bucket, encodeDirObject(exp.Name), ObjectOptions{
- DeletePrefix: true,
- DeletePrefixObject: true, // use prefix delete on exact object (this is an optimization to avoid fan-out calls)
- })
- if err != nil {
- stopFn(exp, err)
- batchLogIf(ctx, fmt.Errorf("Failed to expire %s/%s due to %v (attempts=%d)", exp.Bucket, exp.Name, err, attempts))
- } else {
- stopFn(exp, err)
- success = true
- break
- }
+ if !expireAllVersions(ctx, ri, api, exp, retryAttempts) {
+ return
}
- ri.trackMultipleObjectVersions(exp, success)
}
// DeleteMultiple objects
@@ -525,6 +507,56 @@ func batchObjsForDelete(ctx context.Context, r *BatchJobExpire, ri *batchJobInfo
}
}
+// expireAllVersions purges all versions of the key exp, which the walk
+// matched by its latest version, and counts what the purge removed. It
+// retries a failed purge up to attempts times, each time with the version
+// the walk matched; a purge that stopped at its budget continues without
+// spending an attempt. It does not retry once the key no longer has that
+// latest version, which is not an expiry, or once the purge refused the
+// key. It returns false once ctx ends.
+func expireAllVersions(ctx context.Context, ri *batchJobInfo, api ObjectLayer, exp expireObjInfo, attempts int) bool {
+ var objects, markers int64
+ matched := deleteAllMatch{VersionID: exp.VersionID, ModTime: exp.ModTime}
+ for attempt := 1; attempt <= attempts; attempt++ {
+ select {
+ case <-ctx.Done():
+ ri.trackExpireAll(exp, objects, markers, true)
+ return false
+ default:
+ }
+ stopFn := globalBatchJobsMetrics.trace(batchJobMetricExpire, ri.JobID, attempt)
+ latest := matched
+ var purge deleteAllResult
+ _, err := api.DeleteObject(ctx, exp.Bucket, encodeDirObject(exp.Name), ObjectOptions{
+ DeleteAllVersions: true,
+ DeleteAllLatest: &latest,
+ DeleteAllResult: &purge,
+ })
+ objects += int64(purge.Removed - purge.Markers)
+ markers += int64(purge.Markers)
+ stopFn(exp, err)
+ switch {
+ case err == nil:
+ ri.trackExpireAll(exp, objects, markers, false)
+ return true
+ case errors.Is(err, errDeleteAllChanged):
+ // Changed since the walk matched it: not expired.
+ ri.trackExpireAll(exp, objects, markers, false)
+ return true
+ case errors.Is(err, errDeleteAllBudget) && purge.Removed > 0:
+ attempt-- // progress
+ continue
+ }
+ batchLogIf(ctx, fmt.Errorf("Failed to expire %s/%s due to %v (attempts=%d)", exp.Bucket, exp.Name, err, attempt))
+ var locked ObjectLocked
+ if errors.Is(err, errDeleteAllTiered) || errors.As(err, &locked) {
+ break // refused; a later run of the job checks again
+ }
+ }
+ ri.trackExpireAll(exp, objects, markers, true)
+ return true
+}
+
type expireObjInfo struct {
ObjectInfo
ExpireAll bool
@@ -746,8 +778,8 @@ func (r *BatchJobExpire) Start(ctx context.Context, api ObjectLayer, job BatchJo
<-expireDoneCh // waits for the expire goroutine to complete
wk.Wait() // waits for all expire workers to retire
- ri.Complete = !failed && ri.ObjectsFailed == 0
- ri.Failed = failed || ri.ObjectsFailed > 0
+ ri.Complete = !failed && !ri.expireFailures()
+ ri.Failed = failed || ri.expireFailures()
globalBatchJobsMetrics.save(job.ID, ri)
// Close the saverQuitCh - this also triggers saving in-memory state
diff --git a/cmd/batch-handlers.go b/cmd/batch-handlers.go
index 660e6adfa..dba3c6a20 100644
--- a/cmd/batch-handlers.go
+++ b/cmd/batch-handlers.go
@@ -996,10 +996,17 @@ func (ri *batchJobInfo) updateAfter(ctx context.Context, api ObjectLayer, durati
return nil
}
-// Note: to be used only with batch jobs that affect multiple versions through
-// a single action. e.g batch-expire has an option to expire all versions of an
-// object which matches the given filters.
-func (ri *batchJobInfo) trackMultipleObjectVersions(info expireObjInfo, success bool) {
+// expireFailures reports whether an expiration job failed to remove an
+// object or a delete marker it matched.
+func (ri *batchJobInfo) expireFailures() bool {
+ return ri.ObjectsFailed > 0 || ri.DeleteMarkersFailed > 0
+}
+
+// trackExpireAll counts the objects and delete markers that an expiry of all
+// versions of info removed and, when it failed, those the walk saw that it
+// did not remove, at least one. Only batch-expire uses it, for a rule that
+// expires all versions of an object.
+func (ri *batchJobInfo) trackExpireAll(info expireObjInfo, objects, markers int64, failed bool) {
if ri == nil {
return
}
@@ -1007,15 +1014,20 @@ func (ri *batchJobInfo) trackMultipleObjectVersions(info expireObjInfo, success
ri.mu.Lock()
defer ri.mu.Unlock()
- if success {
+ ri.Objects += objects
+ ri.DeleteMarkers += markers
+ if !failed {
ri.Bucket = info.Bucket
ri.Object = info.Name
- ri.Objects += int64(info.NumVersions) - info.DeleteMarkerCount
- ri.DeleteMarkers += info.DeleteMarkerCount
- } else {
- ri.ObjectsFailed += int64(info.NumVersions) - info.DeleteMarkerCount
- ri.DeleteMarkersFailed += info.DeleteMarkerCount
+ return
}
+ failedObjects := max(int64(info.NumVersions)-info.DeleteMarkerCount-objects, 0)
+ failedMarkers := max(info.DeleteMarkerCount-markers, 0)
+ if failedObjects+failedMarkers == 0 {
+ failedObjects = 1
+ }
+ ri.ObjectsFailed += failedObjects
+ ri.DeleteMarkersFailed += failedMarkers
}
func (ri *batchJobInfo) trackCurrentBucketObject(bucket string, info ObjectInfo, success bool, attempt int) {
diff --git a/cmd/data-scanner.go b/cmd/data-scanner.go
index 77d265367..ca8fc5dc4 100644
--- a/cmd/data-scanner.go
+++ b/cmd/data-scanner.go
@@ -1229,10 +1229,16 @@ func applyExpiryOnNonTransitionedObjects(ctx context.Context, objLayer ObjectLay
opts.Versioned = globalBucketVersioningSys.PrefixEnabled(obj.Bucket, obj.Name)
opts.VersionSuspended = globalBucketVersioningSys.PrefixSuspended(obj.Bucket, obj.Name)
+ // Purge every version of exactly this key, by version ID and under the
+ // object lock, after the pools approved it on the versions they hold;
+ // see erasureServerPools.deleteAllVersions.
+ var purge deleteAllResult
if lcEvent.Action.DeleteAll() {
- opts.DeletePrefix = true
- // use prefix delete on exact object (this is an optimization to avoid fan-out calls)
- opts.DeletePrefixObject = true
+ opts = ObjectOptions{
+ Expiration: ExpirationOptions{Expire: true},
+ DeleteAllVersions: true,
+ DeleteAllResult: &purge,
+ }
}
// ILM-created delete markers and version purges must carry the same
@@ -1263,13 +1269,16 @@ func applyExpiryOnNonTransitionedObjects(ctx context.Context, objLayer ObjectLay
if lcEvent.Action != lifecycle.NoneAction {
numVersions := uint64(1)
if lcEvent.Action.DeleteAll() {
- numVersions = uint64(obj.NumVersions)
+ numVersions = uint64(purge.Removed)
}
timeILM(numVersions)
}
}()
dobj, err = objLayer.DeleteObject(ctx, obj.Bucket, encodeDirObject(obj.Name), opts)
+ if lcEvent.Action.DeleteAll() && (err != nil || purge.Removed == 0) {
+ return deleteAllNotCompleted(ctx, obj, lcEvent, src, purge, err, traceFn)
+ }
if err != nil {
if errors.Is(err, errLifecycleNotApplicable) || isErrObjectNotFound(err) || isErrVersionNotFound(err) {
traceFn(ILMExpiry, nil, nil)
@@ -1310,6 +1319,9 @@ func applyExpiryOnNonTransitionedObjects(ctx context.Context, objLayer ObjectLay
tags := newLifecycleAuditEvent(src, lcEvent).Tags()
tags["version-id"] = dobj.VersionID
+ if lcEvent.Action.DeleteAll() {
+ tags[ilmVersionsRemoved] = strconv.Itoa(purge.Removed)
+ }
// Send audit for the lifecycle delete operation
auditLogLifecycle(ctx, dobj, ILMExpiry, tags, traceFn)
@@ -1336,6 +1348,57 @@ func applyExpiryOnNonTransitionedObjects(ctx context.Context, objLayer ObjectLay
return true
}
+// ilmVersionsRemoved is the audit tag with the number of versions that an
+// expiry of all versions removed; ilmPartial marks one that began to delete
+// and did not finish.
+const (
+ ilmVersionsRemoved = "ilm-versions-removed"
+ ilmPartial = "ilm-partial"
+)
+
+// deleteAllNotCompleted reports an expiry of all versions of obj that did
+// not remove them all. A purge that had nothing to remove, or that failed
+// before it deleted anything, reports what a rejected expiry does. One that
+// failed or stopped after it began to delete is audited as partial, with the
+// versions it removed, but sends no removal event: the key is still there.
+func deleteAllNotCompleted(ctx context.Context, obj ObjectInfo, lcEvent lifecycle.Event, src lcEventSrc, purge deleteAllResult, err error, traceFn func(event string, metadata map[string]string, err error)) bool {
+ switch {
+ case err == nil: // there was nothing to remove
+ traceFn(ILMExpiry, nil, nil)
+ return false
+ case purge.Started:
+ tags := newLifecycleAuditEvent(src, lcEvent).Tags()
+ tags["version-id"] = purge.Latest.VersionID
+ tags[ilmVersionsRemoved] = strconv.Itoa(purge.Removed)
+ tags[ilmPartial] = "true"
+ auditLogInternal(ctx, AuditLogOptions{
+ Event: ILMExpiry,
+ APIName: "ILMExpiry",
+ Bucket: obj.Bucket,
+ Object: obj.Name,
+ VersionID: purge.Latest.VersionID,
+ Error: err.Error(),
+ Tags: tags,
+ })
+ traceFn(ILMExpiry, tags, err)
+ if errors.Is(err, errDeleteAllBudget) {
+ return false // a long history; the next expiry continues
+ }
+ case errors.Is(err, errDeleteAllTiered):
+ // Kept until tiered versions can be purged; see deleteAllVersions.
+ traceFn(ILMExpiry, nil, err)
+ return false
+ case errors.Is(err, errLifecycleNotApplicable) || isErrObjectNotFound(err) || isErrVersionNotFound(err):
+ traceFn(ILMExpiry, nil, nil)
+ return false
+ default:
+ traceFn(ILMExpiry, nil, err)
+ }
+ // Assume it is still there.
+ ilmLogOnceIf(ctx, fmt.Errorf("DeleteObject(%s, %s): %w", obj.Bucket, obj.Name, err), "non-transition-expiry"+obj.Name)
+ return false
+}
+
// Apply object, object version, restored object or restored object version action on the given object
func applyExpiryRule(event lifecycle.Event, src lcEventSrc, obj ObjectInfo) {
globalExpiryState.enqueueByDays(obj, event, src)
diff --git a/cmd/erasure-object.go b/cmd/erasure-object.go
index 8a0de02ad..15249cf48 100644
--- a/cmd/erasure-object.go
+++ b/cmd/erasure-object.go
@@ -1773,6 +1773,13 @@ func (er erasureObjects) deleteObjectVersion(ctx context.Context, bucket, object
// into smaller bulks if some object names are found to be duplicated in the delete list, splitting
// into smaller bulks will avoid holding twice the write lock of the duplicated object names.
func (er erasureObjects) DeleteObjects(ctx context.Context, bucket string, objects []ObjectToDelete, opts ObjectOptions) ([]DeletedObject, []error) {
+ if opts.DeleteAllVersions {
+ errs := make([]error, len(objects))
+ for i := range errs {
+ errs[i] = errDeleteAllUnsupported
+ }
+ return make([]DeletedObject, len(objects)), errs
+ }
if !opts.NoAuditLog {
for _, obj := range objects {
auditObjectErasureSet(ctx, "DeleteObjects", obj.ObjectName, &er)
@@ -2023,6 +2030,9 @@ func (er erasureObjects) deletePrefix(ctx context.Context, bucket, prefix string
// any error as it is not necessary for the handler to reply back a
// response to the client request.
func (er erasureObjects) DeleteObject(ctx context.Context, bucket, object string, opts ObjectOptions) (objInfo ObjectInfo, err error) {
+ if opts.DeleteAllVersions {
+ return objInfo, errDeleteAllUnsupported
+ }
if !opts.NoAuditLog {
auditObjectErasureSet(ctx, "DeleteObject", object, &er)
}
diff --git a/cmd/erasure-server-pool-delete-all-options_test.go b/cmd/erasure-server-pool-delete-all-options_test.go
new file mode 100644
index 000000000..96039ffae
--- /dev/null
+++ b/cmd/erasure-server-pool-delete-all-options_test.go
@@ -0,0 +1,221 @@
+// 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 .
+
+package cmd
+
+import (
+ "context"
+ "errors"
+ "fmt"
+ "slices"
+ "sync/atomic"
+ "testing"
+ "time"
+)
+
+// The parts of the purge of all versions that need its own options.
+
+// A purge stops at its budget before the latest version.
+func TestPurgeAllVersionsStopsAtBudget(t *testing.T) {
+ z := purgeSinglePool(t, 16)
+ p := newPurgeTest(t, z, "purge-budget", MakeBucketOptions{VersioningEnabled: true}, purgeRules)
+ prev := deleteAllBudget
+ deleteAllBudget.versions = 2
+ t.Cleanup(func() { deleteAllBudget = prev })
+ const key = "ea/long"
+ for i, days := range []float64{8, 7, 6, 5} {
+ p.version(0, key, fmt.Sprintf("v%d", i+1), days)
+ }
+ o := observePurges(t)
+ p.expire(0, key)
+ if got := p.held(0, key); !slices.Equal(got, []string{"hot", "hot"}) {
+ t.Errorf("holds %v after a purge stopped at its budget, want the two newest versions", got)
+ }
+ if got := p.get(key); got != "v4" {
+ t.Errorf("reads %s, want v4", got)
+ }
+ events, traces := o.drain()
+ if len(events) != 0 || len(traces) != 1 || traces[0]["ilm-partial"] != "true" || traces[0]["ilm-versions-removed"] != "2" {
+ t.Errorf("stopped purge sent %v and traced %v", events, traces)
+ }
+ p.expire(0, key)
+ if got := p.held(0, key); got != nil {
+ t.Errorf("keeps %v once the purge continued", got)
+ }
+}
+
+// The option admits no other delete.
+func TestPurgeAllVersionsOptions(t *testing.T) {
+ z := purgeSinglePool(t, 16)
+ p := newPurgeTest(t, z, "purge-options", MakeBucketOptions{VersioningEnabled: true}, purgeRules)
+ p.version(0, "ea/key", "v1", 5)
+ match := &deleteAllMatch{}
+ expire := ExpirationOptions{Expire: true}
+ for name, opts := range map[string]ObjectOptions{
+ "no approval": {DeleteAllVersions: true},
+ "two approvals": {DeleteAllVersions: true, Expiration: expire, DeleteAllLatest: match},
+ "prefix": {DeleteAllVersions: true, Expiration: expire, DeletePrefix: true},
+ "prefix object": {DeleteAllVersions: true, Expiration: expire, DeletePrefixObject: true},
+ "version": {DeleteAllVersions: true, Expiration: expire, VersionID: mustGetUUID()},
+ "movement": {DeleteAllVersions: true, Expiration: expire, DataMovement: true},
+ "marker": {DeleteAllVersions: true, Expiration: expire, DeleteMarker: true},
+ "versioned": {DeleteAllVersions: true, Expiration: expire, Versioned: true},
+ "suspended": {DeleteAllVersions: true, Expiration: expire, VersionSuspended: true},
+ } {
+ if _, err := z.DeleteObject(t.Context(), p.bucket, "ea/key", opts); !errors.As(err, new(InvalidArgument)) {
+ t.Errorf("%s: %v", name, err)
+ }
+ }
+ opts := ObjectOptions{DeleteAllVersions: true, Expiration: expire}
+ if _, err := z.serverPools[0].DeleteObject(t.Context(), p.bucket, "ea/key", opts); !errors.As(err, new(NotImplemented)) {
+ t.Errorf("erasure sets: %v", err)
+ }
+ if _, errs := z.DeleteObjects(t.Context(), p.bucket, []ObjectToDelete{{ObjectV: ObjectV{ObjectName: "ea/key"}}}, opts); !errors.As(errs[0], new(NotImplemented)) {
+ t.Errorf("bulk delete: %v", errs[0])
+ }
+ if got := p.held(0, "ea/key"); !slices.Equal(got, []string{"hot"}) {
+ t.Errorf("holds %v after rejected requests", got)
+ }
+ // An absent key is a success that removes nothing.
+ var res deleteAllResult
+ if _, err := z.DeleteObject(t.Context(), p.bucket, "ea/absent", ObjectOptions{DeleteAllVersions: true, Expiration: expire, DeleteAllResult: &res}); err != nil || res.Removed != 0 || !res.Complete {
+ t.Errorf("absent key: %v %+v", err, res)
+ }
+}
+
+type purgeSlowDisk struct {
+ StorageAPI
+ bucket, object string
+}
+
+func (d purgeSlowDisk) ReadXL(ctx context.Context, bucket, object string, data bool) (RawFileInfo, error) {
+ if bucket == d.bucket && object == d.object {
+ time.Sleep(20 * time.Millisecond)
+ }
+ return d.StorageAPI.ReadXL(ctx, bucket, object, data)
+}
+
+// The time budget covers the reads and checks before the first delete.
+func TestPurgeAllVersionsBudgetCoversReads(t *testing.T) {
+ z := purgeSinglePool(t, 16)
+ p := newPurgeTest(t, z, "purge-budget-reads", MakeBucketOptions{VersioningEnabled: true}, "")
+ const key = "bx/slow"
+ p.version(0, key, "v1", 5)
+ exp := p.matched(key)
+ prev := deleteAllBudget
+ deleteAllBudget.duration = 10 * time.Millisecond
+ t.Cleanup(func() { deleteAllBudget = prev })
+ pool := z.serverPools[0]
+ pool.erasureDisksMu.Lock()
+ prior := pool.erasureDisks
+ wrapped := make([][]StorageAPI, len(prior))
+ for i, disks := range prior {
+ wrapped[i] = make([]StorageAPI, len(disks))
+ for j, disk := range disks {
+ wrapped[i][j] = purgeSlowDisk{StorageAPI: disk, bucket: p.bucket, object: key}
+ }
+ }
+ pool.erasureDisks = wrapped
+ pool.erasureDisksMu.Unlock()
+ t.Cleanup(func() { pool.erasureDisksMu.Lock(); pool.erasureDisks = prior; pool.erasureDisksMu.Unlock() })
+ var res deleteAllResult
+ _, err := z.DeleteObject(t.Context(), p.bucket, key, ObjectOptions{
+ DeleteAllVersions: true,
+ DeleteAllLatest: &deleteAllMatch{VersionID: exp.VersionID, ModTime: exp.ModTime},
+ DeleteAllResult: &res,
+ })
+ if !errors.Is(err, errDeleteAllBudget) || res.Started || res.Removed != 0 {
+ t.Errorf("purge after its budget ran out on reads: err=%v %+v", err, res)
+ }
+ if got := p.held(0, key); !slices.Equal(got, []string{"hot"}) {
+ t.Errorf("holds %v", got)
+ }
+}
+
+// A marker that an expiration job failed to remove fails the job.
+func TestPurgeAllVersionsMarkerFailureFailsJob(t *testing.T) {
+ ri := &batchJobInfo{}
+ ri.trackExpireAll(expireObjInfo{ObjectInfo: ObjectInfo{NumVersions: 2}, DeleteMarkerCount: 1}, 1, 0, true)
+ if ri.ObjectsFailed != 0 || ri.DeleteMarkersFailed != 1 || !ri.expireFailures() {
+ t.Errorf("failed=%d/%d failures=%t", ri.ObjectsFailed, ri.DeleteMarkersFailed, ri.expireFailures())
+ }
+}
+
+// purgeSwapDisk replaces itself in its slot once the purge has read it
+// twice, for the history and for the drive check, as a reconnect does.
+type purgeSwapDisk struct {
+ StorageAPI
+ bucket, object string
+ reads *atomic.Int32
+ swap func()
+}
+
+func (d purgeSwapDisk) ReadXL(ctx context.Context, bucket, object string, data bool) (RawFileInfo, error) {
+ rf, err := d.StorageAPI.ReadXL(ctx, bucket, object, data)
+ if bucket == d.bucket && object == d.object && d.reads.Add(1) == 2 {
+ d.swap()
+ }
+ return rf, err
+}
+
+// purgeCountDisk counts the version deletes it receives.
+type purgeCountDisk struct {
+ StorageAPI
+ deletes *atomic.Int32
+}
+
+func (d purgeCountDisk) DeleteVersion(ctx context.Context, volume, path string, fi FileInfo, forceDelMarker bool, opts DeleteOptions) error {
+ d.deletes.Add(1)
+ return d.StorageAPI.DeleteVersion(ctx, volume, path, fi, forceDelMarker, opts)
+}
+
+// The purge deletes on the drive handles it checked, not on a handle that
+// replaced one of them since.
+func TestPurgeAllVersionsDeletesOnCheckedDrives(t *testing.T) {
+ z := purgeSinglePool(t, 16)
+ p := newPurgeTest(t, z, "purge-checked-drives", MakeBucketOptions{VersioningEnabled: true}, "")
+ const key = "bx/swap"
+ p.version(0, key, "v1", 5)
+ exp := p.matched(key)
+ sets := z.serverPools[0]
+ index := sets.getHashedSetIndex(key)
+ var reads, deletes atomic.Int32
+ sets.erasureDisksMu.Lock()
+ original := sets.erasureDisks[index][0]
+ sets.erasureDisks[index][0] = purgeSwapDisk{StorageAPI: original, bucket: p.bucket, object: key, reads: &reads, swap: func() {
+ sets.erasureDisksMu.Lock()
+ sets.erasureDisks[index][0] = purgeCountDisk{StorageAPI: original, deletes: &deletes}
+ sets.erasureDisksMu.Unlock()
+ }}
+ sets.erasureDisksMu.Unlock()
+ t.Cleanup(func() {
+ sets.erasureDisksMu.Lock()
+ sets.erasureDisks[index][0] = original
+ sets.erasureDisksMu.Unlock()
+ })
+ var res deleteAllResult
+ if _, err := z.DeleteObject(t.Context(), p.bucket, key, ObjectOptions{
+ DeleteAllVersions: true,
+ DeleteAllLatest: &deleteAllMatch{VersionID: exp.VersionID, ModTime: exp.ModTime},
+ DeleteAllResult: &res,
+ }); err != nil || !res.Complete {
+ t.Fatalf("purge: %v %+v", err, res)
+ }
+ if reads.Load() != 2 || deletes.Load() != 0 {
+ t.Errorf("reads=%d deletes on the replacement handle=%d", reads.Load(), deletes.Load())
+ }
+}
diff --git a/cmd/erasure-server-pool-delete-all.go b/cmd/erasure-server-pool-delete-all.go
new file mode 100644
index 000000000..50c594270
--- /dev/null
+++ b/cmd/erasure-server-pool-delete-all.go
@@ -0,0 +1,459 @@
+// 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 .
+
+package cmd
+
+import (
+ "context"
+ "errors"
+ "fmt"
+ "slices"
+ "sort"
+ "sync"
+ "time"
+
+ "github.com/pgsty/silo-pkg/v3/sync/errgroup"
+)
+
+// An expiry of all versions of a key (ExpiredObjectAllVersions,
+// DelMarkerExpiration, a batch expiry that keeps no versions) removes the
+// versions of that key and nothing else. On the drives, object a/b lives
+// inside the directory of object a, next to the metadata and data of a. The
+// purge therefore never deletes that directory: it deletes the versions of
+// the key one by one, by version ID, with the version delete of the drives,
+// which removes the directory only once it is empty.
+//
+// It decides and deletes under the object lock that DeleteObject holds, on
+// the authoritative history of the key in all pools (objectVersions):
+// - an ILM expiry needs the latest version to expire with all versions as
+// the scanner evaluates it on all of them (lifecycleEventsAllPools); a
+// batch expiry needs the version it matched to be the latest still;
+// - it deletes exactly the versions it approved, with every copy of them
+// in every pool and the older writes of their IDs, and nothing else;
+// - before it deletes anything it reads every drive of every set that
+// holds a copy, and refuses the whole key when a drive cannot be read or
+// when any revision of an approved version there is transitioned or,
+// when the bucket has Object Lock, protected by retention or a legal
+// hold. It deletes on exactly the drives it read;
+// - it deletes the oldest version first and the latest version last and
+// alone, once every older version is gone from every pool. A purge that
+// stops early keeps the latest version, so it never exposes an older
+// one, and the next expiry continues on a fresh history.
+
+var (
+ // errDeleteAllChanged: the version a batch expiry matched is no longer
+ // the latest version of the key.
+ errDeleteAllChanged = fmt.Errorf("%w: the latest version changed", errLifecycleNotApplicable)
+ // errDeleteAllTiered: a version of the key is in a remote tier. Removing
+ // all versions is left to a later release, which hands the remote
+ // objects to the tier cleanup.
+ errDeleteAllTiered = fmt.Errorf("%w: a version is transitioned", errLifecycleNotApplicable)
+ // errDeleteAllBudget: the purge stopped at its budget, before the
+ // latest version.
+ errDeleteAllBudget = errors.New("purge of all versions stopped at its budget")
+ // errDeleteAllUnsupported: only a single-object delete of the pools
+ // purges all versions. Any other layer would turn the request into
+ // another delete.
+ errDeleteAllUnsupported = NotImplemented{Message: "purge of all versions is a single-object delete of the pools"}
+)
+
+// deleteAllBudget bounds one purge, which holds the object lock while it
+// runs. It stops before the next version once it removed versions versions
+// or ran for duration, its reads and checks included.
+var deleteAllBudget = struct {
+ versions int
+ duration time.Duration
+}{versions: 1000, duration: time.Minute}
+
+// deleteAllRetries bounds how often a version delete that reached its write
+// quorum is repeated on the drives that failed it.
+const deleteAllRetries = 2
+
+// deleteAllMatch identifies the latest version that a batch expiry matched.
+type deleteAllMatch struct {
+ VersionID string
+ ModTime time.Time
+}
+
+// matches reports whether fi is the version m identifies.
+func (m deleteAllMatch) matches(fi FileInfo) bool {
+ return deleteAllVersionID(m.VersionID) == deleteAllVersionID(fi.VersionID) && m.ModTime.Equal(fi.ModTime)
+}
+
+// deleteAllResult is what a purge of all versions confirmed.
+type deleteAllResult struct {
+ // Latest is the latest version the purge approved.
+ Latest ObjectInfo
+ // Removed counts the versions, delete markers included, that are gone
+ // from every pool, Markers the delete markers among them. Free versions
+ // and further copies of a version do not count.
+ Removed, Markers int
+ // Started: the purge began to delete. A failure after that can have
+ // removed copies that Removed does not count.
+ Started bool
+ // Complete: every approved version is gone, the latest included.
+ Complete bool
+}
+
+// deleteAllVersionID returns versionID, "null" for the null version.
+func deleteAllVersionID(versionID string) string {
+ if versionID == "" {
+ return nullVersionID
+ }
+ return versionID
+}
+
+// deleteAllInvalid returns why opts cannot ask for a purge of all versions.
+// The purge is approved either as an ILM expiry or by a batch match, and it
+// only deletes: it never creates a marker, addresses a single version,
+// moves data or deletes a prefix.
+func (opts ObjectOptions) deleteAllInvalid() error {
+ switch {
+ case opts.DeletePrefix || opts.DeletePrefixObject:
+ return errors.New("a purge of all versions is not a prefix delete")
+ case opts.VersionID != "":
+ return errors.New("a purge of all versions addresses no single version")
+ case opts.DataMovement:
+ return errors.New("a purge of all versions moves no data")
+ case opts.DeleteMarker || opts.Versioned || opts.VersionSuspended:
+ return errors.New("a purge of all versions creates no delete marker")
+ case opts.ReplicationRequest || opts.InclFreeVersions || opts.Transition != (TransitionOptions{}) ||
+ opts.CheckPrecondFn != nil || opts.EvalMetadataFn != nil || opts.EvalRetentionBypassFn != nil ||
+ opts.Expiration.CheckedAllPools:
+ return errors.New("a purge of all versions takes no options of other deletes")
+ case opts.Expiration.Expire == (opts.DeleteAllLatest != nil):
+ return errors.New("a purge of all versions needs exactly one approval")
+ }
+ return nil
+}
+
+// deleteAllCopy is the copy of a version in a pool: the revision of it that
+// the authoritative history of the set of the key in that pool accepted.
+// Copies of one write share its ModTime, since data movement keeps it and
+// writes a new data dir. Other writes of a version ID, such as null versions
+// written in several pools, have another ModTime.
+type deleteAllCopy struct {
+ pool int
+ fi FileInfo
+}
+
+// deleteAllVersion is a version of the key, "null" for the null version, and
+// its copies in all pools, the newest write first and, on equal times, the
+// lower pool first. The first copy is the one reads choose.
+type deleteAllVersion struct {
+ id string
+ copies []deleteAllCopy
+}
+
+// deleteAllHistory returns the versions of object in all pools, newest first,
+// with the copies of each, from the authoritative history of each pool. It
+// fails when a pool cannot tell the history.
+func (z *erasureServerPools) deleteAllHistory(ctx context.Context, bucket, object string) ([]deleteAllVersion, error) {
+ poolVersions := make([][]FileInfo, len(z.serverPools))
+ errs := make([]error, len(z.serverPools))
+ var wg sync.WaitGroup
+ for i, pool := range z.serverPools {
+ wg.Add(1)
+ go func(i int, set *erasureObjects) {
+ defer wg.Done()
+ poolVersions[i], errs[i] = set.objectVersions(ctx, bucket, object)
+ }(i, pool.getHashedSet(object))
+ }
+ wg.Wait()
+ for _, err := range errs {
+ if err != nil {
+ return nil, err
+ }
+ }
+
+ var all []deleteAllCopy
+ for pool, versions := range poolVersions {
+ for _, fi := range versions {
+ if fi.ModTime.IsZero() {
+ // A write that cannot be ordered cannot be deleted in order.
+ return nil, toObjectErr(errErasureReadQuorum, bucket, object)
+ }
+ all = append(all, deleteAllCopy{pool: pool, fi: fi})
+ }
+ }
+ // The order of lifecycleEventsAllPools and of reads.
+ sort.SliceStable(all, func(i, j int) bool { return all[i].fi.ModTime.After(all[j].fi.ModTime) })
+ index := make(map[string]int, len(all))
+ var history []deleteAllVersion
+ for _, c := range all {
+ id := deleteAllVersionID(c.fi.VersionID)
+ if i, ok := index[id]; ok {
+ history[i].copies = append(history[i].copies, c)
+ continue
+ }
+ index[id] = len(history)
+ history = append(history, deleteAllVersion{id: id, copies: []deleteAllCopy{c}})
+ }
+ return history, nil
+}
+
+// deleteAllApproved reports whether history holds exactly the versions that
+// the ILM evaluation approved, in the same order, the same writes chosen.
+func deleteAllApproved(history []deleteAllVersion, approved []ObjectInfo) bool {
+ if len(history) != len(approved) {
+ return false
+ }
+ for i, v := range history {
+ got := v.copies[0].fi.ToObjectInfo("", "", false)
+ want := approved[i]
+ if v.id != deleteAllVersionID(want.VersionID) || !got.ModTime.Equal(want.ModTime) ||
+ got.DeleteMarker != want.DeleteMarker || got.ETag != want.ETag || got.Size != want.Size {
+ return false
+ }
+ }
+ return true
+}
+
+// deleteAllVersions deletes every version of object, whose expiry of all
+// versions opts describe, and nothing else, see the comment above. The
+// caller holds the object lock. It returns the approved latest version, and
+// what it confirmed in opts.DeleteAllResult, also on an error. A key without
+// versions is a success that removes nothing.
+func (z *erasureServerPools) deleteAllVersions(ctx context.Context, bucket, object string, opts ObjectOptions) (ObjectInfo, error) {
+ res := opts.DeleteAllResult
+ if res == nil {
+ res = &deleteAllResult{}
+ }
+ *res = deleteAllResult{}
+ start := time.Now()
+ name := decodeDirObject(object)
+ fail := func(err error) (ObjectInfo, error) {
+ if isErrReadQuorum(err) {
+ err = InsufficientWriteQuorum{}
+ }
+ return ObjectInfo{Bucket: bucket, Name: name}, err
+ }
+
+ lr, err := globalBucketObjectLockSys.Get(bucket)
+ if err != nil {
+ return fail(err)
+ }
+
+ // Decide.
+ var approved []ObjectInfo
+ if opts.Expiration.Expire {
+ var all lifecycleAllPools
+ all.versions, all.events, all.err = z.lifecycleEventsAllPools(ctx, bucket, object)
+ if all.err != nil || len(all.versions) > 0 {
+ if err := all.expiryApplies(bucket, object, ObjectOptions{DeletePrefix: true}); err != nil {
+ return fail(err)
+ }
+ }
+ approved = all.versions
+ }
+ // The placement of the approved versions. This read may only confirm
+ // them: whatever differs from the approval refuses the key.
+ history, err := z.deleteAllHistory(ctx, bucket, object)
+ if err != nil {
+ return fail(err)
+ }
+ if opts.Expiration.Expire && !deleteAllApproved(history, approved) {
+ return fail(InsufficientWriteQuorum{})
+ }
+ if len(history) == 0 {
+ res.Complete = true
+ return ObjectInfo{Bucket: bucket, Name: name}, nil
+ }
+ if opts.DeleteAllLatest != nil && !opts.DeleteAllLatest.matches(history[0].copies[0].fi) {
+ return fail(errDeleteAllChanged)
+ }
+ if opts.Expiration.Expire {
+ res.Latest = approved[0]
+ } else {
+ vc, _ := globalBucketVersioningSys.Get(bucket)
+ res.Latest = history[0].copies[0].fi.ToObjectInfo(bucket, object, vc != nil && vc.Versioned(name))
+ }
+ res.Latest.Bucket, res.Latest.Name = bucket, name
+
+ // Check every write before deleting any: the accepted revisions, then
+ // every revision on every drive that the purge will delete on.
+ ids := make(map[string]bool, len(history))
+ held := make([]bool, len(z.serverPools))
+ for _, v := range history {
+ ids[v.id] = true
+ for _, c := range v.copies {
+ held[c.pool] = true
+ if err := deleteAllRefuses(ctx, bucket, object, c.fi, lr.LockEnabled); err != nil {
+ return fail(err)
+ }
+ }
+ }
+ drives := make([][]StorageAPI, len(z.serverPools))
+ for pool := range z.serverPools {
+ if !held[pool] {
+ continue
+ }
+ if drives[pool], err = z.serverPools[pool].getHashedSet(object).deleteAllDrives(ctx, bucket, object, ids, lr.LockEnabled); err != nil {
+ return fail(err)
+ }
+ }
+
+ // Delete the oldest version first. The latest one is the last and is
+ // only reached once every older one is gone from every pool.
+ for i := len(history) - 1; i >= 0; i-- {
+ if err := ctx.Err(); err != nil {
+ return fail(err)
+ }
+ if res.Removed >= deleteAllBudget.versions || time.Since(start) >= deleteAllBudget.duration {
+ return fail(errDeleteAllBudget)
+ }
+ res.Started = true
+ if err := z.deleteAllCopies(ctx, bucket, object, history[i], drives); err != nil {
+ return fail(err)
+ }
+ res.Removed++
+ if history[i].copies[0].fi.Deleted {
+ res.Markers++
+ }
+ }
+ res.Complete = true
+ return res.Latest, nil
+}
+
+// deleteAllRefuses returns why the purge must keep the key that holds the
+// revision fi of an approved version: it is transitioned, whose remote
+// object a later release hands to the tier cleanup, or, with Object Lock,
+// protected by retention or a legal hold.
+func deleteAllRefuses(ctx context.Context, bucket, object string, fi FileInfo, lockEnabled bool) error {
+ if fi.TransitionStatus != "" || fi.TransitionTier != "" || fi.TransitionedObjName != "" {
+ return errDeleteAllTiered
+ }
+ if lockEnabled && enforceRetentionForDeletion(ctx, fi.ToObjectInfo(bucket, object, true)) {
+ return ObjectLocked{Bucket: bucket, Object: decodeDirObject(object), VersionID: deleteAllVersionID(fi.VersionID)}
+ }
+ return nil
+}
+
+// deleteAllDrives reads object on every drive of the set and checks each
+// revision of the versions ids that a drive holds with deleteAllRefuses.
+// A minority of drives can hold a revision other than the accepted one, and
+// heal would later remove what the purge leaves on a drive, so a drive that
+// cannot be read or decoded refuses the key. A drive without the object
+// holds nothing to check. It returns the drive handles it read, which the
+// purge deletes on; a handle that a reconnect installs later is not used. A
+// handle names a drive, not a backend: a backend that replaces the drive
+// behind it starts empty and receives this object only from a write or a
+// heal of it, which take the object lock the purge holds.
+func (er erasureObjects) deleteAllDrives(ctx context.Context, bucket, object string, ids map[string]bool, lockEnabled bool) ([]StorageAPI, error) {
+ disks := slices.Clone(er.getDisks())
+ raws, errs := readAllRawFileInfo(ctx, disks, bucket, object, false)
+ if err := ctx.Err(); err != nil {
+ return nil, err
+ }
+ for i := range disks {
+ if errors.Is(errs[i], errFileNotFound) {
+ continue
+ }
+ if errs[i] != nil {
+ return nil, toObjectErr(errErasureReadQuorum, bucket, object)
+ }
+ var xl xlMetaV2
+ if err := xl.LoadOrConvert(raws[i].Buf); err != nil {
+ return nil, toObjectErr(errErasureReadQuorum, bucket, object)
+ }
+ versions, err := xl.ListVersions(bucket, object, false)
+ if err != nil {
+ return nil, toObjectErr(errErasureReadQuorum, bucket, object)
+ }
+ for _, fi := range versions {
+ if fi.TierFreeVersion() || !ids[deleteAllVersionID(fi.VersionID)] {
+ continue
+ }
+ if err := deleteAllRefuses(ctx, bucket, object, fi, lockEnabled); err != nil {
+ return nil, err
+ }
+ }
+ }
+ return disks, nil
+}
+
+// deleteAllCopies deletes every copy of the version v: the older writes of
+// its ID first, the copy that reads choose last, so that a failure leaves
+// what a read returned before. It deletes on the drives that drives holds
+// for the pool of a copy.
+func (z *erasureServerPools) deleteAllCopies(ctx context.Context, bucket, object string, v deleteAllVersion, drives [][]StorageAPI) error {
+ for i := len(v.copies) - 1; i >= 0; i-- {
+ pool := v.copies[i].pool
+ set := z.serverPools[pool].getHashedSet(object)
+ if err := set.purgeVersion(ctx, bucket, object, v.id, drives[pool]); err != nil {
+ return toObjectErr(err, bucket, decodeDirObject(object), v.id)
+ }
+ }
+ return nil
+}
+
+// purgeVersion physically removes the version versionID of object from
+// disks, the drives of the set that deleteAllDrives checked: no delete
+// marker, no replication state, no free version. A drive that does not hold
+// the version counts as done, so that a retry is idempotent. When the delete
+// reaches its write quorum but fails on some drives, it is repeated on those
+// while the caller still holds the object lock; what remains is left to
+// heal. The caller holds the object lock.
+func (er erasureObjects) purgeVersion(ctx context.Context, bucket, object, versionID string, disks []StorageAPI) error {
+ writeQuorum := len(disks)/2 + 1
+ fi := FileInfo{Name: object, VersionID: versionID}
+ // The purge checked that no drive it deletes on holds a transitioned
+ // revision. A free version would hand a remote object to the tier
+ // cleanup, so none is made.
+ fi.SetSkipTierFreeVersion()
+
+ errs := make([]error, len(disks))
+ for attempt := 0; attempt <= deleteAllRetries; attempt++ {
+ g := errgroup.WithNErrs(len(disks))
+ for index := range disks {
+ if attempt > 0 && (errs[index] == nil || disks[index] == nil) {
+ continue
+ }
+ g.Go(func() error {
+ if disks[index] == nil {
+ return errDiskNotFound
+ }
+ err := disks[index].DeleteVersion(ctx, bucket, object, fi, false, DeleteOptions{})
+ if errors.Is(err, errFileNotFound) || errors.Is(err, errFileVersionNotFound) {
+ return nil
+ }
+ return err
+ }, index)
+ }
+ for index, err := range g.Wait() {
+ if attempt == 0 || (errs[index] != nil && disks[index] != nil) {
+ errs[index] = err
+ }
+ }
+ if err := reduceWriteQuorumErrs(ctx, errs, objectOpIgnoredErrs, writeQuorum); err != nil {
+ er.addPartial(bucket, object, versionID)
+ return err
+ }
+ if !slices.ContainsFunc(errs, func(err error) bool { return err != nil }) {
+ return nil
+ }
+ retry := false
+ for index, err := range errs {
+ retry = retry || err != nil && disks[index] != nil
+ }
+ if !retry {
+ break
+ }
+ }
+ er.addPartial(bucket, object, versionID)
+ return nil
+}
diff --git a/cmd/erasure-server-pool-delete-all_test.go b/cmd/erasure-server-pool-delete-all_test.go
new file mode 100644
index 000000000..b3c861cb1
--- /dev/null
+++ b/cmd/erasure-server-pool-delete-all_test.go
@@ -0,0 +1,1029 @@
+// 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 .
+
+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 = "eavEnabledea/" +
+ "1true" +
+ "dmeEnableddm/" +
+ "1"
+
+// 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("" + rules + "")
+ 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
+ }
+ }
+}
+
+// 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()
+ ctx := p.t.Context()
+ enc := encodeDirObject(name)
+ disk := p.z.serverPools[pool].getHashedSet(enc).getDisks()[0]
+ buf, err := disk.ReadAll(ctx, 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))
+ }
+ 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}
+ item.applyActions(ctx, p.z, objInfos, lr, &sizeSummary{}, nil)
+ var ops []expiryOp
+ for len(queue) > 0 {
+ ops = append(ops, <-queue)
+ }
+ return ops
+}
+
+// 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 !slices.ContainsFunc(ops, func(op expiryOp) bool {
+ task, ok := op.(expiryTask)
+ return ok && task.event.Action.DeleteAll()
+ }) {
+ 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 for walked, the keys its
+// walk matched, and returns the counts of the job.
+func (p *purgeTest) batch(walked ...expireObjInfo) *batchJobInfo {
+ p.t.Helper()
+ globalBatchJobsMetrics.Lock()
+ if globalBatchJobsMetrics.metrics == nil {
+ globalBatchJobsMetrics.metrics = make(map[string]*batchJobInfo)
+ }
+ globalBatchJobsMetrics.Unlock()
+ job := BatchJobRequest{ID: "purge-" + mustGetUUID(), Started: time.Now(), 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)
+ batchObjsForDelete(p.t.Context(), 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("1true"))
+ 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)
+ }
+}
+
+// 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()
+
+ p.expire(0, "ea/tiered-older")
+ 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 {
+ t.Errorf("refused batch purges counted objects=%d failed=%d", ri.Objects, ri.ObjectsFailed)
+ }
+}
+
+// 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.ObjectsFailed == 0 {
+ t.Errorf("job counts objects=%d failed=%d, want 2 and the refusals", ri.Objects, 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)
+ rewrite := func(p *purgeTest, name, versionID string, metadata map[string]string) {
+ t.Helper()
+ disk := z.serverPools[0].getHashedSet(name).getDisks()[0]
+ 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 := z.serverPools[0].getHashedSet("lk/minority").getDisks()[0]
+ 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)
+ }
+}
diff --git a/cmd/erasure-server-pool.go b/cmd/erasure-server-pool.go
index 0811edebc..57cd23ce9 100644
--- a/cmd/erasure-server-pool.go
+++ b/cmd/erasure-server-pool.go
@@ -1234,6 +1234,11 @@ func (z *erasureServerPools) DeleteObject(ctx context.Context, bucket string, ob
if err = checkDelObjArgs(ctx, bucket, object); err != nil {
return objInfo, err
}
+ if opts.DeleteAllVersions {
+ if err := opts.deleteAllInvalid(); err != nil {
+ return objInfo, InvalidArgument{Bucket: bucket, Object: object, Err: err}
+ }
+ }
if !opts.DeletePrefix { // DeletePrefix handles dir object encoding differently.
object = encodeDirObject(object)
@@ -1266,6 +1271,12 @@ func (z *erasureServerPools) DeleteObject(ctx context.Context, bucket string, ob
ctx = lkctx.Context()
defer lk.Unlock(lkctx)
+ // A purge of all versions approves and deletes on the history it reads
+ // itself. The checks below decide by other options and would discard it.
+ if opts.DeleteAllVersions {
+ return z.deleteAllVersions(ctx, bucket, object, opts)
+ }
+
// An ILM expiry is decided on the versions in one pool. Under the lock,
// check it with the versions in all pools; the sets then take it as
// decided, without a recheck by the versions each one holds. They only
@@ -1514,6 +1525,12 @@ func deletesData(obj ObjectToDelete, opts ObjectOptions) bool {
func (z *erasureServerPools) DeleteObjects(ctx context.Context, bucket string, objects []ObjectToDelete, opts ObjectOptions) ([]DeletedObject, []error) {
derrs := make([]error, len(objects))
dobjects := make([]DeletedObject, len(objects))
+ if opts.DeleteAllVersions {
+ for i := range derrs {
+ derrs[i] = errDeleteAllUnsupported
+ }
+ return dobjects, derrs
+ }
objSets := set.NewStringSet()
for i := range derrs {
objects[i].ObjectName = encodeDirObject(objects[i].ObjectName)
diff --git a/cmd/erasure-sets.go b/cmd/erasure-sets.go
index daf7d8e99..b20a1ce90 100644
--- a/cmd/erasure-sets.go
+++ b/cmd/erasure-sets.go
@@ -763,6 +763,9 @@ func (s *erasureSets) deletePrefix(ctx context.Context, bucket string, prefix st
// DeleteObject - deletes an object from the hashedSet based on the object name.
func (s *erasureSets) DeleteObject(ctx context.Context, bucket string, object string, opts ObjectOptions) (objInfo ObjectInfo, err error) {
+ if opts.DeleteAllVersions {
+ return objInfo, errDeleteAllUnsupported
+ }
if opts.DeletePrefix && !opts.DeletePrefixObject {
err := s.deletePrefix(ctx, bucket, object)
return ObjectInfo{}, err
diff --git a/cmd/object-api-interface.go b/cmd/object-api-interface.go
index 307268adb..cd75dd0bf 100644
--- a/cmd/object-api-interface.go
+++ b/cmd/object-api-interface.go
@@ -103,6 +103,17 @@ type ObjectOptions struct {
DeletePrefix bool // set true to enforce a prefix deletion, only application for DeleteObject API,
DeletePrefixObject bool // set true when object's erasure set is resolvable by object name (using getHashedSetIndex)
+ // DeleteAllVersions asks DeleteObject to purge every version of exactly
+ // this key by version ID, for an ILM or batch expiry of all versions;
+ // see erasureServerPools.deleteAllVersions. In memory only.
+ DeleteAllVersions bool
+ // DeleteAllLatest is the latest version that a batch expiry matched.
+ // The purge of all versions only runs while it is still the latest.
+ DeleteAllLatest *deleteAllMatch
+ // DeleteAllResult, when set, receives what a purge of all versions
+ // confirmed, also when it fails part way.
+ DeleteAllResult *deleteAllResult
+
// Pools-layer version resolver; the caller holds the shared object lock.
replicaObjectInfo GetObjectInfoFn