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 <rh@vonng.com>
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Feng Ruohang
2026-11-05 00:16:27 +08:00
parent f321f7f1c7
commit 9f7b49369d
8 changed files with 1940 additions and 55 deletions
+10 -2
View File
@@ -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
+6 -1
View File
@@ -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)
File diff suppressed because it is too large Load Diff
@@ -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 {
+482 -29
View File
@@ -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
}
+10 -3
View File
@@ -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]
}
+4
View File
@@ -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
+5 -4
View File
@@ -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{