diff --git a/cmd/erasure-object.go b/cmd/erasure-object.go index 5c9ee5d80..8ceae79d4 100644 --- a/cmd/erasure-object.go +++ b/cmd/erasure-object.go @@ -2151,7 +2151,7 @@ func (er erasureObjects) DeleteObject(ctx context.Context, bucket, object string } } - if opts.Expiration.Expire { + if opts.Expiration.Expire && !opts.Expiration.CheckedAllPools { if gerr == nil { evt := evalActionFromLifecycle(ctx, *lc, rcfg, replcfg, goi) var isErr bool diff --git a/cmd/erasure-server-pool-lifecycle.go b/cmd/erasure-server-pool-lifecycle.go new file mode 100644 index 000000000..800b39d92 --- /dev/null +++ b/cmd/erasure-server-pool-lifecycle.go @@ -0,0 +1,252 @@ +// Copyright (c) 2026 Denis Babchuk +// +// 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" + "sort" + "sync" + "time" + + "github.com/minio/minio/internal/bucket/lifecycle" +) + +// ILM decides on the versions of an object that one pool holds: the scanner +// and the drive heal read the drives of one set, decommission and rebalance +// walk one pool, and a GET or HEAD of a version reads the pool that holds it. +// IsLatest, NumVersions and SuccessorModTime count only these. An object can +// have versions in several pools: from two concurrent first writes, from +// overwrites that went to another pool, from data movement. The expiries +// these decisions run delete in every pool, so a version that is the latest +// in its own pool but not of the object had a newer version hidden under a +// delete marker or deleted with all versions, and a delete marker without +// noncurrent versions in its pool was removed over the versions of another +// pool. Deciding on fewer versions only delays the expiry of a noncurrent +// version: its successor in the pool is not older than the one of the object, +// and fewer newer versions count against NewerNoncurrentVersions. +// +// Before an expiry deletes in any pool, it is evaluated again with the +// versions of the object in all pools, as the scanner would evaluate it if +// one pool held them all. An expiry that no longer applies is not run, as the +// recheck of a set does it: the scanner queues it again once the rule +// applies to the object. + +// objectVersions returns the versions of object that a read quorum of the +// drives of the set agree on, newest first, without free versions. It returns +// no versions when the set does not hold the object, and an error when too +// few drives answer to tell. +func (er erasureObjects) objectVersions(ctx context.Context, bucket, object string) ([]FileInfo, error) { + disks := er.getDisks() + rawFileInfos, errs := readAllRawFileInfo(ctx, disks, bucket, object, false) + readQuorum := (len(disks) + 1) / 2 + + var ( + metas [][]xlMetaV2ShallowVersion + metaV uint8 + unknown int + ) + for i, rf := range rawFileInfos { + if errs[i] != nil { + if !errors.Is(errs[i], errFileNotFound) && !errors.Is(errs[i], errVolumeNotFound) { + unknown++ + } + continue + } + var xl xlMetaV2 + if err := xl.LoadOrConvert(rf.Buf); err != nil { + unknown++ + continue + } + metas = append(metas, xl.versions) + metaV = xl.metaV + } + if len(metas) < readQuorum { + if len(metas)+unknown < readQuorum { + return nil, nil // not in this set + } + return nil, toObjectErr(errErasureReadQuorum, bucket, object) + } + + merged := mergeXLV2Versions(readQuorum, false, 0, metas...) + if len(merged) == 0 { + for _, versions := range metas { + if len(versions) > 0 { + return nil, toObjectErr(errErasureReadQuorum, bucket, object) + } + } + return nil, nil + } + all, err := xlMetaV2{versions: merged, metaV: metaV}.ListVersions(bucket, object, false) + if err != nil { + return nil, toObjectErr(err, bucket, object) + } + versions := all[:0] + for _, fi := range all { + if !fi.TierFreeVersion() { + versions = append(versions, fi) + } + } + return versions, nil +} + +// lifecycleEventsAllPools returns the versions of object in all pools, +// newest first, and the lifecycle event of each as the scanner evaluates it +// on them. A version with copies in several pools, as an interrupted move +// leaves it, counts once. +func (z *erasureServerPools) lifecycleEventsAllPools(ctx context.Context, bucket, object string) ([]ObjectInfo, []lifecycle.Event, error) { + lc, err := globalLifecycleSys.Get(bucket) + if err != nil { + if errors.Is(err, BucketLifecycleNotFound{Bucket: bucket}) { + return nil, nil, nil + } + return nil, nil, err + } + lr, err := globalBucketObjectLockSys.Get(bucket) + if err != nil { + return nil, nil, err + } + rcfg, err := getReplicationConfig(ctx, bucket) + if err != nil { + return nil, nil, err + } + + 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, nil, err + } + } + + var all []FileInfo + for _, versions := range poolVersions { + all = append(all, versions...) + } + // Each pool lists its versions newest first. On equal times the lower + // pool comes first, as reads choose it. + sort.SliceStable(all, func(i, j int) bool { return all[i].ModTime.After(all[j].ModTime) }) + seen := make(map[string]struct{}, len(all)) + versions := all[:0] + for _, fi := range all { + if _, ok := seen[fi.VersionID]; ok { + continue + } + seen[fi.VersionID] = struct{}{} + versions = append(versions, fi) + } + + vc, _ := globalBucketVersioningSys.Get(bucket) + versioned := vc != nil && vc.Versioned(decodeDirObject(object)) + objInfos := make([]ObjectInfo, len(versions)) + objOpts := make([]lifecycle.ObjectOpts, len(versions)) + for i := range versions { + fi := versions[i] + fi.IsLatest = i == 0 + fi.NumVersions = len(versions) + fi.SuccessorModTime = time.Time{} + if i > 0 { + fi.SuccessorModTime = versions[i-1].ModTime + } + objInfos[i] = fi.ToObjectInfo(bucket, object, versioned) + objOpts[i] = objInfos[i].ToLifecycleOpts() + } + events, err := lifecycle.NewEvaluator(*lc).WithLockRetention(&lr).WithReplicationConfig(rcfg).Eval(objOpts) + if err != nil { + return nil, nil, err + } + return objInfos, events, nil +} + +// lifecycleAllPools holds the versions of an object in all pools and their +// lifecycle events, see lifecycleEventsAllPools. +type lifecycleAllPools struct { + versions []ObjectInfo + events []lifecycle.Event + err error +} + +// checkExpiryAllPools returns nil when the ILM expiry that opts describe +// still applies to object with the versions it has in all pools, see +// lifecycleAllPools.expiryApplies. +func (z *erasureServerPools) checkExpiryAllPools(ctx context.Context, bucket, object string, opts ObjectOptions) error { + var all lifecycleAllPools + all.versions, all.events, all.err = z.lifecycleEventsAllPools(ctx, bucket, object) + return all.expiryApplies(bucket, object, opts) +} + +// expiryApplies returns nil when the ILM expiry that opts describe applies +// to the versions, and otherwise the error a set returns for an expiry that +// no longer applies: +// - an expiry of all versions (DeletePrefix) needs the latest version to +// expire with all versions; +// - a delete marker (no VersionID) hides the latest version, which has to +// expire; +// - a version (VersionID) has to expire by itself: a noncurrent version, +// or a delete marker without noncurrent versions. +func (all lifecycleAllPools) expiryApplies(bucket, object string, opts ObjectOptions) error { + if all.err != nil { + if _, ok := all.err.(InsufficientReadQuorum); ok { + return InsufficientWriteQuorum{} + } + return all.err + } + name := decodeDirObject(object) + switch { + case opts.DeletePrefix: + if len(all.events) > 0 && all.events[0].Action.DeleteAll() { + return nil + } + case opts.VersionID == "": + if len(all.events) > 0 && (all.events[0].Action == lifecycle.DeleteAction || all.events[0].Action == lifecycle.DeleteAllVersionsAction) { + return nil + } + default: + for i, oi := range all.versions { + if oi.VersionID == opts.VersionID || oi.VersionID == "" && opts.VersionID == nullVersionID { + if all.events[i].Action == lifecycle.DeleteVersionAction { + return nil + } + break + } + } + return VersionNotFound{Bucket: bucket, Object: name, VersionID: opts.VersionID} + } + return ObjectNotFound{Bucket: bucket, Object: name} +} + +// expiryNeedsAllPools reports whether an ILM expiry in opts deletes or hides +// versions by a rule that depends on the other versions of the object. A +// restored copy expires by its own restore date. Without versions each copy +// of an object is a write of its own, and a set rechecks the rule on its +// copy (deleteObjectFromAllPools). +func (z *erasureServerPools) expiryNeedsAllPools(opts ObjectOptions) bool { + if !opts.Expiration.Expire || z.SinglePool() || opts.Transition.ExpireRestored { + return false + } + return opts.DeletePrefix || opts.VersionID != "" || opts.Versioned || opts.VersionSuspended +} diff --git a/cmd/erasure-server-pool-lifecycle_test.go b/cmd/erasure-server-pool-lifecycle_test.go new file mode 100644 index 000000000..0e346e33f --- /dev/null +++ b/cmd/erasure-server-pool-lifecycle_test.go @@ -0,0 +1,416 @@ +// Copyright (c) 2026 Denis Babchuk +// +// 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 ( + "fmt" + "io" + "slices" + "strings" + "testing" + "time" + + "github.com/minio/minio/internal/bucket/lifecycle" + "github.com/minio/minio/internal/bucket/versioning" +) + +// The scanner, the drive heal and a GET or HEAD of a version evaluate ILM on +// the versions of a key in one pool: IsLatest, NumVersions and SuccessorModTime +// count only these. A key can have versions in several pools: two concurrent +// first writes, overwrites that went to another pool. The expiry they decide +// runs in every pool. + +// splitILM lays out the versions of keys in the two pools of migrationILM. +type splitILM struct { + *migrationILM + ids map[string]string // object and label -> version ID + bodies map[string]string // body -> label +} + +func newSplitILM(t *testing.T, status versioning.State, rules string) *splitILM { + t.Helper() + return &splitILM{migrationILM: newMigrationILM(t, status, false, rules), ids: map[string]string{}, bodies: map[string]string{}} +} + +// splitVersion is a version of a key in a test layout. +type splitVersion struct { + pools []int // the pools that hold a copy of it + label string + days float64 // its age + marker bool + tiered bool // its data is in a remote tier + null bool // a null version (unversioned or suspended bucket) + size int // its size, at least the size of its label +} + +func (s *splitILM) write(object string, v splitVersion) { + s.t.Helper() + ctx := s.t.Context() + mtime := UTCNow().Add(-time.Duration(v.days * float64(24*time.Hour))) + versionID := mustGetUUID() + if v.null { + versionID = "" + } + for _, pool := range v.pools { + if v.marker { + _, err := s.z.serverPools[pool].DeleteObject(ctx, s.bucket, object, + ObjectOptions{Versioned: true, VersionID: versionID, DeleteMarker: true, MTime: mtime}) + if err != nil { + s.t.Fatal(err) + } + continue + } + opts := ObjectOptions{Versioned: !v.null, VersionID: versionID, MTime: mtime} + if v.null && s.status() == versioning.Suspended { + opts = ObjectOptions{VersionSuspended: true, MTime: mtime} + } + body := object + " " + v.label + if len(body) < v.size { + body += strings.Repeat(".", v.size-len(body)) + } + putConsistencyObject(s.t, s.z, s.bucket, object, pool, body, opts) + s.bodies[body] = v.label + if v.tiered { + vid := versionID + if vid == "" { + vid = nullVersionID + } + infos, errs := readAllFileInfo(ctx, s.z.serverPools[pool].getHashedSet(object).getDisks(), "", s.bucket, object, vid, false, false) + if errs[0] != nil { + s.t.Fatal(errs[0]) + } + fi := infos[0] + fi.TransitionStatus, fi.TransitionTier, fi.TransitionedObjName = lifecycle.TransitionComplete, "WARM", "remote-"+v.label + if err := s.z.serverPools[pool].DecomTieredObject(ctx, s.bucket, object, fi, ObjectOptions{}); err != nil { + s.t.Fatal(err) + } + } + } + if v.marker { + s.markers[s.key(object, versionID)] = v.label + } else { + s.labels[s.key(object, versionID)] = v.label + } + s.ids[object+" "+v.label] = versionID + s.track(object) +} + +func (s *splitILM) status() versioning.State { + vc, _ := globalBucketVersioningSys.Get(s.bucket) + if vc == nil { + return "" + } + return vc.Status +} + +// scan applies ILM to every object in a pool as the scanner of a drive of +// that pool does, from the xl.meta of that drive, and runs the queued +// expiries as an expiry worker does. +func (s *splitILM) scan(pool int) { + s.t.Helper() + ctx := s.t.Context() + lc, err := globalLifecycleSys.Get(s.bucket) + if err != nil { + s.t.Fatal(err) + } + lr, err := globalBucketObjectLockSys.Get(s.bucket) + if err != nil { + s.t.Fatal(err) + } + vc, _ := globalBucketVersioningSys.Get(s.bucket) + for _, object := range s.objects { + disk := s.z.serverPools[pool].getHashedSet(object).getDisks()[0] + rf, err := disk.ReadXL(ctx, s.bucket, object, false) + if err != nil { + continue // not in this pool + } + fivs, err := getFileInfoVersions(rf.Buf, s.bucket, object, false) + if err != nil { + s.t.Fatal(err) + } + objInfos := make([]ObjectInfo, len(fivs.Versions)) + for i, fi := range fivs.Versions { + objInfos[i] = fi.ToObjectInfo(s.bucket, object, vc != nil && vc.Versioned(object)) + } + item := scannerItem{bucket: s.bucket, objectName: object, lifeCycle: lc} + item.applyActions(ctx, s.z, objInfos, lr, &sizeSummary{}, nil) + } + s.runQueue() +} + +// head applies ILM to a version as a HEAD or GET with its versionId does. +func (s *splitILM) head(object, label string) lifecycle.Action { + s.t.Helper() + ctx := s.t.Context() + oi, err := s.z.GetObjectInfo(ctx, s.bucket, object, ObjectOptions{VersionID: s.ids[object+" "+label]}) + if err != nil { + s.t.Fatal(err) + } + lc, err := globalLifecycleSys.Get(s.bucket) + if err != nil { + s.t.Fatal(err) + } + lr, err := globalBucketObjectLockSys.Get(s.bucket) + if err != nil { + s.t.Fatal(err) + } + evt := evalActionFromLifecycle(ctx, *lc, lr, nil, oi) + if evt.Action.Delete() { + applyExpiryRule(evt, lcEventSrc_s3HeadObject, oi) + } + s.runQueue() + return evt.Action +} + +// runQueue runs the queued expiries with an expiry worker. +func (s *splitILM) runQueue() { + done := make(chan struct{}) + go func() { + defer close(done) + globalExpiryState.Worker(s.queue) + }() + s.queue <- nil // the worker returns once it ran every expiry queued before + <-done +} + +// get returns the label of the version a GET of object reads, or "-" when it +// reads as deleted. +func (s *splitILM) get(object string) string { + s.t.Helper() + ctx := s.t.Context() + oi, err := s.z.GetObjectInfo(ctx, s.bucket, object, ObjectOptions{}) + if err != nil { + if isErrObjectNotFound(err) || isErrVersionNotFound(err) || isErrMethodNotAllowed(err) { + return "-" + } + return fmt.Sprintf("<%v>", err) + } + if oi.VersionID != "" && oi.VersionID != nullVersionID { + return s.label(object, oi.VersionID, false) + } + // Copies of a null version in two pools are told apart by their data. + gr, err := s.z.GetObjectNInfo(ctx, s.bucket, object, nil, nil, ObjectOptions{}) + if err != nil { + return fmt.Sprintf("<%v>", err) + } + defer gr.Close() + data, err := io.ReadAll(gr) + if err != nil { + s.t.Fatal(err) + } + if l, ok := s.bodies[string(data)]; ok { + return l + } + return "?" + string(data) +} + +// views returns the versions of each object in each pool, newest first. +func (s *splitILM) views() []map[string][]string { + views := make([]map[string][]string, len(s.z.serverPools)) + for pool := range s.z.serverPools { + views[pool] = s.view(pool) + } + return views +} + +type splitKey struct { + name string + versions []splitVersion + before string // what a GET reads before ILM ran + after string // what a GET reads once ILM ran + want [][]string // versions in each pool once ILM ran, nil to skip +} + +// run lays out keys, runs two scanner cycles over all pools and checks that +// a GET of each key never reads anything but before or after, and the +// versions each pool holds at the end. The second cycle must not change +// anything. +func (s *splitILM) run(keys []splitKey) { + s.t.Helper() + for _, k := range keys { + for _, v := range k.versions { + s.write(k.name, v) + } + if got := s.get(k.name); got != k.before { + s.t.Fatalf("%s: fixture reads %s, want %s", k.name, got, k.before) + } + } + var first []map[string][]string + for cycle := 1; cycle <= 2; cycle++ { + for pool := range s.z.serverPools { + s.scan(pool) + for _, k := range keys { + if got := s.get(k.name); got != k.before && got != k.after { + s.t.Errorf("%s: reads %s after the scanner of pool %d (cycle %d), want %s", k.name, got, pool, cycle, k.after) + } + } + } + if cycle == 1 { + first = s.views() + } + } + last := s.views() + for _, k := range keys { + if got := s.get(k.name); got != k.after { + s.t.Errorf("%s: reads %s once ILM ran, want %s", k.name, got, k.after) + } + for pool := range last { + if !slices.Equal(first[pool][k.name], last[pool][k.name]) { + s.t.Errorf("%s: the second scanner cycle changed pool %d from %v to %v", k.name, pool, first[pool][k.name], last[pool][k.name]) + } + if k.want != nil && !slices.Equal(last[pool][k.name], k.want[pool]) { + s.t.Errorf("%s: pool %d holds %v, want %v", k.name, pool, last[pool][k.name], k.want[pool]) + } + } + } +} + +const splitILMRules = "expEnabledexp-" + + "1" + + "eavEnabledeav-" + + "1true" + + "edmEnablededm-" + + "true" + + "dmeEnableddme-" + + "1" + + "ncvEnabledncv-" + + "1" + + "exsEnabledexs-100" + + "1" + + "nnvEnablednnv-" + + "11" + +func splitAt(label string, days float64, pools ...int) splitVersion { + return splitVersion{pools: pools, label: label, days: days} +} + +func splitMarker(label string, days float64, pools ...int) splitVersion { + return splitVersion{pools: pools, label: label, days: days, marker: true} +} + +func splitTiered(label string, days float64, pools ...int) splitVersion { + return splitVersion{pools: pools, label: label, days: days, tiered: true} +} + +func splitNull(label string, days float64, pools ...int) splitVersion { + return splitVersion{pools: pools, label: label, days: days, null: true} +} + +// splitILMOnePool are keys in one pool. ILM applies to them as usual. +func splitILMOnePool(pool int) []splitKey { + in := func(vs ...string) [][]string { + want := [][]string{nil, nil} + want[pool] = vs + return want + } + return []splitKey{ + {"exp-one", []splitVersion{splitAt("v1", 5, pool), splitAt("v2", 4, pool)}, "v2", "-", in("DM", "v2", "v1")}, + {"exp-one-tiered", []splitVersion{splitAt("v1", 5, pool), splitTiered("v2", 4, pool)}, "v2", "-", in("DM", "v2", "v1")}, + {"eav-one", []splitVersion{splitAt("v1", 5, pool), splitAt("v2", 4, pool)}, "v2", "-", in()}, + {"edm-one", []splitVersion{splitMarker("dm", 3, pool)}, "-", "-", in()}, + {"dme-one", []splitVersion{splitAt("v1", 5, pool), splitMarker("dm", 3, pool)}, "-", "-", in()}, + {"ncv-one", []splitVersion{splitAt("v1", 5, pool), splitAt("v2", 4, pool), splitAt("v3", 0.01, pool)}, "v3", "v3", in("v3", "v2")}, + {"nnv-one", []splitVersion{splitAt("v1", 5, pool), splitAt("v2", 4, pool), splitAt("v3", 3, pool), splitAt("v4", 2, pool)}, "v4", "v4", in("v4", "v3")}, + } +} + +func TestPoolsLifecycleSplitObject(t *testing.T) { + t.Run("versioned", func(t *testing.T) { + s := newSplitILM(t, versioning.Enabled, splitILMRules) + s.run([]splitKey{ + // Expiration with Days applies to the current version v2 only. + {"exp-old-in-0", []splitVersion{splitAt("v1", 5, 0), splitAt("v2", 0.01, 1)}, "v2", "v2", [][]string{{"v1"}, {"v2"}}}, + {"exp-old-in-1", []splitVersion{splitAt("v2", 0.01, 0), splitAt("v1", 5, 1)}, "v2", "v2", [][]string{{"v2"}, {"v1"}}}, + {"exp-tiered-old-in-0", []splitVersion{splitTiered("v1", 5, 0), splitAt("v2", 0.01, 1)}, "v2", "v2", [][]string{{"v1"}, {"v2"}}}, + // v2 is old enough: one delete marker hides the key. + {"exp-both-old", []splitVersion{splitAt("v1", 5, 0), splitAt("v2", 4, 1)}, "v2", "-", [][]string{{"DM", "v1"}, {"v2"}}}, + // The rule applies to v2 only by its size. + {"exs-big-in-1", []splitVersion{splitAt("v1", 5, 0), {pools: []int{1}, label: "v2", days: 4, size: 200}}, "v2", "-", [][]string{{"DM", "v1"}, {"v2"}}}, + // The marker hides v1 and is not a marker without noncurrent + // versions; v1 is not the current version. + {"exp-marker-in-0", []splitVersion{splitAt("v1", 5, 1), splitMarker("dm", 3, 0)}, "-", "-", [][]string{{"dm"}, {"v1"}}}, + // ExpiredObjectAllVersions applies to the current version v2 only. + {"eav-old-in-0", []splitVersion{splitAt("v1", 5, 0), splitAt("v2", 0.01, 1)}, "v2", "v2", [][]string{{"v1"}, {"v2"}}}, + {"eav-old-in-1", []splitVersion{splitAt("v2", 0.01, 0), splitAt("v1", 5, 1)}, "v2", "v2", [][]string{{"v2"}, {"v1"}}}, + {"eav-tiered-old-in-0", []splitVersion{splitTiered("v1", 5, 0), splitAt("v2", 0.01, 1)}, "v2", "v2", [][]string{{"v1"}, {"v2"}}}, + {"eav-both-old", []splitVersion{splitAt("v1", 5, 0), splitAt("v2", 4, 1)}, "v2", "-", [][]string{nil, nil}}, + // A copy of v1 in both pools, as an interrupted move leaves it. + {"eav-copies", []splitVersion{splitAt("v1", 5, 0, 1)}, "v1", "-", [][]string{nil, nil}}, + // The marker hides v1 in the other pool. + {"edm-marker-in-0", []splitVersion{splitAt("v1", 5, 1), splitMarker("dm", 3, 0)}, "-", "-", [][]string{{"dm"}, {"v1"}}}, + {"edm-marker-in-1", []splitVersion{splitAt("v1", 5, 0), splitMarker("dm", 3, 1)}, "-", "-", [][]string{{"v1"}, {"dm"}}}, + {"edm-copies", []splitVersion{splitMarker("dm", 3, 0, 1)}, "-", "-", [][]string{nil, nil}}, + // v2 in the other pool is the current version, the marker does + // not expire. + {"dme-new-in-1", []splitVersion{splitMarker("dm", 3, 0), splitAt("v2", 0.01, 1)}, "v2", "v2", [][]string{{"dm"}, {"v2"}}}, + {"dme-new-in-0", []splitVersion{splitAt("v2", 0.01, 0), splitMarker("dm", 3, 1)}, "v2", "v2", [][]string{{"v2"}, {"dm"}}}, + // The marker is the current version and expires with v1. + {"dme-old-in-0", []splitVersion{splitAt("v1", 5, 0), splitMarker("dm", 3, 1)}, "-", "-", [][]string{nil, nil}}, + // v1 is noncurrent since v2 was written, v2 since v3 was. + {"ncv-split", []splitVersion{splitAt("v1", 5, 0), splitAt("v2", 4, 1), splitAt("v3", 2, 0)}, "v3", "v3", [][]string{{"v3"}, {"v2"}}}, + }) + }) + t.Run("versioned/one-pool", func(t *testing.T) { + for pool := range 2 { + s := newSplitILM(t, versioning.Enabled, splitILMRules) + s.run(splitILMOnePool(pool)) + } + }) + t.Run("versioned/head-version", func(t *testing.T) { + s := newSplitILM(t, versioning.Enabled, splitILMRules) + keys := []splitKey{ + {"exp-old-in-0", []splitVersion{splitAt("v1", 5, 0), splitAt("v2", 0.01, 1)}, "v2", "v2", [][]string{{"v1"}, {"v2"}}}, + {"eav-old-in-0", []splitVersion{splitAt("v1", 5, 0), splitAt("v2", 0.01, 1)}, "v2", "v2", [][]string{{"v1"}, {"v2"}}}, + } + for _, k := range keys { + for _, v := range k.versions { + s.write(k.name, v) + } + act := s.head(k.name, "v1") + t.Logf("%s: HEAD of v1 evaluates %s", k.name, act) + if got := s.get(k.name); got != k.after { + t.Errorf("%s: reads %s after a HEAD of v1, want %s", k.name, got, k.after) + } + views := s.views() + for pool := range views { + if !slices.Equal(views[pool][k.name], k.want[pool]) { + t.Errorf("%s: pool %d holds %v, want %v", k.name, pool, views[pool][k.name], k.want[pool]) + } + } + } + }) + t.Run("suspended", func(t *testing.T) { + s := newSplitILM(t, versioning.Suspended, splitILMRules) + s.run([]splitKey{ + {"exp-null-old-in-0", []splitVersion{splitNull("v1", 5, 0), splitAt("v2", 0.01, 1)}, "v2", "v2", [][]string{{"v1"}, {"v2"}}}, + }) + }) + t.Run("unversioned", func(t *testing.T) { + s := newSplitILM(t, "", splitILMRules) + s.run([]splitKey{ + {"eav-old-in-0", []splitVersion{splitNull("old", 5, 0), splitNull("new", 0.01, 1)}, "new", "new", nil}, + {"exp-old-in-0", []splitVersion{splitNull("old", 5, 0), splitNull("new", 0.01, 1)}, "new", "new", nil}, + }) + views := s.views() + for object, want := range map[string]int{"eav-old-in-0": 1, "exp-old-in-0": 0} { + if len(views[0][object]) != want || len(views[1][object]) != 1 { + t.Errorf("%s: pools hold %v and %v, want %d and 1 copies", object, views[0][object], views[1][object], want) + } + } + }) +} diff --git a/cmd/erasure-server-pool.go b/cmd/erasure-server-pool.go index d1e83bb71..c79a0f5b5 100644 --- a/cmd/erasure-server-pool.go +++ b/cmd/erasure-server-pool.go @@ -1248,6 +1248,16 @@ func (z *erasureServerPools) DeleteObject(ctx context.Context, bucket string, ob ctx = lkctx.Context() defer lk.Unlock(lkctx) + // 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. + if z.expiryNeedsAllPools(opts) { + if err = z.checkExpiryAllPools(ctx, bucket, object, opts); err != nil { + return ObjectInfo{Name: decodeDirObject(object), VersionID: opts.VersionID}, err + } + opts.Expiration.CheckedAllPools = true + } + if opts.DeletePrefix { return ObjectInfo{}, z.deletePrefix(ctx, bucket, object) } @@ -1517,10 +1527,28 @@ func (z *erasureServerPools) DeleteObjects(ctx context.Context, bucket string, o batch = make([]ObjectToDelete, 0, len(objects)) batchIdx = make([]int, 0, len(objects)) data, dataIdx := make([]ObjectToDelete, 0, len(objects)), make([]int, 0, len(objects)) + ilm := make(map[string]lifecycleAllPools) for i, obj := range objects { if derrs[i] != nil { continue } + // A version that ILM expires is checked with the versions of + // its object in all pools, see checkExpiryAllPools. When the + // expiry no longer applies, nothing is deleted and nothing is + // left to replicate. + vopts := ObjectOptions{Expiration: opts.Expiration, VersionID: obj.VersionID} + if obj.VersionID != "" && z.expiryNeedsAllPools(vopts) { + all, ok := ilm[obj.ObjectName] + if !ok { + all.versions, all.events, all.err = z.lifecycleEventsAllPools(ctx, bucket, obj.ObjectName) + ilm[obj.ObjectName] = all + } + if err := all.expiryApplies(bucket, obj.ObjectName, vopts); err != nil { + dobjects[i] = DeletedObject{ObjectName: obj.ObjectName, VersionID: obj.VersionID} + derrs[i] = err + continue + } + } if deletesData(obj, opts) { if z.copyWithoutDeleteQuorum(ctx, bucket, obj.ObjectName, obj.VersionID) { derrs[i] = InsufficientWriteQuorum{} diff --git a/cmd/object-api-interface.go b/cmd/object-api-interface.go index 964a01149..658e30f9b 100644 --- a/cmd/object-api-interface.go +++ b/cmd/object-api-interface.go @@ -159,6 +159,10 @@ type WalkOptions struct { // ExpirationOptions represents object options for object expiration at objectLayer. type ExpirationOptions struct { Expire bool + // CheckedAllPools: the expiry was checked with the versions of the + // object in all pools (checkExpiryAllPools). A set does not check it + // again with the versions it holds alone. + CheckedAllPools bool } // TransitionOptions represents object options for transition ObjectLayer operation diff --git a/cmd/object-api-interface_gen.go b/cmd/object-api-interface_gen.go index 919cdb174..8637f5498 100644 --- a/cmd/object-api-interface_gen.go +++ b/cmd/object-api-interface_gen.go @@ -79,10 +79,13 @@ func (z BucketOptions) Msgsize() (s int) { // MarshalMsg implements msgp.Marshaler func (z ExpirationOptions) MarshalMsg(b []byte) (o []byte, err error) { o = msgp.Require(b, z.Msgsize()) - // map header, size 1 + // map header, size 2 // string "Expire" - o = append(o, 0x81, 0xa6, 0x45, 0x78, 0x70, 0x69, 0x72, 0x65) + o = append(o, 0x82, 0xa6, 0x45, 0x78, 0x70, 0x69, 0x72, 0x65) o = msgp.AppendBool(o, z.Expire) + // string "CheckedAllPools" + o = append(o, 0xaf, 0x43, 0x68, 0x65, 0x63, 0x6b, 0x65, 0x64, 0x41, 0x6c, 0x6c, 0x50, 0x6f, 0x6f, 0x6c, 0x73) + o = msgp.AppendBool(o, z.CheckedAllPools) return } @@ -110,6 +113,12 @@ func (z *ExpirationOptions) UnmarshalMsg(bts []byte) (o []byte, err error) { err = msgp.WrapError(err, "Expire") return } + case "CheckedAllPools": + z.CheckedAllPools, bts, err = msgp.ReadBoolBytes(bts) + if err != nil { + err = msgp.WrapError(err, "CheckedAllPools") + return + } default: bts, err = msgp.Skip(bts) if err != nil { @@ -124,7 +133,7 @@ func (z *ExpirationOptions) UnmarshalMsg(bts []byte) (o []byte, err error) { // Msgsize returns an upper bound estimate of the number of bytes occupied by the serialized message func (z ExpirationOptions) Msgsize() (s int) { - s = 1 + 7 + msgp.BoolSize + s = 1 + 7 + msgp.BoolSize + 16 + msgp.BoolSize return } diff --git a/cmd/object-handlers-common.go b/cmd/object-handlers-common.go index ddfc8dde3..acd7ebcb2 100644 --- a/cmd/object-handlers-common.go +++ b/cmd/object-handlers-common.go @@ -467,6 +467,7 @@ func deleteObjectVersions(ctx context.Context, o ObjectLayer, bucket string, toD deletedObjs, errs := o.DeleteObjects(ctx, bucket, toDel, ObjectOptions{ PrefixEnabledFn: vc.PrefixEnabled, VersionSuspended: vc.Suspended(), + Expiration: ExpirationOptions{Expire: true}, }) for i, dobj := range deletedObjs {