Fix Object Lock replica reconciliation boundaries

Reconcile CopyObject lock and tag revisions with the addressed destination,
including missing versions and self-referential copies. Preserve active
COMPLIANCE across pool orderings, lifecycle decisions and set-level writes.

Require retention authorization for source retention revisions, advance local
lock and tag revisions monotonically, and fail force-delete before option
parsing when bucket lock configuration cannot be read. Preserve authorized
same-content replication and existing encryption and tier cleanup behavior.

Signed-off-by: Feng Ruohang <rh@vonng.com>
(cherry picked from commit a08c3d149dcccad74dd68ab374c02f56379181be)

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
 	}
 }