From 9f7b49369d4cbc1a5506af4bff05df05c8c8b6d3 Mon Sep 17 00:00:00 2001 From: Feng Ruohang Date: Thu, 5 Nov 2026 00:16:27 +0800 Subject: [PATCH] fix(ilm): require an authoritative cross-pool history before lifecycle expiry The merged all-pool check from #279 decided on the versions that a read quorum of each set merged non-strictly. A drive that missed one metadata update could therefore supply the revision used for the decision, and a partially unreadable set could drop a committed version from the history. Both let an expiry delete data the rule did not cover, including a version under legal hold. - objectVersions accepts a revision only when max(n/2+1, the version's metadata read quorum) drives hold it, identifies revisions by the header plus a sha256 over a canonical, length-prefixed encoding of the normalized metadata, filters non-accepted revisions before the strict merge, and fails closed on omitted versions, too many unreadable drives, or cancellation. - Lock fields are read with the evaluator's key precedence. Cross-pool copies that disagree on hold or retention must be strictly ordered by valid timestamps; copies with different ModTime are separate writes and an older protected write refuses the object. The set safeguard re-checks protection and pending purges on its fresh read for checked version deletes. - Duplicate copies with semantically different replication or purge state refuse the object. - Batch expiry announces and audits only completed or already-absent removals. Reviewed adversarially over seven rounds; the final round approved with no remaining findings in scope. All-versions expiry through deletePrefix and tiered delete-all dispatch are tracked separately. Signed-off-by: Feng Ruohang Co-Authored-By: Claude Opus 5.5 --- cmd/erasure-object.go | 12 +- cmd/erasure-server-pool-consistency.go | 7 +- ...re-server-pool-lifecycle-authority_test.go | 1399 +++++++++++++++++ ...asure-server-pool-lifecycle-safety_test.go | 40 +- cmd/erasure-server-pool-lifecycle.go | 511 +++++- cmd/erasure-server-pool.go | 13 +- cmd/object-api-interface.go | 4 + cmd/object-handlers-common.go | 9 +- 8 files changed, 1940 insertions(+), 55 deletions(-) create mode 100644 cmd/erasure-server-pool-lifecycle-authority_test.go diff --git a/cmd/erasure-object.go b/cmd/erasure-object.go index 64658aa50..9307aae81 100644 --- a/cmd/erasure-object.go +++ b/cmd/erasure-object.go @@ -2118,6 +2118,11 @@ func (er erasureObjects) DeleteObject(ctx context.Context, bucket, object string return ObjectInfo{}, toObjectErr(er.deletePrefix(ctx, bucket, object), bucket, object) } + // An ILM version expiry that the pools checked with the versions in all + // pools; the set still guards the protection it reads, see + // checkedExpiryRefused. + checkedExpiry := opts.Expiration.Expire && opts.Expiration.CheckedAllPools && opts.VersionID != "" && !opts.Transition.ExpireRestored + storageDisks := er.getDisks() versionFound := true objInfo = ObjectInfo{VersionID: opts.VersionID} // version id needed in Delete API response. @@ -2126,8 +2131,9 @@ func (er erasureObjects) DeleteObject(ctx context.Context, bucket, object string if gerr != nil && goi.Name == "" { if _, ok := gerr.(InsufficientReadQuorum); ok { // A restored-copy expiry needs fresh metadata to confirm its target - // and deadline; ordinary unversioned DELETE may still try a blind delete. - if opts.Transition.ExpireRestored || opts.Versioned || opts.VersionSuspended || countOnlineDisks(storageDisks) < len(storageDisks)/2+1 { + // and deadline, a checked expiry to guard its protection; ordinary + // unversioned DELETE may still try a blind delete. + if opts.Transition.ExpireRestored || checkedExpiry || opts.Versioned || opts.VersionSuspended || countOnlineDisks(storageDisks) < len(storageDisks)/2+1 { // Add an MRF heal for next time. er.addPartial(bucket, object, opts.VersionID) return objInfo, InsufficientWriteQuorum{} @@ -2197,6 +2203,8 @@ func (er erasureObjects) DeleteObject(ctx context.Context, bucket, object string } } } + } else if checkedExpiry && checkedExpiryRefused(*lc, rcfg, replcfg, goi, opts.Expiration.ApprovedProtection) { + return goi, errLifecycleNotApplicable } // Determine whether to mark object deleted for replication diff --git a/cmd/erasure-server-pool-consistency.go b/cmd/erasure-server-pool-consistency.go index 320fb57af..7d449b60a 100644 --- a/cmd/erasure-server-pool-consistency.go +++ b/cmd/erasure-server-pool-consistency.go @@ -671,7 +671,9 @@ func (z *erasureServerPools) deleteObjectReconciled(ctx context.Context, bucket, // deleteObjectsReconciled removes data while retaining each item's authoritative // copy until every other pool has completed its cleanup. The caller holds the // pools-layer write locks and separates marker creation from this batch. -func (z *erasureServerPools) deleteObjectsReconciled(ctx context.Context, bucket string, objects []ObjectToDelete, opts ObjectOptions) ([]DeletedObject, []error) { +// approved holds the Object Lock state each checked ILM expiry was approved +// with, see ExpirationOptions.ApprovedProtection. +func (z *erasureServerPools) deleteObjectsReconciled(ctx context.Context, bucket string, objects []ObjectToDelete, approved []objectLockState, opts ObjectOptions) ([]DeletedObject, []error) { deleted := make([]DeletedObject, len(objects)) errs := make([]error, len(objects)) primary := make([]int, len(objects)) @@ -765,6 +767,9 @@ func (z *erasureServerPools) deleteObjectsReconciled(ctx context.Context, bucket // versions against all pools. Empty IDs and user deletes have no such // authorization; restored-copy expiry uses the single-object path. deleteOpts.Expiration.CheckedAllPools = opts.Expiration.Expire && obj.VersionID != "" + if deleteOpts.Expiration.CheckedAllPools { + deleteOpts.Expiration.ApprovedProtection = approved[i] + } deleteOpts.DeleteReplication = obj.ReplicationState() if opts.PrefixEnabledFn != nil { deleteOpts.Versioned = opts.PrefixEnabledFn(obj.ObjectName) diff --git a/cmd/erasure-server-pool-lifecycle-authority_test.go b/cmd/erasure-server-pool-lifecycle-authority_test.go new file mode 100644 index 000000000..d75525356 --- /dev/null +++ b/cmd/erasure-server-pool-lifecycle-authority_test.go @@ -0,0 +1,1399 @@ +// 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" + "crypto/sha256" + "encoding/json" + "errors" + "fmt" + "maps" + "slices" + "strings" + "testing" + "time" + + "github.com/google/uuid" + "github.com/minio/madmin-go/v3" + "github.com/minio/minio/internal/bucket/lifecycle" + "github.com/minio/minio/internal/bucket/replication" + "github.com/minio/minio/internal/bucket/versioning" + "github.com/minio/minio/internal/config/storageclass" + "github.com/minio/minio/internal/event" + xhttp "github.com/minio/minio/internal/http" + "github.com/minio/minio/internal/pubsub" +) + +// An all-pools ILM approval is only as sound as the history each set reports +// and the protection it reads under the lock. These cases leave drives with +// stale, partial or unreadable metadata, as failed or interrupted writes and +// offline drives do, and check that no expiry deletes what they hide. + +const authorityNoncurrentRule = `ncvEnabledncv-1` + +var authorityHoldKey = strings.ToLower(xhttp.AmzObjectLockLegalHold) + +func newLockedSplitILM(t *testing.T, rules string) *splitILM { + t.Helper() + return &splitILM{migrationILM: newMigrationILM(t, versioning.Enabled, true, rules), ids: map[string]string{}, bodies: map[string]string{}} +} + +// authorityParity pins new objects to 12+4 on the 16 drives of a set. Other +// tests can leave another storage class configured. Without one, 4 is the +// default; Update cannot restore an unconfigured class, so leave it then. +func authorityParity(t *testing.T) { + t.Helper() + if globalStorageClass.GetParityForSC("") < 0 { + return + } + previous := globalStorageClass + globalStorageClass.Update(storageclass.Config{Standard: storageclass.StorageClass{Parity: 4}}) + t.Cleanup(func() { globalStorageClass.Update(previous) }) +} + +// authorityDisk fails the history scan (ReadXL) or the version removal of +// one key on a drive. +type authorityDisk struct { + StorageAPI + bucket, key string + failRead, failWrite bool +} + +func (d authorityDisk) ReadXL(ctx context.Context, bucket, key string, data bool) (RawFileInfo, error) { + if d.failRead && bucket == d.bucket && key == d.key { + return RawFileInfo{}, errDiskNotFound + } + return d.StorageAPI.ReadXL(ctx, bucket, key, data) +} + +func (d authorityDisk) DeleteVersion(ctx context.Context, volume, path string, fi FileInfo, forceDelMarker bool, opts DeleteOptions) error { + if d.failWrite && volume == d.bucket && path == d.key { + return errDiskNotFound + } + return d.StorageAPI.DeleteVersion(ctx, volume, path, fi, forceDelMarker, opts) +} + +func (d authorityDisk) DeleteVersions(ctx context.Context, volume string, versions []FileInfoVersions, opts DeleteOptions) []error { + errs := make([]error, len(versions)) + var pass []FileInfoVersions + var idx []int + for i, v := range versions { + if d.failWrite && volume == d.bucket && v.Name == d.key { + errs[i] = errDiskNotFound + continue + } + pass = append(pass, v) + idx = append(idx, i) + } + if len(pass) > 0 { + for j, err := range d.StorageAPI.DeleteVersions(ctx, volume, pass, opts) { + errs[idx[j]] = err + } + } + return errs +} + +// wrapAuthorityDisks makes the drives of pool that drive selects fail reads +// or writes of key. +func wrapAuthorityDisks(t *testing.T, s *splitILM, pool int, key string, failRead, failWrite bool, drive func(int) bool) { + t.Helper() + p := s.z.serverPools[pool] + p.erasureDisksMu.Lock() + prior := p.erasureDisks + wrapped := make([][]StorageAPI, len(prior)) + for i, disks := range prior { + wrapped[i] = append([]StorageAPI(nil), disks...) + for j := range disks { + if drive(j) { + wrapped[i][j] = authorityDisk{StorageAPI: disks[j], bucket: s.bucket, key: key, failRead: failRead, failWrite: failWrite} + } + } + } + p.erasureDisks = wrapped + p.erasureDisksMu.Unlock() + t.Cleanup(func() { + p.erasureDisksMu.Lock() + p.erasureDisks = prior + p.erasureDisksMu.Unlock() + }) +} + +func authorityDrives(from, to int) func(int) bool { + return func(i int) bool { return i >= from && i < to } +} + +// removeAuthorityCopies removes version id of key from the drives of pool +// that drive selects, as a write that did not reach them leaves it. +func removeAuthorityCopies(t *testing.T, s *splitILM, pool int, key, id string, drive func(int) bool) { + t.Helper() + for i, disk := range s.z.serverPools[pool].getHashedSet(key).getDisks() { + if !drive(i) { + continue + } + fi := FileInfo{VersionID: id} + fi.SetSkipTierFreeVersion() + if err := disk.DeleteVersion(t.Context(), s.bucket, key, fi, false, DeleteOptions{}); err != nil { + t.Fatal(err) + } + } +} + +func authorityHeader(t *testing.T, disk StorageAPI, bucket, key, id string) xlMetaV2VersionHeader { + t.Helper() + rf, err := disk.ReadXL(t.Context(), bucket, key, false) + if err != nil { + t.Fatal(err) + } + var xl xlMetaV2 + if err := xl.LoadOrConvert(rf.Buf); err != nil { + t.Fatal(err) + } + for _, ver := range xl.versions { + if uuid.UUID(ver.header.VersionID).String() == id { + return ver.header + } + } + t.Fatalf("drive has no version %s", id) + return xlMetaV2VersionHeader{} +} + +func authorityExpireVersion(t *testing.T, s *splitILM, key, id string) error { + t.Helper() + _, err := s.z.DeleteObject(t.Context(), s.bucket, key, ObjectOptions{Versioned: true, VersionID: id, Expiration: ExpirationOptions{Expire: true}}) + return err +} + +func authorityPresent(t *testing.T, s *splitILM, pool int, key, id string) { + t.Helper() + if _, err := s.z.serverPools[pool].GetObjectInfo(t.Context(), s.bucket, key, ObjectOptions{VersionID: id}); err != nil { + t.Fatalf("pool %d lost version %s: %v", pool, id, err) + } +} + +// minorityRevision writes the versions old and new of key to pool 0, applies +// metadata to old, and leaves its last drive with the revision before, as a +// drive that missed the update does. It reports whether that stale revision +// sorts before the fifteen others. +func minorityRevision(t *testing.T, s *splitILM, key string, metadata map[string]string) (id string, staleFirst bool) { + t.Helper() + ctx := t.Context() + s.write(key, splitAt("old", 5, 0)) + s.write(key, splitAt("new", 3, 0)) + id = s.ids[key+" old"] + disks := s.z.serverPools[0].getHashedSet(key).getDisks() + last := disks[len(disks)-1] + stale, err := last.ReadVersion(ctx, "", s.bucket, key, id, ReadOptions{}) + if err != nil { + t.Fatal(err) + } + setLifecycleCopyMetadata(t, s, key, id, 0, metadata, false) + if err := last.WriteMetadata(ctx, "", s.bucket, key, stale); err != nil { + t.Fatal(err) + } + return id, authorityHeader(t, last, s.bucket, key, id).sortsBefore(authorityHeader(t, disks[0], s.bucket, key, id)) +} + +// One drive missed a legal hold that the other fifteen committed. Its stale +// revision must not be selected, and must not keep the fifteen from +// deciding, whichever signature sorts first. +func TestPoolsLifecycleMinorityLockRevision(t *testing.T) { + authorityParity(t) + s := newLockedSplitILM(t, authorityNoncurrentRule) + ctx := t.Context() + for attempt := range 32 { + key := fmt.Sprintf("ncv-lock-%d", attempt) + stamp := UTCNow().Format(time.RFC3339Nano) + id, staleFirst := minorityRevision(t, s, key, poolLockMetadata("", "ON", stamp, stamp)) + versions, events, err := s.z.lifecycleEventsAllPools(ctx, s.bucket, key) + switch { + case err != nil: + t.Errorf("attempt %d (stale sorts first: %t): history: %v", attempt, staleFirst, err) + case len(versions) != 2: + t.Fatalf("attempt %d: merged %d versions", attempt, len(versions)) + case versions[1].UserDefined[authorityHoldKey] != "ON" || events[1].Action != lifecycle.NoneAction: + t.Errorf("attempt %d (stale sorts first: %t): merged hold=%q event=%s", attempt, staleFirst, versions[1].UserDefined[authorityHoldKey], events[1].Action) + } + if err := authorityExpireVersion(t, s, key, id); !errors.Is(err, errLifecycleNotApplicable) { + t.Errorf("attempt %d (stale sorts first: %t): expiry of a held version: %v", attempt, staleFirst, err) + } + oi, err := s.z.serverPools[0].GetObjectInfo(ctx, s.bucket, key, ObjectOptions{VersionID: id}) + if err != nil || oi.UserDefined[authorityHoldKey] != "ON" { + t.Fatalf("held version changed: hold=%q err=%v", oi.UserDefined[authorityHoldKey], err) + } + if staleFirst { + return + } + } + t.Fatal("no attempt where the stale revision sorts first") +} + +// The same layout without protection: a drive missed an ordinary metadata +// update of a version that expires. The fifteen still decide its expiry. +func TestPoolsLifecycleMinorityStaleRevision(t *testing.T) { + authorityParity(t) + s := newSplitILM(t, versioning.Enabled, splitILMRules) + ctx := t.Context() + for attempt := range 32 { + key := fmt.Sprintf("ncv-stale-%d", attempt) + id, staleFirst := minorityRevision(t, s, key, map[string]string{"x-amz-meta-note": fmt.Sprint(attempt)}) + _, events, err := s.z.lifecycleEventsAllPools(ctx, s.bucket, key) + if err != nil || len(events) != 2 || events[1].Action != lifecycle.DeleteVersionAction { + t.Errorf("attempt %d (stale sorts first: %t): events=%v err=%v", attempt, staleFirst, events, err) + } + if err := authorityExpireVersion(t, s, key, id); err != nil { + t.Fatalf("attempt %d (stale sorts first: %t): expiry: %v", attempt, staleFirst, err) + } + for i, disk := range s.z.serverPools[0].getHashedSet(key).getDisks() { + if _, err := disk.ReadVersion(ctx, "", s.bucket, key, id, ReadOptions{}); !errors.Is(err, errFileVersionNotFound) { + t.Fatalf("attempt %d: drive %d still holds the expired version: %v", attempt, i, err) + } + } + if staleFirst { + return + } + } + t.Fatal("no attempt where the stale revision sorts first") +} + +// authorityRevision encodes a 12+4 object version as drive index holds it, +// with the legal hold status hold. +func authorityRevision(t *testing.T, id [16]byte, modTime int64, index int, hold string) xlMetaV2ShallowVersion { + t.Helper() + v := xlMetaV2Version{Type: ObjectType, ObjectV2: &xlMetaV2Object{ + VersionID: id, DataDir: [16]byte{1}, ErasureAlgorithm: ReedSolomon, ErasureM: 12, ErasureN: 4, + ErasureBlockSize: blockSizeV2, ErasureIndex: index, ErasureDist: make([]uint8, 16), BitrotChecksumAlgo: HighwayHash, + PartNumbers: []int{1}, PartSizes: []int64{5}, Size: 5, ModTime: modTime, + MetaUser: map[string]string{authorityHoldKey: hold, "etag": "e", "content-type": "text/plain"}, + MetaSys: map[string][]byte{"x-minio-internal-a": {1}, "x-minio-internal-b": {2}, "x-minio-internal-c": {3}}, + }} + for i := range v.ObjectV2.ErasureDist { + v.ObjectV2.ErasureDist[i] = uint8(i + 1) + } + meta, err := v.MarshalMsg(nil) + if err != nil { + t.Fatal(err) + } + return xlMetaV2ShallowVersion{header: v.header(), meta: meta} +} + +// The header Signature is a 32-bit fold of weak hashes, so two revisions can +// share a header. Revisions are told apart by their metadata: drives that +// hold byte-for-byte different metadata under one header are not one +// revision, and the merged history only carries the accepted metadata. +func TestPoolsLifecycleRevisionIdentity(t *testing.T) { + newID, oldID := [16]byte{2}, [16]byte{1} + history := func(off func(drive int) bool) [][]xlMetaV2ShallowVersion { + metas := make([][]xlMetaV2ShallowVersion, 16) + for i := range metas { + old := authorityRevision(t, oldID, 1, i+1, "ON") + if off(i) { + // The stale revision under the header of the held one. + header := old.header + old = authorityRevision(t, oldID, 1, i+1, "OFF") + old.header = header + } + metas[i] = []xlMetaV2ShallowVersion{authorityRevision(t, newID, 2, i+1, ""), old} + } + return metas + } + metaVs := slices.Repeat([]uint8{xlMetaVersion}, 16) + readQuorum := func(xlMetaV2ShallowVersion, uint8) int { return 12 } + + t.Run("revision-key", func(t *testing.T) { + on, onOther := authorityRevision(t, oldID, 1, 1, "ON"), authorityRevision(t, oldID, 1, 7, "ON") + off := authorityRevision(t, oldID, 1, 1, "OFF") + off.header = on.header + if versionRevision(on, xlMetaVersion) != versionRevision(onOther, xlMetaVersion) { + t.Error("one revision on two drives has two keys") + } + if versionRevision(on, xlMetaVersion) == versionRevision(off, xlMetaVersion) { + t.Error("different metadata under one header has one key") + } + }) + t.Run("stale-minority-first", func(t *testing.T) { + merged, err := authoritativeHistory(t.Context(), history(func(i int) bool { return i == 0 }), metaVs, xlMetaVersion, 9, readQuorum) + if err != nil || len(merged) != 2 { + t.Fatalf("history: err=%v versions=%d", err, len(merged)) + } + var v xlMetaV2Version + if _, err := v.unmarshalV(xlMetaVersion, merged[1].meta); err != nil { + t.Fatal(err) + } + if hold := v.ObjectV2.MetaUser[authorityHoldKey]; hold != "ON" { + t.Fatalf("merged history carries hold=%q", hold) + } + }) + t.Run("split-8-8", func(t *testing.T) { + if merged, err := authoritativeHistory(t.Context(), history(func(i int) bool { return i%2 == 0 }), metaVs, xlMetaVersion, 9, readQuorum); !errors.Is(err, errErasureReadQuorum) { + t.Fatalf("8/8 metadata split: %d versions, err=%v", len(merged), err) + } + }) + // The user and the system map are separate sections of the digest: an + // entry that moves between them changes what the evaluator reads. + t.Run("map-boundary", func(t *testing.T) { + for _, entry := range [][2]string{ + {authorityHoldKey, "ON"}, + {strings.ToLower(xhttp.AmzObjectLockRetainUntilDate), "2099-01-01T00:00:00Z"}, + {xhttp.AmzObjectTagging, "expire=yes"}, + } { + ver := authorityRevision(t, oldID, 1, 1, "") + var v xlMetaV2Version + if _, err := v.unmarshalV(xlMetaVersion, ver.meta); err != nil { + t.Fatal(err) + } + v.ObjectV2.MetaUser, v.ObjectV2.MetaSys = map[string]string{entry[0]: entry[1]}, nil + user, err := v.MarshalMsg(nil) + if err != nil { + t.Fatal(err) + } + v.ObjectV2.MetaUser, v.ObjectV2.MetaSys = nil, map[string][]byte{entry[0]: []byte(entry[1])} + sys, err := v.MarshalMsg(nil) + if err != nil { + t.Fatal(err) + } + if versionRevision(xlMetaV2ShallowVersion{header: ver.header, meta: user}, xlMetaVersion) == + versionRevision(xlMetaV2ShallowVersion{header: ver.header, meta: sys}, xlMetaVersion) { + t.Errorf("%s moved between the maps keeps the revision", entry[0]) + } + } + }) + t.Run("canceled", func(t *testing.T) { + ctx, cancel := context.WithCancel(t.Context()) + cancel() + if _, err := authoritativeHistory(ctx, history(func(int) bool { return false }), metaVs, xlMetaVersion, 9, readQuorum); !errors.Is(err, context.Canceled) { + t.Fatalf("canceled history: %v", err) + } + }) +} + +// authorityMovedHold serves a drive whose metadata of version id carries the +// legal hold in the system map instead of the user map, under the header it +// had: the evaluator no longer reads the hold there. +type authorityMovedHold struct { + StorageAPI + bucket, key, id string +} + +func (d authorityMovedHold) ReadXL(ctx context.Context, bucket, key string, data bool) (RawFileInfo, error) { + rf, err := d.StorageAPI.ReadXL(ctx, bucket, key, data) + if err != nil || bucket != d.bucket || key != d.key { + return rf, err + } + var xl xlMetaV2 + if err = xl.LoadOrConvert(rf.Buf); err != nil { + return rf, err + } + for i := range xl.versions { + if uuid.UUID(xl.versions[i].header.VersionID).String() != d.id { + continue + } + var v xlMetaV2Version + if _, err = v.unmarshalV(xl.metaV, xl.versions[i].meta); err != nil { + return rf, err + } + v.ObjectV2.MetaSys[authorityHoldKey] = []byte(v.ObjectV2.MetaUser[authorityHoldKey]) + delete(v.ObjectV2.MetaUser, authorityHoldKey) + if xl.versions[i].meta, err = v.MarshalMsg(nil); err != nil { + return rf, err + } + } + rf.Buf, err = xl.AppendTo(nil) + return rf, err +} + +// One drive's metadata moves the hold of a held version to the system map +// under the same header. Its revision is not the held one, and the bulk +// expiry keeps the version. +func TestPoolsLifecycleMovedHoldKeepsVersion(t *testing.T) { + s := newLockedSplitILM(t, authorityNoncurrentRule) + ctx := t.Context() + const key = "ncv-moved-hold" + s.write(key, splitAt("old", 5, 0)) + s.write(key, splitAt("new", 3, 0)) + id := s.ids[key+" old"] + stamp := UTCNow().Format(time.RFC3339Nano) + setLifecycleCopyMetadata(t, s, key, id, 0, map[string]string{authorityHoldKey: "ON", ReservedMetadataPrefixLower + ObjectLockLegalHoldTimestamp: stamp}, false) + p := s.z.serverPools[0] + p.erasureDisksMu.Lock() + prior := p.erasureDisks + wrapped := make([][]StorageAPI, len(prior)) + for i, disks := range prior { + wrapped[i] = append([]StorageAPI(nil), disks...) + wrapped[i][0] = authorityMovedHold{StorageAPI: disks[0], bucket: s.bucket, key: key, id: id} + } + p.erasureDisks = wrapped + p.erasureDisksMu.Unlock() + t.Cleanup(func() { p.erasureDisksMu.Lock(); p.erasureDisks = prior; p.erasureDisksMu.Unlock() }) + + versions, events, err := s.z.lifecycleEventsAllPools(ctx, s.bucket, key) + if err != nil || versions[1].UserDefined[authorityHoldKey] != "ON" || events[1].Action != lifecycle.NoneAction { + t.Errorf("history: events=%v err=%v", events, err) + } + if err := authorityDeletePath(t, s, key, id, "bulk"); !errors.Is(err, errLifecycleNotApplicable) { + t.Errorf("bulk expiry of a held version: %v", err) + } + authorityPresent(t, s, 0, key, id) +} + +// A legacy version can carry several content checksums for each part. The +// map is encoded in iteration order, and must not change the revision of +// identical metadata; a different checksum does change it (control). +func TestPoolsLifecycleLegacyPartChecksums(t *testing.T) { + legacy := func(index int, sha256Value string) []byte { + m := getSampleXLMeta(1) + m.Stat.ModTime, m.Stat.Size, m.Parts[0].Size = time.Unix(1700000000, 0).UTC(), 20, 20 + m.Erasure.DataBlocks, m.Erasure.ParityBlocks, m.Erasure.Index = 12, 4, index + m.Erasure.Distribution = []int{1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16} + m.Parts[0].Checksums = map[string]string{"CRC32": "AAAAAA==", "CRC32C": "AAAAAQ==", "SHA1": "AAAAAg==", "SHA256": sha256Value} + buf, err := json.Marshal(m) + if err != nil { + t.Fatal(err) + } + return buf + } + revision := func(buf []byte) revisionKey { + var xl xlMetaV2 + if err := xl.LoadOrConvert(buf); err != nil { + t.Fatal(err) + } + return versionRevision(xl.versions[0], xl.metaV) + } + keys := make(map[revisionKey]bool) + for i := range 32 { + keys[revision(legacy(i%16+1, "AAAAAw=="))] = true + } + if len(keys) != 1 { + t.Fatalf("identical legacy metadata has %d revisions", len(keys)) + } + if revision(legacy(1, "AAAABA==")) == revision(legacy(1, "AAAAAw==")) { + t.Fatal("a different part checksum keeps the revision") + } + disks := make([]StorageAPI, 16) + for i := range disks { + disks[i] = authorityRawDisk{buf: legacy(i+1, "AAAAAw==")} + } + er := erasureObjects{getDisks: func() []StorageAPI { return disks }, defaultParityCount: 4} + if versions, err := er.objectVersions(t.Context(), "bucket", "key"); err != nil || len(versions) != 1 { + t.Fatalf("healthy legacy history: %d versions, err=%v", len(versions), err) + } +} + +// authorityRawDisk serves the same xl.meta for every key. +type authorityRawDisk struct { + StorageAPI + buf []byte +} + +func (d authorityRawDisk) ReadXL(context.Context, string, string, bool) (RawFileInfo, error) { + return RawFileInfo{Buf: d.buf}, nil +} + +// A legacy (xl.json) version stores the bitrot checksum of each drive's own +// shard. Healthy drives differ there and still hold one revision. +func TestPoolsLifecycleLegacyHistory(t *testing.T) { + for _, distinct := range []bool{false, true} { + name := "same-checksums" // control + if distinct { + name = "per-shard-checksums" + } + t.Run(name, func(t *testing.T) { + disks := make([]StorageAPI, 16) + for i := range disks { + m := getSampleXLMeta(1) + m.Stat.ModTime, m.Stat.Size, m.Parts[0].Size = time.Unix(1700000000, 0).UTC(), 20, 20 + m.Erasure.DataBlocks, m.Erasure.ParityBlocks, m.Erasure.Index = 12, 4, i+1 + m.Erasure.Distribution = []int{1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16} + shard := []byte{1, 1} + if distinct { + shard[0] = byte(i + 1) + } + sum := sha256.Sum256(shard) + m.Erasure.Checksums[0] = ChecksumInfo{PartNumber: 1, Algorithm: SHA256, Hash: sum[:]} + buf, err := json.Marshal(m) + if err != nil { + t.Fatal(err) + } + disks[i] = authorityRawDisk{buf: buf} + } + er := erasureObjects{getDisks: func() []StorageAPI { return disks }, defaultParityCount: 4} + if versions, err := er.objectVersions(t.Context(), "bucket", "key"); err != nil || len(versions) != 1 { + t.Fatalf("healthy legacy history: %d versions, err=%v", len(versions), err) + } + }) + } +} + +// 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) { + s := newSplitILM(t, versioning.Enabled, splitILMRules) + ctx := t.Context() + const key = "eav-partial-history" + s.write(key, splitAt("old", 5, 0, 1)) + s.write(key, splitAt("fresh", 0, 1)) + id := s.ids[key+" fresh"] + removeAuthorityCopies(t, s, 1, key, id, authorityDrives(0, 1)) + wrapAuthorityDisks(t, s, 1, key, true, false, authorityDrives(1, 9)) + + 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}}) + if !isErrWriteQuorum(err) { + t.Errorf("DeleteAll on a partial history: %v", err) + } + authorityPresent(t, s, 1, key, id) +} + +// Eight drives hold one revision of a version and eight another. Neither is +// authoritative, so neither may decide the expiry. +func TestPoolsLifecycleConflictingRevisions(t *testing.T) { + s := newLockedSplitILM(t, authorityNoncurrentRule) + ctx := t.Context() + const key = "ncv-split-revision" + s.write(key, splitAt("old", 5, 0)) + s.write(key, splitAt("new", 3, 0)) + id := s.ids[key+" old"] + disks := s.z.serverPools[0].getHashedSet(key).getDisks() + half := len(disks) / 2 + stale := make([]FileInfo, len(disks)) + for i := half; i < len(disks); i++ { + fi, err := disks[i].ReadVersion(ctx, "", s.bucket, key, id, ReadOptions{}) + if err != nil { + t.Fatal(err) + } + stale[i] = fi + } + stamp := UTCNow().Format(time.RFC3339Nano) + setLifecycleCopyMetadata(t, s, key, id, 0, poolLockMetadata("", "ON", stamp, stamp), false) + for i := half; i < len(disks); i++ { + if err := disks[i].WriteMetadata(ctx, "", s.bucket, key, stale[i]); err != nil { + t.Fatal(err) + } + } + + if _, _, err := s.z.lifecycleEventsAllPools(ctx, s.bucket, key); !isErrReadQuorum(err) { + t.Errorf("8/8 revisions: %v", err) + } + if err := authorityExpireVersion(t, s, key, id); !isErrWriteQuorum(err) { + t.Errorf("expiry on 8/8 revisions: %v", err) + } + for i, disk := range disks { + if _, err := disk.ReadVersion(ctx, "", s.bucket, key, id, ReadOptions{}); err != nil { + t.Fatalf("drive %d lost the version: %v", i, err) + } + } +} + +// Every drive answers and the old version is complete, but a fresh version +// that some drives show reached too few of them to be authoritative. +func TestPoolsLifecycleMissingVersion(t *testing.T) { + s := newSplitILM(t, versioning.Enabled, splitILMRules) + ctx := t.Context() + const key = "eav-missing-version" + s.write(key, splitAt("old", 5, 0, 1)) + s.write(key, splitAt("fresh", 0, 1)) + id := s.ids[key+" fresh"] + removeAuthorityCopies(t, s, 1, key, id, authorityDrives(0, 9)) + + 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}}) + if !isErrWriteQuorum(err) { + t.Errorf("DeleteAll: %v", err) + } + disks := s.z.serverPools[1].getHashedSet(key).getDisks() + for i := 9; i < len(disks); i++ { + if _, err := disks[i].ReadVersion(ctx, "", s.bucket, key, id, ReadOptions{}); err != nil { + t.Fatalf("drive %d lost the fresh version: %v", i, err) + } + } +} + +// 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) { + s := newSplitILM(t, versioning.Enabled, splitILMRules) + ctx := t.Context() + const key = "eav-minority-only" + s.write(key, splitAt("old", 5, 0)) + s.write(key, splitAt("fresh", 0, 1)) + id := s.ids[key+" fresh"] + removeAuthorityCopies(t, s, 1, key, id, authorityDrives(0, 9)) + disks := s.z.serverPools[1].getHashedSet(key).getDisks() + if _, err := disks[0].ReadXL(ctx, s.bucket, key, false); !errors.Is(err, errFileNotFound) { + t.Fatalf("fixture: drive 0 still holds the key: %v", err) + } + 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) { + t.Errorf("DeleteAll: %v", err) + } + for i := 9; i < len(disks); i++ { + if _, err := disks[i].ReadVersion(ctx, "", s.bucket, key, id, ReadOptions{}); err != nil { + t.Fatalf("drive %d lost the fresh version: %v", i, err) + } + } + }) + + // Control: two versions on seven drives each, none reaches a quorum. + t.Run("disjoint-minorities", func(t *testing.T) { + s := newSplitILM(t, versioning.Enabled, splitILMRules) + ctx := t.Context() + const key = "eav-disjoint" + s.write(key, splitAt("old", 5, 0)) + s.write(key, splitAt("a", 2, 1)) + s.write(key, splitAt("b", 1, 1)) + removeAuthorityCopies(t, s, 1, key, s.ids[key+" a"], authorityDrives(7, 16)) + removeAuthorityCopies(t, s, 1, key, s.ids[key+" b"], func(i int) bool { return i < 7 || i >= 14 }) + 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) { + t.Fatalf("DeleteAll: %v", err) + } + }) +} + +// A set without an unknown drive less than half of its drives still tells +// its history; half of them unknown could hide a version. +func TestPoolsLifecycleUnknownDriveThreshold(t *testing.T) { + for _, tc := range []struct { + name string + pool int // the pool with unreadable drives + unknown int + refused bool + }{ + {"history/7", 0, 7, false}, + {"history/8", 0, 8, true}, + {"absent/7", 1, 7, false}, + {"absent/8", 1, 8, true}, + } { + t.Run(tc.name, func(t *testing.T) { + s := newSplitILM(t, versioning.Enabled, splitILMRules) + ctx := t.Context() + const key = "ncv-threshold" + // Empty versions need only a simple majority of the drives, so + // seven unknown drives leave enough copies. + var ids []string + for _, days := range []int{5, 3} { + id := mustGetUUID() + putConsistencyObject(t, s.z, s.bucket, key, 0, "", ObjectOptions{Versioned: true, VersionID: id, MTime: UTCNow().AddDate(0, 0, -days)}) + ids = append(ids, id) + } + wrapAuthorityDisks(t, s, tc.pool, key, true, false, authorityDrives(0, tc.unknown)) + err := authorityExpireVersion(t, s, key, ids[0]) + if tc.refused { + if !isErrWriteQuorum(err) { + t.Errorf("%d unknown drives: %v", tc.unknown, err) + } + authorityPresent(t, s, 0, key, ids[0]) + return + } + if err != nil { + t.Fatalf("%d unknown drives: %v", tc.unknown, err) + } + if _, err := s.z.serverPools[0].GetObjectInfo(ctx, s.bucket, key, ObjectOptions{VersionID: ids[0]}); !isErrVersionNotFound(err) { + t.Fatalf("expired version remains: %v", err) + } + }) + } +} + +// Control: free versions are tier GC records. Residue of one on a few drives +// must not block an expiry. +func TestPoolsLifecycleFreeVersionResidue(t *testing.T) { + for _, tc := range []struct { + name string + residue int // the pool with the residue + }{ + {"beside-history", 0}, + {"alone", 1}, + } { + t.Run(tc.name, func(t *testing.T) { + s := newSplitILM(t, versioning.Enabled, splitILMRules) + ctx := t.Context() + const key = "ncv-free-residue" + s.write(key, splitTiered("gone", 6, tc.residue)) + s.write(key, splitAt("old", 5, 0)) + s.write(key, splitAt("new", 3, 0)) + gone := s.ids[key+" gone"] + disks := s.z.serverPools[tc.residue].getHashedSet(key).getDisks() + fvID := mustGetUUID() + for i, disk := range disks { + fi := FileInfo{VersionID: gone} + if i < 3 { + fi.SetTierFreeVersionID(fvID) + } else { + fi.SetSkipTierFreeVersion() + } + if err := disk.DeleteVersion(ctx, s.bucket, key, fi, false, DeleteOptions{}); err != nil { + t.Fatal(err) + } + } + if h := authorityHeader(t, disks[0], s.bucket, key, fvID); !h.FreeVersion() { + t.Fatalf("fixture lacks a free version: %v", h) + } + versions, events, err := s.z.lifecycleEventsAllPools(ctx, s.bucket, key) + if err != nil || len(versions) != 2 || events[1].Action != lifecycle.DeleteVersionAction { + t.Fatalf("history with free-version residue: %d versions, events=%v, err=%v", len(versions), events, err) + } + if err := authorityExpireVersion(t, s, key, s.ids[key+" old"]); err != nil { + t.Fatalf("expiry beside free-version residue: %v", err) + } + }) + } +} + +// An interrupted 12+4 write left nine matching copies of a newer version: +// a majority of the drives, but less than its read quorum. It must not make +// the last committed version noncurrent. +func TestPoolsLifecycleReducedParityPartialVersion(t *testing.T) { + for _, path := range []string{"single", "bulk"} { + t.Run(path, func(t *testing.T) { + authorityParity(t) + s := newSplitILM(t, versioning.Enabled, splitILMRules) + ctx := t.Context() + const key = "ncv-partial-write" + s.write(key, splitAt("committed", 5, 0)) + s.write(key, splitAt("partial", 3, 0)) + committed, partial := s.ids[key+" committed"], s.ids[key+" partial"] + oi, err := s.z.serverPools[0].GetObjectInfo(ctx, s.bucket, key, ObjectOptions{VersionID: partial}) + if err != nil || oi.DataBlocks != 12 || oi.ParityBlocks != 4 { + t.Fatalf("fixture is not 12+4: data=%d parity=%d err=%v", oi.DataBlocks, oi.ParityBlocks, err) + } + removeAuthorityCopies(t, s, 0, key, partial, authorityDrives(9, 16)) + + if _, _, err := s.z.lifecycleEventsAllPools(ctx, s.bucket, key); !isErrReadQuorum(err) { + t.Errorf("history with a partial version: %v", err) + } + if err := authorityDeletePath(t, s, key, committed, path); !isErrWriteQuorum(err) { + t.Errorf("expiry of the last committed version: %v", err) + } + authorityPresent(t, s, 0, key, committed) + }) + } +} + +// Control: two drives do not answer, every version still has its quorum. +func TestPoolsLifecycleDegradedHistory(t *testing.T) { + authorityParity(t) + s := newSplitILM(t, versioning.Enabled, splitILMRules) + ctx := t.Context() + const key = "ncv-degraded" + s.write(key, splitAt("old", 5, 0)) + s.write(key, splitAt("new", 3, 0)) + id := s.ids[key+" old"] + wrapAuthorityDisks(t, s, 0, key, true, false, authorityDrives(0, 2)) + if err := authorityExpireVersion(t, s, key, id); err != nil { + t.Fatalf("degraded expiry: %v", err) + } + if _, err := s.z.serverPools[0].GetObjectInfo(ctx, s.bucket, key, ObjectOptions{VersionID: id}); !isErrVersionNotFound(err) { + t.Fatalf("expired version remains: %v", err) + } +} + +// authorityPoolLock sets the Object Lock metadata of the copy of version id +// in pool. +func authorityPoolLock(t *testing.T, s *splitILM, key, id string, pool int, metadata map[string]string) { + t.Helper() + metadata = maps.Clone(metadata) + if metadata[strings.ToLower(xhttp.AmzObjectLockMode)] != "" { + metadata[strings.ToLower(xhttp.AmzObjectLockRetainUntilDate)] = UTCNow().AddDate(1, 0, 0).Format(time.RFC3339) + } else { + metadata[strings.ToLower(xhttp.AmzObjectLockRetainUntilDate)] = "" + } + setLifecycleCopyMetadata(t, s, key, id, pool, metadata, false) +} + +// authorityDeletePath expires version id by the single path or by the bulk +// path, which takes the ordinary or the tiered route as the version is. +func authorityDeletePath(t *testing.T, s *splitILM, key, id, path string) error { + t.Helper() + if path == "single" { + return authorityExpireVersion(t, s, key, id) + } + _, errs := s.z.DeleteObjects(t.Context(), s.bucket, []ObjectToDelete{{ObjectV: ObjectV{ObjectName: key, VersionID: id}}}, ObjectOptions{Versioned: true, Expiration: ExpirationOptions{Expire: true}}) + return errs[0] +} + +var authorityPaths = []string{"single", "bulk", "tiered-bulk"} + +// The copies of a version in two pools carry protection revisions that do not +// tell which is the newest. Refuse instead of guessing. +func TestPoolsLifecycleAmbiguousProtection(t *testing.T) { + stamp := UTCNow().Add(-time.Hour).Format(time.RFC3339Nano) + old := UTCNow().Add(-2 * time.Hour).Format(time.RFC3339Nano) + for _, tc := range []struct { + name string + copies [2]map[string]string + }{ + {"hold-equal-stamps", [2]map[string]string{poolLockMetadata("", "OFF", old, stamp), poolLockMetadata("", "ON", old, stamp)}}, + {"hold-missing-stamp", [2]map[string]string{poolLockMetadata("", "ON", old, ""), poolLockMetadata("", "OFF", old, stamp)}}, + {"hold-malformed-stamp", [2]map[string]string{poolLockMetadata("", "ON", old, "yesterday"), poolLockMetadata("", "OFF", old, stamp)}}, + {"retention-equal-stamps", [2]map[string]string{poolLockMetadata("", "OFF", stamp, old), poolLockMetadata("GOVERNANCE", "OFF", stamp, old)}}, + } { + for _, path := range authorityPaths { + t.Run(tc.name+"/"+path, func(t *testing.T) { + s := newLockedSplitILM(t, authorityNoncurrentRule) + const key = "ncv-ambiguous" + v := splitAt("old", 5, 0, 1) + v.tiered = path == "tiered-bulk" + s.write(key, v) + s.write(key, splitAt("new", 3, 0, 1)) + id := s.ids[key+" old"] + for pool, metadata := range tc.copies { + authorityPoolLock(t, s, key, id, pool, metadata) + } + if _, _, err := s.z.lifecycleEventsAllPools(t.Context(), s.bucket, key); !errors.Is(err, errLifecycleNotApplicable) { + t.Errorf("ambiguous protection: %v", err) + } + if err := authorityDeletePath(t, s, key, id, path); !errors.Is(err, errLifecycleNotApplicable) { + t.Errorf("expiry with ambiguous protection: %v", err) + } + for pool := range s.z.serverPools { + authorityPresent(t, s, pool, key, id) + } + }) + } + } +} + +// Protection stored under canonical-case keys counts as the evaluator reads +// it, in the comparison of copies, in the merged version and in the set. +func TestPoolsLifecycleCanonicalProtection(t *testing.T) { + older := UTCNow().Add(-2 * time.Hour).Format(time.RFC3339Nano) + recent := UTCNow().Add(-time.Hour).Format(time.RFC3339Nano) + until := UTCNow().AddDate(1, 0, 0).Format(time.RFC3339) + hold := func(status, stamp string) map[string]string { + return map[string]string{xhttp.AmzObjectLockLegalHold: status, ReservedMetadataPrefixLower + ObjectLockLegalHoldTimestamp: stamp} + } + retention := func(mode, stamp string) map[string]string { + m := map[string]string{ReservedMetadataPrefixLower + ObjectLockRetentionTimestamp: stamp} + if mode != "" { + m[xhttp.AmzObjectLockMode], m[xhttp.AmzObjectLockRetainUntilDate] = mode, until + } + return m + } + for _, tc := range []struct { + name string + copies [2]map[string]string + expires bool + }{ + {"hold-equal-stamps", [2]map[string]string{hold("OFF", recent), hold("ON", recent)}, false}, + {"hold-newer", [2]map[string]string{hold("OFF", older), hold("ON", recent)}, false}, + {"retention-equal-stamps", [2]map[string]string{retention("", recent), retention("COMPLIANCE", recent)}, false}, + {"retention-newer", [2]map[string]string{retention("", older), retention("GOVERNANCE", recent)}, false}, + {"hold-released", [2]map[string]string{hold("ON", older), hold("OFF", recent)}, true}, // control + } { + for _, path := range []string{"single", "bulk"} { + t.Run(tc.name+"/"+path, func(t *testing.T) { + s := newLockedSplitILM(t, authorityNoncurrentRule) + const key = "ncv-canonical" + id, mtime := mustGetUUID(), UTCNow().AddDate(0, 0, -5) + for pool, metadata := range tc.copies { + putConsistencyObject(t, s.z, s.bucket, key, pool, "old", ObjectOptions{Versioned: true, VersionID: id, MTime: mtime, UserDefined: maps.Clone(metadata)}) + } + s.write(key, splitAt("new", 3, 0, 1)) + _, events, err := s.z.lifecycleEventsAllPools(t.Context(), s.bucket, key) + if !tc.expires && err == nil && events[1].Action != lifecycle.NoneAction { + t.Errorf("protected version approved: %s", events[1].Action) + } + err = authorityDeletePath(t, s, key, id, path) + if tc.expires { + if err != nil { + t.Fatalf("released version: %v", err) + } + if _, err := s.z.serverPools[0].GetObjectInfo(t.Context(), s.bucket, key, ObjectOptions{VersionID: id}); !isErrVersionNotFound(err) { + t.Fatalf("released version remains: %v", err) + } + return + } + if !errors.Is(err, errLifecycleNotApplicable) { + t.Errorf("expiry of a protected version: %v", err) + } + for pool := range s.z.serverPools { + authorityPresent(t, s, pool, key, id) + } + }) + } + } +} + +// Control: a strictly newer release in another pool still expires the +// version by the ordinary bulk path too, beside the single and tiered bulk +// paths of TestPoolsLifecycleDuplicateLockMetadata. +func TestPoolsLifecycleProtectionReleaseBulk(t *testing.T) { + old := UTCNow().Add(-2 * time.Hour).Format(time.RFC3339Nano) + recent := UTCNow().Add(-time.Hour).Format(time.RFC3339Nano) + for _, tc := range []struct { + name string + copies [2]map[string]string + }{ + {"hold-released", [2]map[string]string{poolLockMetadata("", "ON", old, old), poolLockMetadata("", "OFF", old, recent)}}, + {"retention-released", [2]map[string]string{poolLockMetadata("GOVERNANCE", "OFF", old, old), poolLockMetadata("", "OFF", recent, old)}}, + } { + t.Run(tc.name, func(t *testing.T) { + s := newLockedSplitILM(t, authorityNoncurrentRule) + const key = "ncv-released" + s.write(key, splitAt("old", 5, 0, 1)) + s.write(key, splitAt("new", 3, 0, 1)) + id := s.ids[key+" old"] + for pool, metadata := range tc.copies { + authorityPoolLock(t, s, key, id, pool, metadata) + } + if err := authorityDeletePath(t, s, key, id, "bulk"); err != nil { + t.Fatalf("released version: %v", err) + } + for pool := range s.z.serverPools { + if _, err := s.z.serverPools[pool].GetObjectInfo(t.Context(), s.bucket, key, ObjectOptions{VersionID: id}); !isErrVersionNotFound(err) { + t.Errorf("pool %d still holds the released version: %v", pool, err) + } + } + }) + } +} + +// A set takes an all-pools approval only with the protection it was approved +// with. Protection that the set reads and the approval does not strictly +// supersede keeps the version. +func TestPoolsLifecycleCheckedExpirySafeguard(t *testing.T) { + old := UTCNow().Add(-2 * time.Hour).Format(time.RFC3339Nano) + recent := UTCNow().Add(-time.Hour).Format(time.RFC3339Nano) + for _, tc := range []struct { + name string + local map[string]string + approved objectLockState + deleted bool + }{ + {"hold-newer-locally", poolLockMetadata("", "ON", old, recent), objectLockState{legalHold: "OFF", legalHoldTimestamp: old}, false}, + {"hold-missing-local-stamp", poolLockMetadata("", "ON", old, ""), objectLockState{legalHold: "OFF", legalHoldTimestamp: recent}, false}, + {"hold-equal-stamps", poolLockMetadata("", "ON", old, recent), objectLockState{legalHold: "OFF", legalHoldTimestamp: recent}, false}, + {"hold-missing-approved-stamp", poolLockMetadata("", "ON", old, old), objectLockState{legalHold: "OFF"}, false}, + {"retention-newer-locally", poolLockMetadata("GOVERNANCE", "OFF", recent, old), objectLockState{retentionTimestamp: old, legalHold: "OFF", legalHoldTimestamp: old}, false}, + // Controls: the approval strictly supersedes the local protection. + {"hold-released", poolLockMetadata("", "ON", old, old), objectLockState{legalHold: "OFF", legalHoldTimestamp: recent}, true}, + {"retention-released", poolLockMetadata("GOVERNANCE", "OFF", old, old), objectLockState{retentionTimestamp: recent, legalHold: "OFF", legalHoldTimestamp: old}, true}, + } { + t.Run(tc.name, func(t *testing.T) { + s := newLockedSplitILM(t, authorityNoncurrentRule) + ctx := t.Context() + const key = "ncv-safeguard" + s.write(key, splitAt("old", 5, 0)) + s.write(key, splitAt("new", 3, 0)) + id := s.ids[key+" old"] + authorityPoolLock(t, s, key, id, 0, tc.local) + opts := ObjectOptions{Versioned: true, VersionID: id, Expiration: ExpirationOptions{Expire: true, CheckedAllPools: true}} + opts.Expiration.ApprovedProtection = tc.approved + _, err := s.z.serverPools[0].DeleteObject(ctx, s.bucket, key, opts) + if !tc.deleted { + if !errors.Is(err, errLifecycleNotApplicable) { + t.Errorf("protected version: %v", err) + } + authorityPresent(t, s, 0, key, id) + return + } + if err != nil { + t.Fatalf("released version: %v", err) + } + if _, err := s.z.serverPools[0].GetObjectInfo(ctx, s.bucket, key, ObjectOptions{VersionID: id}); !isErrVersionNotFound(err) { + t.Fatalf("released version remains: %v", err) + } + }) + } +} + +// A pending purge vetoes an approved expiry in the set, also when the read +// returns it with MethodNotAllowed, as for a delete marker. +func TestPoolsLifecycleCheckedExpiryPendingPurge(t *testing.T) { + for _, marker := range []bool{false, true} { + name := "version" + if marker { + name = "delete-marker" + } + t.Run(name, func(t *testing.T) { + s := newSplitILM(t, versioning.Enabled, splitILMRules) + ctx := t.Context() + const key = "ncv-pending-purge" + label := "old" + s.write(key, splitAt("first", 6, 0)) + if marker { + label = "dm" + s.write(key, splitMarker(label, 5, 0)) + } else { + s.write(key, splitAt(label, 5, 0)) + } + s.write(key, splitAt("new", 3, 0)) + id := s.ids[key+" "+label] + target := newILMTestTarget(t) + setILMTestReplicationConfig(t, s.bucket, target.arn, target.bucket) + oi, err := s.z.serverPools[0].GetObjectInfo(ctx, s.bucket, key, ObjectOptions{VersionID: id}) + if oi.DeleteMarker != marker || (err != nil) != marker || marker && !isErrMethodNotAllowed(err) { + t.Fatalf("fixture: marker=%t err=%v", oi.DeleteMarker, err) + } + _, err = s.z.serverPools[0].DeleteObject(ctx, s.bucket, key, ObjectOptions{ + VersionID: id, Versioned: true, MTime: oi.ModTime, + DeleteReplication: ReplicationState{ + VersionPurgeStatusInternal: target.arn + "=PENDING;", + PurgeTargets: map[string]VersionPurgeStatusType{target.arn: replication.VersionPurgePending}, + }, + }) + if err != nil { + t.Fatal(err) + } + pending, err := s.z.serverPools[0].GetObjectInfo(ctx, s.bucket, key, ObjectOptions{VersionID: id}) + if !isErrMethodNotAllowed(err) || pending.VersionPurgeStatus != replication.VersionPurgePending { + t.Fatalf("fixture: purge=%q err=%v", pending.VersionPurgeStatus, err) + } + _, err = s.z.serverPools[0].DeleteObject(ctx, s.bucket, key, ObjectOptions{Versioned: true, VersionID: id, Expiration: ExpirationOptions{Expire: true, CheckedAllPools: true}}) + if !errors.Is(err, errLifecycleNotApplicable) { + t.Errorf("approved expiry of a pending purge: %v", err) + } + after, err := s.z.serverPools[0].GetObjectInfo(ctx, s.bucket, key, ObjectOptions{VersionID: id}) + if !isErrMethodNotAllowed(err) || after.VersionPurgeStatus != replication.VersionPurgePending { + t.Fatalf("pending purge changed: purge=%q err=%v", after.VersionPurgeStatus, err) + } + }) + } +} + +// The copy of a version in another pool carries a pending purge that the +// merge would drop with the replication state of the first copy. +func TestPoolsLifecycleDuplicatePendingPurge(t *testing.T) { + for _, path := range []string{"single", "bulk"} { + t.Run(path, func(t *testing.T) { + s := newSplitILM(t, versioning.Enabled, splitILMRules) + ctx := t.Context() + const key = "ncv-pending-copy" + s.write(key, splitAt("old", 5, 0, 1)) + s.write(key, splitAt("new", 3, 0, 1)) + id := s.ids[key+" old"] + target := newILMTestTarget(t) + setILMTestReplicationConfig(t, s.bucket, target.arn, target.bucket) + oi, err := s.z.serverPools[1].GetObjectInfo(ctx, s.bucket, key, ObjectOptions{VersionID: id}) + if err != nil { + t.Fatal(err) + } + _, err = s.z.serverPools[1].DeleteObject(ctx, s.bucket, key, ObjectOptions{ + VersionID: id, Versioned: true, MTime: oi.ModTime, + DeleteReplication: ReplicationState{ + VersionPurgeStatusInternal: target.arn + "=PENDING;", + PurgeTargets: map[string]VersionPurgeStatusType{target.arn: replication.VersionPurgePending}, + }, + }) + if err != nil { + t.Fatal(err) + } + files, err := s.z.serverPools[1].getHashedSet(key).objectVersions(ctx, s.bucket, key) + if err != nil || len(files) != 2 || files[1].ToObjectInfo(s.bucket, key, true).VersionPurgeStatus != replication.VersionPurgePending { + t.Fatalf("fixture lacks the pending copy: %d versions, err=%v", len(files), err) + } + if _, _, err := s.z.lifecycleEventsAllPools(ctx, s.bucket, key); !errors.Is(err, errLifecycleNotApplicable) { + t.Errorf("copies with a different purge state: %v", err) + } + if err := authorityDeletePath(t, s, key, id, path); !errors.Is(err, errLifecycleNotApplicable) { + t.Errorf("expiry beside a pending purge: %v", err) + } + stored, err := s.z.serverPools[0].GetObjectInfo(ctx, s.bucket, key, ObjectOptions{VersionID: id}) + if err != nil || stored.VersionPurgeStatus != "" { + t.Fatalf("first copy changed: purge=%q err=%v", stored.VersionPurgeStatus, err) + } + if _, err := s.z.serverPools[1].GetObjectInfo(ctx, s.bucket, key, ObjectOptions{VersionID: id}); !isErrMethodNotAllowed(err) { + t.Fatalf("pending copy changed: %v", err) + } + }) + } +} + +// Copies of a version compare their per-target replication state as parsed: +// the order of the serialized targets does not count, a target does. +func TestPoolsLifecycleDuplicateReplicationTargets(t *testing.T) { + for _, tc := range []struct { + name string + statuses [2]string + refused bool + }{ + {"reordered", [2]string{"arn:a=COMPLETED;arn:b=COMPLETED;", "arn:b=COMPLETED;arn:a=COMPLETED;"}, false}, // control + {"other-targets", [2]string{"arn:a=COMPLETED;", "arn:a=COMPLETED;arn:b=COMPLETED;"}, true}, + {"other-status", [2]string{"arn:a=COMPLETED;arn:b=COMPLETED;", "arn:a=COMPLETED;arn:b=FAILED;"}, true}, + } { + t.Run(tc.name, func(t *testing.T) { + s := newSplitILM(t, versioning.Enabled, splitILMRules) + ctx := t.Context() + const key = "ncv-replicated-copies" + id, mtime := mustGetUUID(), UTCNow().AddDate(0, 0, -5) + for pool, status := range tc.statuses { + putConsistencyObject(t, s.z, s.bucket, key, pool, "old", ObjectOptions{ + Versioned: true, VersionID: id, MTime: mtime, + UserDefined: map[string]string{ReservedMetadataPrefixLower + ReplicationStatus: status}, + }) + } + s.write(key, splitAt("new", 3, 0, 1)) + for pool, status := range tc.statuses { + oi, err := s.z.serverPools[pool].GetObjectInfo(ctx, s.bucket, key, ObjectOptions{VersionID: id}) + if err != nil || oi.ReplicationStatusInternal != status { + t.Fatalf("fixture pool %d: status=%q err=%v", pool, oi.ReplicationStatusInternal, err) + } + } + _, events, err := s.z.lifecycleEventsAllPools(ctx, s.bucket, key) + if tc.refused { + if !errors.Is(err, errLifecycleNotApplicable) { + t.Errorf("copies with other replication targets: %v", err) + } + if err := authorityExpireVersion(t, s, key, id); !errors.Is(err, errLifecycleNotApplicable) { + t.Errorf("expiry: %v", err) + } + authorityPresent(t, s, 0, key, id) + return + } + if err != nil || events[1].Action != lifecycle.DeleteVersionAction { + t.Fatalf("equivalent copies: events=%v err=%v", events, err) + } + if err := authorityExpireVersion(t, s, key, id); err != nil { + t.Fatalf("expiry of equivalent copies: %v", err) + } + }) + } +} + +// Null versions written in two pools of a suspended bucket share a version +// ID without being one write. Copies with another ModTime are older writes: +// the newest decides, unless an older one is still protected. Copies with +// the same ModTime are duplicates and must still agree. +func TestPoolsLifecycleNullGenerations(t *testing.T) { + hold := func(status, stamp string) map[string]string { + return map[string]string{authorityHoldKey: status, ReservedMetadataPrefixLower + ObjectLockLegalHoldTimestamp: stamp} + } + older := UTCNow().Add(-2 * time.Hour).Format(time.RFC3339Nano) + recent := UTCNow().Add(-time.Hour).Format(time.RFC3339Nano) + for _, tc := range []struct { + name string + lock bool + old, new map[string]string + sameModTime bool + wantNewest bool // evaluated on the newer write, else refused + replicateOld bool + }{ + {"older-unprotected", false, hold("OFF", ""), nil, false, true, true}, + {"older-held", true, hold("ON", older), hold("OFF", recent), false, false, false}, + {"same-modtime-duplicates", false, hold("OFF", ""), nil, true, false, false}, + } { + t.Run(tc.name, func(t *testing.T) { + s := &splitILM{migrationILM: newMigrationILM(t, versioning.Suspended, tc.lock, splitILMRules), ids: map[string]string{}, bodies: map[string]string{}} + ctx := t.Context() + const key = "exp-null-generations" + oldTime, newTime := UTCNow().AddDate(0, 0, -5), UTCNow().AddDate(0, 0, -3) + if tc.sameModTime { + newTime = oldTime + } + oldMeta := maps.Clone(tc.old) + if tc.replicateOld { + oldMeta[ReservedMetadataPrefixLower+ReplicationStatus] = "arn:a=COMPLETED;" + } + putConsistencyObject(t, s.z, s.bucket, key, 0, "old", ObjectOptions{VersionSuspended: true, MTime: oldTime, UserDefined: oldMeta}) + newest := putConsistencyObject(t, s.z, s.bucket, key, 1, "new", ObjectOptions{VersionSuspended: true, MTime: newTime, UserDefined: maps.Clone(tc.new)}) + + versions, events, err := s.z.lifecycleEventsAllPools(ctx, s.bucket, key) + opts := ObjectOptions{VersionSuspended: true, Expiration: ExpirationOptions{Expire: true}} + if !tc.wantNewest { + if !errors.Is(err, errLifecycleNotApplicable) { + t.Errorf("history: %d versions, events=%v, err=%v", len(versions), events, err) + } + if _, err := s.z.DeleteObject(ctx, s.bucket, key, opts); !errors.Is(err, errLifecycleNotApplicable) { + t.Errorf("expiry: %v", err) + } + for pool := range s.z.serverPools { + oi, err := s.z.serverPools[pool].GetObjectInfo(ctx, s.bucket, key, ObjectOptions{VersionID: nullVersionID}) + if err != nil || oi.DeleteMarker { + t.Fatalf("pool %d lost its null version: marker=%t err=%v", pool, oi.DeleteMarker, err) + } + } + return + } + if err != nil || len(versions) != 1 || !versions[0].ModTime.Equal(newest.ModTime) || events[0].Action != lifecycle.DeleteAction { + t.Fatalf("history: %d versions, events=%v, err=%v", len(versions), events, err) + } + if _, err := s.z.DeleteObject(ctx, s.bucket, key, opts); err != nil { + t.Fatalf("expiry of the newest null write: %v", err) + } + }) + } +} + +// authoritySideEffects collects the version IDs that removal notifications +// and ILM expiry audit traces report. +func authoritySideEffects(t *testing.T, run func()) (notified, traced []string) { + t.Helper() + done := make(chan struct{}) + defer close(done) + events := make(chan event.Event, 16) + if err := globalHTTPListen.Subscribe(pubsub.MaskFromMaskable(event.ObjectRemovedDelete), events, done, nil); err != nil { + t.Fatal(err) + } + traces := make(chan madmin.TraceInfo, 16) + if err := globalTrace.Subscribe(madmin.TraceILM, traces, done, nil); err != nil { + t.Fatal(err) + } + run() + // Both publishers deliver synchronously, so everything sent is buffered. + for { + select { + case ev := <-events: + notified = append(notified, ev.S3.Object.VersionID) + case tr := <-traces: + if tr.FuncName == ILMExpiry { + traced = append(traced, tr.Custom["version-id"]) + } + default: + slices.Sort(notified) + slices.Sort(traced) + return notified, traced + } + } +} + +// A batch expiry that another pool cannot read is not a removal. +func TestPoolsLifecycleBatchQuorumNotAnnounced(t *testing.T) { + s := newSplitILM(t, versioning.Enabled, splitILMRules) + const key = "ncv-quorum-event" + s.write(key, splitAt("old", 5, 0)) + s.write(key, splitAt("new", 3, 0)) + id := s.ids[key+" old"] + lifecycleReadDisks(t, s, key, 1, errDiskNotFound) + result := &lifecycleBatchResult{ObjectLayer: s.z} + notified, traced := authoritySideEffects(t, func() { + deleteObjectVersions(t.Context(), result, s.bucket, []ObjectToDelete{{ObjectV: ObjectV{ObjectName: key, VersionID: id}}}, []lifecycle.Event{{Action: lifecycle.DeleteVersionAction}}) + }) + if !isErrWriteQuorum(result.errs[0]) { + t.Fatalf("fixture result: %v", result.errs[0]) + } + authorityPresent(t, s, 0, key, id) + if len(notified) != 0 || len(traced) != 0 { + t.Fatalf("unapplied expiry announced: notifications=%v traces=%v", notified, traced) + } +} + +// Each entry of a batch gets the side effects of its own result: a completed +// or already absent delete is announced and audited, a refused or failed one +// is not. +func TestPoolsLifecycleBatchSideEffects(t *testing.T) { + for _, path := range []string{"single-pool", "bulk", "tiered-bulk"} { + t.Run(path, func(t *testing.T) { + s := newSplitILM(t, versioning.Enabled, splitILMRules) + const key, failing = "ncv-side-effects", "ncv-side-effects-failing" + for _, k := range []string{key, failing} { + v := splitAt("old", 5, 0) + v.tiered = path == "tiered-bulk" + s.write(k, v) + s.write(k, splitAt("new", 3, 0)) + } + if path == "single-pool" { + prior := s.z.serverPools + t.Cleanup(func() { s.z.serverPools = prior }) + s.z.serverPools = s.z.serverPools[:1] + } + wrapAuthorityDisks(t, s, 0, failing, false, true, authorityDrives(0, 9)) + success, absent := s.ids[key+" old"], mustGetUUID() + entries := []ObjectToDelete{ + {ObjectV: ObjectV{ObjectName: key, VersionID: success}}, + {ObjectV: ObjectV{ObjectName: failing, VersionID: s.ids[failing+" old"]}}, + {ObjectV: ObjectV{ObjectName: key, VersionID: absent}}, + } + if path != "single-pool" { + // A single pool has no all-pools check to refuse a bulk entry. + entries = append(entries, ObjectToDelete{ObjectV: ObjectV{ObjectName: key, VersionID: s.ids[key+" new"]}}) + } + evts := make([]lifecycle.Event, len(entries)) + for i := range evts { + evts[i] = lifecycle.Event{Action: lifecycle.DeleteVersionAction} + } + result := &lifecycleBatchResult{ObjectLayer: s.z} + notified, traced := authoritySideEffects(t, func() { + deleteObjectVersions(t.Context(), result, s.bucket, entries, evts) + }) + if result.errs[0] != nil || !isErrWriteQuorum(result.errs[1]) || + result.errs[2] != nil && !isErrVersionNotFound(result.errs[2]) || + len(entries) > 3 && !errors.Is(result.errs[3], errLifecycleNotApplicable) { + t.Fatalf("fixture results: %v", result.errs) + } + want := []string{success, absent} + slices.Sort(want) + if !slices.Equal(notified, want) { + t.Errorf("removal notifications for %v, want %v", notified, want) + } + if !slices.Equal(traced, want) { + t.Errorf("ILM expiry audit for %v, want %v", traced, want) + } + }) + } +} + +// A single pool keeps the explicit refusal rather than a not-found result. +func TestPoolsLifecycleSinglePoolRefusal(t *testing.T) { + s := newSplitILM(t, versioning.Enabled, splitILMRules) + s.write("exp-new", splitAt("fresh", 0, 0)) + prior := s.z.serverPools + t.Cleanup(func() { s.z.serverPools = prior }) + s.z.serverPools = s.z.serverPools[:1] + _, err := s.z.DeleteObject(t.Context(), s.bucket, "exp-new", ObjectOptions{Versioned: true, Expiration: ExpirationOptions{Expire: true}}) + if !errors.Is(err, errLifecycleNotApplicable) || isErrObjectNotFound(err) || isErrVersionNotFound(err) { + t.Fatalf("single-pool stale expiry: %v", err) + } +} + +// Migration leaves a version to the logical expiry. When that expiry is +// refused at retirement while the source still holds the version, the source +// is kept: a refusal is never absence. Decommission and rebalance both +// retire through removeMovedVersions. +func TestPoolsMigrationLifecycleRefusedExpiryKeepsSource(t *testing.T) { + s := newSplitILM(t, versioning.Enabled, splitILMRules) + ctx := t.Context() + const key = "exp-moved" + s.write(key, splitAt("old", 5, 0, 1)) + s.write(key, splitAt("new", 0.01, 1)) + id := s.ids[key+" old"] + source, err := s.z.serverPools[0].GetObjectInfo(ctx, s.bucket, key, ObjectOptions{VersionID: id}) + if err != nil { + t.Fatal(err) + } + _, updatedAt, err := globalBucketMetadataSys.GetLifecycleConfig(s.bucket) + if err != nil { + t.Fatal(err) + } + version := FileInfo{Name: key, VersionID: id, ModTime: source.ModTime} + err = s.z.removeMovedVersions(ctx, s.bucket, key, 0, []FileInfo{version}, []migrationExpiredVersion{{ + version: version, + event: lifecycle.Event{Action: lifecycle.DeleteAction}, + lifecycleUpdatedAt: updatedAt, + source: lcEventSrc_Decom, + }}) + if _, ok := err.(PreConditionFailed); !ok { + t.Fatalf("retirement after a refused expiry: %v", err) + } + authorityPresent(t, s, 0, key, id) + if views := s.views(); !slices.Equal(views[0][key], []string{"old"}) || !slices.Equal(views[1][key], []string{"new", "old"}) { + t.Fatalf("refused expiry changed the pools: %v", views) + } +} diff --git a/cmd/erasure-server-pool-lifecycle-safety_test.go b/cmd/erasure-server-pool-lifecycle-safety_test.go index 27791813f..0ee319350 100644 --- a/cmd/erasure-server-pool-lifecycle-safety_test.go +++ b/cmd/erasure-server-pool-lifecycle-safety_test.go @@ -94,23 +94,31 @@ func TestPoolsLifecycleDuplicateLockMetadata(t *testing.T) { setLifecycleCopyMetadata(t, s, key, id, pool, metadata, tc.newerPool1 && pool == 1) } versions, events, err := s.z.lifecycleEventsAllPools(t.Context(), s.bucket, key) - if err != nil { + switch { + case tc.newerPool1: + // Another ModTime makes the copies separate writes. The + // held older one refuses the object instead of merging. + if !errors.Is(err, errLifecycleNotApplicable) { + t.Fatalf("held older write: %v", err) + } + case err != nil: t.Fatal(err) - } - if len(versions) != 2 || versions[1].VersionID != id || versions[1].NumVersions != 2 { - t.Fatalf("wrong version history: %+v", versions) - } - state := storedObjectLockState(versions[1].UserDefined) - if state.mode != tc.mode || state.legalHold != tc.hold { - t.Errorf("merged state=%+v, want mode=%s hold=%s", state, tc.mode, tc.hold) - } - want := lifecycle.NoneAction - if tc.delete { - want = lifecycle.DeleteVersionAction - } - // Read-side expiry rechecks consume the same event producer. - if events[1].Action != want { - t.Errorf("expiry event=%s want=%s", events[1].Action, want) + default: + if len(versions) != 2 || versions[1].VersionID != id || versions[1].NumVersions != 2 { + t.Fatalf("wrong version history: %+v", versions) + } + state := storedObjectLockState(versions[1].UserDefined) + if state.mode != tc.mode || state.legalHold != tc.hold { + t.Errorf("merged state=%+v, want mode=%s hold=%s", state, tc.mode, tc.hold) + } + want := lifecycle.NoneAction + if tc.delete { + want = lifecycle.DeleteVersionAction + } + // Read-side expiry rechecks consume the same event producer. + if events[1].Action != want { + t.Errorf("expiry event=%s want=%s", events[1].Action, want) + } } opts := ObjectOptions{Versioned: true, Expiration: ExpirationOptions{Expire: true}} if batch { diff --git a/cmd/erasure-server-pool-lifecycle.go b/cmd/erasure-server-pool-lifecycle.go index 6df11bedb..597cdf249 100644 --- a/cmd/erasure-server-pool-lifecycle.go +++ b/cmd/erasure-server-pool-lifecycle.go @@ -19,12 +19,20 @@ package cmd import ( "context" + "crypto/sha256" + "encoding/binary" "errors" + "maps" + "slices" "sort" + "strings" "sync" "time" "github.com/minio/minio/internal/bucket/lifecycle" + objectlock "github.com/minio/minio/internal/bucket/object/lock" + "github.com/minio/minio/internal/bucket/replication" + xhttp "github.com/minio/minio/internal/http" ) // An ordinary ILM rejection is not evidence that the version is absent. In @@ -51,17 +59,25 @@ var errLifecycleNotApplicable = errors.New("lifecycle expiry no longer applies") // 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. +// objectVersions returns the versions of object that the drives of the set +// hold authoritatively, newest first, without free versions. It returns no +// versions when the set does not hold the object, and an error when the +// drives cannot tell its complete history. +// +// Expiry deletes by this history, so a revision of a version, its header +// and its metadata (versionRevision), counts only when more than half of +// the configured drives and the read quorum of its metadata hold it. A +// version that some drive shows without such a revision, or too many +// unreadable drives, could hide what an expiry must not delete. Free +// versions are tier GC records and do not count. 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 + majority := len(disks)/2 + 1 var ( metas [][]xlMetaV2ShallowVersion + metaVs []uint8 metaV uint8 unknown int ) @@ -78,23 +94,23 @@ func (er erasureObjects) objectVersions(ctx context.Context, bucket, object stri continue } metas = append(metas, xl.versions) + metaVs = append(metaVs, xl.metaV) metaV = xl.metaV } - if len(metas) < readQuorum { - if len(metas)+unknown < readQuorum { - return nil, nil // not in this set - } + // Without an unknown drive the history is complete, also on one drive. + if unknown > 0 && unknown >= len(disks)/2 { 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 + quorumMetas, quorumErrs := make([]FileInfo, len(disks)), make([]error, len(disks)) + merged, err := authoritativeHistory(ctx, metas, metaVs, metaV, majority, func(ver xlMetaV2ShallowVersion, metaV uint8) int { + return er.versionReadQuorum(ctx, bucket, object, ver, metaV, quorumMetas, quorumErrs) + }) + if err != nil { + return nil, toObjectErr(err, bucket, object) + } + if merged == nil { + return nil, nil // not in this set } all, err := xlMetaV2{versions: merged, metaV: metaV}.ListVersions(bucket, object, false) if err != nil { @@ -109,10 +125,253 @@ func (er erasureObjects) objectVersions(ctx context.Context, bucket, object stri return versions, nil } +// revisionKey identifies a revision of a version: its header and a strong +// digest of its metadata. The header Signature is a 32-bit fold of weak +// hashes; two revisions can share it. +type revisionKey struct { + header xlMetaV2VersionHeader + digest [sha256.Size]byte +} + +// versionRevision returns the revision of ver, whose metadata is encoded as +// metaV describes. As Signature does, it leaves out the fields that differ +// between the drives of one write: the erasure index, empty part lists and, +// for legacy versions, the release and the per-shard checksums. The digest +// input is canonical: each map, including the part checksums of a legacy +// version, is its own section with its entry count and sorted, +// length-prefixed entries, since msgp encodes maps in iteration order and an +// entry must not move between maps unnoticed. The header of a +// legacy version carries a signature over its per-shard checksums, so the +// key replaces it with one from the digest. +func versionRevision(ver xlMetaV2ShallowVersion, metaV uint8) revisionKey { + h := sha256.New() + writeUint := func(n int) { + var b [8]byte + binary.LittleEndian.PutUint64(b[:], uint64(n)) + h.Write(b[:]) + } + writeField := func(b []byte) { + writeUint(len(b)) + h.Write(b) + } + section := func(name string, entries int) { + writeField([]byte(name)) + writeUint(entries) + } + writeStrings := func(name string, m map[string]string) { + section(name, len(m)) + for _, k := range slices.Sorted(maps.Keys(m)) { + writeField([]byte(k)) + writeField([]byte(m[k])) + } + } + writeBytes := func(name string, m map[string][]byte) { + section(name, len(m)) + for _, k := range slices.Sorted(maps.Keys(m)) { + writeField([]byte(k)) + writeField(m[k]) + } + } + key := revisionKey{header: ver.header} + var v xlMetaV2Version + if _, err := v.unmarshalV(metaV, ver.meta); err == nil { + var ( + bts []byte + err error = errFileCorrupt + ) + section("type", 1) + writeField([]byte{byte(v.Type)}) + switch { + case v.Type == ObjectType && v.ObjectV2 != nil: + c := *v.ObjectV2 + c.ErasureIndex = 0 + if !slices.ContainsFunc(c.PartETags, func(tag string) bool { return tag != "" }) { + c.PartETags = nil + } + if len(c.PartActualSizes) == 0 { + c.PartActualSizes = nil + } + writeStrings("user", c.MetaUser) + writeBytes("sys", c.MetaSys) + c.MetaUser, c.MetaSys = nil, nil + bts, err = c.MarshalMsg(nil) + case v.Type == DeleteType && v.DeleteMarker != nil: + c := *v.DeleteMarker + writeBytes("sys", c.MetaSys) + c.MetaSys = nil + bts, err = c.MarshalMsg(nil) + case v.Type == LegacyType && v.ObjectV1 != nil: + c := *v.ObjectV1 + c.Erasure.Index = 0 + c.Erasure.Checksums = nil + c.Minio.Release = "" + writeStrings("meta", c.Meta) + c.Meta = nil + // Part checksums are the only other map: one section per part. + section("parts", len(c.Parts)) + c.Parts = slices.Clone(c.Parts) + for i := range c.Parts { + writeStrings("part-checksums", c.Parts[i].Checksums) + c.Parts[i].Checksums = nil + } + bts, err = c.MarshalMsg(nil) + } + if err == nil { + section("fields", 1) + writeField(bts) + h.Sum(key.digest[:0]) + if v.Type == LegacyType { + copy(key.header.Signature[:], key.digest[:]) + } + return key + } + h.Reset() + } + // Undecodable metadata only matches the same bytes. + section("raw", 1) + writeField(ver.meta) + h.Sum(key.digest[:0]) + return key +} + +// authoritativeHistory returns the versions of metas, the version lists of +// the drives that answered, as merged newest first, when every version +// that is not a free version has a revision that at least majority drives +// and its readQuorum hold. It returns no versions when none is live, +// errErasureReadQuorum when the drives cannot tell the history, and the +// error of ctx once it ends. metaVs are the metadata versions of metas, +// metaV the one the merged result is decoded with. +func authoritativeHistory(ctx context.Context, metas [][]xlMetaV2ShallowVersion, metaVs []uint8, metaV uint8, majority int, readQuorum func(xlMetaV2ShallowVersion, uint8) int) ([]xlMetaV2ShallowVersion, error) { + // Count the drives that hold each revision; a drive counts once for each + // version. + type revision struct { + ver xlMetaV2ShallowVersion + metaV uint8 + count int + } + keys := make([][]revisionKey, len(metas)) + revisions := make(map[[16]byte]map[revisionKey]*revision) + live := make(map[[16]byte]bool) + for i, versions := range metas { + keys[i] = make([]revisionKey, len(versions)) + seen := make(map[[16]byte]bool, len(versions)) + for j, ver := range versions { + // Decoding and hashing every history has a cost; stop with the + // request. + if err := ctx.Err(); err != nil { + return nil, err + } + id := ver.header.VersionID + if !ver.header.FreeVersion() { + live[id] = true + } + if seen[id] { + continue + } + seen[id] = true + key := versionRevision(ver, metaVs[i]) + keys[i][j] = key + if revisions[id] == nil { + revisions[id] = make(map[revisionKey]*revision) + } + r := revisions[id][key] + if r == nil { + r = &revision{ver: ver, metaV: metaVs[i]} + revisions[id][key] = r + } + r.count++ + } + } + accepted := make(map[[16]byte]revisionKey, len(live)) + for id := range live { + var best *revision + var bestKey revisionKey + for key, r := range revisions[id] { + if !key.header.FreeVersion() && (best == nil || r.count > best.count) { + best, bestKey = r, key + } + } + // Only one revision can hold more than half of the drives. + if best == nil || best.count < majority || best.count < readQuorum(best.ver, best.metaV) { + return nil, errErasureReadQuorum + } + accepted[id] = bestKey + } + + // The strict merge orders the history. It has to hold exactly the + // accepted revisions. A minority revision that sorts first would reset + // its count, and one that shares the header could stand in for the + // accepted one, so the merge only sees the accepted revision of each + // version and free versions as they are. It sees the headers of the + // keys, which do not differ between the drives of a legacy version. + filtered := make([][]xlMetaV2ShallowVersion, len(metas)) + for i, versions := range metas { + for j, ver := range versions { + key, ok := accepted[ver.header.VersionID] + switch { + case ver.header.FreeVersion(): + case ok && keys[i][j] == key: + ver.header = key.header + default: + continue + } + filtered[i] = append(filtered[i], ver) + } + } + if err := ctx.Err(); err != nil { + return nil, err + } + merged := mergeXLV2Versions(majority, true, 0, filtered...) + emitted := make(map[[16]byte]bool, len(accepted)) + for _, ver := range merged { + if err := ctx.Err(); err != nil { + return nil, err + } + if ver.header.FreeVersion() { + continue + } + id := ver.header.VersionID + if key, ok := accepted[id]; !ok || emitted[id] || versionRevision(ver, metaV) != key { + return nil, errErasureReadQuorum + } + emitted[id] = true + } + if len(emitted) != len(accepted) { + return nil, errErasureReadQuorum + } + if len(accepted) == 0 { + return nil, nil + } + return merged, nil +} + +// versionReadQuorum returns the read quorum of the metadata of ver, as +// objectQuorumFromMeta derives it when all drives of the set hold it, or +// more than the drives when its metadata cannot be read. metaArr and errs +// have one entry for each drive; errs stays nil. +func (er erasureObjects) versionReadQuorum(ctx context.Context, bucket, object string, ver xlMetaV2ShallowVersion, metaV uint8, metaArr []FileInfo, errs []error) int { + var v xlMetaV2Version + if _, err := v.unmarshalV(metaV, ver.meta); err != nil { + return len(metaArr) + 1 + } + fi, err := v.ToFileInfo(bucket, object, false) + if err != nil { + return len(metaArr) + 1 + } + for i := range metaArr { + metaArr[i] = fi + } + readQuorum, _, err := objectQuorumFromMeta(ctx, metaArr, errs, er.defaultParityCount) + if err != nil { + return len(metaArr) + 1 + } + return readQuorum +} + // 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. +// leaves it, counts once, and so do other writes of its version ID. func (z *erasureServerPools) lifecycleEventsAllPools(ctx context.Context, bucket, object string) ([]ObjectInfo, []lifecycle.Event, error) { lc, err := globalLifecycleSys.Get(bucket) if err != nil { @@ -169,14 +428,36 @@ func (z *erasureServerPools) lifecycleEventsAllPools(ctx context.Context, bucket versions = append(versions, []PoolObjInfo{copy}) } + e := lifecycle.NewEvaluator(*lc).WithLockRetention(&lr).WithReplicationConfig(rcfg) objInfos := make([]ObjectInfo, len(versions)) objOpts := make([]lifecycle.ObjectOpts, len(versions)) for i, copies := range versions { + // Copies with another ModTime are other writes of the version ID, + // such as null versions written in several pools, not duplicates of + // one write. The newest is the version. Data movement writes a new + // data dir, so only ModTime tells the duplicates of a write apart. + // An older write that is still protected keeps the whole object. + newest := 1 + for newest < len(copies) && copies[newest].ObjInfo.ModTime.Equal(copies[0].ObjInfo.ModTime) { + newest++ + } + for _, older := range copies[newest:] { + if opts := older.ObjInfo.ToLifecycleOpts(); e.IsObjectLocked(opts) || e.IsPendingReplication(opts) { + return nil, nil, errLifecycleNotApplicable + } + } + copies = copies[:newest] oi := copies[0].ObjInfo if len(copies) > 1 { + // The copies must tell which protection is the newest, and no + // copy may carry replication state that the merge would drop. + if !lockRevisionsOrdered(copies) || !replicationStatesAgree(copies) { + return nil, nil, errLifecycleNotApplicable + } // Count the version once, but retain independently ordered lock // and tag revisions, including explicit removals, from every copy. oi = mergedPoolObjectInfo(copies) + setLifecycleLockState(oi.UserDefined, newestLockState(copies)) } oi.IsLatest = i == 0 oi.NumVersions = len(versions) @@ -187,13 +468,160 @@ func (z *erasureServerPools) lifecycleEventsAllPools(ctx context.Context, bucket objInfos[i] = oi objOpts[i] = oi.ToLifecycleOpts() } - events, err := lifecycle.NewEvaluator(*lc).WithLockRetention(&lr).WithReplicationConfig(rcfg).Eval(objOpts) + events, err := e.Eval(objOpts) if err != nil { return nil, nil, err } return objInfos, events, nil } +// lockRevisionTime parses the timestamp of an Object Lock revision. Unlike +// olderThan, a missing, malformed or zero timestamp orders nothing. +func lockRevisionTime(stamp string) (time.Time, bool) { + t, err := time.Parse(time.RFC3339Nano, stamp) + return t, err == nil && !t.IsZero() +} + +// lockRevisionNewer reports whether the revision stamped newer strictly +// supersedes the one stamped older. +func lockRevisionNewer(newer, older string) bool { + n, ok := lockRevisionTime(newer) + o, ok2 := lockRevisionTime(older) + return ok && ok2 && n.After(o) +} + +// lifecycleLockState returns the Object Lock state of metadata as the +// lifecycle evaluator reads it (objectlock.GetObjectRetentionMeta and +// GetObjectLegalHoldMeta): a lowercase key before its canonical form. The +// stored state (storedObjectLockState) only has lowercase keys. +func lifecycleLockState(metadata map[string]string) objectLockState { + get := func(canonical string) string { + if v, ok := metadata[strings.ToLower(canonical)]; ok { + return v + } + return metadata[canonical] + } + return objectLockState{ + mode: get(objectlock.AmzObjectLockMode), + retainUntil: get(objectlock.AmzObjectLockRetainUntilDate), + retentionTimestamp: get(ReservedMetadataPrefix + ObjectLockRetentionTimestamp), + legalHold: get(objectlock.AmzObjectLockLegalHold), + legalHoldTimestamp: get(ReservedMetadataPrefix + ObjectLockLegalHoldTimestamp), + } +} + +// setLifecycleLockState sets state in metadata so that the evaluator reads +// it: lowercase keys take precedence over canonical ones. +func setLifecycleLockState(metadata map[string]string, state objectLockState) { + set := func(canonical, value string) { + lower := strings.ToLower(canonical) + _, hasLower := metadata[lower] + _, hasCanonical := metadata[canonical] + if value != "" || hasLower || hasCanonical { + metadata[lower] = value + } + } + set(objectlock.AmzObjectLockMode, state.mode) + set(objectlock.AmzObjectLockRetainUntilDate, state.retainUntil) + set(ReservedMetadataPrefix+ObjectLockRetentionTimestamp, state.retentionTimestamp) + set(objectlock.AmzObjectLockLegalHold, state.legalHold) + set(ReservedMetadataPrefix+ObjectLockLegalHoldTimestamp, state.legalHoldTimestamp) +} + +// newestLockState returns the newest hold and the newest retention of the +// copies of a version, which lockRevisionsOrdered found equal or ordered. +func newestLockState(copies []PoolObjInfo) objectLockState { + state := lifecycleLockState(copies[0].ObjInfo.UserDefined) + for _, copy := range copies[1:] { + next := lifecycleLockState(copy.ObjInfo.UserDefined) + if lockRevisionNewer(next.legalHoldTimestamp, state.legalHoldTimestamp) { + state.legalHold, state.legalHoldTimestamp = next.legalHold, next.legalHoldTimestamp + } + if lockRevisionNewer(next.retentionTimestamp, state.retentionTimestamp) { + state.mode, state.retainUntil, state.retentionTimestamp = next.mode, next.retainUntil, next.retentionTimestamp + } + } + return state +} + +// checkedExpiryRefused reports whether a set must keep a version whose ILM +// expiry the pools approved with the Object Lock state approved, by the +// metadata goi the set reads under the same lock. A pending purge always +// keeps it: its status orders nothing. The approved hold or retention only +// replaces the one the set reads when it is strictly newer; otherwise the +// protection the set reads stands, also when it is newer than the approval. +func checkedExpiryRefused(lc lifecycle.Lifecycle, lr objectlock.Retention, rcfg *replication.Config, goi ObjectInfo, approved objectLockState) bool { + e := lifecycle.NewEvaluator(lc).WithLockRetention(&lr).WithReplicationConfig(rcfg) + obj := goi.ToLifecycleOpts() + if e.IsPendingReplication(obj) { + return true + } + local := lifecycleLockState(goi.UserDefined) + obj.UserDefined = maps.Clone(goi.UserDefined) + if obj.UserDefined == nil { + obj.UserDefined = make(map[string]string) + } + if lockRevisionNewer(approved.legalHoldTimestamp, local.legalHoldTimestamp) { + obj.UserDefined[strings.ToLower(xhttp.AmzObjectLockLegalHold)] = approved.legalHold + } + if lockRevisionNewer(approved.retentionTimestamp, local.retentionTimestamp) { + obj.UserDefined[strings.ToLower(xhttp.AmzObjectLockMode)] = approved.mode + obj.UserDefined[strings.ToLower(xhttp.AmzObjectLockRetainUntilDate)] = approved.retainUntil + } + return e.IsObjectLocked(obj) +} + +// lockRevisionsOrdered reports whether the copies of a version agree on the +// legal hold and on the retention, each on its own, or carry revisions with +// distinct valid timestamps, so that the newest one is known. +func lockRevisionsOrdered(copies []PoolObjInfo) bool { + ordered := func(revision func(objectLockState) (value, stamp string)) bool { + stamps := make(map[[2]string]string, len(copies)) + for _, copy := range copies { + value, stamp := revision(lifecycleLockState(copy.ObjInfo.UserDefined)) + stamps[[2]string{value, stamp}] = stamp + } + if len(stamps) == 1 { + return true + } + seen := make([]time.Time, 0, len(stamps)) + for _, stamp := range stamps { + t, ok := lockRevisionTime(stamp) + if !ok { + return false + } + for _, other := range seen { + if t.Equal(other) { + return false + } + } + seen = append(seen, t) + } + return true + } + return ordered(func(s objectLockState) (string, string) { + return s.legalHold, s.legalHoldTimestamp + }) && ordered(func(s objectLockState) (string, string) { + return s.mode + "\x00" + s.retainUntil, s.retentionTimestamp + }) +} + +// replicationStatesAgree reports whether the copies of a version carry the +// same replication and purge state. The per-target states are compared as +// parsed, regardless of their order. +func replicationStatesAgree(copies []PoolObjInfo) bool { + first := copies[0].ObjInfo + for _, copy := range copies[1:] { + oi := copy.ObjInfo + if oi.ReplicationStatus != first.ReplicationStatus || oi.VersionPurgeStatus != first.VersionPurgeStatus || + !maps.Equal(replicationStatusesMap(oi.ReplicationStatusInternal), replicationStatusesMap(first.ReplicationStatusInternal)) || + !maps.Equal(versionPurgeStatusesMap(oi.VersionPurgeStatusInternal), versionPurgeStatusesMap(first.VersionPurgeStatusInternal)) { + return false + } + } + return true +} + // lifecycleAllPools holds the versions of an object in all pools and their // lifecycle events, see lifecycleEventsAllPools. type lifecycleAllPools struct { @@ -204,11 +632,37 @@ type lifecycleAllPools struct { // 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 { +// lifecycleAllPools.expiryApplies, and the Object Lock state with which it +// approved the expiry of a version. +func (z *erasureServerPools) checkExpiryAllPools(ctx context.Context, bucket, object string, opts ObjectOptions) (objectLockState, error) { var all lifecycleAllPools all.versions, all.events, all.err = z.lifecycleEventsAllPools(ctx, bucket, object) - return all.expiryApplies(bucket, object, opts) + if err := all.expiryApplies(bucket, object, opts); err != nil { + return objectLockState{}, err + } + return all.protection(opts.VersionID), nil +} + +// version returns the index of the version versionID, or -1. +func (all lifecycleAllPools) version(versionID string) int { + for i, oi := range all.versions { + if oi.VersionID == versionID || oi.VersionID == "" && versionID == nullVersionID { + return i + } + } + return -1 +} + +// protection returns the Object Lock state of the version versionID that +// the evaluator decided on, to be checked against the state the set reads. +func (all lifecycleAllPools) protection(versionID string) objectLockState { + if versionID == "" { + return objectLockState{} + } + if i := all.version(versionID); i >= 0 { + return lifecycleLockState(all.versions[i].UserDefined) + } + return objectLockState{} } // expiryApplies returns nil when the ILM expiry that opts describe applies @@ -237,13 +691,11 @@ func (all lifecycleAllPools) expiryApplies(bucket, object string, opts ObjectOpt 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 - } - return errLifecycleNotApplicable + if i := all.version(opts.VersionID); i >= 0 { + if all.events[i].Action == lifecycle.DeleteVersionAction { + return nil } + return errLifecycleNotApplicable } return VersionNotFound{Bucket: bucket, Object: name, VersionID: opts.VersionID} } @@ -328,5 +780,6 @@ func (z *erasureServerPools) migrationExpiryApplies(ctx context.Context, oi Obje if !z.expiryNeedsAllPools(opts) { return true } - return z.checkExpiryAllPools(ctx, oi.Bucket, encodeDirObject(oi.Name), opts) == nil + _, err := z.checkExpiryAllPools(ctx, oi.Bucket, encodeDirObject(oi.Name), opts) + return err == nil } diff --git a/cmd/erasure-server-pool.go b/cmd/erasure-server-pool.go index 2f9f0d7b4..849ad6c67 100644 --- a/cmd/erasure-server-pool.go +++ b/cmd/erasure-server-pool.go @@ -1250,12 +1250,15 @@ func (z *erasureServerPools) DeleteObject(ctx context.Context, bucket string, ob // 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. + // decided, without a recheck by the versions each one holds. They only + // keep a version whose protection they read differently. if z.expiryNeedsAllPools(opts) { - if err = z.checkExpiryAllPools(ctx, bucket, object, opts); err != nil { + var approved objectLockState + if approved, err = z.checkExpiryAllPools(ctx, bucket, object, opts); err != nil { return ObjectInfo{Name: decodeDirObject(object), VersionID: opts.VersionID}, err } opts.Expiration.CheckedAllPools = true + opts.Expiration.ApprovedProtection = approved } if opts.DeletePrefix { @@ -1527,6 +1530,7 @@ 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)) + approved := make([]objectLockState, 0, len(objects)) ilm := make(map[string]lifecycleAllPools) for i, obj := range objects { if derrs[i] != nil { @@ -1536,6 +1540,7 @@ func (z *erasureServerPools) DeleteObjects(ctx context.Context, bucket string, o // its object in all pools, see checkExpiryAllPools. When the // expiry no longer applies, nothing is deleted and nothing is // left to replicate. + var protection objectLockState vopts := ObjectOptions{Expiration: opts.Expiration, VersionID: obj.VersionID} if obj.VersionID != "" && z.expiryNeedsAllPools(vopts) { all, ok := ilm[obj.ObjectName] @@ -1548,6 +1553,7 @@ func (z *erasureServerPools) DeleteObjects(ctx context.Context, bucket string, o derrs[i] = err continue } + protection = all.protection(obj.VersionID) } if deletesData(obj, opts) { if z.copyWithoutDeleteQuorum(ctx, bucket, obj.ObjectName, obj.VersionID) { @@ -1556,13 +1562,14 @@ func (z *erasureServerPools) DeleteObjects(ctx context.Context, bucket string, o } data = append(data, obj) dataIdx = append(dataIdx, i) + approved = append(approved, protection) continue } batch = append(batch, obj) batchIdx = append(batchIdx, i) } if len(data) > 0 { - deleted, errs := z.deleteObjectsReconciled(ctx, bucket, data, opts) + deleted, errs := z.deleteObjectsReconciled(ctx, bucket, data, approved, opts) for j, i := range dataIdx { dobjects[i], derrs[i] = deleted[j], errs[j] } diff --git a/cmd/object-api-interface.go b/cmd/object-api-interface.go index 658e30f9b..d40ff3ce7 100644 --- a/cmd/object-api-interface.go +++ b/cmd/object-api-interface.go @@ -163,6 +163,10 @@ type ExpirationOptions struct { // object in all pools (checkExpiryAllPools). A set does not check it // again with the versions it holds alone. CheckedAllPools bool + // ApprovedProtection: the Object Lock state of the version with which + // the check approved its expiry. The set compares it with the state it + // reads (checkedExpiryRefused). In memory only. + ApprovedProtection objectLockState `msg:"-"` } // TransitionOptions represents object options for transition ObjectLayer operation diff --git a/cmd/object-handlers-common.go b/cmd/object-handlers-common.go index 903e82731..41441882a 100644 --- a/cmd/object-handlers-common.go +++ b/cmd/object-handlers-common.go @@ -19,7 +19,6 @@ package cmd import ( "context" - "errors" "fmt" "net/http" "regexp" @@ -472,9 +471,11 @@ func deleteObjectVersions(ctx context.Context, o ObjectLayer, bucket string, toD }) for i, dobj := range deletedObjs { - if errors.Is(errs[i], errLifecycleNotApplicable) { - // The recheck kept this version. Do not audit or announce a - // removal that did not happen; the next scan re-evaluates it. + if errs[i] != nil && !isErrObjectNotFound(errs[i]) && !isErrVersionNotFound(errs[i]) { + // Only a completed delete, or one confirmed already absent, is + // a removal. After a refusal, a quorum, configuration or any + // other error the version may remain; do not audit or announce + // it. The next scan re-evaluates it. continue } oi := ObjectInfo{