diff --git a/cmd/auth-handler.go b/cmd/auth-handler.go
index ebf753988..e4d408444 100644
--- a/cmd/auth-handler.go
+++ b/cmd/auth-handler.go
@@ -795,11 +795,12 @@ func isPutActionAllowedWithRequestTags(ctx context.Context, atype authType, buck
func isPutActionAllowedWithCred(bucketName, objectName string, r *http.Request, action policy.Action, requestTags *string, cred auth.Credentials, owner bool) APIErrorCode {
// Do not check for PutObjectRetentionAction permission,
- // if mode and retain until date are not set.
+ // if neither retention values nor a replica retention revision are set.
// Can happen when bucket has default lock config set
if action == policy.PutObjectRetentionAction &&
r.Header.Get(xhttp.AmzObjectLockMode) == "" &&
- r.Header.Get(xhttp.AmzObjectLockRetainUntilDate) == "" {
+ r.Header.Get(xhttp.AmzObjectLockRetainUntilDate) == "" &&
+ r.Header.Get(xhttp.MinIOSourceObjectRetentionTimestamp) == "" {
return ErrNone
}
diff --git a/cmd/bucket-object-lock.go b/cmd/bucket-object-lock.go
index 25fdcaa1e..d7c74d036 100644
--- a/cmd/bucket-object-lock.go
+++ b/cmd/bucket-object-lock.go
@@ -354,6 +354,8 @@ func checkPutObjectLockAllowed(ctx context.Context, rq *http.Request, bucket, ob
var legalHold objectlock.ObjectLegalHold
retentionRequested := objectlock.IsObjectLockRetentionRequested(rq.Header)
+ // A timestamp without a value is a replicated retention removal.
+ retentionRequested = retentionRequested || replicaTrusted && rq.Header.Get(xhttp.MinIOSourceObjectRetentionTimestamp) != ""
legalHoldRequested := objectlock.IsObjectLockLegalHoldRequested(rq.Header)
retentionCfg, err := globalBucketObjectLockSys.Get(bucket)
@@ -472,6 +474,103 @@ func storedObjectLockState(metadata map[string]string) objectLockState {
}
}
+// reducePoolLockState selects each field independently using a total order,
+// so neither pool enumeration nor the starting copy can change the result.
+// Active COMPLIANCE takes precedence over revisions; among those copies the
+// longest retention wins. Only expired COMPLIANCE may lose to a removal.
+func reducePoolLockState(copies []PoolObjInfo, readState func(map[string]string) objectLockState) objectLockState {
+ var now time.Time
+ clockRead := false
+ state := readState(copies[0].ObjInfo.UserDefined)
+ for _, copy := range copies[1:] {
+ next := readState(copy.ObjInfo.UserDefined)
+ if !clockRead && (state.mode == string(objectlock.RetCompliance) || next.mode == string(objectlock.RetCompliance)) {
+ var err error
+ now, err = objectlock.UTCNowNTP()
+ if err != nil {
+ now = time.Time{} // An unavailable clock cannot prove retention expired.
+ }
+ clockRead = true
+ }
+ if comparePoolRetention(next, state, now) > 0 {
+ state.mode, state.retainUntil, state.retentionTimestamp = next.mode, next.retainUntil, next.retentionTimestamp
+ }
+ order := compareLockRevision(next.legalHoldTimestamp, state.legalHoldTimestamp)
+ // Equal or unordered conflicting holds resolve to ON, then OFF, then
+ // absence. A strictly newer valid OFF still releases a prior hold.
+ if order == 0 {
+ order = strings.Compare(next.legalHold, state.legalHold)
+ }
+ if order == 0 {
+ order = strings.Compare(next.legalHoldTimestamp, state.legalHoldTimestamp)
+ }
+ if order > 0 {
+ state.legalHold, state.legalHoldTimestamp = next.legalHold, next.legalHoldTimestamp
+ }
+ }
+ return state
+}
+
+func compareLockRevision(a, b string) int {
+ ta, aok := lockRevisionTime(a)
+ tb, bok := lockRevisionTime(b)
+ if aok != bok {
+ if aok {
+ return 1
+ }
+ return -1
+ }
+ if !aok {
+ return 0
+ }
+ return ta.Compare(tb)
+}
+
+func comparePoolRetention(a, b objectLockState, now time.Time) int {
+ au, ae := time.Parse(time.RFC3339Nano, a.retainUntil)
+ bu, be := time.Parse(time.RFC3339Nano, b.retainUntil)
+ ap := a.mode == string(objectlock.RetCompliance) && (ae != nil || now.IsZero() || au.After(now))
+ bp := b.mode == string(objectlock.RetCompliance) && (be != nil || now.IsZero() || bu.After(now))
+ if ap != bp {
+ if ap {
+ return 1
+ }
+ return -1
+ }
+ if ap { // An unreadable COMPLIANCE deadline cannot be shortened either.
+ if (ae != nil) != (be != nil) {
+ if ae != nil {
+ return 1
+ }
+ return -1
+ }
+ if ae == nil && au.Compare(bu) != 0 {
+ return au.Compare(bu)
+ }
+ }
+ if order := compareLockRevision(a.retentionTimestamp, b.retentionTimestamp); order != 0 {
+ return order
+ }
+ // Ties and legacy unordered metadata must also have a stable winner.
+ // A retained value takes precedence over an unordered absence.
+ if order := strings.Compare(a.mode, b.mode); order != 0 {
+ return order
+ }
+ if (ae != nil) != (be != nil) {
+ if ae != nil {
+ return 1
+ }
+ return -1
+ }
+ if ae == nil && be == nil && au.Compare(bu) != 0 {
+ return au.Compare(bu)
+ }
+ if order := strings.Compare(a.retainUntil, b.retainUntil); order != 0 {
+ return order
+ }
+ return strings.Compare(a.retentionTimestamp, b.retentionTimestamp)
+}
+
// olderThan reports whether a stored replication timestamp is missing,
// unreadable, or earlier than the source timestamp, in which case the
// replica update wins. A zero source timestamp never wins.
@@ -543,14 +642,20 @@ func (s objectLockState) restoreLegalHold(metadata map[string]string) {
// (an older incoming value must not win over a newer stored one just because the
// read timed out).
func replicaStoredLock(ctx context.Context, getObjectInfo GetObjectInfoFn, bucket, object, versionID string) (objectLockState, error) {
- oi, err := getObjectInfo(ctx, bucket, object, ObjectOptions{VersionID: versionID})
+ oi, err := replicaStoredObjectInfo(ctx, getObjectInfo, bucket, object, versionID)
+ return storedObjectLockState(oi.UserDefined), err
+}
+
+// The same destination lookup supplies both Object Lock and tag revisions.
+func replicaStoredObjectInfo(ctx context.Context, getObjectInfo GetObjectInfoFn, bucket, object, versionID string) (ObjectInfo, error) {
+ oi, err := getObjectInfo(ctx, bucket, object, ObjectOptions{VersionID: versionID, ReplicationRequest: true})
switch {
case err == nil:
- return storedObjectLockState(oi.UserDefined), nil
+ return oi, nil
case isErrObjectNotFound(err) || isErrVersionNotFound(err):
- return objectLockState{}, nil
+ return ObjectInfo{}, nil
default:
- return objectLockState{}, err
+ return ObjectInfo{}, err
}
}
diff --git a/cmd/erasure-object.go b/cmd/erasure-object.go
index fec815c73..53cde55f6 100644
--- a/cmd/erasure-object.go
+++ b/cmd/erasure-object.go
@@ -133,8 +133,25 @@ func (er erasureObjects) CopyObject(ctx context.Context, srcBucket, srcObject, d
}
if dstOpts.ReplicaLockReconcile {
- reconcileStoredObjectLock(srcInfo.UserDefined, storedObjectLockState(fi.Metadata))
- reconcileStoredObjectTags(srcInfo.UserDefined, fi.Metadata[xhttp.AmzObjectTagging], fi.Metadata[ReservedMetadataPrefixLower+TaggingTimestamp])
+ stored := fi.ToObjectInfo(dstBucket, dstObject, dstOpts.Versioned || dstOpts.VersionSuspended)
+ if srcInfo.versionOnly || dstOpts.replicaObjectInfo != nil {
+ // fi describes the source of the new reference, not the version
+ // being replaced. A pooled lookup also preserves protection merged
+ // from another copy, even when this set holds a newer weak revision.
+ getInfo := er.getObjectInfo
+ if dstOpts.replicaObjectInfo != nil {
+ getInfo = dstOpts.replicaObjectInfo
+ }
+ stored, err = replicaStoredObjectInfo(ctx, func(ctx context.Context, bucket, object string, opts ObjectOptions) (ObjectInfo, error) {
+ opts.NoLock = true
+ return getInfo(ctx, bucket, object, opts)
+ }, dstBucket, dstObject, dstOpts.VersionID)
+ if err != nil {
+ return oi, err
+ }
+ }
+ reconcileStoredObjectLock(srcInfo.UserDefined, storedObjectLockState(stored.UserDefined))
+ reconcileStoredObjectTags(srcInfo.UserDefined, stored.UserTags, stored.UserDefined[ReservedMetadataPrefixLower+TaggingTimestamp])
}
filterOnlineDisksInplace(fi, metaArr, onlineDisks)
@@ -2574,7 +2591,7 @@ func (er erasureObjects) PutObjectTags(ctx context.Context, bucket, object strin
fi.Metadata[xhttp.AmzObjectTagging] = tags
fi.ReplicationState = opts.PutReplicationState()
- stamp := monotonicTaggingTimestamp(opts.UserDefined[ReservedMetadataPrefixLower+TaggingTimestamp], fi.Metadata[ReservedMetadataPrefixLower+TaggingTimestamp])
+ stamp := monotonicMetadataTimestamp(opts.UserDefined[ReservedMetadataPrefixLower+TaggingTimestamp], fi.Metadata[ReservedMetadataPrefixLower+TaggingTimestamp])
maps.Copy(fi.Metadata, opts.UserDefined)
if stamp != "" {
fi.Metadata[ReservedMetadataPrefixLower+TaggingTimestamp] = stamp
diff --git a/cmd/erasure-server-pool-consistency.go b/cmd/erasure-server-pool-consistency.go
index 1d1414023..2e8550503 100644
--- a/cmd/erasure-server-pool-consistency.go
+++ b/cmd/erasure-server-pool-consistency.go
@@ -383,19 +383,7 @@ func (z *erasureServerPools) healObjectInPool(ctx context.Context, er *erasureOb
// mergePoolLockState resolves each independently ordered field. An ordered
// removal is a value in its own right; an absent unordered field is not one.
func mergePoolLockState(copies []PoolObjInfo) objectLockState {
- state := storedObjectLockState(copies[0].ObjInfo.UserDefined)
- for _, copy := range copies[1:] {
- next := storedObjectLockState(copy.ObjInfo.UserDefined)
- retentionTime, _ := time.Parse(time.RFC3339Nano, next.retentionTimestamp)
- if state.retentionIsOlderThan(retentionTime) || (state.retentionTimestamp == "" && state.mode == "" && next.mode != "") {
- state.mode, state.retainUntil, state.retentionTimestamp = next.mode, next.retainUntil, next.retentionTimestamp
- }
- holdTime, _ := time.Parse(time.RFC3339Nano, next.legalHoldTimestamp)
- if state.legalHoldIsOlderThan(holdTime) || (state.legalHoldTimestamp == "" && state.legalHold == "" && next.legalHold != "") {
- state.legalHold, state.legalHoldTimestamp = next.legalHold, next.legalHoldTimestamp
- }
- }
- return state
+ return reducePoolLockState(copies, storedObjectLockState)
}
func replaceObjectLockMetadata(metadata map[string]string, state objectLockState) {
@@ -547,10 +535,10 @@ func reconcileStoredObjectTags(metadata map[string]string, storedTags, storedTim
}
}
-// Local tagging mutations must advance the revision they overwrite, even when
+// Local metadata mutations must advance the revision they overwrite, even when
// a request's clock or lock acquisition order is behind the stored revision.
// Replica writes use reconcileStoredObjectTags instead of minting a revision.
-func monotonicTaggingTimestamp(incoming, stored string) string {
+func monotonicMetadataTimestamp(incoming, stored string) string {
requested, err := time.Parse(time.RFC3339Nano, incoming)
if err != nil {
return incoming
diff --git a/cmd/erasure-server-pool-lifecycle-authority_test.go b/cmd/erasure-server-pool-lifecycle-authority_test.go
index 82a13dde2..a9ab17835 100644
--- a/cmd/erasure-server-pool-lifecycle-authority_test.go
+++ b/cmd/erasure-server-pool-lifecycle-authority_test.go
@@ -1260,6 +1260,7 @@ func TestPoolsLifecycleCheckedExpirySafeguard(t *testing.T) {
{"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},
+ {"compliance-cannot-release", poolLockMetadata("COMPLIANCE", "OFF", old, old), objectLockState{retentionTimestamp: recent, 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},
diff --git a/cmd/erasure-server-pool-lifecycle-safety_test.go b/cmd/erasure-server-pool-lifecycle-safety_test.go
index 3e987a564..d6dc04c1b 100644
--- a/cmd/erasure-server-pool-lifecycle-safety_test.go
+++ b/cmd/erasure-server-pool-lifecycle-safety_test.go
@@ -70,6 +70,8 @@ func TestPoolsLifecycleDuplicateLockMetadata(t *testing.T) {
{name: "independent-fields", copies: [2]map[string]string{poolLockMetadata("GOVERNANCE", "OFF", recent, old), poolLockMetadata("", "ON", old, recent)}, mode: "GOVERNANCE", hold: "ON"},
{name: "hold-released", copies: [2]map[string]string{poolLockMetadata("", "ON", old, old), poolLockMetadata("", "OFF", old, recent)}, hold: "OFF", delete: true},
{name: "retention-released", copies: [2]map[string]string{poolLockMetadata("GOVERNANCE", "OFF", old, old), poolLockMetadata("", "OFF", recent, old)}, hold: "OFF", delete: true},
+ {name: "compliance-before-removal", copies: [2]map[string]string{poolLockMetadata("COMPLIANCE", "OFF", old, old), poolLockMetadata("", "OFF", recent, old)}, mode: "COMPLIANCE", hold: "OFF"},
+ {name: "compliance-after-removal", copies: [2]map[string]string{poolLockMetadata("", "OFF", recent, old), poolLockMetadata("COMPLIANCE", "OFF", old, old)}, mode: "COMPLIANCE", hold: "OFF"},
} {
for _, batch := range []bool{false, true} {
path := "single"
diff --git a/cmd/erasure-server-pool-lifecycle.go b/cmd/erasure-server-pool-lifecycle.go
index 58b757097..8a97153af 100644
--- a/cmd/erasure-server-pool-lifecycle.go
+++ b/cmd/erasure-server-pool-lifecycle.go
@@ -587,20 +587,10 @@ func setLifecycleLockState(metadata map[string]string, state objectLockState) {
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.
+// newestLockState orders independently revised fields without weakening a
+// COMPLIANCE retention that any copy still protects.
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
+ return reducePoolLockState(copies, lifecycleLockState)
}
// checkedExpiryRefused reports whether a set must keep a version whose ILM
@@ -623,7 +613,7 @@ func checkedExpiryRefused(lc lifecycle.Lifecycle, lr objectlock.Retention, rcfg
if lockRevisionNewer(approved.legalHoldTimestamp, local.legalHoldTimestamp) {
obj.UserDefined[strings.ToLower(xhttp.AmzObjectLockLegalHold)] = approved.legalHold
}
- if lockRevisionNewer(approved.retentionTimestamp, local.retentionTimestamp) {
+ if lockRevisionNewer(approved.retentionTimestamp, local.retentionTimestamp) && !local.complianceWouldWeaken(approved) {
obj.UserDefined[strings.ToLower(xhttp.AmzObjectLockMode)] = approved.mode
obj.UserDefined[strings.ToLower(xhttp.AmzObjectLockRetainUntilDate)] = approved.retainUntil
}
diff --git a/cmd/erasure-server-pool.go b/cmd/erasure-server-pool.go
index 6a9dbdd6e..0bff3529d 100644
--- a/cmd/erasure-server-pool.go
+++ b/cmd/erasure-server-pool.go
@@ -1723,6 +1723,7 @@ func (z *erasureServerPools) CopyObject(ctx context.Context, srcBucket, srcObjec
stored := mergedPoolObjectInfo(copies)
reconcileStoredObjectLock(srcInfo.UserDefined, storedObjectLockState(stored.UserDefined))
reconcileStoredObjectTags(srcInfo.UserDefined, stored.UserTags, stored.UserDefined[ReservedMetadataPrefixLower+TaggingTimestamp])
+ dstOpts.replicaObjectInfo = z.replicaObjectInfo
idx := copies[0].Index
oi, err := z.serverPools[idx].CopyObject(ctx, srcBucket, srcObject, dstBucket, dstObject, srcInfo, srcOpts, dstOpts)
if err == nil {
@@ -3373,7 +3374,7 @@ func (z *erasureServerPools) PutObjectTags(ctx context.Context, bucket, object s
// revision beyond all copies, so the returned value and every pool agree.
if stamp := opts.UserDefined[ReservedMetadataPrefixLower+TaggingTimestamp]; stamp != "" {
for _, copy := range copies {
- stamp = monotonicTaggingTimestamp(stamp, copy.ObjInfo.UserDefined[ReservedMetadataPrefixLower+TaggingTimestamp])
+ stamp = monotonicMetadataTimestamp(stamp, copy.ObjInfo.UserDefined[ReservedMetadataPrefixLower+TaggingTimestamp])
}
opts.UserDefined = cloneMSS(opts.UserDefined)
opts.UserDefined[ReservedMetadataPrefixLower+TaggingTimestamp] = stamp
diff --git a/cmd/object-handlers.go b/cmd/object-handlers.go
index ee2973961..bac280fb7 100644
--- a/cmd/object-handlers.go
+++ b/cmd/object-handlers.go
@@ -1847,12 +1847,27 @@ func (api objectAPIHandlers) CopyObjectHandler(w http.ResponseWriter, r *http.Re
srcInfo.PutObjReader = pReader
- // Object Lock state as stored on disk, captured before the metadata
- // directive rebuilds the map. A replica update is applied only when its
- // source timestamp is newer than the stored one, and a stale update must
- // leave the stored state in place instead of erasing it.
- storedLock := storedObjectLockState(srcInfo.UserDefined)
- storedTagTimestamp := srcInfo.UserDefined[ReservedMetadataPrefixLower+TaggingTimestamp]
+ // Reconcile against the addressed destination version under its write
+ // lock. The source can be a different object or version, including when
+ // the destination version has never been written.
+ getObjectInfo := func(_ context.Context, bucket, object string, opts ObjectOptions) (ObjectInfo, error) {
+ opts.NoLock = true
+ return objectAPI.GetObjectInfo(lockCtx, bucket, object, opts)
+ }
+ var storedInfo ObjectInfo
+ if replicaTrusted && !remoteCallRequired {
+ storedInfo, err = replicaStoredObjectInfo(lockCtx, getObjectInfo, dstBucket, dstObject, dstOpts.VersionID)
+ if err != nil {
+ writeErrorResponse(ctx, w, toAPIError(ctx, err), r.URL)
+ return
+ }
+ }
+ storedLock := storedObjectLockState(storedInfo.UserDefined)
+ storedTagTimestamp := storedInfo.UserDefined[ReservedMetadataPrefixLower+TaggingTimestamp]
+ localTagTimestamp := ""
+ if cpSrcDstSame {
+ localTagTimestamp = srcInfo.UserDefined[ReservedMetadataPrefixLower+TaggingTimestamp]
+ }
srcInfo.UserDefined, err = getCpObjMetadataFromHeader(ctx, r, srcInfo.UserDefined, allowReplicationMetadata)
if err != nil {
@@ -1877,9 +1892,9 @@ func (api objectAPIHandlers) CopyObjectHandler(w http.ResponseWriter, r *http.Re
// the captured state even if metadata REPLACE rebuilt the map.
srcInfo.UserDefined[ReservedMetadataPrefixLower+TaggingTimestamp] = srcTimestamp.UTC().Format(time.RFC3339Nano)
srcInfo.UserDefined[xhttp.AmzObjectTagging] = objTags
- reconcileStoredObjectTags(srcInfo.UserDefined, srcInfo.UserTags, storedTagTimestamp)
+ reconcileStoredObjectTags(srcInfo.UserDefined, storedInfo.UserTags, storedTagTimestamp)
} else {
- srcInfo.UserDefined[xhttp.AmzObjectTagging] = srcInfo.UserTags
+ srcInfo.UserDefined[xhttp.AmzObjectTagging] = storedInfo.UserTags
if storedTagTimestamp != "" {
srcInfo.UserDefined[ReservedMetadataPrefixLower+TaggingTimestamp] = storedTagTimestamp
} else {
@@ -1888,7 +1903,7 @@ func (api objectAPIHandlers) CopyObjectHandler(w http.ResponseWriter, r *http.Re
}
} else {
srcInfo.UserDefined[xhttp.AmzObjectTagging] = objTags
- srcInfo.UserDefined[ReservedMetadataPrefixLower+TaggingTimestamp] = UTCNow().Format(time.RFC3339Nano)
+ srcInfo.UserDefined[ReservedMetadataPrefixLower+TaggingTimestamp] = monotonicMetadataTimestamp(UTCNow().Format(time.RFC3339Nano), localTagTimestamp)
}
// SSE-C rotation snapshots reserved metadata before the tag decision. Its
// later merge must not put the old timestamp back over the accepted state.
@@ -1896,14 +1911,12 @@ func (api objectAPIHandlers) CopyObjectHandler(w http.ResponseWriter, r *http.Re
srcInfo.UserDefined = filterReplicationStatusMetadata(srcInfo.UserDefined)
srcInfo.UserDefined = objectlock.FilterObjectLockMetadata(srcInfo.UserDefined, true, true)
+ // COPY may have preserved the source's reserved timestamps. Only the
+ // addressed destination and the incoming update may order a replica.
+ delete(srcInfo.UserDefined, ReservedMetadataPrefixLower+ObjectLockRetentionTimestamp)
+ delete(srcInfo.UserDefined, ReservedMetadataPrefixLower+ObjectLockLegalHoldTimestamp)
retPerms := isPutActionAllowed(ctx, getRequestAuthType(r), dstBucket, dstObject, r, policy.PutObjectRetentionAction)
holdPerms := isPutActionAllowed(ctx, getRequestAuthType(r), dstBucket, dstObject, r, policy.PutObjectLegalHoldAction)
- // The destination is locked above: look it up without its lock, in the
- // locks' context.
- getObjectInfo := func(_ context.Context, bucket, object string, opts ObjectOptions) (ObjectInfo, error) {
- opts.NoLock = true
- return objectAPI.GetObjectInfo(lockCtx, bucket, object, opts)
- }
// apply default bucket configuration/governance headers for dest side.
retentionMode, retentionDate, legalHold, s3Err := checkPutObjectLockAllowed(ctx, r, dstBucket, dstObject, getObjectInfo, retPerms, holdPerms, replicaTrusted)
@@ -3158,6 +3171,22 @@ func (api objectAPIHandlers) DeleteObjectHandler(w http.ResponseWriter, r *http.
}
}
+ // Check before delOpts reads bucket versioning: unavailable bucket metadata
+ // must fail the request before a force-delete can reach the object layer.
+ if forceDelete, _ := strconv.ParseBool(r.Header.Get(xhttp.MinIOForceDelete)); forceDelete {
+ rcfg, err := globalBucketObjectLockSys.Get(bucket)
+ if err != nil {
+ writeErrorResponse(ctx, w, toAPIError(ctx, err), r.URL)
+ return
+ }
+ if rcfg.LockEnabled {
+ apiErr := toAPIError(ctx, errInvalidArgument)
+ apiErr.Description = "force-delete is forbidden on Object Locking enabled buckets"
+ writeErrorResponse(ctx, w, apiErr, r.URL)
+ return
+ }
+ }
+
opts, err := delOpts(ctx, r, bucket, object)
if err != nil {
writeErrorResponse(ctx, w, toAPIError(ctx, err), r.URL)
@@ -3178,14 +3207,6 @@ func (api objectAPIHandlers) DeleteObjectHandler(w http.ResponseWriter, r *http.
}
}
- rcfg, _ := globalBucketObjectLockSys.Get(bucket)
- if rcfg.LockEnabled && opts.DeletePrefix {
- apiErr := toAPIError(ctx, errInvalidArgument)
- apiErr.Description = "force-delete is forbidden on Object Locking enabled buckets"
- writeErrorResponse(ctx, w, apiErr, r.URL)
- return
- }
-
os := newObjSweeper(bucket, object).WithVersion(opts.VersionID).WithVersioning(opts.Versioned, opts.VersionSuspended)
opts.SetEvalMetadataFn(func(oi *ObjectInfo, gerr error) (dsc ReplicateDecision, err error) {
@@ -3384,7 +3405,7 @@ func (api objectAPIHandlers) PutObjectLegalHoldHandler(w http.ResponseWriter, r
VersionID: opts.VersionID,
EvalMetadataFn: func(oi *ObjectInfo, gerr error) (ReplicateDecision, error) {
oi.UserDefined[strings.ToLower(xhttp.AmzObjectLockLegalHold)] = strings.ToUpper(string(legalHold.Status))
- oi.UserDefined[ReservedMetadataPrefixLower+ObjectLockLegalHoldTimestamp] = UTCNow().Format(time.RFC3339Nano)
+ oi.UserDefined[ReservedMetadataPrefixLower+ObjectLockLegalHoldTimestamp] = monotonicMetadataTimestamp(UTCNow().Format(time.RFC3339Nano), oi.UserDefined[ReservedMetadataPrefixLower+ObjectLockLegalHoldTimestamp])
dsc := mustReplicate(ctx, bucket, object, oi.getMustReplicateOptions(replication.MetadataReplicationType, opts))
if dsc.ReplicateAny() {
@@ -3561,7 +3582,7 @@ func (api objectAPIHandlers) PutObjectRetentionHandler(w http.ResponseWriter, r
oi.UserDefined[strings.ToLower(xhttp.AmzObjectLockMode)] = ""
oi.UserDefined[strings.ToLower(xhttp.AmzObjectLockRetainUntilDate)] = ""
}
- oi.UserDefined[ReservedMetadataPrefixLower+ObjectLockRetentionTimestamp] = UTCNow().Format(time.RFC3339Nano)
+ oi.UserDefined[ReservedMetadataPrefixLower+ObjectLockRetentionTimestamp] = monotonicMetadataTimestamp(UTCNow().Format(time.RFC3339Nano), oi.UserDefined[ReservedMetadataPrefixLower+ObjectLockRetentionTimestamp])
dsc = mustReplicate(ctx, bucket, object, oi.getMustReplicateOptions(replication.MetadataReplicationType, opts))
if dsc.ReplicateAny() {
oi.UserDefined[ReservedMetadataPrefixLower+ReplicationTimestamp] = UTCNow().Format(time.RFC3339Nano)
diff --git a/cmd/object-lock-reconciliation_test.go b/cmd/object-lock-reconciliation_test.go
new file mode 100644
index 000000000..1624f7021
--- /dev/null
+++ b/cmd/object-lock-reconciliation_test.go
@@ -0,0 +1,419 @@
+// 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/md5"
+ "encoding/base64"
+ "fmt"
+ "maps"
+ "net/http"
+ "net/url"
+ "strings"
+ "testing"
+ "time"
+
+ "github.com/minio/minio/internal/auth"
+ xhttp "github.com/minio/minio/internal/http"
+)
+
+func TestPoolLockReductionOrderIndependent(t *testing.T) {
+ const old = objectLockTestStamp0900
+ const recent = objectLockTestStamp1000
+ const newest = objectLockTestStamp1100
+ compliance := objectLockState{mode: "COMPLIANCE", retainUntil: "2035-01-01T00:00:00Z", retentionTimestamp: old}
+ for _, tc := range []struct {
+ name string
+ states [3]objectLockState
+ want objectLockState
+ }{
+ {"compliance-before-newer-removal", [3]objectLockState{compliance, {retentionTimestamp: recent}, {mode: "GOVERNANCE", retainUntil: "2036-01-01T00:00:00Z", retentionTimestamp: newest}}, compliance},
+ {"longer-compliance-before-newer-shortening", [3]objectLockState{compliance, {mode: "COMPLIANCE", retainUntil: "2034-01-01T00:00:00Z", retentionTimestamp: newest}, {retentionTimestamp: recent}}, compliance},
+ {"unordered-compliance", [3]objectLockState{{mode: "COMPLIANCE", retainUntil: compliance.retainUntil, retentionTimestamp: "invalid"}, {retentionTimestamp: newest}, {}}, objectLockState{mode: "COMPLIANCE", retainUntil: compliance.retainUntil, retentionTimestamp: "invalid"}},
+ {"expired-compliance-can-release", [3]objectLockState{{mode: "COMPLIANCE", retainUntil: "2000-01-01T00:00:00Z", retentionTimestamp: old}, {retentionTimestamp: newest}, {retentionTimestamp: recent}}, objectLockState{retentionTimestamp: newest}},
+ {"governance-can-release", [3]objectLockState{{mode: "GOVERNANCE", retainUntil: compliance.retainUntil, retentionTimestamp: old}, {retentionTimestamp: newest}, {}}, objectLockState{retentionTimestamp: newest}},
+ {"unordered-hold-conflict", [3]objectLockState{{legalHold: "OFF"}, {legalHold: "ON"}, {}}, objectLockState{legalHold: "ON"}},
+ {"newer-hold-release", [3]objectLockState{{legalHold: "ON", legalHoldTimestamp: old}, {legalHold: "OFF", legalHoldTimestamp: newest}, {legalHold: "ON", legalHoldTimestamp: "invalid"}}, objectLockState{legalHold: "OFF", legalHoldTimestamp: newest}},
+ {"independent-fields", [3]objectLockState{compliance, {retentionTimestamp: newest, legalHold: "ON", legalHoldTimestamp: recent}, {legalHold: "OFF", legalHoldTimestamp: old}}, objectLockState{mode: compliance.mode, retainUntil: compliance.retainUntil, retentionTimestamp: old, legalHold: "ON", legalHoldTimestamp: recent}},
+ } {
+ t.Run(tc.name, func(t *testing.T) {
+ for _, order := range [][3]int{{0, 1, 2}, {0, 2, 1}, {1, 0, 2}, {1, 2, 0}, {2, 0, 1}, {2, 1, 0}} {
+ var copies []PoolObjInfo
+ for _, i := range order {
+ meta := make(map[string]string)
+ tc.states[i].restoreRetention(meta)
+ tc.states[i].restoreLegalHold(meta)
+ copies = append(copies, PoolObjInfo{Index: i, ObjInfo: ObjectInfo{UserDefined: meta}})
+ }
+ for name, reduce := range map[string]func([]PoolObjInfo) objectLockState{"merge": mergePoolLockState, "lifecycle": newestLockState} {
+ if got := reduce(copies); got != tc.want {
+ t.Errorf("%s order=%v: got %+v want %+v", name, order, got, tc.want)
+ }
+ }
+ }
+ })
+ }
+}
+
+// All requests copy the same bytes. Revisions belong to the addressed
+// destination, even if the source is the latest version of the same key.
+func TestAPICopyReplicaDestinationState(t *testing.T) {
+ defer DetectTestLeak(t)()
+ ExecObjectLayerAPITest(ExecObjectLayerAPITestArgs{
+ t: t, makeBucketOptions: MakeBucketOptions{LockEnabled: true},
+ objAPITest: func(obj ObjectLayer, backend, bucket string, router http.Handler, cred auth.Credentials, t *testing.T) {
+ for _, sameKey := range []bool{true, false} {
+ for _, missing := range []bool{false, true} {
+ for _, directive := range []string{copyDirective, replaceDirective} {
+ name := fmt.Sprintf("copy-destination/%t/%t/%s", sameKey, missing, directive)
+ source := name
+ if !sameKey {
+ source += "-source"
+ }
+ targetMeta := poolLockMetadata("GOVERNANCE", "ON", objectLockTestStamp1000, objectLockTestStamp1000)
+ targetMeta[xhttp.AmzObjectTagging] = "state=destination"
+ targetMeta[r5TagStamp] = objectLockTestStamp1000
+ targetID := mustGetUUID()
+ if !missing {
+ targetID = putObjectLockVersion(t, obj, bucket, name, targetMeta)
+ }
+ sourceMeta := poolLockMetadata("", "OFF", objectLockTestStamp1100, objectLockTestStamp1100)
+ sourceMeta[xhttp.AmzObjectTagging] = "state=source"
+ sourceMeta[r5TagStamp] = objectLockTestStamp1100
+ sourceID := putObjectLockVersion(t, obj, bucket, source, sourceMeta)
+ headers := map[string]string{
+ xhttp.AmzCopySource: "/" + bucket + "/" + source + "?versionId=" + sourceID,
+ xhttp.AmzMetadataDirective: directive, xhttp.AmzTagDirective: replaceDirective,
+ xhttp.MinIOSourceReplicationRequest: "true", xhttp.AmzBucketReplicationStatus: "REPLICA",
+ }
+ // No source revision means no change, including no
+ // inheritance from a different source for a first write.
+ rec := r5Request(t, router, cred, http.MethodPut, "/"+bucket+"/"+name+"?versionId="+targetID, "", headers)
+ if rec.Code != http.StatusOK {
+ t.Fatalf("%s %s: %d %s", backend, name, rec.Code, rec.Body.String())
+ }
+ want := storedObjectLockState(targetMeta)
+ wantTags, wantStamp := "state=destination", objectLockTestStamp1000
+ if missing {
+ want = objectLockState{}
+ wantTags = ""
+ wantStamp = ""
+ }
+ got := r5Stored(t, obj, bucket, name, targetID, wantTags, wantStamp)
+ if state := storedObjectLockState(got.UserDefined); state != want {
+ t.Fatalf("%s absent revision: got %+v want %+v", name, state, want)
+ }
+ // The incoming update is older than the source but may
+ // still win at a new destination. A source lookup loses it.
+ headers[xhttp.MinIOSourceObjectRetentionTimestamp] = objectLockTestStamp0930
+ headers[xhttp.MinIOSourceObjectLegalHoldTimestamp] = objectLockTestStamp0930
+ headers[xhttp.AmzObjectLockMode] = "GOVERNANCE"
+ headers[xhttp.AmzObjectLockRetainUntilDate] = objectLockTestRetainUntil
+ headers[xhttp.AmzObjectLockLegalHold] = "ON"
+ headers[xhttp.MinIOSourceTaggingTimestamp] = objectLockTestStamp0930
+ headers[xhttp.AmzObjectTagging] = "state=incoming"
+ rec = r5Request(t, router, cred, http.MethodPut, "/"+bucket+"/"+name+"?versionId="+targetID, "", headers)
+ if rec.Code != http.StatusOK {
+ t.Fatalf("incoming %s: %d %s", name, rec.Code, rec.Body.String())
+ }
+ if missing {
+ want = objectLockState{mode: "GOVERNANCE", retainUntil: "2030-01-01T00:00:00.000Z", retentionTimestamp: objectLockTestStamp0930, legalHold: "ON", legalHoldTimestamp: objectLockTestStamp0930}
+ wantTags = "state=incoming"
+ wantStamp = objectLockTestStamp0930
+ }
+ got = r5Stored(t, obj, bucket, name, targetID, wantTags, wantStamp)
+ if state := storedObjectLockState(got.UserDefined); state != want {
+ t.Fatalf("%s stale revision: got %+v want %+v", name, state, want)
+ }
+ // Authorized same-content release needs no governance
+ // bypass header; it must not be rejected by the data guard.
+ delete(headers, xhttp.AmzObjectLockMode)
+ delete(headers, xhttp.AmzObjectLockRetainUntilDate)
+ headers[xhttp.MinIOSourceObjectRetentionTimestamp] = "2026-09-03T10:30:00Z"
+ headers[xhttp.MinIOSourceObjectLegalHoldTimestamp] = "2026-09-03T10:30:00Z"
+ headers[xhttp.AmzObjectLockLegalHold] = "OFF"
+ rec = r5Request(t, router, cred, http.MethodPut, "/"+bucket+"/"+name+"?versionId="+targetID, "", headers)
+ if rec.Code != http.StatusOK {
+ t.Fatalf("release %s: %d %s", name, rec.Code, rec.Body.String())
+ }
+ got = r5Stored(t, obj, bucket, name, targetID, wantTags, wantStamp)
+ want = objectLockState{retentionTimestamp: "2026-09-03T10:30:00Z", legalHold: "OFF", legalHoldTimestamp: "2026-09-03T10:30:00Z"}
+ if state := storedObjectLockState(got.UserDefined); state != want {
+ t.Fatalf("%s release: got %+v want %+v", name, state, want)
+ }
+ }
+ }
+ }
+ },
+ })
+}
+
+func TestReplicaReferenceCopyDestinationState(t *testing.T) {
+ z, bucket := consistencyPools(t)
+ for _, missing := range []bool{false, true} {
+ key := fmt.Sprintf("reference-copy-%t", missing)
+ id := mustGetUUID()
+ targetMeta := poolLockMetadata("GOVERNANCE", "ON", objectLockTestStamp1000, objectLockTestStamp1000)
+ targetMeta[xhttp.AmzObjectTagging] = "state=destination"
+ targetMeta[r5TagStamp] = objectLockTestStamp1000
+ if !missing {
+ putConsistencyObject(t, z, bucket, key, 0, "data", ObjectOptions{Versioned: true, VersionID: id, UserDefined: targetMeta})
+ }
+ sourceMeta := poolLockMetadata("", "OFF", objectLockTestStamp1100, objectLockTestStamp1100)
+ sourceMeta[xhttp.AmzObjectTagging] = "state=source"
+ sourceMeta[r5TagStamp] = objectLockTestStamp1100
+ src := putConsistencyObject(t, z, bucket, key, 0, "data", ObjectOptions{Versioned: true, UserDefined: sourceMeta})
+ src.metadataOnly = true
+ src.UserDefined = make(map[string]string)
+ _, err := z.CopyObject(t.Context(), bucket, key, bucket, key, src, ObjectOptions{VersionID: src.VersionID}, ObjectOptions{Versioned: true, VersionID: id, ReplicaLockReconcile: true, ReplicationRequest: true})
+ if err != nil {
+ t.Fatal(err)
+ }
+ want := storedObjectLockState(targetMeta)
+ wantTags, wantStamp := "state=destination", objectLockTestStamp1000
+ if missing {
+ want = objectLockState{}
+ wantTags = ""
+ wantStamp = ""
+ }
+ got := r5Stored(t, z, bucket, key, id, wantTags, wantStamp)
+ if state := storedObjectLockState(got.UserDefined); state != want {
+ t.Fatalf("missing=%t: got %+v want %+v", missing, state, want)
+ }
+ }
+}
+
+// The set must not undo the all-pools decision with a newer weak revision
+// from its own copy before the protected competing copy is retired.
+func TestPoolsReplicaComplianceMetadataCopyOrder(t *testing.T) {
+ z, bucket := consistencyPools(t)
+ for _, protectedPool := range []int{0, 1} {
+ key := fmt.Sprintf("compliance-copy-%d", protectedPool)
+ id := mustGetUUID()
+ stamp := UTCNow().Add(-time.Hour)
+ var source ObjectInfo
+ for pool := range 2 {
+ meta := poolLockMetadata("", "OFF", objectLockTestStamp1100, objectLockTestStamp1000)
+ if pool == protectedPool {
+ meta = poolLockMetadata("COMPLIANCE", "OFF", objectLockTestStamp0900, objectLockTestStamp1000)
+ }
+ oi := putConsistencyObject(t, z, bucket, key, pool, "data", ObjectOptions{Versioned: true, VersionID: id, MTime: stamp, UserDefined: meta})
+ if pool == 0 {
+ source = oi
+ }
+ }
+ source.metadataOnly = true
+ source.UserDefined = map[string]string{}
+ _, err := z.CopyObject(t.Context(), bucket, key, bucket, key, source, ObjectOptions{VersionID: id}, ObjectOptions{VersionID: id, Versioned: true, ReplicaLockReconcile: true, ReplicationRequest: true})
+ if err != nil {
+ t.Fatal(err)
+ }
+ got := readObjectLockFields(t, z, bucket, key, id)
+ if got.mode != "COMPLIANCE" || got.retainUntil != "2030-01-01T00:00:00Z" {
+ t.Fatalf("protected pool %d lost COMPLIANCE: %+v", protectedPool, got)
+ }
+ }
+}
+
+func TestAPIReplicaRetentionTimestampAuthorization(t *testing.T) {
+ defer DetectTestLeak(t)()
+ ExecObjectLayerAPITest(ExecObjectLayerAPITestArgs{
+ t: t, makeBucketOptions: MakeBucketOptions{LockEnabled: true},
+ objAPITest: func(obj ObjectLayer, backend, bucket string, router http.Handler, _ auth.Credentials, t *testing.T) {
+ basic := `"s3:GetObject","s3:PutObject","s3:ReplicateObject"`
+ noRetention := newObjectAttributesAuthzUser(t, backend, bucket, basic)
+ withRetention := newObjectAttributesAuthzUser(t, backend, bucket, basic+`,"s3:PutObjectRetention"`)
+ for _, operation := range []string{"put", "copy", "multipart"} {
+ for _, authorized := range []bool{false, true} {
+ for _, timestamp := range []bool{false, true} {
+ key := fmt.Sprintf("replica-permission/%s/%t/%t", operation, authorized, timestamp)
+ meta := poolLockMetadata("GOVERNANCE", "", objectLockTestStamp0900, "")
+ id := putObjectLockVersion(t, obj, bucket, key, meta)
+ headers := map[string]string{xhttp.MinIOSourceReplicationRequest: "true", xhttp.AmzBucketReplicationStatus: "REPLICA"}
+ if timestamp {
+ headers[xhttp.MinIOSourceObjectRetentionTimestamp] = objectLockTestStamp1000
+ }
+ method, path, body := http.MethodPut, "/"+bucket+"/"+key+"?versionId="+id, "data"
+ switch operation {
+ case "copy":
+ headers[xhttp.AmzCopySource] = "/" + bucket + "/" + key + "?versionId=" + id
+ headers[xhttp.AmzMetadataDirective] = replaceDirective
+ body = ""
+ case "multipart":
+ method = http.MethodPost
+ path += "&uploads"
+ body = ""
+ }
+ cred := noRetention
+ if authorized {
+ cred = withRetention
+ }
+ rec := r5Request(t, router, cred, method, path, body, headers)
+ want := http.StatusOK
+ if timestamp && !authorized {
+ want = http.StatusForbidden
+ }
+ if rec.Code != want {
+ t.Fatalf("%s: %d want %d: %s", key, rec.Code, want, rec.Body.String())
+ }
+ if want == http.StatusForbidden || !timestamp {
+ got := readObjectLockFields(t, obj, bucket, key, id)
+ if got.mode != "GOVERNANCE" || got.retentionStamp != objectLockTestStamp0900 {
+ t.Fatalf("unrequested/denied retention changed: %+v", got)
+ }
+ }
+ }
+ }
+ }
+ },
+ })
+}
+
+func TestAPILocalObjectLockTimestampsAdvance(t *testing.T) {
+ defer DetectTestLeak(t)()
+ ExecObjectLayerAPITest(ExecObjectLayerAPITestArgs{
+ t: t, makeBucketOptions: MakeBucketOptions{LockEnabled: true},
+ objAPITest: func(obj ObjectLayer, backend, bucket string, router http.Handler, cred auth.Credentials, t *testing.T) {
+ future := UTCNow().Add(48 * time.Hour)
+ meta := poolLockMetadata("GOVERNANCE", "ON", future.Format(time.RFC3339Nano), future.Format(time.RFC3339Nano))
+ id := putObjectLockVersion(t, obj, bucket, "local-lock", meta)
+ for n := 1; n <= 2; n++ {
+ if status := putRetention(t, router, cred, bucket, "local-lock", id, "GOVERNANCE", time.Date(2035, 1, n, 0, 0, 0, 0, time.UTC), false); status != http.StatusOK {
+ t.Fatalf("%s retention: %d", backend, status)
+ }
+ body := []byte(`OFF`)
+ sum := md5.Sum(body)
+ rec := retentionAuthzRequest(t, router, cred, http.MethodPut, makeTestTargetURL("", bucket, "local-lock", url.Values{"legal-hold": {""}, "versionId": {id}}), body, map[string]string{xhttp.ContentMD5: base64.StdEncoding.EncodeToString(sum[:])})
+ if rec.Code != http.StatusOK {
+ t.Fatalf("%s hold: %d %s", backend, rec.Code, rec.Body.String())
+ }
+ got := readObjectLockFields(t, obj, bucket, "local-lock", id)
+ want := future.Add(time.Duration(n) * time.Nanosecond).Format(time.RFC3339Nano)
+ if got.retentionStamp != want || got.legalHoldStamp != want {
+ t.Fatalf("local mutation %d regressed: %+v want %s", n, got, want)
+ }
+ }
+ },
+ })
+}
+
+type lockBoundaryReadFailure struct {
+ ObjectLayer
+ bucket, key, version string
+ metadataError error
+ destinationError error
+ deleteCalls int
+}
+
+func (f *lockBoundaryReadFailure) GetObjectNInfo(ctx context.Context, bucket, key string, rs *HTTPRangeSpec, h http.Header, opts ObjectOptions) (*GetObjectReader, error) {
+ if f.metadataError != nil && bucket == minioMetaBucket && strings.HasPrefix(key, pathJoin(bucketMetaPrefix, f.bucket)+"/") {
+ return nil, f.metadataError
+ }
+ return f.ObjectLayer.GetObjectNInfo(ctx, bucket, key, rs, h, opts)
+}
+func (f *lockBoundaryReadFailure) GetObjectInfo(ctx context.Context, bucket, key string, opts ObjectOptions) (ObjectInfo, error) {
+ if f.destinationError != nil && bucket == f.bucket && key == f.key && opts.VersionID == f.version {
+ return ObjectInfo{}, f.destinationError
+ }
+ return f.ObjectLayer.GetObjectInfo(ctx, bucket, key, opts)
+}
+func (f *lockBoundaryReadFailure) DeleteObject(ctx context.Context, bucket, key string, opts ObjectOptions) (ObjectInfo, error) {
+ f.deleteCalls++
+ return f.ObjectLayer.DeleteObject(ctx, bucket, key, opts)
+}
+
+func TestAPICopyReplicaDestinationReadFailure(t *testing.T) {
+ ExecObjectLayerAPITest(ExecObjectLayerAPITestArgs{t: t, makeBucketOptions: MakeBucketOptions{LockEnabled: true}, objAPITest: func(obj ObjectLayer, backend, bucket string, _ http.Handler, cred auth.Credentials, t *testing.T) {
+ id := putObjectLockVersion(t, obj, bucket, "destination", poolLockMetadata("GOVERNANCE", "ON", objectLockTestStamp1000, objectLockTestStamp1000))
+ source := putObjectLockVersion(t, obj, bucket, "source", nil)
+ f := &lockBoundaryReadFailure{ObjectLayer: obj, bucket: bucket, key: "destination", version: id, destinationError: errErasureReadQuorum}
+ router := initTestAPIEndPoints(f, []string{"CopyObject"})
+ headers := map[string]string{xhttp.AmzCopySource: "/" + bucket + "/source?versionId=" + source, xhttp.MinIOSourceReplicationRequest: "true", xhttp.AmzBucketReplicationStatus: "REPLICA"}
+ rec := r5Request(t, router, cred, http.MethodPut, "/"+bucket+"/destination?versionId="+id, "", headers)
+ if rec.Code < 500 {
+ t.Fatalf("%s read failure allowed copy: %d %s", backend, rec.Code, rec.Body.String())
+ }
+ if got := readObjectLockFields(t, obj, bucket, "destination", id); got.legalHold != "ON" || got.mode != "GOVERNANCE" {
+ t.Fatalf("failed read changed state: %+v", got)
+ }
+ }})
+}
+
+func TestAPIForceDeleteLockConfigFailure(t *testing.T) {
+ for _, locked := range []bool{false, true} {
+ t.Run(fmt.Sprintf("locked=%t", locked), func(t *testing.T) {
+ ExecObjectLayerAPITest(ExecObjectLayerAPITestArgs{t: t, makeBucketOptions: MakeBucketOptions{LockEnabled: locked}, objAPITest: func(obj ObjectLayer, backend, bucket string, _ http.Handler, cred auth.Credentials, t *testing.T) {
+ saved, err := globalBucketMetadataSys.Get(bucket)
+ if err != nil {
+ t.Fatal(err)
+ }
+ restoreInit := markBucketMetadataInitialized(t)
+ defer restoreInit()
+ previous := newObjectLayerFn()
+ defer setObjectLayer(previous)
+ defer globalBucketMetadataSys.Set(bucket, saved)
+ for _, failure := range []string{"quorum", "unavailable", "cache-miss"} {
+ key := "force-" + failure + "/data"
+ putObjectLockVersion(t, obj, bucket, key, nil)
+ f := &lockBoundaryReadFailure{ObjectLayer: obj, bucket: bucket}
+ globalBucketMetadataSys.Remove(bucket)
+ switch failure {
+ case "quorum":
+ f.metadataError = errErasureReadQuorum
+ case "unavailable":
+ f.metadataError = errServerNotInitialized
+ }
+ router := initTestAPIEndPoints(f, []string{"DeleteObject"})
+ rec := r5Request(t, router, cred, http.MethodDelete, "/"+bucket+"/force-"+failure, "", map[string]string{xhttp.MinIOForceDelete: "true"})
+ setObjectLayer(previous)
+ globalBucketMetadataSys.Set(bucket, saved)
+ allowed := failure == "cache-miss" && !locked
+ if allowed {
+ if rec.Code != http.StatusNoContent || f.deleteCalls != 1 {
+ t.Fatalf("%s missing unlocked cache: %d calls=%d %s", backend, rec.Code, f.deleteCalls, rec.Body.String())
+ }
+ } else {
+ if rec.Code < 400 || f.deleteCalls != 0 {
+ t.Fatalf("%s %s: force-delete proceeded: %d calls=%d %s", backend, failure, rec.Code, f.deleteCalls, rec.Body.String())
+ }
+ if _, err := obj.GetObjectInfo(t.Context(), bucket, key, ObjectOptions{}); err != nil {
+ t.Fatalf("rejected force-delete lost data: %v", err)
+ }
+ }
+ }
+ }})
+ })
+ }
+}
+
+func TestAPILocalCopyTagTimestampAdvances(t *testing.T) {
+ ExecObjectLayerAPITest(ExecObjectLayerAPITestArgs{t: t, objAPITest: func(obj ObjectLayer, backend, bucket string, router http.Handler, cred auth.Credentials, t *testing.T) {
+ stamp := UTCNow().Add(48 * time.Hour)
+ meta := map[string]string{r5TagStamp: stamp.Format(time.RFC3339Nano), xhttp.AmzObjectTagging: "key=old"}
+ _, err := obj.PutObject(t.Context(), bucket, "local-copy", mustGetPutObjReader(t, strings.NewReader("data"), 4, "", ""), ObjectOptions{UserDefined: maps.Clone(meta)})
+ if err != nil {
+ t.Fatal(err)
+ }
+ rec := r5Request(t, router, cred, http.MethodPut, "/"+bucket+"/local-copy", "", map[string]string{xhttp.AmzCopySource: "/" + bucket + "/local-copy", xhttp.AmzMetadataDirective: replaceDirective, xhttp.AmzTagDirective: replaceDirective})
+ if rec.Code != http.StatusOK {
+ t.Fatalf("%s copy: %d %s", backend, rec.Code, rec.Body.String())
+ }
+ r5Stored(t, obj, bucket, "local-copy", "", "", stamp.Add(time.Nanosecond).Format(time.RFC3339Nano))
+ }})
+}