diff --git a/cmd/bucket-lifecycle-incomplete.go b/cmd/bucket-lifecycle-incomplete.go index c50b67d54..5f8bb30aa 100644 --- a/cmd/bucket-lifecycle-incomplete.go +++ b/cmd/bucket-lifecycle-incomplete.go @@ -33,11 +33,23 @@ import ( // could not be read, or another error. It is counted by reason, traced with // the reason, and, unless it was refused, audited and logged. It never sends // a removal notification: what it addressed is still there. +// +// An expiry of all versions of a key can also be refused because a version +// is in a remote tier, or because the latest version changed since a batch +// matched it, and it can stop at its budget after it removed some versions. const ( ilmIncompleteNotApplicable = "not_applicable" ilmIncompleteLocked = "locked" ilmIncompleteHistoryUnavailable = "history_unavailable" ilmIncompleteError = "error" + ilmIncompleteTierUnsupported = "tier_unsupported" + ilmIncompleteBudget = "budget" + ilmIncompleteChanged = "changed" + // ilmIncompleteTierSkipped counts the scanner's observations of a key + // whose expiry of all versions it did not queue, because a version it + // holds is in a remote tier and the purge would refuse it. Each drive's + // scanner of the key observes it, once per cycle. + ilmIncompleteTierSkipped = "tier_unsupported_skipped" ) // ilmIncompleteReasons are the reasons expiry_incomplete_total counts, a @@ -47,6 +59,10 @@ var ilmIncompleteReasons = []string{ ilmIncompleteLocked, ilmIncompleteHistoryUnavailable, ilmIncompleteError, + ilmIncompleteTierUnsupported, + ilmIncompleteBudget, + ilmIncompleteChanged, + ilmIncompleteTierSkipped, } // ilmIncompleteStats counts incomplete ILM expiries by reason. @@ -85,6 +101,12 @@ func ilmExpiryNotDone(err error) (reason string, counted bool) { switch { case err == nil || isErrObjectNotFound(err) || isErrVersionNotFound(err): return "", false + case errors.Is(err, errDeleteAllTiered): + return ilmIncompleteTierUnsupported, true + case errors.Is(err, errDeleteAllChanged): + return ilmIncompleteChanged, true + case errors.Is(err, errDeleteAllBudget): + return ilmIncompleteBudget, true case errors.Is(err, errLifecycleNotApplicable): return ilmIncompleteNotApplicable, true } @@ -101,7 +123,11 @@ func ilmExpiryNotDone(err error) (reason string, counted bool) { // ilmExpiryRefused reports whether reason is a refusal, which changes // nothing, rather than a failure. func ilmExpiryRefused(reason string) bool { - return reason == ilmIncompleteNotApplicable || reason == ilmIncompleteLocked + switch reason { + case ilmIncompleteNotApplicable, ilmIncompleteLocked, ilmIncompleteTierUnsupported, ilmIncompleteChanged: + return true + } + return false } // reportILMExpiryNotDone reports an ILM expiry of oi that ended with err. It diff --git a/cmd/data-scanner.go b/cmd/data-scanner.go index 0f7584590..71ed3b48b 100644 --- a/cmd/data-scanner.go +++ b/cmd/data-scanner.go @@ -24,10 +24,12 @@ import ( "errors" "fmt" "io/fs" + "maps" "math" "math/rand" "os" "path" + "slices" "strconv" "strings" "sync" @@ -1118,8 +1120,31 @@ eventLoop: size := actualSz switch event.Action { case lifecycle.DeleteAllVersionsAction, lifecycle.DelMarkerDeleteAllVersionsAction: - remainingVersions = 0 - applyExpiryRule(event, lcEventSrc_Scanner, oi) + // An expiry of a hot latest version purges the key, which refuses + // a key with a version in a remote tier: do not queue it when this + // pool holds one. A tiered version only another pool holds is + // not known here; the purge decides on it. A transitioned latest + // version takes the transitioned expiry, as before. + refused := oi.TransitionedObject.Status == "" && slices.ContainsFunc(objInfos, ObjectInfo.deleteAllTierRefuses) + if refused { + globalILMIncomplete.add(ilmIncompleteTierSkipped) + } else { + remainingVersions = 0 + applyExpiryRule(event, lcEventSrc_Scanner, oi) + } + // The versions hold their space until the purge removes them, + // which can refuse the key or stop at its budget: account each + // at its size, with no other action. The alerts on many or + // large versions count those of a key that is not queued. + for _, v := range objInfos[idx:] { + sz, _ := v.GetActualSize() + if fn != nil { + fn(v, sz, sz, sizeS) + } + if refused { + cumulativeSize += sz + } + } break eventLoop case lifecycle.DeleteAction, lifecycle.DeleteRestoredAction, lifecycle.DeleteRestoredVersionAction: @@ -1320,16 +1345,24 @@ const ( // 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. +// +// Each is counted and traced with its reason, as reportILMExpiryNotDone +// reports other expiries. A refusal because a version is in a remote tier is +// also logged, since it lasts until such versions can be purged. 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 { + logID := "non-transition-expiry" + obj.Name switch { case err == nil: // there was nothing to remove traceFn(ILMExpiry, nil, nil) return false case purge.Started: + reason, _ := ilmExpiryNotDone(err) + globalILMIncomplete.add(reason) tags := newLifecycleAuditEvent(src, lcEvent).Tags() tags["version-id"] = purge.Latest.VersionID tags[ilmVersionsRemoved] = strconv.Itoa(purge.Removed) tags[ilmPartial] = "true" + tags["ilm-incomplete-reason"] = reason auditLogInternal(ctx, AuditLogOptions{ Event: ILMExpiry, APIName: "ILMExpiry", @@ -1339,22 +1372,22 @@ func deleteAllNotCompleted(ctx context.Context, obj ObjectInfo, lcEvent lifecycl Error: err.Error(), Tags: tags, }) - traceFn(ILMExpiry, tags, err) + trace := maps.Clone(tags) + trace["result"], trace["reason"] = "partial", reason + traceFn(ILMExpiry, trace, err) if errors.Is(err, errDeleteAllBudget) { return false // a long history; the next expiry continues } + // Assume it is still there. + ilmLogOnceIf(ctx, fmt.Errorf("ILM expiry of all versions of %s/%s stopped after it removed %d: %w", obj.Bucket, obj.Name, purge.Removed, err), logID) + return false 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) + ilmLogOnceIf(ctx, fmt.Errorf("ILM expiry of all versions of %s/%s refused: a version is in a remote tier, and a purge of all versions does not remove tiered versions yet", obj.Bucket, obj.Name), logID) + } + if !reportILMExpiryNotDone(ctx, obj, lcEvent, src, err, logID, traceFn) { + traceFn(ILMExpiry, nil, nil) // gone already } - // 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 } diff --git a/cmd/data-scanner_test.go b/cmd/data-scanner_test.go index c45ad1e79..ff66d988c 100644 --- a/cmd/data-scanner_test.go +++ b/cmd/data-scanner_test.go @@ -240,8 +240,10 @@ func TestApplyNewerNoncurrentVersionsLimit(t *testing.T) { }, }, { + // Every version is accounted at its size until the purge + // removes it. objInfos: allVersExpObjInfos, - wants: nil, + wants: allVersExpObjInfos, wantExpired: []ObjectToDelete{{ObjectV: ObjectV{ObjectName: obj, VersionID: allVersExpObjInfos[0].VersionID}}}, }, { diff --git a/cmd/erasure-server-pool-delete-all-reasons_test.go b/cmd/erasure-server-pool-delete-all-reasons_test.go new file mode 100644 index 000000000..9224a1c8a --- /dev/null +++ b/cmd/erasure-server-pool-delete-all-reasons_test.go @@ -0,0 +1,297 @@ +// Copyright (c) 2026 PGSTY +// +// 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 ( + "errors" + "slices" + "strings" + "testing" + + "github.com/minio/minio/internal/bucket/lifecycle" +) + +// An expiry of all versions that does not remove the key is counted by why, +// traced with that reason, and sends no removal event. The scanner does not +// queue one that the purge would refuse for a tiered version it holds, and +// it accounts the versions of a key under such an expiry at their sizes. + +// incompleteDelta runs f and returns by how much each reason of +// expiry_incomplete_total grew. +func incompleteDelta(f func()) map[string]int64 { + before := globalILMIncomplete.snapshot() + f() + after := globalILMIncomplete.snapshot() + delta := map[string]int64{} + for r, n := range after { + if d := n - before[r]; d != 0 { + delta[r] = d + } + } + return delta +} + +func TestPurgeAllVersionsRefusalReasons(t *testing.T) { + z := purgeSinglePool(t, 16) + p := newPurgeTest(t, z, "purge-reasons", MakeBucketOptions{VersioningEnabled: true}, purgeRules) + o := observePurges(t) + expect := func(what string, ops []expiryOp, reason, result string) map[string]string { + t.Helper() + o.drain() + delta := incompleteDelta(func() { p.run(ops) }) + if delta[reason] != 1 { + t.Errorf("%s: counted %v, want one %s", what, delta, reason) + } + events, traces := o.drain() + if len(events) != 0 { + t.Errorf("%s: sent %v", what, events) + } + if len(traces) != 1 || traces[0]["result"] != result || traces[0]["reason"] != reason { + t.Fatalf("%s: traced %v, want one %s expiry for %s", what, traces, result, reason) + } + return traces[0] + } + + // A version in a remote tier: refused, the key kept. + p.tiered(0, "ea/tiered", "v1", 6) + p.version(0, "ea/tiered", "v2", 5) + expect("tiered", []expiryOp{p.deleteAllTask(0, "ea/tiered", lifecycle.DeleteAllVersionsAction)}, ilmIncompleteTierUnsupported, "refused") + if got := p.held(0, "ea/tiered"); !slices.Equal(got, []string{"hot", "tiered"}) { + t.Errorf("tiered key holds %v", got) + } + + // A drive that cannot be read: failed, the key kept. + p.version(0, "ea/unreadable", "v1", 5) + ops := []expiryOp{p.deleteAllTask(0, "ea/unreadable", lifecycle.DeleteAllVersionsAction)} + sets := z.serverPools[0] + index := sets.getHashedSetIndex("ea/unreadable") + sets.erasureDisksMu.Lock() + prior := sets.erasureDisks[index][0] + sets.erasureDisks[index][0] = authorityDisk{StorageAPI: prior, bucket: p.bucket, key: "ea/unreadable", failRead: true} + sets.erasureDisksMu.Unlock() + expect("unreadable drive", ops, ilmIncompleteHistoryUnavailable, "failed") + sets.erasureDisksMu.Lock() + sets.erasureDisks[index][0] = prior + sets.erasureDisksMu.Unlock() + if got := p.held(0, "ea/unreadable"); !slices.Equal(got, []string{"hot"}) { + t.Errorf("key behind an unreadable drive holds %v", got) + } + + // The budget: partial, with the versions removed so far; the next + // expiry continues. + for i, label := range []string{"v1", "v2", "v3"} { + p.version(0, "ea/budget", label, float64(6-i)) + } + saved := deleteAllBudget + deleteAllBudget.versions = 1 + tr := expect("budget", []expiryOp{p.deleteAllTask(0, "ea/budget", lifecycle.DeleteAllVersionsAction)}, ilmIncompleteBudget, "partial") + deleteAllBudget = saved + if tr[ilmPartial] != "true" || tr[ilmVersionsRemoved] != "1" { + t.Errorf("budget: traced %v, want a partial expiry of 1 version", tr) + } + if got := p.held(0, "ea/budget"); !slices.Equal(got, []string{"hot", "hot"}) { + t.Errorf("after the budget the key holds %v, want the two newer versions", got) + } + p.expire(0, "ea/budget") + if got := p.held(0, "ea/budget"); got != nil { + t.Errorf("the next expiry left %v", got) + } + + // Only a batch expiry names the latest version it matched; a changed + // one is a refusal too. + for err, want := range map[error]string{ + errDeleteAllChanged: ilmIncompleteChanged, + errDeleteAllTiered: ilmIncompleteTierUnsupported, + errDeleteAllBudget: ilmIncompleteBudget, + errDeleteAllHistoryUnavailable: ilmIncompleteHistoryUnavailable, + errLifecycleNotApplicable: ilmIncompleteNotApplicable, + } { + reason, counted := ilmExpiryNotDone(err) + if !counted || reason != want { + t.Errorf("%v: reason %q counted %v, want %q", err, reason, counted, want) + } + if refused := ilmExpiryRefused(reason); refused != (want == ilmIncompleteChanged || want == ilmIncompleteTierUnsupported || want == ilmIncompleteNotApplicable) { + t.Errorf("%s: refused %v", reason, refused) + } + } + // To an API caller the unreadable drive is a write quorum error. + if !isErrWriteQuorum(errDeleteAllHistoryUnavailable) || toAPIErrorCode(t.Context(), errDeleteAllHistoryUnavailable) != ErrSlowDownWrite { + t.Errorf("%v is not a write quorum error to callers", errDeleteAllHistoryUnavailable) + } +} + +// A drive whose metadata does not decode refuses the purge, as an +// unreadable one does. Object heal does not rewrite such metadata; once the +// drive holds readable metadata again, the next purge removes the key. +func TestPurgeAllVersionsRefusesCorruptDrive(t *testing.T) { + z := purgeSinglePool(t, 16) + p := newPurgeTest(t, z, "purge-corrupt", MakeBucketOptions{VersioningEnabled: true}, purgeRules) + ctx := t.Context() + const key = "ea/corrupt" + p.put(0, key, "null", daysAgo(7), ObjectOptions{}) + p.version(0, key, "v1", 6) + p.version(0, key, "v2", 5) + ops := []expiryOp{p.deleteAllTask(0, key, lifecycle.DeleteAllVersionsAction)} + + disks := z.serverPools[0].getHashedSet(key).getDisks() + bad := disks[len(disks)-1] + metaPath := pathJoin(key, xlStorageFormatFile) + good, err := bad.ReadAll(ctx, p.bucket, metaPath) + if err != nil { + t.Fatal(err) + } + buf := slices.Clone(good) + buf[len(buf)/2] ^= 0xff + if err := bad.WriteAll(ctx, p.bucket, metaPath, buf); err != nil { + t.Fatal(err) + } + if raw, err := bad.ReadXL(ctx, p.bucket, key, false); err != nil { + t.Fatalf("the corrupted metadata is not read: %v", err) + } else if err := (&xlMetaV2{}).LoadOrConvert(raw.Buf); err == nil { + t.Fatal("the corrupted metadata still decodes") + } + + delta := incompleteDelta(func() { p.run(ops) }) + if delta[ilmIncompleteHistoryUnavailable] != 1 { + t.Errorf("counted %v, want one %s", delta, ilmIncompleteHistoryUnavailable) + } + if got := p.get(key); got != "v2" { + t.Fatalf("the refused purge left %s readable", got) + } + + if err := bad.WriteAll(ctx, p.bucket, metaPath, good); err != nil { + t.Fatal(err) + } + p.run([]expiryOp{p.deleteAllTask(0, key, lifecycle.DeleteAllVersionsAction)}) + if got := p.held(0, key); got != nil { + t.Errorf("the purge after the repair left %v", got) + } +} + +// The scanner of a pool that holds a tiered version of a key whose latest +// version is hot does not queue the expiry of all versions, which the purge +// would refuse, and counts that it skipped it. It accounts every version of +// a key under an expiry of all versions at its size, queued or not. +func TestScannerSkipsDeleteAllForTieredKey(t *testing.T) { + z := purgeSinglePool(t, 16) + p := newPurgeTest(t, z, "purge-scan-tiered", MakeBucketOptions{VersioningEnabled: true}, purgeRules) + ctx := t.Context() + + p.tiered(0, "ea/tiered", "v1", 6) + p.version(0, "ea/tiered", "v2", 5) + p.tiered(0, "dm/tiered", "v1", 6) + p.marker(0, "dm/tiered", 5) + // A tiered latest version takes the transitioned expiry. + p.version(0, "ea/tiered-latest", "v1", 6) + p.tiered(0, "ea/tiered-latest", "v2", 5) + // A free version is not a tiered version. + fv := p.tiered(0, "ea/free", "v1", 6) + if _, err := z.DeleteObject(ctx, p.bucket, "ea/free", ObjectOptions{Versioned: true, VersionID: fv.VersionID}); err != nil { + t.Fatal(err) + } + p.version(0, "ea/free", "v2", 5) + p.version(0, "ea/hot", "v1", 6) + p.version(0, "ea/hot", "v2", 5) + + for _, tc := range []struct { + key string + queued bool + }{ + {"ea/tiered", false}, + {"dm/tiered", false}, + {"ea/tiered-latest", true}, + {"ea/free", true}, + {"ea/hot", true}, + } { + var ops []expiryOp + var sizes map[string]int64 + delta := incompleteDelta(func() { ops, sizes = p.scan(0, tc.key) }) + if got := queuesDeleteAll(ops); got != tc.queued { + t.Errorf("%s: queued an expiry of all versions %v, want %v: %v", tc.key, got, tc.queued, ops) + } + skipped := int64(0) + if !tc.queued { + skipped = 1 + } + if delta[ilmIncompleteTierSkipped] != skipped { + t.Errorf("%s: counted %v, want %d %s", tc.key, delta, skipped, ilmIncompleteTierSkipped) + } + for _, oi := range p.scanned(0, tc.key) { + size, _ := oi.GetActualSize() + if got := sizes[deleteAllVersionID(oi.VersionID)]; got != size { + t.Errorf("%s: version %s accounted at %d, want its size %d", tc.key, oi.VersionID, got, size) + } + } + } + + // The skipped keys keep their versions and remote objects, and the + // queued hot key is purged. + remotes := p.remotes() + for _, key := range []string{"ea/tiered", "dm/tiered", "ea/free", "ea/hot"} { + if ops := p.scanQueue(0, key); len(ops) > 0 { + p.run(ops) + } + } + for key, want := range map[string][]string{ + "ea/tiered": {"hot", "tiered"}, + "dm/tiered": {"DM", "tiered"}, + "ea/free": {"free"}, + "ea/hot": nil, + } { + if got := p.held(0, key); !slices.Equal(got, want) { + t.Errorf("%s holds %v, want %v", key, got, want) + } + } + if got := p.remotes(); got != remotes { + t.Errorf("tier holds %d remote objects, want %d", got, remotes) + } +} + +// A lifecycle expiry never deletes a prefix: the old combination of options +// that removed the keys under a key is refused, and a force delete of a +// prefix still works. +func TestPoolsDeleteObjectRejectsExpirePrefix(t *testing.T) { + z := purgeSinglePool(t, 16) + p := newPurgeTest(t, z, "purge-prefix-expire", MakeBucketOptions{VersioningEnabled: true}, "") + ctx := t.Context() + p.version(0, "pfx/key", "v1", 5) + p.version(0, "pfx/key/child", "v1", 5) + for _, opts := range []ObjectOptions{ + {DeletePrefix: true, Expiration: ExpirationOptions{Expire: true}}, + {DeletePrefix: true, DeletePrefixObject: true, Expiration: ExpirationOptions{Expire: true}}, + } { + _, err := z.DeleteObject(ctx, p.bucket, "pfx/key", opts) + var invalid InvalidArgument + if !errors.As(err, &invalid) || !strings.Contains(err.Error(), "never deletes a prefix") { + t.Errorf("%+v: err = %v, want InvalidArgument", opts, err) + } + } + for _, key := range []string{"pfx/key", "pfx/key/child"} { + if got := p.held(0, key); !slices.Equal(got, []string{"hot"}) { + t.Errorf("%s holds %v after a refused delete", key, got) + } + } + if _, err := z.DeleteObject(ctx, p.bucket, "pfx/key", ObjectOptions{DeletePrefix: true}); err != nil { + t.Fatalf("force delete of the prefix: %v", err) + } + for _, key := range []string{"pfx/key", "pfx/key/child"} { + if got := p.get(key); got != "-" { + t.Errorf("%s reads %s after a force delete", key, got) + } + } +} diff --git a/cmd/erasure-server-pool-delete-all.go b/cmd/erasure-server-pool-delete-all.go index 42fc17101..47b8ba229 100644 --- a/cmd/erasure-server-pool-delete-all.go +++ b/cmd/erasure-server-pool-delete-all.go @@ -69,6 +69,10 @@ var ( // errDeleteAllBudget: the purge stopped at its budget, before the // latest version. errDeleteAllBudget = errors.New("purge of all versions stopped at its budget") + // errDeleteAllHistoryUnavailable: a drive that may hold a revision of + // an approved version could not be read or decoded. Callers see a + // write quorum error. + errDeleteAllHistoryUnavailable = fmt.Errorf("a drive that may hold a version of the key cannot be read or decoded: %w", InsufficientWriteQuorum{}) // errDeleteAllUnsupported: only a single-object delete of the pools // purges all versions. Any other layer would turn the request into // another delete. @@ -377,12 +381,31 @@ func (w *deleteAllWrites) add(fi FileInfo) { w.versions = append(w.versions, fi.ToObjectInfo(w.bucket, w.name, true)) } +// tierRefusesDeleteAll reports whether a version with these transition +// fields makes the purge refuse its key: any version that is not a free +// version and is transitioned, being transitioned, or restored. A free +// version only records a remote object for the tier cleanup. +func tierRefusesDeleteAll(freeVersion bool, status, tier, remote string) bool { + return !freeVersion && (status != "" || tier != "" || remote != "") +} + +// deleteAllTierRefuses: see tierRefusesDeleteAll. +func (fi FileInfo) deleteAllTierRefuses() bool { + return tierRefusesDeleteAll(fi.TierFreeVersion(), fi.TransitionStatus, fi.TransitionTier, fi.TransitionedObjName) +} + +// deleteAllTierRefuses: see tierRefusesDeleteAll. +func (oi ObjectInfo) deleteAllTierRefuses() bool { + t := oi.TransitionedObject + return tierRefusesDeleteAll(t.FreeVersion, t.Status, t.Tier, t.Name) +} + // 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 != "" { + if fi.deleteAllTierRefuses() { return errDeleteAllTiered } if lockEnabled && enforceRetentionForDeletion(ctx, fi.ToObjectInfo(bucket, object, true)) { @@ -413,15 +436,15 @@ func (er erasureObjects) deleteAllDrives(ctx context.Context, bucket, object str continue } if errs[i] != nil { - return nil, toObjectErr(errErasureReadQuorum, bucket, object) + return nil, errDeleteAllHistoryUnavailable } var xl xlMetaV2 if err := xl.LoadOrConvert(raws[i].Buf); err != nil { - return nil, toObjectErr(errErasureReadQuorum, bucket, object) + return nil, errDeleteAllHistoryUnavailable } versions, err := xl.ListVersions(bucket, object, false) if err != nil { - return nil, toObjectErr(errErasureReadQuorum, bucket, object) + return nil, errDeleteAllHistoryUnavailable } for _, fi := range versions { if fi.TierFreeVersion() || !ids[deleteAllVersionID(fi.VersionID)] { diff --git a/cmd/erasure-server-pool-delete-all_test.go b/cmd/erasure-server-pool-delete-all_test.go index eb6ef2ecf..31787b5f9 100644 --- a/cmd/erasure-server-pool-delete-all_test.go +++ b/cmd/erasure-server-pool-delete-all_test.go @@ -232,14 +232,13 @@ func (p *purgeTest) childIn(pool int, parent string, same bool) string { } } -// 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 { +// scanned returns the versions of name that the first drive of its set in +// pool holds, as the scanner of that drive reads them. +func (p *purgeTest) scanned(pool int, name string) []ObjectInfo { p.t.Helper() - 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)) + buf, err := disk.ReadAll(p.t.Context(), p.bucket, pathJoin(enc, xlStorageFormatFile)) if err != nil { p.t.Fatalf("read %s: %v", name, err) } @@ -252,6 +251,24 @@ func (p *purgeTest) scanQueue(pool int, name string) []expiryOp { for i, fi := range fivs.Versions { objInfos[i] = fi.ToObjectInfo(p.bucket, name, vc != nil && vc.Versioned(name)) } + return objInfos +} + +// scanQueue applies the lifecycle of the bucket to the versions that pool +// holds of name, as the scanner does, and returns what it queued. +func (p *purgeTest) scanQueue(pool int, name string) []expiryOp { + p.t.Helper() + ops, _ := p.scan(pool, name) + return ops +} + +// scan is scanQueue that also returns the size the scanner accounts for +// each version, by version ID. +func (p *purgeTest) scan(pool int, name string) ([]expiryOp, map[string]int64) { + p.t.Helper() + ctx := p.t.Context() + enc := encodeDirObject(name) + objInfos := p.scanned(pool, name) lc, err := globalLifecycleSys.Get(p.bucket) if err != nil { p.t.Fatal(err) @@ -268,12 +285,31 @@ func (p *purgeTest) scanQueue(pool int, name string) []expiryOp { 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) + sizes := map[string]int64{} + item.applyActions(ctx, p.z, objInfos, lr, &sizeSummary{}, func(oi ObjectInfo, sz, _ int64, _ *sizeSummary) { + sizes[deleteAllVersionID(oi.VersionID)] += sz + }) var ops []expiryOp for len(queue) > 0 { ops = append(ops, <-queue) } - return ops + return ops, sizes +} + +// deleteAllTask is the expiry of all versions of name that a scanner queues +// for the latest version that pool holds, as the scanner of a drive or a +// pool that holds no tiered version of the key queues it. +func (p *purgeTest) deleteAllTask(pool int, name string, action lifecycle.Action) expiryOp { + p.t.Helper() + return expiryTask{objInfo: p.scanned(pool, name)[0], event: lifecycle.Event{Action: action}, src: lcEventSrc_Scanner} +} + +// queuesDeleteAll reports whether ops holds an expiry of all versions. +func queuesDeleteAll(ops []expiryOp) bool { + return slices.ContainsFunc(ops, func(op expiryOp) bool { + task, ok := op.(expiryTask) + return ok && task.event.Action.DeleteAll() + }) } // expire runs what the scanner queues for name in pool with an expiry @@ -281,10 +317,7 @@ func (p *purgeTest) scanQueue(pool int, name string) []expiryOp { 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() - }) { + if !queuesDeleteAll(ops) { p.t.Fatalf("the scanner queued no expiry of all versions of %s: %v", name, ops) } p.run(ops) @@ -754,7 +787,12 @@ func TestPurgeAllVersionsRefusesTieredKeys(t *testing.T) { p.tiered(0, "bx/tiered-latest", "v2", 5) remotes := p.remotes() - p.expire(0, "ea/tiered-older") + // The scanner of a pool that holds the tiered version does not queue + // the expiry; one that does not hold it queues it, and the purge refuses. + if ops := p.scanQueue(0, "ea/tiered-older"); queuesDeleteAll(ops) { + t.Errorf("the scanner queued an expiry of all versions that the purge refuses: %v", ops) + } + p.run([]expiryOp{p.deleteAllTask(0, "ea/tiered-older", lifecycle.DeleteAllVersionsAction)}) ri := p.batch(p.matched("bx/tiered-older"), p.matched("bx/tiered-latest")) for name, want := range map[string][]string{ "ea/tiered-older": {"hot", "tiered"}, @@ -900,9 +938,15 @@ func TestPurgeAllVersionsRetiresOlderWritesFirst(t *testing.T) { // on: a transitioned or a protected one refuses the key. func TestPurgeAllVersionsChecksEveryDrive(t *testing.T) { z := purgeSinglePool(t, 16) + // The minority revision is on the last drive; the scanner of the first + // drive, which the test runs, does not see it and queues the expiry. + minority := func(name string) StorageAPI { + disks := z.serverPools[0].getHashedSet(name).getDisks() + return disks[len(disks)-1] + } rewrite := func(p *purgeTest, name, versionID string, metadata map[string]string) { t.Helper() - disk := z.serverPools[0].getHashedSet(name).getDisks()[0] + disk := minority(name) if err := disk.UpdateMetadata(t.Context(), p.bucket, name, FileInfo{VersionID: versionID, Metadata: metadata}, UpdateMetadataOpts{}); err != nil { t.Fatal(err) } @@ -930,7 +974,7 @@ func TestPurgeAllVersionsChecksEveryDrive(t *testing.T) { 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] + disk := minority("lk/minority") if fi, err := disk.ReadVersion(t.Context(), "", locked.bucket, "lk/minority", v.VersionID, ReadOptions{}); err != nil || fi.Metadata[strings.ToLower(xhttp.AmzObjectLockLegalHold)] != string(objectlock.LegalHoldOn) { t.Errorf("the held revision: %v", err) diff --git a/cmd/erasure-server-pool-lifecycle-authority_test.go b/cmd/erasure-server-pool-lifecycle-authority_test.go index 196bde257..3c3f3e5d5 100644 --- a/cmd/erasure-server-pool-lifecycle-authority_test.go +++ b/cmd/erasure-server-pool-lifecycle-authority_test.go @@ -539,6 +539,10 @@ func TestPoolsLifecycleLegacyHistory(t *testing.T) { } } +// authorityDeleteAll is the expiry of all versions of a key as the expiry +// worker sends it: a purge of all versions, approved by the lifecycle. +var authorityDeleteAll = ObjectOptions{DeleteAllVersions: true, Expiration: ExpirationOptions{Expire: true}} + // A fresh version committed to 15 of 16 drives, with eight other drives // unreadable: only seven answering drives show it. func TestPoolsLifecyclePartialHistory(t *testing.T) { @@ -554,7 +558,7 @@ func TestPoolsLifecyclePartialHistory(t *testing.T) { if _, _, err := s.z.lifecycleEventsAllPools(ctx, s.bucket, key); !isErrReadQuorum(err) { t.Errorf("partial history: %v", err) } - _, err := s.z.DeleteObject(ctx, s.bucket, key, ObjectOptions{Versioned: true, DeletePrefix: true, DeletePrefixObject: true, Expiration: ExpirationOptions{Expire: true}}) + _, err := s.z.DeleteObject(ctx, s.bucket, key, authorityDeleteAll) if !isErrWriteQuorum(err) { t.Errorf("DeleteAll on a partial history: %v", err) } @@ -615,7 +619,7 @@ func TestPoolsLifecycleMissingVersion(t *testing.T) { if _, _, err := s.z.lifecycleEventsAllPools(ctx, s.bucket, key); !isErrReadQuorum(err) { t.Errorf("history without an observed version: %v", err) } - _, err := s.z.DeleteObject(ctx, s.bucket, key, ObjectOptions{Versioned: true, DeletePrefix: true, DeletePrefixObject: true, Expiration: ExpirationOptions{Expire: true}}) + _, err := s.z.DeleteObject(ctx, s.bucket, key, authorityDeleteAll) if !isErrWriteQuorum(err) { t.Errorf("DeleteAll: %v", err) } @@ -630,7 +634,6 @@ func TestPoolsLifecycleMissingVersion(t *testing.T) { // The completeness checks also hold where a set has fewer answering drives // than a read quorum or no version reaches one. func TestPoolsLifecycleHistoryEarlyReturns(t *testing.T) { - deleteAll := ObjectOptions{Versioned: true, DeletePrefix: true, DeletePrefixObject: true, Expiration: ExpirationOptions{Expire: true}} // Seven drives hold the key, the other nine do not have it at all. t.Run("minority-copy-only", func(t *testing.T) { @@ -648,7 +651,7 @@ func TestPoolsLifecycleHistoryEarlyReturns(t *testing.T) { if _, _, err := s.z.lifecycleEventsAllPools(ctx, s.bucket, key); !isErrReadQuorum(err) { t.Errorf("minority copy: %v", err) } - if _, err := s.z.DeleteObject(ctx, s.bucket, key, deleteAll); !isErrWriteQuorum(err) { + if _, err := s.z.DeleteObject(ctx, s.bucket, key, authorityDeleteAll); !isErrWriteQuorum(err) { t.Errorf("DeleteAll: %v", err) } for i := 9; i < len(disks); i++ { @@ -671,7 +674,7 @@ func TestPoolsLifecycleHistoryEarlyReturns(t *testing.T) { if _, _, err := s.z.lifecycleEventsAllPools(ctx, s.bucket, key); !isErrReadQuorum(err) { t.Fatalf("disjoint minorities: %v", err) } - if _, err := s.z.DeleteObject(ctx, s.bucket, key, deleteAll); !isErrWriteQuorum(err) { + if _, err := s.z.DeleteObject(ctx, s.bucket, key, authorityDeleteAll); !isErrWriteQuorum(err) { t.Fatalf("DeleteAll: %v", err) } }) diff --git a/cmd/erasure-server-pool.go b/cmd/erasure-server-pool.go index 0fa01112b..4d141d711 100644 --- a/cmd/erasure-server-pool.go +++ b/cmd/erasure-server-pool.go @@ -1239,6 +1239,11 @@ func (z *erasureServerPools) DeleteObject(ctx context.Context, bucket string, ob return objInfo, InvalidArgument{Bucket: bucket, Object: object, Err: err} } } + // A lifecycle expiry removes all versions of one key as a purge of all + // versions. A prefix delete would also remove the keys under it. + if opts.DeletePrefix && opts.Expiration.Expire { + return objInfo, InvalidArgument{Bucket: bucket, Object: object, Err: errors.New("a lifecycle expiry never deletes a prefix")} + } if !opts.DeletePrefix { // DeletePrefix handles dir object encoding differently. object = encodeDirObject(object)