astelm/minio
1
0
Fork 0
mirror of https://github.com/pgsty/minio.git synced 2026-09-26 04:45:59 +03:00
Code Issues Packages Projects Releases Wiki Activity

fix: apply replicated Object Lock updates only when newer than the stored state

Browse Source 
A replicated CopyObject uses the REPLACE metadata directive, so the map the
handler compared replication timestamps against had already been rebuilt
from the request and filtered of Object Lock keys: the stored retention and
legal-hold timestamps were never seen, every replica update was applied
regardless of order, and the legal-hold timestamp was written under the
retention key. A stale replica could turn a newer legal hold off or shorten
a newer retention.

Capture the stored Object Lock state before the metadata is rebuilt, apply a
replica update only when its source timestamp is newer, put the stored state
back when the update is stale, and keep each timestamp under its own key.
Inherited from upstream; recorded in the advisory ledger.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01PvgysXDmhPBBimCReYtA8q
Signed-off-by: Feng Ruohang <rh@vonng.com>
This commit is contained in:
Feng Ruohang
2026-09-03 00:45:02 +08:00
parent 6e112d1856
commit f4c1286c9d
4 changed files with 145 additions and 17 deletions
Download Patch File Download Diff File Expand all files Collapse all files
cmd/object-handlers.go
+18 -17
View File
@@ -1664,6 +1664,12 @@ 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)
srcInfo.UserDefined, err = getCpObjMetadataFromHeader(ctx, r, srcInfo.UserDefined, allowReplicationMetadata)
if err != nil {
writeErrorResponse(ctx, w, toAPIError(ctx, err), r.URL)
@@ -1707,17 +1713,14 @@ func (api objectAPIHandlers) CopyObjectHandler(w http.ResponseWriter, r *http.Re
// apply default bucket configuration/governance headers for dest side.
retentionMode, retentionDate, legalHold, s3Err := checkPutObjectLockAllowed(ctx, r, dstBucket, dstObject, getObjectInfo, retPerms, holdPerms, replicaTrusted)
if s3Err == ErrNone && retentionMode.Valid() {
lastretentionTimestamp := srcInfo.UserDefined[ReservedMetadataPrefixLower+ObjectLockRetentionTimestamp]
if dstOpts.ReplicationRequest {
srcTimestamp := dstOpts.ReplicationSourceRetentionTimestamp
if !srcTimestamp.IsZero() {
ondiskTimestamp, err := time.Parse(time.RFC3339Nano, lastretentionTimestamp)
// update retention metadata only if replica timestamp is newer than what's on disk
if err != nil || (err == nil && ondiskTimestamp.Before(srcTimestamp)) {
srcInfo.UserDefined[strings.ToLower(xhttp.AmzObjectLockMode)] = string(retentionMode)
srcInfo.UserDefined[strings.ToLower(xhttp.AmzObjectLockRetainUntilDate)] = amztime.ISO8601Format(retentionDate.UTC())
srcInfo.UserDefined[ReservedMetadataPrefixLower+ObjectLockRetentionTimestamp] = srcTimestamp.UTC().Format(time.RFC3339Nano)
}
if storedLock.retentionIsOlderThan(srcTimestamp) {
srcInfo.UserDefined[strings.ToLower(xhttp.AmzObjectLockMode)] = string(retentionMode)
srcInfo.UserDefined[strings.ToLower(xhttp.AmzObjectLockRetainUntilDate)] = amztime.ISO8601Format(retentionDate.UTC())
srcInfo.UserDefined[ReservedMetadataPrefixLower+ObjectLockRetentionTimestamp] = srcTimestamp.UTC().Format(time.RFC3339Nano)
} else {
storedLock.restoreRetention(srcInfo.UserDefined)
}
} else {
srcInfo.UserDefined[strings.ToLower(xhttp.AmzObjectLockMode)] = string(retentionMode)
@@ -1727,19 +1730,17 @@ func (api objectAPIHandlers) CopyObjectHandler(w http.ResponseWriter, r *http.Re
}
if s3Err == ErrNone && legalHold.Status.Valid() {
lastLegalHoldTimestamp := srcInfo.UserDefined[ReservedMetadataPrefixLower+ObjectLockLegalHoldTimestamp]
if dstOpts.ReplicationRequest {
srcTimestamp := dstOpts.ReplicationSourceLegalholdTimestamp
if !srcTimestamp.IsZero() {
ondiskTimestamp, err := time.Parse(time.RFC3339Nano, lastLegalHoldTimestamp)
// update legalhold metadata only if replica timestamp is newer than what's on disk
if err != nil || (err == nil && ondiskTimestamp.Before(srcTimestamp)) {
srcInfo.UserDefined[strings.ToLower(xhttp.AmzObjectLockLegalHold)] = string(legalHold.Status)
srcInfo.UserDefined[ReservedMetadataPrefixLower+ObjectLockRetentionTimestamp] = srcTimestamp.Format(time.RFC3339Nano)
}
if storedLock.legalHoldIsOlderThan(srcTimestamp) {
srcInfo.UserDefined[strings.ToLower(xhttp.AmzObjectLockLegalHold)] = string(legalHold.Status)
srcInfo.UserDefined[ReservedMetadataPrefixLower+ObjectLockLegalHoldTimestamp] = srcTimestamp.UTC().Format(time.RFC3339Nano)
} else {
storedLock.restoreLegalHold(srcInfo.UserDefined)
}
} else {
srcInfo.UserDefined[strings.ToLower(xhttp.AmzObjectLockLegalHold)] = string(legalHold.Status)
srcInfo.UserDefined[ReservedMetadataPrefixLower+ObjectLockLegalHoldTimestamp] = UTCNow().Format(time.RFC3339Nano)
}
}
if s3Err != ErrNone {