From 2b228f678c2961fdc4fdefdaba7b61e72701565d Mon Sep 17 00:00:00 2001 From: Feng Ruohang Date: Thu, 5 Nov 2026 03:24:42 +0800 Subject: [PATCH] fix(ilm): purge only the expired key's own versions in delete-all An expiry of all versions (ExpiredObjectAllVersions, DelMarkerExpiration, and batch expiration with the default purge) asked the pools layer for a prefix delete. The pools layer rebuilt the options, so the request ran as a recursive delete of the key's directory in every erasure set of every pool. Object a/b is stored inside the directory of a, so every key under a/ was removed with it, however new, with one audit entry and one notification for a alone. The rebuilt options also skipped the lifecycle recheck, so on a single pool a version written after the decision was deleted too, and tiered versions lost their metadata without free versions. The expiry worker, the migration expiry and batch expiration now ask for a purge with the in-memory ObjectOptions.DeleteAllVersions: - DeleteObject hands it to the purge under the object lock, before the all-pools check. ILM decides with the unchanged lifecycleEventsAllPools, on a single pool too; batch needs the version its walk matched to still be the latest. A second read of each pool's authoritative history records every copy, older writes of the same ID included, and must equal the approval. Anything else deletes nothing. - Before deleting, the purge reads every drive of each set that holds a copy. It refuses the key when a drive cannot be read, when any revision of an approved version there is transitioned, or, with Object Lock, when retention or a legal hold protects one. It then deletes on exactly the drive handles it checked. - Versions are deleted by ID with the drives' version delete, which only removes the object directory once it is empty: oldest first, for each version older writes before the copy reads choose, and the latest last, only once every older version is gone from every pool. No marker, replication state or free version is written. The purge stops at the first error, repeats a quorum-successful delete on the drives that missed it while the lock is held, and stops before the latest version at a budget of 1000 versions or one minute, reads included. - The ILM audit entry records how many versions were removed. A purge that failed after it began to delete is audited as partial and sends no removal event. Batch counts come from the purge; a key whose latest version changed is not expired, and a failed delete marker fails the job. The option is rejected with conflicting options and by every layer below the pools. Force delete and internal prefix deletes keep DeletePrefix. Delete-all of a transitioned latest version keeps its current path, and a key with tiered older versions is refused. On a single pool the purge relies on the lock-domain fix for writer exclusion. Reviewed adversarially in two design rounds and four patch rounds; the last approved with nits, which are applied. Co-Authored-By: Claude Opus 5.5 Signed-off-by: Feng Ruohang --- cmd/batch-expire.go | 82 +- cmd/batch-handlers.go | 32 +- cmd/data-scanner.go | 71 +- cmd/erasure-object.go | 10 + ...ure-server-pool-delete-all-options_test.go | 221 ++++ cmd/erasure-server-pool-delete-all.go | 459 ++++++++ cmd/erasure-server-pool-delete-all_test.go | 1029 +++++++++++++++++ cmd/erasure-server-pool.go | 17 + cmd/erasure-sets.go | 3 + cmd/object-api-interface.go | 11 + 10 files changed, 1896 insertions(+), 39 deletions(-) create mode 100644 cmd/erasure-server-pool-delete-all-options_test.go create mode 100644 cmd/erasure-server-pool-delete-all.go create mode 100644 cmd/erasure-server-pool-delete-all_test.go 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 9307aae81..2ea927d6f 100644 --- a/cmd/erasure-object.go +++ b/cmd/erasure-object.go @@ -1774,6 +1774,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) @@ -2024,6 +2031,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 849ad6c67..698c5c8d7 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) @@ -1248,6 +1253,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 @@ -1496,6 +1507,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 d40ff3ce7..cef70b699 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