From 109d824e5f67d365b72f843e6fa66751ab267a16 Mon Sep 17 00:00:00 2001 From: Feng Ruohang Date: Sat, 5 Sep 2026 17:45:12 +0800 Subject: [PATCH] fix: retransmit and re-order Object Lock for SSE-C replicas (single erasure set) Issue #120 routes an existing SSE-C replica through PutObjectHandler and NewMultipartUploadHandler. On main those handlers assigned the incoming retention and legal hold directly, without the source-timestamp ordering #111 added to CopyObjectHandler and without persisting the ordering timestamps, so in active-active replication a retransmit carrying an older value could overwrite a destination version's newer lock state. Share #111's ordering decision as applyReplicatedObjectLock in cmd/bucket-object-lock.go and call it from CopyObject, PUT and multipart initiation. A request that is not an actual trusted replica keeps ordinary write semantics (a validated value is applied and stamped now); only a real replica update is ordered against the stored version, so a marker-only peer write no longer drops a validated hold or default retention. CopyObject keeps its SSE-C key-rotation encMetadata reconciliation inline. putReplicationOpts now emits a stored retention ordering timestamp even when the value keys are absent, so a removal recorded on the retransmit PUT path still replicates onward. replicateAll marks Failed and carries the error when putReplicationOpts fails. The handler decision is made against the version as it stands then, which a concurrent lock update can outrun before the write commits, and for multipart across the whole initiation-to-completion span. Close that window under the object write lock the receiving erasure set holds: a trusted SSE-C replica full write sets ObjectOptions.ReplicaLockReconcile, and erasureObjects.PutObject and CompleteMultipartUpload re-run the ordering (reconcileStoredObjectLock, which orders retention and legal hold independently by their reserved timestamps) against the destination version read on that set before committing. Persisted upload metadata records the null version as an empty VersionID, so completion looks that up as the null version rather than the latest. The reconcile runs only against an existing version; a not-found destination keeps the write's own accepted lock, including a pre-upgrade upload that persisted values without ordering timestamps, and a non-not-found read error fails the write. Scoped to the SSE-C paths this issue enables; CopyObject is left as #111 wrote it. Scope: this orders Object Lock against the destination version under the write lock and is correct for a single erasure set. A multi-pool deployment -- where a version can have duplicate copies across pools, object ModTime ties do not track per-field lock timestamps, and the object namespace lock is per-pool -- needs a cross-pool lock-safe reconcile and is deliberately out of scope here, tracked in pgsty/silo#TBD-multipool-lock. Tests: TestAPISSECReplicaRetransmitObjectLockOrdering and its multipart sibling; TestAPIReplicaMultipartNewerHoldSurvivesCompletion and TestReplicaPutObjectLockReconcileUnderWriteLock (a hold or retention reaching the version after the handler decision, or after multipart initiation, survives the commit; a pre-upgrade upload on an absent version keeps its lock); TestReplicaLockReconcileNullVersion (a null-version completion reconciles the null version, not a coexisting UUID version, and an absent null version keeps its accepted lock); TestAPIReplicaMarkerOnlyAppliesObjectLock; TestReplicaStoredLock; the timestamp-only putReplicationOpts round trip; and the retransmit, exemption and target-head tests. The #111 CopyObject replica suite and the existing #120 suite stay green, as do the PUT/multipart handler and object-layer regression suites. Compatibility: the shared helper preserves #111's CopyObject behavior; a non-replica PUT or multipart initiation that sets Object Lock now also stamps the reserved ordering timestamp, matching CopyObject since #111; only trusted SSE-C replica writes take the in-lock reconcile. Refs pgsty/silo#120 Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_01L7qJqWwy8oFA6aCXWRzXQe Signed-off-by: Feng Ruohang --- cmd/bucket-object-lock.go | 114 +++ cmd/bucket-replication.go | 19 +- cmd/erasure-multipart.go | 66 +- cmd/erasure-object.go | 32 +- cmd/object-api-interface.go | 1 + cmd/object-handlers.go | 78 +- cmd/object-multipart-handlers.go | 41 +- cmd/replication-ssec-retransmit_test.go | 1000 +++++++++++++++++++++++ 8 files changed, 1268 insertions(+), 83 deletions(-) diff --git a/cmd/bucket-object-lock.go b/cmd/bucket-object-lock.go index e8cb31114..e5828e2dd 100644 --- a/cmd/bucket-object-lock.go +++ b/cmd/bucket-object-lock.go @@ -25,6 +25,7 @@ import ( "strings" "time" + "github.com/minio/minio/internal/amztime" "github.com/minio/minio/internal/auth" objectlock "github.com/minio/minio/internal/bucket/object/lock" xhttp "github.com/minio/minio/internal/http" @@ -407,3 +408,116 @@ func (s objectLockState) restoreLegalHold(metadata map[string]string) { } metadata[strings.ToLower(xhttp.AmzObjectLockLegalHold)] = s.legalHold } + +// replicaStoredLock reads the Object Lock state stored on the addressed version +// so a trusted replica write can order its update against it. A missing object +// or version yields an empty state, which is correct for the first write of a +// version; any other read error is returned so the caller fails the write rather +// than ordering an incoming update against lock state it merely failed to read +// (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}) + switch { + case err == nil: + return storedObjectLockState(oi.UserDefined), nil + case isErrObjectNotFound(err) || isErrVersionNotFound(err): + return objectLockState{}, nil + default: + return objectLockState{}, err + } +} + +// applyReplicatedObjectLock writes the retention and legal-hold decision into +// metadata for a PUT, CopyObject, or multipart-initiation request. A request +// that is not an actual trusted replica -- a normal user write, or a trusted +// peer that carried the replication marker without REPLICA status -- takes +// ordinary write semantics: a validated value is applied and stamped now, and a +// missing value is left as is. Only an actual replica update is ordered against +// the state already stored on the addressed version, so a stale value cannot +// overwrite a newer one and a full retransmit cannot roll a destination back. +// The stored argument is meaningful only for a replica; callers pass an empty +// state otherwise. Only the two Object Lock keys and their reserved ordering +// timestamps are touched; any encryption-metadata reconciliation stays with the +// caller. +func applyReplicatedObjectLock(metadata map[string]string, stored objectLockState, + replicaTrusted bool, + retentionMode objectlock.RetMode, retentionDate objectlock.RetentionDate, + legalHold objectlock.ObjectLegalHold, srcRetentionTimestamp, srcLegalholdTimestamp time.Time, +) { + switch { + case !replicaTrusted: + // Ordinary write semantics: apply a validated retention and stamp it now; + // a missing value carries no instruction, so leave the metadata as it is. + if retentionMode.Valid() { + metadata[strings.ToLower(xhttp.AmzObjectLockMode)] = string(retentionMode) + metadata[strings.ToLower(xhttp.AmzObjectLockRetainUntilDate)] = amztime.ISO8601Format(retentionDate.UTC()) + metadata[ReservedMetadataPrefixLower+ObjectLockRetentionTimestamp] = UTCNow().Format(time.RFC3339Nano) + } + case !stored.retentionIsOlderThan(srcRetentionTimestamp): + // The stored update is at least as new as this replica's, or the replica + // carries no ordering timestamp: keep what is stored. This is also how a + // stale retransmit is rejected. + stored.restoreRetention(metadata) + default: + // The replica update wins. A removal carries no value but still records + // the source timestamp that orders it. + if retentionMode.Valid() { + metadata[strings.ToLower(xhttp.AmzObjectLockMode)] = string(retentionMode) + metadata[strings.ToLower(xhttp.AmzObjectLockRetainUntilDate)] = amztime.ISO8601Format(retentionDate.UTC()) + } + metadata[ReservedMetadataPrefixLower+ObjectLockRetentionTimestamp] = srcRetentionTimestamp.UTC().Format(time.RFC3339Nano) + } + + // Legal hold has no removal in S3: an explicitly empty header is already + // rejected as an invalid status, so the only value-less shape that gets here + // is an absent one, which conveys no legal-hold change. Only a valid status + // can win. + switch { + case !replicaTrusted: + if legalHold.Status.Valid() { + metadata[strings.ToLower(xhttp.AmzObjectLockLegalHold)] = string(legalHold.Status) + metadata[ReservedMetadataPrefixLower+ObjectLockLegalHoldTimestamp] = UTCNow().Format(time.RFC3339Nano) + } + case legalHold.Status.Valid() && stored.legalHoldIsOlderThan(srcLegalholdTimestamp): + metadata[strings.ToLower(xhttp.AmzObjectLockLegalHold)] = string(legalHold.Status) + metadata[ReservedMetadataPrefixLower+ObjectLockLegalHoldTimestamp] = srcLegalholdTimestamp.UTC().Format(time.RFC3339Nano) + default: + stored.restoreLegalHold(metadata) + } +} + +// reconcileStoredObjectLock re-orders the Object Lock already written into +// metadata against the state currently stored on the destination version, both +// compared by their reserved ordering timestamps. It runs inside the object +// layer under the namespace write lock that guards the version replacement, +// after the destination version is read and before the new one is committed, so +// a replica update whose ordering was decided at handler time (or, for multipart, +// at initiation) cannot overwrite a newer lock update that reached the version in +// between. metadata already carries the incoming update with its source +// timestamps; a stored value that is not older than the incoming one is put back, +// which for a stored removal means clearing the incoming value and keeping only +// the removal's timestamp. Only the two lock keys and their reserved timestamps +// move; a non-replica write never sets the flag that invokes this. +func reconcileStoredObjectLock(metadata map[string]string, stored objectLockState) { + incoming := storedObjectLockState(metadata) + + incomingRetentionTS, _ := time.Parse(time.RFC3339Nano, incoming.retentionTimestamp) + if !stored.retentionIsOlderThan(incomingRetentionTS) { + // The stored retention is at least as new as the incoming one (or the + // incoming update is unordered): drop the incoming value and put the stored + // state back, which may itself be a removal (value keys absent, timestamp + // present). + delete(metadata, strings.ToLower(xhttp.AmzObjectLockMode)) + delete(metadata, strings.ToLower(xhttp.AmzObjectLockRetainUntilDate)) + delete(metadata, ReservedMetadataPrefixLower+ObjectLockRetentionTimestamp) + stored.restoreRetention(metadata) + } + + incomingLegalHoldTS, _ := time.Parse(time.RFC3339Nano, incoming.legalHoldTimestamp) + if incoming.legalHold == "" || !stored.legalHoldIsOlderThan(incomingLegalHoldTS) { + delete(metadata, strings.ToLower(xhttp.AmzObjectLockLegalHold)) + delete(metadata, ReservedMetadataPrefixLower+ObjectLockLegalHoldTimestamp) + stored.restoreLegalHold(metadata) + } +} diff --git a/cmd/bucket-replication.go b/cmd/bucket-replication.go index e78ab2edd..1b42e970c 100644 --- a/cmd/bucket-replication.go +++ b/cmd/bucket-replication.go @@ -877,8 +877,8 @@ func putReplicationOpts(ctx context.Context, sc string, objInfo ObjectInfo) (put if hasMode { putOpts.Mode = minio.RetentionMode(mode) } - // A removed retention is stored as an empty mode and date; it is sent as - // a value-less update that still carries its ordering timestamp. + // A removed retention is stored as an empty or absent mode and date; it is + // sent as a value-less update that still carries its ordering timestamp. if hasRetainDate && retainDateStr != "" { rdate, err := amztime.ISO8601Parse(retainDateStr) if err != nil { @@ -886,10 +886,14 @@ func putReplicationOpts(ctx context.Context, sc string, objInfo ObjectInfo) (put } putOpts.RetainUntilDate = rdate } - if hasMode || hasRetainDate { - // set retention timestamp in opts + retainTmstampStr, hasRetainTmstamp := objInfo.UserDefined[ReservedMetadataPrefixLower+ObjectLockRetentionTimestamp] + if hasMode || hasRetainDate || hasRetainTmstamp { + // Send the ordering timestamp whenever the version carries one, even for a + // removal whose value keys are absent (the shape a retransmit PUT leaves), + // so the next hop can order the removal instead of keeping obsolete + // retention. retTimestamp := objInfo.ModTime - if retainTmstampStr, ok := objInfo.UserDefined[ReservedMetadataPrefixLower+ObjectLockRetentionTimestamp]; ok { + if hasRetainTmstamp { var err error retTimestamp, err = time.Parse(time.RFC3339Nano, retainTmstampStr) if err != nil { @@ -1618,8 +1622,11 @@ applyAction: } else { putOpts, isMP, err := putReplicationOpts(ctx, tgt.StorageClass, objInfo) if err != nil { - rinfo.Err = err + // rinfo was primed Completed above; a failure to build the write + // options means nothing reached the target, so mark it Failed and + // carry the error instead of reporting a phantom success. rinfo.ReplicationStatus = replication.Failed + rinfo.Err = err replLogIf(ctx, fmt.Errorf("failed to set replicate options for object %s/%s(%s) (target %s) err:%w", bucket, objInfo.Name, objInfo.VersionID, tgt.EndpointURL(), err)) sendEvent(eventArgs{ EventName: event.ObjectReplicationNotTracked, diff --git a/cmd/erasure-multipart.go b/cmd/erasure-multipart.go index 35c790805..c50c44b0f 100644 --- a/cmd/erasure-multipart.go +++ b/cmd/erasure-multipart.go @@ -1114,7 +1114,7 @@ func (er erasureObjects) CompleteMultipartUpload(ctx context.Context, bucket str auditObjectErasureSet(ctx, "CompleteMultipartUpload", object, &er) } - if opts.CheckPrecondFn != nil { + if opts.CheckPrecondFn != nil || opts.ReplicaLockReconcile { if !opts.NoLock { ns := er.NewNSLock(bucket, object) lkctx, err := ns.GetLock(ctx, globalOperationTimeout) @@ -1126,18 +1126,24 @@ func (er erasureObjects) CompleteMultipartUpload(ctx context.Context, bucket str opts.NoLock = true } - obj, err := er.getObjectInfo(ctx, bucket, object, opts) - if err == nil && opts.CheckPrecondFn(obj) { - return ObjectInfo{}, PreConditionFailed{} - } - if err != nil && !isErrVersionNotFound(err) && !isErrObjectNotFound(err) { - return ObjectInfo{}, err - } + // The Object Lock reconcile below needs the version being committed, read + // after checkUploadIDExists, so only the precondition read happens here; + // both run under this same write lock, held until the version is renamed + // into place. + if opts.CheckPrecondFn != nil { + obj, err := er.getObjectInfo(ctx, bucket, object, opts) + if err == nil && opts.CheckPrecondFn(obj) { + return ObjectInfo{}, PreConditionFailed{} + } + if err != nil && !isErrVersionNotFound(err) && !isErrObjectNotFound(err) { + return ObjectInfo{}, err + } - // if object doesn't exist return error for If-Match conditional requests - // If-None-Match should be allowed to proceed for non-existent objects - if err != nil && opts.HasIfMatch && (isErrObjectNotFound(err) || isErrVersionNotFound(err)) { - return ObjectInfo{}, err + // if object doesn't exist return error for If-Match conditional requests + // If-None-Match should be allowed to proceed for non-existent objects + if err != nil && opts.HasIfMatch && (isErrObjectNotFound(err) || isErrVersionNotFound(err)) { + return ObjectInfo{}, err + } } } @@ -1149,6 +1155,42 @@ func (er erasureObjects) CompleteMultipartUpload(ctx context.Context, bucket str return oi, toObjectErr(err, bucket, object, uploadID) } + // A trusted SSE-C replica completion re-orders the Object Lock carried in the + // upload metadata against the version it is about to replace, read on this + // erasure set under the write lock held above, so a hold or retention that + // reached the version after this upload was initiated is not rolled back at + // completion (issue #120). Scoped to SSE-C uploads, the only ones this issue + // routes through completion. + // + // Scope: correct for a single erasure set. A multi-pool deployment (duplicate + // versions across pools, ModTime ties, cross-pool lock authority) is out of + // scope and tracked in pgsty/silo#TBD-multipool-lock. + if opts.ReplicaLockReconcile && crypto.SSEC.IsEncrypted(fi.Metadata) { + // A persisted upload records the null version as an empty VersionID; look + // it up as the null version so the reconcile reads the addressed version's + // stored lock, not the latest version's. + lookupVersionID := fi.VersionID + if lookupVersionID == "" { + lookupVersionID = nullVersionID + } + curr, gerr := er.getObjectInfo(ctx, bucket, object, ObjectOptions{ + VersionID: lookupVersionID, + Versioned: opts.Versioned, + VersionSuspended: opts.VersionSuspended, + NoLock: true, + }) + switch { + case gerr == nil: + reconcileStoredObjectLock(fi.Metadata, storedObjectLockState(curr.UserDefined)) + case isErrVersionNotFound(gerr) || isErrObjectNotFound(gerr): + // No existing version to order against: keep the upload's own accepted + // lock, including a pre-upgrade upload that persisted values without + // their ordering timestamps. + default: + return oi, toObjectErr(gerr, bucket, object) + } + } + uploadIDPath := er.getUploadIDDir(bucket, object, uploadID) onlineDisks := er.getDisks() writeQuorum := fi.WriteQuorum(er.defaultWQuorum()) diff --git a/cmd/erasure-object.go b/cmd/erasure-object.go index db5e77ed2..a6293edc2 100644 --- a/cmd/erasure-object.go +++ b/cmd/erasure-object.go @@ -1268,7 +1268,7 @@ func (er erasureObjects) putObject(ctx context.Context, bucket string, object st data := r.Reader - if opts.CheckPrecondFn != nil { + if opts.CheckPrecondFn != nil || opts.ReplicaLockReconcile { if !opts.NoLock { ns := er.NewNSLock(bucket, object) lkctx, err := ns.GetLock(ctx, globalOperationTimeout) @@ -1281,17 +1281,33 @@ func (er erasureObjects) putObject(ctx context.Context, bucket string, object st } obj, err := er.getObjectInfo(ctx, bucket, object, opts) - if err == nil && opts.CheckPrecondFn(obj) { - return objInfo, PreConditionFailed{} - } + // A destination read that fails for a reason other than not-found must not + // be taken as a passed precondition or as absent lock state. if err != nil && !isErrVersionNotFound(err) && !isErrObjectNotFound(err) { return objInfo, err } + if opts.CheckPrecondFn != nil { + if err == nil && opts.CheckPrecondFn(obj) { + return objInfo, PreConditionFailed{} + } + // if object doesn't exist return error for If-Match conditional requests + // If-None-Match should be allowed to proceed for non-existent objects + if err != nil && opts.HasIfMatch && (isErrObjectNotFound(err) || isErrVersionNotFound(err)) { + return objInfo, err + } + } - // if object doesn't exist return error for If-Match conditional requests - // If-None-Match should be allowed to proceed for non-existent objects - if err != nil && opts.HasIfMatch && (isErrObjectNotFound(err) || isErrVersionNotFound(err)) { - return objInfo, err + // Order this trusted SSE-C replica's Object Lock against the addressed + // version's stored state, read on this erasure set under the write lock, + // so a value that lost the ordering cannot overwrite a newer one committed + // after the handler decided (issue #120). Only reconcile against an + // existing version; on not-found the write's own accepted lock is kept. + // + // Scope: correct for a single erasure set. A multi-pool deployment + // (duplicate versions across pools, ModTime ties, cross-pool lock + // authority) is out of scope and tracked in pgsty/silo#TBD-multipool-lock. + if opts.ReplicaLockReconcile && err == nil { + reconcileStoredObjectLock(opts.UserDefined, storedObjectLockState(obj.UserDefined)) } } diff --git a/cmd/object-api-interface.go b/cmd/object-api-interface.go index f8664310d..2f64ded4f 100644 --- a/cmd/object-api-interface.go +++ b/cmd/object-api-interface.go @@ -99,6 +99,7 @@ type ObjectOptions struct { ReplicationSourceTaggingTimestamp time.Time // set if MinIOSourceTaggingTimestamp received ReplicationSourceLegalholdTimestamp time.Time // set if MinIOSourceObjectLegalholdTimestamp received ReplicationSourceRetentionTimestamp time.Time // set if MinIOSourceObjectRetentionTimestamp received + ReplicaLockReconcile bool // set for a trusted SSE-C replica full write/completion: re-order Object Lock against the destination version read under the write lock (single erasure set; see pgsty/silo#TBD-multipool-lock) DeletePrefix bool // set true to enforce a prefix deletion, only application for DeleteObject API, DeletePrefixObject bool // set true when object's erasure set is resolvable by object name (using getHashedSetIndex) diff --git a/cmd/object-handlers.go b/cmd/object-handlers.go index fcb834821..789d0fe2a 100644 --- a/cmd/object-handlers.go +++ b/cmd/object-handlers.go @@ -1771,50 +1771,9 @@ 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 { - // A replica update is ordered by its source timestamp alone, whether or - // not it still carries a value: an update newer than the stored state - // applies, and a removal is just an update with no value. An update that - // is stale, or that carries no source timestamp at all and is therefore - // unordered, leaves the stored state in place instead of erasing it. - switch { - case !dstOpts.ReplicationRequest: - if retentionMode.Valid() { - srcInfo.UserDefined[strings.ToLower(xhttp.AmzObjectLockMode)] = string(retentionMode) - srcInfo.UserDefined[strings.ToLower(xhttp.AmzObjectLockRetainUntilDate)] = amztime.ISO8601Format(retentionDate.UTC()) - srcInfo.UserDefined[ReservedMetadataPrefixLower+ObjectLockRetentionTimestamp] = UTCNow().Format(time.RFC3339Nano) - } - case !replicaTrusted && !retentionMode.Valid(): - // A trusted peer that did not mark this request as a replica sends - // no replicated state, so a missing value carries no instruction and - // the rebuilt metadata is left as it is. - case !storedLock.retentionIsOlderThan(dstOpts.ReplicationSourceRetentionTimestamp): - storedLock.restoreRetention(srcInfo.UserDefined) - default: - if retentionMode.Valid() { - srcInfo.UserDefined[strings.ToLower(xhttp.AmzObjectLockMode)] = string(retentionMode) - srcInfo.UserDefined[strings.ToLower(xhttp.AmzObjectLockRetainUntilDate)] = amztime.ISO8601Format(retentionDate.UTC()) - } - srcInfo.UserDefined[ReservedMetadataPrefixLower+ObjectLockRetentionTimestamp] = dstOpts.ReplicationSourceRetentionTimestamp.UTC().Format(time.RFC3339Nano) - } - - // Legal hold has no removal in S3: an explicitly empty header is already - // rejected as an invalid status, so the only value-less shape that gets - // here is an absent one, and that conveys no legal-hold change even when - // an orphaned timestamp comes with it. Only a valid status can win. - switch { - case !dstOpts.ReplicationRequest: - if legalHold.Status.Valid() { - srcInfo.UserDefined[strings.ToLower(xhttp.AmzObjectLockLegalHold)] = string(legalHold.Status) - srcInfo.UserDefined[ReservedMetadataPrefixLower+ObjectLockLegalHoldTimestamp] = UTCNow().Format(time.RFC3339Nano) - } - case !replicaTrusted && !legalHold.Status.Valid(): - // As above: a marker-only request carries no legal-hold update. - case legalHold.Status.Valid() && storedLock.legalHoldIsOlderThan(dstOpts.ReplicationSourceLegalholdTimestamp): - srcInfo.UserDefined[strings.ToLower(xhttp.AmzObjectLockLegalHold)] = string(legalHold.Status) - srcInfo.UserDefined[ReservedMetadataPrefixLower+ObjectLockLegalHoldTimestamp] = dstOpts.ReplicationSourceLegalholdTimestamp.UTC().Format(time.RFC3339Nano) - default: - storedLock.restoreLegalHold(srcInfo.UserDefined) - } + applyReplicatedObjectLock(srcInfo.UserDefined, storedLock, replicaTrusted, + retentionMode, retentionDate, legalHold, + dstOpts.ReplicationSourceRetentionTimestamp, dstOpts.ReplicationSourceLegalholdTimestamp) if replicaTrusted { // An SSE-C key rotation snapshots every stored reserved key into @@ -2282,12 +2241,31 @@ func (api objectAPIHandlers) PutObjectHandler(w http.ResponseWriter, r *http.Req getObjectInfo := objectAPI.GetObjectInfo retentionMode, retentionDate, legalHold, s3Err := checkPutObjectLockAllowed(ctx, r, bucket, object, getObjectInfo, retPerms, holdPerms, isReplicaTrusted(ctx)) - if s3Err == ErrNone && retentionMode.Valid() { - metadata[strings.ToLower(xhttp.AmzObjectLockMode)] = string(retentionMode) - metadata[strings.ToLower(xhttp.AmzObjectLockRetainUntilDate)] = amztime.ISO8601Format(retentionDate.UTC()) - } - if s3Err == ErrNone && legalHold.Status.Valid() { - metadata[strings.ToLower(xhttp.AmzObjectLockLegalHold)] = string(legalHold.Status) + if s3Err == ErrNone { + // A trusted replica write addressing a specific version can be a full + // retransmit over an existing version whose lock state is newer than the + // source snapshot (issue #120). Order the incoming update against what is + // stored so a stale value cannot overwrite it; a non-replica write (and a + // marker-only peer write) has no stored state to order against and takes + // the helper's ordinary-write branch. + var storedLock objectLockState + if isReplicaTrusted(ctx) && opts.VersionID != "" { + var lerr error + if storedLock, lerr = replicaStoredLock(ctx, getObjectInfo, bucket, object, opts.VersionID); lerr != nil { + writeErrorResponse(ctx, w, toAPIError(ctx, lerr), r.URL) + return + } + } + applyReplicatedObjectLock(metadata, storedLock, isReplicaTrusted(ctx), + retentionMode, retentionDate, legalHold, + opts.ReplicationSourceRetentionTimestamp, opts.ReplicationSourceLegalholdTimestamp) + // The decision above orders against the version as read here; let the + // object layer re-run it against the version read under the write lock + // that guards the replacement, so a newer hold or retention committed in + // between is not rolled back (issue #120). Scoped to the SSE-C replica + // retransmit this issue enables, keyed on the incoming write's restored + // SSE-C seal, the same predicate as the duplicate-version exemption. + opts.ReplicaLockReconcile = isReplicaTrusted(ctx) && opts.VersionID != "" && crypto.SSEC.IsEncrypted(metadata) } if s3Err != ErrNone { writeErrorResponse(ctx, w, errorCodes.ToAPIErr(s3Err), r.URL) diff --git a/cmd/object-multipart-handlers.go b/cmd/object-multipart-handlers.go index 22d2839f1..983644223 100644 --- a/cmd/object-multipart-handlers.go +++ b/cmd/object-multipart-handlers.go @@ -34,7 +34,6 @@ import ( "github.com/minio/minio-go/v7" "github.com/minio/minio-go/v7/pkg/encrypt" "github.com/minio/minio-go/v7/pkg/tags" - "github.com/minio/minio/internal/amztime" sse "github.com/minio/minio/internal/bucket/encryption" objectlock "github.com/minio/minio/internal/bucket/object/lock" "github.com/minio/minio/internal/bucket/replication" @@ -260,12 +259,32 @@ func (api objectAPIHandlers) NewMultipartUploadHandler(w http.ResponseWriter, r getObjectInfo := objectAPI.GetObjectInfo retentionMode, retentionDate, legalHold, s3Err := checkPutObjectLockAllowed(ctx, r, bucket, object, getObjectInfo, retPerms, holdPerms, replicaTrusted) - if s3Err == ErrNone && retentionMode.Valid() { - metadata[strings.ToLower(xhttp.AmzObjectLockMode)] = string(retentionMode) - metadata[strings.ToLower(xhttp.AmzObjectLockRetainUntilDate)] = amztime.ISO8601Format(retentionDate.UTC()) - } - if s3Err == ErrNone && legalHold.Status.Valid() { - metadata[strings.ToLower(xhttp.AmzObjectLockLegalHold)] = string(legalHold.Status) + if s3Err == ErrNone { + // A trusted replica NewMultipartUpload addressing a specific version can + // be a full retransmit over an existing version whose lock state is newer + // than the source snapshot (issue #120). Order the incoming update against + // what is stored so a stale value cannot overwrite it. opts is built below + // (its ServerSideEncryption depends on the encMetadata merge that has not + // happened yet), so read the replica ordering inputs the way + // putOptsFromHeaders will; a malformed timestamp fails the request when + // opts is built, so a parse error here is left as a zero time. + var ( + storedLock objectLockState + srcRetentionTS, srcLegalholdTS time.Time + ) + if replicaTrusted { + srcRetentionTS, _ = time.Parse(time.RFC3339, strings.TrimSpace(r.Header.Get(xhttp.MinIOSourceObjectRetentionTimestamp))) + srcLegalholdTS, _ = time.Parse(time.RFC3339, strings.TrimSpace(r.Header.Get(xhttp.MinIOSourceObjectLegalHoldTimestamp))) + if versionID := strings.TrimSpace(r.Form.Get(xhttp.VersionID)); versionID != "" { + var lerr error + if storedLock, lerr = replicaStoredLock(ctx, getObjectInfo, bucket, object, versionID); lerr != nil { + writeErrorResponse(ctx, w, toAPIError(ctx, lerr), r.URL) + return + } + } + } + applyReplicatedObjectLock(metadata, storedLock, replicaTrusted, + retentionMode, retentionDate, legalHold, srcRetentionTS, srcLegalholdTS) } if s3Err != ErrNone { writeErrorResponse(ctx, w, errorCodes.ToAPIErr(s3Err), r.URL) @@ -1166,6 +1185,14 @@ func (api objectAPIHandlers) CompleteMultipartUploadHandler(w http.ResponseWrite } opts.Versioned = versioned opts.VersionSuspended = suspended + // A replicated multipart completion carries the internal replication marker + // (the sender does not re-assert REPLICA status on Complete, so this is keyed + // on trusted replication, matching completeMultipartOpts). The object layer + // re-orders the Object Lock it carries against the destination version read + // under the write lock, but only for an SSE-C upload -- the scope this issue + // enables -- so a marker-only non-SSE-C completion keeps ordinary write + // semantics (issue #120). + opts.ReplicaLockReconcile = trustedReplication // First, we compute the ETag of the multipart object. // The ETag of a multi-part object is always: diff --git a/cmd/replication-ssec-retransmit_test.go b/cmd/replication-ssec-retransmit_test.go index e894b8fa0..bbb8712b9 100644 --- a/cmd/replication-ssec-retransmit_test.go +++ b/cmd/replication-ssec-retransmit_test.go @@ -12,6 +12,7 @@ package cmd import ( "bytes" + "context" "crypto/md5" "encoding/base64" "encoding/xml" @@ -792,3 +793,1002 @@ func testAPISSECReplicaWriteExemptionIsKeyedOnTheIncomingWrite(obj ObjectLayer, } }) } + +// Retain-until values in the millisecond form ISO8601Format round-trips to, so +// a value applied through the handler reads back byte-for-byte. +const ( + retransmitRetainUntilNewer = "2031-01-01T00:00:00.000Z" + retransmitRetainUntilStale = "2028-01-01T00:00:00.000Z" +) + +// TestAPISSECReplicaRetransmitObjectLockOrdering proves that the full SSE-C +// replica retransmit orders the Object Lock update it carries against the state +// already stored on the addressed version, the same way the metadata CopyObject +// path does. Before issue #120 routed these writes through PutObjectHandler the +// handler applied the incoming retention and legal hold directly and never +// persisted the ordering timestamps, so a retransmit carrying an older value +// could overwrite a destination version's newer one. See pgsty/silo#120. +func TestAPISSECReplicaRetransmitObjectLockOrdering(t *testing.T) { + defer DetectTestLeak(t)() + ExecObjectLayerAPITest(ExecObjectLayerAPITestArgs{ + t: t, + objAPITest: testAPISSECReplicaRetransmitObjectLockOrdering, + makeBucketOptions: MakeBucketOptions{LockEnabled: true}, + }) +} + +func testAPISSECReplicaRetransmitObjectLockOrdering(obj ObjectLayer, instanceType, bucketName string, + apiRouter http.Handler, credentials auth.Credentials, t *testing.T, +) { + previousTLS := globalIsTLS + globalIsTLS = true + defer func() { globalIsTLS = previousTLS }() + + key := bytes.Repeat([]byte{0x45}, 32) + keyMD5 := md5.Sum(key) + sseHeaders := map[string]string{ + xhttp.AmzServerSideEncryptionCustomerAlgorithm: xhttp.AmzEncryptionAES, + xhttp.AmzServerSideEncryptionCustomerKey: base64.StdEncoding.EncodeToString(key), + xhttp.AmzServerSideEncryptionCustomerKeyMD5: base64.StdEncoding.EncodeToString(keyMD5[:]), + } + + // seed writes a fresh SSE-C source version with no Object Lock state and + // returns its version id. + seed := func(t *testing.T, object string) string { + t.Helper() + putCopyChecksumSource(t, apiRouter, credentials, bucketName, object, + bytes.Repeat([]byte("ssec-lock-ordering-"), 64), sseHeaders) + info, err := obj.GetObjectInfo(t.Context(), bucketName, object, ObjectOptions{}) + if err != nil { + t.Fatal(err) + } + return info.VersionID + } + + // retransmit sends the full SSE-C replica retransmit the sender emits for the + // addressed version, over its raw ciphertext, carrying exactly the given + // Object Lock headers on top of the replica seal. The credential is the admin + // user so the retention and legal-hold permission checks pass; the request is + // still a trusted replica because it carries the marker and REPLICA status. + retransmit := func(t *testing.T, object, versionID string, lockHeaders map[string]string) { + t.Helper() + gr, err := obj.GetObjectNInfo(t.Context(), bucketName, object, nil, http.Header{}, ObjectOptions{ + ReplicationRequest: true, + VersionID: versionID, + }) + if err != nil { + t.Fatal(err) + } + srcInfo := gr.ObjInfo + cipher, err := io.ReadAll(gr) + gr.Close() + if err != nil { + t.Fatal(err) + } + opts, _, err := putReplicationOpts(t.Context(), "", srcInfo) + if err != nil { + t.Fatal(err) + } + opts.Internal.SourceMTime = time.Time{} + hdrs := make(map[string]string) + for name, values := range opts.Header() { + if len(values) > 0 { + hdrs[name] = values[0] + } + } + hdrs[xhttp.MinIOSourceReplicationRequest] = "true" + hdrs[xhttp.AmzBucketReplicationStatus] = "REPLICA" + hdrs[xhttp.MinIOSourceETag] = srcInfo.ETag + // The case owns the lock instruction: drop any lock header the sender + // derived from the source version. + for _, name := range []string{ + xhttp.AmzObjectLockMode, xhttp.AmzObjectLockRetainUntilDate, xhttp.AmzObjectLockLegalHold, + xhttp.MinIOSourceObjectRetentionTimestamp, xhttp.MinIOSourceObjectLegalHoldTimestamp, + } { + delete(hdrs, name) + } + maps.Copy(hdrs, lockHeaders) + + req, err := newTestSignedRequestV4(http.MethodPut, + getPutObjectURL("", bucketName, object)+"?versionId="+versionID, int64(len(cipher)), + bytes.NewReader(cipher), credentials.AccessKey, credentials.SecretKey, hdrs) + if err != nil { + t.Fatal(err) + } + rec := httptest.NewRecorder() + apiRouter.ServeHTTP(rec, req) + if rec.Code != http.StatusOK { + t.Fatalf("%s: replica retransmit PUT status %d, want 200: %s", instanceType, rec.Code, rec.Body.String()) + } + } + + t.Run("older-legal-hold-off-does-not-clear-newer-on", func(t *testing.T) { + object := "ssec-lock-ordering/legal-hold" + versionID := seed(t, object) + + // Establish the newer legal hold ON. Its timestamp persistence is proven + // by the retention sibling test, so here only require the value took + // effect before the older OFF arrives, so the clobber below is what tells + // a fixed handler from a broken one. + retransmit(t, object, versionID, map[string]string{ + xhttp.AmzObjectLockLegalHold: "ON", + xhttp.MinIOSourceObjectLegalHoldTimestamp: objectLockTestStamp1000, + }) + if got := readObjectLockFields(t, obj, bucketName, object, versionID); got.legalHold != "ON" { + t.Fatalf("%s: seeding legal hold ON failed: got %+v", instanceType, got) + } + + // A retransmit carrying an older legal-hold OFF must not clear it. + retransmit(t, object, versionID, map[string]string{ + xhttp.AmzObjectLockLegalHold: "OFF", + xhttp.MinIOSourceObjectLegalHoldTimestamp: objectLockTestStamp0900, + }) + want := objectLockFields{legalHold: "ON", legalHoldStamp: objectLockTestStamp1000} + if got := readObjectLockFields(t, obj, bucketName, object, versionID); got != want { + t.Fatalf("%s: an older legal-hold OFF cleared the newer ON: got %+v, want %+v", instanceType, got, want) + } + }) + + t.Run("newer-retention-applies-and-persists-its-timestamp", func(t *testing.T) { + object := "ssec-lock-ordering/retention-applies" + versionID := seed(t, object) + + retransmit(t, object, versionID, map[string]string{ + xhttp.AmzObjectLockMode: "GOVERNANCE", + xhttp.AmzObjectLockRetainUntilDate: retransmitRetainUntilNewer, + xhttp.MinIOSourceObjectRetentionTimestamp: objectLockTestStamp1000, + }) + // The value applies and, crucially, the ordering timestamp is persisted so + // a later stale update can be recognized as older. + want := objectLockFields{ + mode: "GOVERNANCE", retainUntil: retransmitRetainUntilNewer, retentionStamp: objectLockTestStamp1000, + } + if got := readObjectLockFields(t, obj, bucketName, object, versionID); got != want { + t.Fatalf("%s: newer retention did not apply and persist its timestamp: got %+v, want %+v", instanceType, got, want) + } + }) + + t.Run("stale-retention-update-is-ignored", func(t *testing.T) { + object := "ssec-lock-ordering/retention-stale" + versionID := seed(t, object) + + // Establish the newer retention first; its timestamp persistence is proven + // by the sibling test above, so here only require the value took effect, so + // the stale overwrite below is what tells a fixed handler from a broken one. + retransmit(t, object, versionID, map[string]string{ + xhttp.AmzObjectLockMode: "GOVERNANCE", + xhttp.AmzObjectLockRetainUntilDate: retransmitRetainUntilNewer, + xhttp.MinIOSourceObjectRetentionTimestamp: objectLockTestStamp1000, + }) + if got := readObjectLockFields(t, obj, bucketName, object, versionID); got.mode != "GOVERNANCE" || got.retainUntil != retransmitRetainUntilNewer { + t.Fatalf("%s: seeding the newer retention failed: got %+v", instanceType, got) + } + + // A retransmit carrying an older retention with a different date must be + // ignored; the newer date and its ordering timestamp survive. + retransmit(t, object, versionID, map[string]string{ + xhttp.AmzObjectLockMode: "GOVERNANCE", + xhttp.AmzObjectLockRetainUntilDate: retransmitRetainUntilStale, + xhttp.MinIOSourceObjectRetentionTimestamp: objectLockTestStamp0900, + }) + want := objectLockFields{ + mode: "GOVERNANCE", retainUntil: retransmitRetainUntilNewer, retentionStamp: objectLockTestStamp1000, + } + if got := readObjectLockFields(t, obj, bucketName, object, versionID); got != want { + t.Fatalf("%s: a stale retention update overwrote the newer one: got %+v, want %+v", instanceType, got, want) + } + }) +} + +// TestAPISSECReplicaRetransmitMultipartObjectLockOrdering proves the ordering +// fix also covers the multipart initiation path #120 routes an SSE-C replica +// through: NewMultipartUploadHandler reads the addressed version's stored lock +// state and orders the incoming update against it, so a large-object retransmit +// carrying an older legal-hold OFF cannot clear a newer ON. See pgsty/silo#120. +func TestAPISSECReplicaRetransmitMultipartObjectLockOrdering(t *testing.T) { + defer DetectTestLeak(t)() + ExecObjectLayerAPITest(ExecObjectLayerAPITestArgs{ + t: t, + objAPITest: testAPISSECReplicaRetransmitMultipartObjectLockOrdering, + makeBucketOptions: MakeBucketOptions{LockEnabled: true}, + }) +} + +func testAPISSECReplicaRetransmitMultipartObjectLockOrdering(obj ObjectLayer, instanceType, bucketName string, + apiRouter http.Handler, credentials auth.Credentials, t *testing.T, +) { + previousTLS := globalIsTLS + globalIsTLS = true + defer func() { globalIsTLS = previousTLS }() + + key := bytes.Repeat([]byte{0x46}, 32) + keyMD5 := md5.Sum(key) + sseHeaders := map[string]string{ + xhttp.AmzServerSideEncryptionCustomerAlgorithm: xhttp.AmzEncryptionAES, + xhttp.AmzServerSideEncryptionCustomerKey: base64.StdEncoding.EncodeToString(key), + xhttp.AmzServerSideEncryptionCustomerKeyMD5: base64.StdEncoding.EncodeToString(keyMD5[:]), + } + + // replicaSealHeaders returns the sender's replica seal and marker headers for + // the addressed version, with every Object Lock header stripped so the case + // owns the lock instruction. + replicaSealHeaders := func(t *testing.T, srcInfo ObjectInfo) map[string]string { + t.Helper() + opts, _, err := putReplicationOpts(t.Context(), "", srcInfo) + if err != nil { + t.Fatal(err) + } + opts.Internal.SourceMTime = time.Time{} + hdrs := make(map[string]string) + for name, values := range opts.Header() { + if len(values) > 0 { + hdrs[name] = values[0] + } + } + hdrs[xhttp.MinIOSourceReplicationRequest] = "true" + hdrs[xhttp.AmzBucketReplicationStatus] = "REPLICA" + hdrs[xhttp.MinIOSourceETag] = srcInfo.ETag + for _, name := range []string{ + xhttp.AmzObjectLockMode, xhttp.AmzObjectLockRetainUntilDate, xhttp.AmzObjectLockLegalHold, + xhttp.MinIOSourceObjectRetentionTimestamp, xhttp.MinIOSourceObjectLegalHoldTimestamp, + } { + delete(hdrs, name) + } + return hdrs + } + + object := "ssec-lock-ordering/multipart" + // A single-part SSE-C source is enough; the retransmit re-uploads its bytes + // as one multipart part and commits the addressed version. + putCopyChecksumSource(t, apiRouter, credentials, bucketName, object, + bytes.Repeat([]byte("ssec-multipart-lock-ordering-"), 64), sseHeaders) + base, err := obj.GetObjectInfo(t.Context(), bucketName, object, ObjectOptions{}) + if err != nil { + t.Fatal(err) + } + versionID := base.VersionID + + // Establish the newer legal hold ON through the single-part PUT retransmit. + { + gr, err := obj.GetObjectNInfo(t.Context(), bucketName, object, nil, http.Header{}, ObjectOptions{ + ReplicationRequest: true, VersionID: versionID, + }) + if err != nil { + t.Fatal(err) + } + srcInfo := gr.ObjInfo + cipher, rerr := io.ReadAll(gr) + gr.Close() + if rerr != nil { + t.Fatal(rerr) + } + hdrs := replicaSealHeaders(t, srcInfo) + hdrs[xhttp.AmzObjectLockLegalHold] = "ON" + hdrs[xhttp.MinIOSourceObjectLegalHoldTimestamp] = objectLockTestStamp1000 + req, rerr := newTestSignedRequestV4(http.MethodPut, + getPutObjectURL("", bucketName, object)+"?versionId="+versionID, int64(len(cipher)), + bytes.NewReader(cipher), credentials.AccessKey, credentials.SecretKey, hdrs) + if rerr != nil { + t.Fatal(rerr) + } + rec := httptest.NewRecorder() + apiRouter.ServeHTTP(rec, req) + if rec.Code != http.StatusOK { + t.Fatalf("%s: seeding legal hold ON via PUT status %d: %s", instanceType, rec.Code, rec.Body.String()) + } + if got := readObjectLockFields(t, obj, bucketName, object, versionID); got.legalHold != "ON" { + t.Fatalf("%s: seeding legal hold ON failed: got %+v", instanceType, got) + } + } + + // A multipart retransmit carrying an older legal-hold OFF must not clear it. + gr, err := obj.GetObjectNInfo(t.Context(), bucketName, object, nil, http.Header{}, ObjectOptions{ + ReplicationRequest: true, VersionID: versionID, + }) + if err != nil { + t.Fatal(err) + } + srcInfo := gr.ObjInfo + cipher, err := io.ReadAll(gr) + gr.Close() + if err != nil { + t.Fatal(err) + } + actualSize, err := srcInfo.GetActualSize() + if err != nil { + t.Fatal(err) + } + + newHdrs := replicaSealHeaders(t, srcInfo) + newHdrs[xhttp.AmzObjectLockLegalHold] = "OFF" + newHdrs[xhttp.MinIOSourceObjectLegalHoldTimestamp] = objectLockTestStamp0900 + newReq, err := newTestSignedRequestV4(http.MethodPost, + getNewMultipartURL("", bucketName, object)+"&versionId="+versionID, 0, nil, + credentials.AccessKey, credentials.SecretKey, newHdrs) + if err != nil { + t.Fatal(err) + } + newRec := httptest.NewRecorder() + apiRouter.ServeHTTP(newRec, newReq) + if newRec.Code != http.StatusOK { + t.Fatalf("%s: replica NewMultipartUpload status %d: %s", instanceType, newRec.Code, newRec.Body.String()) + } + var init InitiateMultipartUploadResponse + if err = xmlDecoder(newRec.Body, &init, int64(newRec.Body.Len())); err != nil { + t.Fatal(err) + } + + partReq, err := newTestSignedRequestV4(http.MethodPut, + getPutObjectPartURL("", bucketName, object, init.UploadID, "1"), int64(len(cipher)), + bytes.NewReader(cipher), credentials.AccessKey, credentials.SecretKey, + map[string]string{xhttp.MinIOSourceReplicationRequest: "true"}) + if err != nil { + t.Fatal(err) + } + partRec := httptest.NewRecorder() + apiRouter.ServeHTTP(partRec, partReq) + if partRec.Code != http.StatusOK { + t.Fatalf("%s: replica PutObjectPart status %d: %s", instanceType, partRec.Code, partRec.Body.String()) + } + + completeBody, err := xml.Marshal(CompleteMultipartUpload{Parts: []CompletePart{ + {PartNumber: 1, ETag: canonicalizeETag(partRec.Header()[xhttp.ETag][0])}, + }}) + if err != nil { + t.Fatal(err) + } + completeReq, err := newTestSignedRequestV4(http.MethodPost, + getCompleteMultipartUploadURL("", bucketName, object, init.UploadID), int64(len(completeBody)), + bytes.NewReader(completeBody), credentials.AccessKey, credentials.SecretKey, map[string]string{ + xhttp.MinIOSourceReplicationRequest: "true", + xhttp.MinIOSourceMTime: srcInfo.ModTime.Format(time.RFC3339Nano), + xhttp.MinIOSourceETag: srcInfo.ETag, + xhttp.MinIOReplicationActualObjectSize: strconv.FormatInt(actualSize, 10), + }) + if err != nil { + t.Fatal(err) + } + completeRec := httptest.NewRecorder() + apiRouter.ServeHTTP(completeRec, completeReq) + if completeRec.Code != http.StatusOK { + t.Fatalf("%s: replica CompleteMultipartUpload status %d: %s", instanceType, completeRec.Code, completeRec.Body.String()) + } + + want := objectLockFields{legalHold: "ON", legalHoldStamp: objectLockTestStamp1000} + if got := readObjectLockFields(t, obj, bucketName, object, versionID); got != want { + t.Fatalf("%s: an older legal-hold OFF via multipart cleared the newer ON: got %+v, want %+v", instanceType, got, want) + } +} + +// TestReplicaStoredLock verifies how a replica write reads the destination +// version's stored lock state before ordering its update: a present version +// yields its state, a missing object or version yields an empty state so a first +// write is not blocked, and any other read error (a quorum loss, a timeout) is +// returned so the caller fails the write instead of ordering an incoming update +// against lock state it merely could not read. See pgsty/silo#120. +func TestReplicaStoredLock(t *testing.T) { + fixed := func(oi ObjectInfo, err error) GetObjectInfoFn { + return func(context.Context, string, string, ObjectOptions) (ObjectInfo, error) { return oi, err } + } + stored := ObjectInfo{UserDefined: map[string]string{ + strings.ToLower(xhttp.AmzObjectLockLegalHold): "ON", + ReservedMetadataPrefixLower + ObjectLockLegalHoldTimestamp: objectLockTestStamp1000, + }} + + t.Run("present-version-returns-its-state", func(t *testing.T) { + got, err := replicaStoredLock(context.Background(), fixed(stored, nil), "b", "o", "v") + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if got.legalHold != "ON" || got.legalHoldTimestamp != objectLockTestStamp1000 { + t.Fatalf("stored lock state = %+v, want legal hold ON stamped %q", got, objectLockTestStamp1000) + } + }) + + for name, notFound := range map[string]error{ + "object-not-found": ObjectNotFound{Bucket: "b", Object: "o"}, + "version-not-found": VersionNotFound{Bucket: "b", Object: "o", VersionID: "v"}, + } { + t.Run(name+"-is-empty-state", func(t *testing.T) { + got, err := replicaStoredLock(context.Background(), fixed(ObjectInfo{}, notFound), "b", "o", "v") + if err != nil { + t.Fatalf("a not-found read must not error: %v", err) + } + if got != (objectLockState{}) { + t.Fatalf("a not-found read must yield empty state, got %+v", got) + } + }) + } + + t.Run("transient-read-error-propagates", func(t *testing.T) { + boom := InsufficientReadQuorum{} + got, err := replicaStoredLock(context.Background(), fixed(ObjectInfo{}, boom), "b", "o", "v") + if err == nil { + t.Fatal("a transient read error must propagate, not be treated as absent lock state") + } + if isErrObjectNotFound(err) || isErrVersionNotFound(err) { + t.Fatalf("transient error misclassified as not-found: %v", err) + } + if got != (objectLockState{}) { + t.Fatalf("on a read error the caller must get empty state and fail the write, got %+v", got) + } + }) +} + +// TestPutReplicationOptsRetentionRemovalTimestampOnly asserts that a version +// whose retention was removed on the retransmit PUT path -- stored as an +// ordering timestamp with the value keys absent, not empty -- still builds +// replication options that carry the removal timestamp, so the next hop can +// order the removal instead of keeping obsolete retention. See pgsty/silo#120. +func TestPutReplicationOptsRetentionRemovalTimestampOnly(t *testing.T) { + removedAt := time.Date(2026, 9, 5, 10, 0, 0, 0, time.UTC) + oi := ObjectInfo{ + Bucket: "b", Name: "o", VersionID: "v1", ModTime: removedAt.Add(-time.Hour), + UserDefined: map[string]string{ + // Only the reserved ordering timestamp; no mode/date keys at all. + ReservedMetadataPrefixLower + ObjectLockRetentionTimestamp: removedAt.Format(time.RFC3339Nano), + }, + } + opts, _, err := putReplicationOpts(t.Context(), "", oi) + if err != nil { + t.Fatalf("putReplicationOpts on a timestamp-only removal: %v", err) + } + if opts.Mode != "" || !opts.RetainUntilDate.IsZero() { + t.Errorf("removal sent as a retention: mode %q date %v", opts.Mode, opts.RetainUntilDate) + } + if !opts.Internal.RetentionTimestamp.Equal(removedAt) { + t.Errorf("removal timestamp %v, want %v", opts.Internal.RetentionTimestamp, removedAt) + } + if hdr := opts.Header(); hdr.Get(xhttp.AmzObjectLockMode) != "" || hdr.Get(xhttp.AmzObjectLockRetainUntilDate) != "" || + hdr.Get(xhttp.MinIOSourceObjectRetentionTimestamp) == "" { + t.Errorf("removal headers %v: want no lock value and a retention timestamp", hdr) + } +} + +// TestAPIReplicaMarkerOnlyAppliesObjectLock guards the regression the shared +// ordering helper could introduce: a trusted peer write that carries the +// internal replication marker but NOT REPLICA status (replicationRequest true, +// replicaTrusted false) must keep ordinary write semantics and apply its +// validated Object Lock, not restore an empty stored state. It covers an +// explicit legal hold and a bucket-default retention, on both the PUT and the +// multipart-initiation paths. See pgsty/silo#120 (Codex finding 3). +func TestAPIReplicaMarkerOnlyAppliesObjectLock(t *testing.T) { + defer DetectTestLeak(t)() + ExecObjectLayerAPITest(ExecObjectLayerAPITestArgs{ + t: t, + objAPITest: testAPIReplicaMarkerOnlyAppliesObjectLock, + makeBucketOptions: MakeBucketOptions{LockEnabled: true}, + }) +} + +func testAPIReplicaMarkerOnlyAppliesObjectLock(obj ObjectLayer, instanceType, bucketName string, + apiRouter http.Handler, credentials auth.Credentials, t *testing.T, +) { + // A trusted peer credential holding ReplicateObject plus the lock permissions + // checkPutObjectLockAllowed enforces even for a marker-only write. + peer := newObjectAttributesAuthzUser(t, instanceType, bucketName, + `"s3:GetObject","s3:PutObject","s3:ReplicateObject","s3:PutObjectRetention","s3:PutObjectLegalHold"`) + + // Bucket default retention, so a marker-only write with no lock headers still + // has a validated retention to apply. + lockCfg := []byte(`EnabledGOVERNANCE30`) + if _, err := globalBucketMetadataSys.Update(t.Context(), bucketName, objectLockConfig, lockCfg); err != nil { + t.Fatalf("%s: configure bucket default retention: %v", instanceType, err) + } + + // markerOnly returns the trusted-but-not-REPLICA header set plus extra: the + // internal marker with no REPLICA status. + markerOnly := func(extra map[string]string) map[string]string { + h := map[string]string{xhttp.MinIOSourceReplicationRequest: "true"} + maps.Copy(h, extra) + return h + } + data := bytes.Repeat([]byte("marker-only-lock-"), 32) + + markerOnlyMPU := func(t *testing.T, object string, lockHeaders map[string]string) { + t.Helper() + newReq, err := newTestSignedRequestV4(http.MethodPost, getNewMultipartURL("", bucketName, object), 0, nil, + peer.AccessKey, peer.SecretKey, markerOnly(lockHeaders)) + if err != nil { + t.Fatal(err) + } + newRec := httptest.NewRecorder() + apiRouter.ServeHTTP(newRec, newReq) + if newRec.Code != http.StatusOK { + t.Fatalf("%s: marker-only NewMultipartUpload %d: %s", instanceType, newRec.Code, newRec.Body.String()) + } + var init InitiateMultipartUploadResponse + if err = xmlDecoder(newRec.Body, &init, int64(newRec.Body.Len())); err != nil { + t.Fatal(err) + } + partReq, err := newTestSignedRequestV4(http.MethodPut, + getPutObjectPartURL("", bucketName, object, init.UploadID, "1"), int64(len(data)), + bytes.NewReader(data), peer.AccessKey, peer.SecretKey, nil) + if err != nil { + t.Fatal(err) + } + partRec := httptest.NewRecorder() + apiRouter.ServeHTTP(partRec, partReq) + if partRec.Code != http.StatusOK { + t.Fatalf("%s: marker-only PutObjectPart %d: %s", instanceType, partRec.Code, partRec.Body.String()) + } + body, err := xml.Marshal(CompleteMultipartUpload{Parts: []CompletePart{ + {PartNumber: 1, ETag: canonicalizeETag(partRec.Header()[xhttp.ETag][0])}, + }}) + if err != nil { + t.Fatal(err) + } + completeReq, err := newTestSignedRequestV4(http.MethodPost, + getCompleteMultipartUploadURL("", bucketName, object, init.UploadID), int64(len(body)), + bytes.NewReader(body), peer.AccessKey, peer.SecretKey, nil) + if err != nil { + t.Fatal(err) + } + completeRec := httptest.NewRecorder() + apiRouter.ServeHTTP(completeRec, completeReq) + if completeRec.Code != http.StatusOK { + t.Fatalf("%s: marker-only CompleteMultipartUpload %d: %s", instanceType, completeRec.Code, completeRec.Body.String()) + } + } + markerOnlyPUT := func(t *testing.T, object string, lockHeaders map[string]string) { + t.Helper() + req, err := newTestSignedRequestV4(http.MethodPut, getPutObjectURL("", bucketName, object), int64(len(data)), + bytes.NewReader(data), peer.AccessKey, peer.SecretKey, markerOnly(lockHeaders)) + if err != nil { + t.Fatal(err) + } + rec := httptest.NewRecorder() + apiRouter.ServeHTTP(rec, req) + if rec.Code != http.StatusOK { + t.Fatalf("%s: marker-only PUT %d: %s", instanceType, rec.Code, rec.Body.String()) + } + } + lockOf := func(t *testing.T, object string) objectLockState { + t.Helper() + info, err := obj.GetObjectInfo(t.Context(), bucketName, object, ObjectOptions{}) + if err != nil { + t.Fatal(err) + } + return storedObjectLockState(info.UserDefined) + } + + // An explicit legal hold on a marker-only write must be applied, not dropped. + // A legal-hold request suppresses the bucket default retention, so the version + // carries only the hold. + explicitHold := map[string]string{xhttp.AmzObjectLockLegalHold: "ON"} + + t.Run("put-explicit-legal-hold", func(t *testing.T) { + object := "marker-only/put-legal-hold" + markerOnlyPUT(t, object, explicitHold) + if got := lockOf(t, object); got.legalHold != "ON" { + t.Fatalf("%s: marker-only PUT dropped the explicit legal hold: got %+v", instanceType, got) + } + }) + t.Run("mpu-explicit-legal-hold", func(t *testing.T) { + object := "marker-only/mpu-legal-hold" + markerOnlyMPU(t, object, explicitHold) + if got := lockOf(t, object); got.legalHold != "ON" { + t.Fatalf("%s: marker-only multipart dropped the explicit legal hold: got %+v", instanceType, got) + } + }) + t.Run("put-bucket-default-retention", func(t *testing.T) { + object := "marker-only/put-default-retention" + markerOnlyPUT(t, object, nil) + if got := lockOf(t, object); got.mode != "GOVERNANCE" || got.retainUntil == "" { + t.Fatalf("%s: marker-only PUT dropped the bucket default retention: got %+v", instanceType, got) + } + }) + t.Run("mpu-bucket-default-retention", func(t *testing.T) { + object := "marker-only/mpu-default-retention" + markerOnlyMPU(t, object, nil) + if got := lockOf(t, object); got.mode != "GOVERNANCE" || got.retainUntil == "" { + t.Fatalf("%s: marker-only multipart dropped the bucket default retention: got %+v", instanceType, got) + } + }) +} + +// TestAPIReplicaMultipartNewerHoldSurvivesCompletion verifies that a legal hold +// that reaches a destination version AFTER a replica multipart upload was +// initiated is not rolled back when that upload completes. The initiation +// resolves the lock against the version as it then stands, but completion +// re-orders the carried lock against the version read under the namespace write +// lock that guards the replacement, so a newer hold (with its newer timestamp) +// survives. See pgsty/silo#120 (Codex finding 1). +func TestAPIReplicaMultipartNewerHoldSurvivesCompletion(t *testing.T) { + defer DetectTestLeak(t)() + ExecObjectLayerAPITest(ExecObjectLayerAPITestArgs{ + t: t, + objAPITest: testAPIReplicaMultipartNewerHoldSurvivesCompletion, + makeBucketOptions: MakeBucketOptions{LockEnabled: true}, + }) +} + +func testAPIReplicaMultipartNewerHoldSurvivesCompletion(obj ObjectLayer, instanceType, bucketName string, + apiRouter http.Handler, credentials auth.Credentials, t *testing.T, +) { + previousTLS := globalIsTLS + globalIsTLS = true + defer func() { globalIsTLS = previousTLS }() + + key := bytes.Repeat([]byte{0x47}, 32) + keyMD5 := md5.Sum(key) + sseHeaders := map[string]string{ + xhttp.AmzServerSideEncryptionCustomerAlgorithm: xhttp.AmzEncryptionAES, + xhttp.AmzServerSideEncryptionCustomerKey: base64.StdEncoding.EncodeToString(key), + xhttp.AmzServerSideEncryptionCustomerKeyMD5: base64.StdEncoding.EncodeToString(keyMD5[:]), + } + object := "mpu-lock-boundary/obj" + + // seal returns the sender's SSE-C replica seal and marker headers for the + // addressed version, with every Object Lock header stripped so the case owns + // the lock instruction. + seal := func(t *testing.T, srcInfo ObjectInfo) map[string]string { + t.Helper() + opts, _, err := putReplicationOpts(t.Context(), "", srcInfo) + if err != nil { + t.Fatal(err) + } + opts.Internal.SourceMTime = time.Time{} + hdrs := make(map[string]string) + for name, values := range opts.Header() { + if len(values) > 0 { + hdrs[name] = values[0] + } + } + hdrs[xhttp.MinIOSourceReplicationRequest] = "true" + hdrs[xhttp.AmzBucketReplicationStatus] = "REPLICA" + hdrs[xhttp.MinIOSourceETag] = srcInfo.ETag + for _, name := range []string{ + xhttp.AmzObjectLockMode, xhttp.AmzObjectLockRetainUntilDate, xhttp.AmzObjectLockLegalHold, + xhttp.MinIOSourceObjectRetentionTimestamp, xhttp.MinIOSourceObjectLegalHoldTimestamp, + } { + delete(hdrs, name) + } + return hdrs + } + rawOf := func(t *testing.T, versionID string) (ObjectInfo, []byte) { + t.Helper() + gr, err := obj.GetObjectNInfo(t.Context(), bucketName, object, nil, http.Header{}, ObjectOptions{ + ReplicationRequest: true, VersionID: versionID, + }) + if err != nil { + t.Fatal(err) + } + srcInfo := gr.ObjInfo + cipher, rerr := io.ReadAll(gr) + gr.Close() + if rerr != nil { + t.Fatal(rerr) + } + return srcInfo, cipher + } + + // A destination SSE-C version with no lock. + putCopyChecksumSource(t, apiRouter, credentials, bucketName, object, + bytes.Repeat([]byte("mpu-lock-boundary-"), 64), sseHeaders) + base, err := obj.GetObjectInfo(t.Context(), bucketName, object, ObjectOptions{}) + if err != nil { + t.Fatal(err) + } + versionID := base.VersionID + srcInfo, cipher := rawOf(t, versionID) + + // Initiate an SSE-C replica multipart carrying legal hold OFF stamped 09:00. + // The destination has no lock yet, so the decision made now is OFF@09:00. + newHdrs := seal(t, srcInfo) + newHdrs[xhttp.AmzObjectLockLegalHold] = "OFF" + newHdrs[xhttp.MinIOSourceObjectLegalHoldTimestamp] = objectLockTestStamp0900 + newReq, err := newTestSignedRequestV4(http.MethodPost, + getNewMultipartURL("", bucketName, object)+"&versionId="+versionID, 0, nil, + credentials.AccessKey, credentials.SecretKey, newHdrs) + if err != nil { + t.Fatal(err) + } + newRec := httptest.NewRecorder() + apiRouter.ServeHTTP(newRec, newReq) + if newRec.Code != http.StatusOK { + t.Fatalf("%s: replica NewMultipartUpload %d: %s", instanceType, newRec.Code, newRec.Body.String()) + } + var init InitiateMultipartUploadResponse + if err = xmlDecoder(newRec.Body, &init, int64(newRec.Body.Len())); err != nil { + t.Fatal(err) + } + + // After initiation, a newer legal hold ON@10:00 lands on the same version + // through an independent SSE-C replica PUT retransmit. + putHdrs := seal(t, srcInfo) + putHdrs[xhttp.AmzObjectLockLegalHold] = "ON" + putHdrs[xhttp.MinIOSourceObjectLegalHoldTimestamp] = objectLockTestStamp1000 + putReq, err := newTestSignedRequestV4(http.MethodPut, + getPutObjectURL("", bucketName, object)+"?versionId="+versionID, int64(len(cipher)), + bytes.NewReader(cipher), credentials.AccessKey, credentials.SecretKey, putHdrs) + if err != nil { + t.Fatal(err) + } + putRec := httptest.NewRecorder() + apiRouter.ServeHTTP(putRec, putReq) + if putRec.Code != http.StatusOK { + t.Fatalf("%s: interleaving replica PUT %d: %s", instanceType, putRec.Code, putRec.Body.String()) + } + if got := readObjectLockFields(t, obj, bucketName, object, versionID); got.legalHold != "ON" { + t.Fatalf("%s: the interleaving PUT did not set the newer ON: got %+v", instanceType, got) + } + + // Finish the multipart upload the sender's way and prove the newer ON survives. + partReq, err := newTestSignedRequestV4(http.MethodPut, + getPutObjectPartURL("", bucketName, object, init.UploadID, "1"), int64(len(cipher)), + bytes.NewReader(cipher), credentials.AccessKey, credentials.SecretKey, + map[string]string{xhttp.MinIOSourceReplicationRequest: "true"}) + if err != nil { + t.Fatal(err) + } + partRec := httptest.NewRecorder() + apiRouter.ServeHTTP(partRec, partReq) + if partRec.Code != http.StatusOK { + t.Fatalf("%s: replica PutObjectPart %d: %s", instanceType, partRec.Code, partRec.Body.String()) + } + actualSize, err := srcInfo.GetActualSize() + if err != nil { + t.Fatal(err) + } + body, err := xml.Marshal(CompleteMultipartUpload{Parts: []CompletePart{ + {PartNumber: 1, ETag: canonicalizeETag(partRec.Header()[xhttp.ETag][0])}, + }}) + if err != nil { + t.Fatal(err) + } + completeReq, err := newTestSignedRequestV4(http.MethodPost, + getCompleteMultipartUploadURL("", bucketName, object, init.UploadID), int64(len(body)), + bytes.NewReader(body), credentials.AccessKey, credentials.SecretKey, map[string]string{ + xhttp.MinIOSourceReplicationRequest: "true", + xhttp.MinIOSourceMTime: srcInfo.ModTime.Format(time.RFC3339Nano), + xhttp.MinIOSourceETag: srcInfo.ETag, + xhttp.MinIOReplicationActualObjectSize: strconv.FormatInt(actualSize, 10), + }) + if err != nil { + t.Fatal(err) + } + completeRec := httptest.NewRecorder() + apiRouter.ServeHTTP(completeRec, completeReq) + if completeRec.Code != http.StatusOK { + t.Fatalf("%s: replica CompleteMultipartUpload %d: %s", instanceType, completeRec.Code, completeRec.Body.String()) + } + + // The completion re-ordered the carried OFF@09:00 against the ON@10:00 that + // reached the version after initiation: the newer hold and its timestamp win. + want := objectLockFields{legalHold: "ON", legalHoldStamp: objectLockTestStamp1000} + if got := readObjectLockFields(t, obj, bucketName, object, versionID); got != want { + t.Fatalf("%s: a newer legal hold that arrived after initiation was rolled back at completion: got %+v, want %+v", + instanceType, got, want) + } +} + +// TestReplicaPutObjectLockReconcileUnderWriteLock exercises the PUT counterpart +// of the multipart reconcile: a replica full write reaches the object layer +// carrying the Object Lock its handler resolved, but the addressed version has +// since taken a newer lock update. PutObject must re-order the incoming lock +// against the version read under the write lock, so the newer stored value is +// kept. Driving the object layer directly stands in for the handler-read / +// backend-commit interleave without a timing race. See pgsty/silo#120. +func TestReplicaPutObjectLockReconcileUnderWriteLock(t *testing.T) { + defer DetectTestLeak(t)() + ExecObjectLayerAPITest(ExecObjectLayerAPITestArgs{ + t: t, + objAPITest: testReplicaPutObjectLockReconcileUnderWriteLock, + makeBucketOptions: MakeBucketOptions{LockEnabled: true}, + }) +} + +func testReplicaPutObjectLockReconcileUnderWriteLock(obj ObjectLayer, instanceType, bucketName string, + _ http.Handler, _ auth.Credentials, t *testing.T, +) { + ctx := t.Context() + + seedVersion := func(t *testing.T, object string, meta map[string]string) string { + t.Helper() + info, err := obj.PutObject(ctx, bucketName, object, + mustGetPutObjReader(t, bytes.NewReader([]byte("data")), 4, "", ""), + ObjectOptions{Versioned: true, UserDefined: meta}) + if err != nil { + t.Fatal(err) + } + return info.VersionID + } + // replicaWrite overwrites the addressed version the way a replica retransmit + // reaches the object layer, with the in-lock reconcile enabled. + replicaWrite := func(t *testing.T, object, versionID string, meta map[string]string) { + t.Helper() + _, err := obj.PutObject(ctx, bucketName, object, + mustGetPutObjReader(t, bytes.NewReader([]byte("data")), 4, "", ""), + ObjectOptions{Versioned: true, VersionID: versionID, ReplicaLockReconcile: true, UserDefined: meta}) + if err != nil { + t.Fatal(err) + } + } + + t.Run("older-legal-hold-off-does-not-clear-newer-on", func(t *testing.T) { + object := "reconcile/put-legal-hold" + versionID := seedVersion(t, object, map[string]string{ + strings.ToLower(xhttp.AmzObjectLockLegalHold): "ON", + ReservedMetadataPrefixLower + ObjectLockLegalHoldTimestamp: objectLockTestStamp1000, + }) + replicaWrite(t, object, versionID, map[string]string{ + strings.ToLower(xhttp.AmzObjectLockLegalHold): "OFF", + ReservedMetadataPrefixLower + ObjectLockLegalHoldTimestamp: objectLockTestStamp0900, + }) + want := objectLockFields{legalHold: "ON", legalHoldStamp: objectLockTestStamp1000} + if got := readObjectLockFields(t, obj, bucketName, object, versionID); got != want { + t.Fatalf("%s: in-lock reconcile let an older OFF overwrite the newer stored ON: got %+v, want %+v", + instanceType, got, want) + } + }) + + t.Run("stale-retention-does-not-overwrite-newer", func(t *testing.T) { + object := "reconcile/put-retention" + versionID := seedVersion(t, object, map[string]string{ + strings.ToLower(xhttp.AmzObjectLockMode): "GOVERNANCE", + strings.ToLower(xhttp.AmzObjectLockRetainUntilDate): retransmitRetainUntilNewer, + ReservedMetadataPrefixLower + ObjectLockRetentionTimestamp: objectLockTestStamp1000, + }) + replicaWrite(t, object, versionID, map[string]string{ + strings.ToLower(xhttp.AmzObjectLockMode): "GOVERNANCE", + strings.ToLower(xhttp.AmzObjectLockRetainUntilDate): retransmitRetainUntilStale, + ReservedMetadataPrefixLower + ObjectLockRetentionTimestamp: objectLockTestStamp0900, + }) + want := objectLockFields{ + mode: "GOVERNANCE", retainUntil: retransmitRetainUntilNewer, retentionStamp: objectLockTestStamp1000, + } + if got := readObjectLockFields(t, obj, bucketName, object, versionID); got != want { + t.Fatalf("%s: in-lock reconcile let a stale retention overwrite the newer stored one: got %+v, want %+v", + instanceType, got, want) + } + }) + + t.Run("newer-incoming-hold-applies", func(t *testing.T) { + object := "reconcile/put-newer-applies" + versionID := seedVersion(t, object, map[string]string{ + strings.ToLower(xhttp.AmzObjectLockLegalHold): "OFF", + ReservedMetadataPrefixLower + ObjectLockLegalHoldTimestamp: objectLockTestStamp0900, + }) + replicaWrite(t, object, versionID, map[string]string{ + strings.ToLower(xhttp.AmzObjectLockLegalHold): "ON", + ReservedMetadataPrefixLower + ObjectLockLegalHoldTimestamp: objectLockTestStamp1000, + }) + want := objectLockFields{legalHold: "ON", legalHoldStamp: objectLockTestStamp1000} + if got := readObjectLockFields(t, obj, bucketName, object, versionID); got != want { + t.Fatalf("%s: in-lock reconcile did not apply the newer incoming hold: got %+v, want %+v", + instanceType, got, want) + } + }) + + t.Run("pre-upgrade-shape-on-absent-version-is-preserved", func(t *testing.T) { + // A pre-upgrade upload persisted validated lock values WITHOUT their + // ordering timestamps. Completing it while the destination version is + // absent must keep those values: there is nothing to order against, so the + // reconcile is skipped rather than deleting the accepted lock. The same + // not-found handling guards CompleteMultipartUpload. + object := "reconcile/put-absent-version" + absentVersion := mustGetUUID() + _, err := obj.PutObject(ctx, bucketName, object, + mustGetPutObjReader(t, bytes.NewReader([]byte("data")), 4, "", ""), + ObjectOptions{Versioned: true, VersionID: absentVersion, ReplicaLockReconcile: true, UserDefined: map[string]string{ + strings.ToLower(xhttp.AmzObjectLockMode): "GOVERNANCE", + strings.ToLower(xhttp.AmzObjectLockRetainUntilDate): retransmitRetainUntilNewer, + strings.ToLower(xhttp.AmzObjectLockLegalHold): "ON", + }}) + if err != nil { + t.Fatal(err) + } + want := objectLockFields{mode: "GOVERNANCE", retainUntil: retransmitRetainUntilNewer, legalHold: "ON"} + if got := readObjectLockFields(t, obj, bucketName, object, absentVersion); got != want { + t.Fatalf("%s: a pre-upgrade lock (no ordering timestamps) on an absent version was stripped: got %+v, want %+v", + instanceType, got, want) + } + }) +} + +// TestReplicaLockReconcileNullVersion covers the null version, which persisted +// upload metadata records as an empty VersionID. The completion reconcile must +// order the incoming lock against the null version's own stored state -- looked +// up as the null version, not the latest -- so a retransmit addressing the null +// version cannot be reconciled against an unrelated UUID version, and a null +// version absent while a UUID version exists is not mistaken for present. Single +// erasure set. See pgsty/silo#120. +func TestReplicaLockReconcileNullVersion(t *testing.T) { + defer DetectTestLeak(t)() + ExecObjectLayerAPITest(ExecObjectLayerAPITestArgs{ + t: t, + objAPITest: testReplicaLockReconcileNullVersion, + makeBucketOptions: MakeBucketOptions{VersioningEnabled: true}, + }) +} + +func testReplicaLockReconcileNullVersion(obj ObjectLayer, instanceType, bucketName string, + _ http.Handler, _ auth.Credentials, t *testing.T, +) { + ctx := t.Context() + + // completeNullMPU runs an SSE-C replica multipart upload addressing the null + // version (VersionSuspended) carrying uploadLock, through the reconcile. + completeNullMPU := func(t *testing.T, object string, uploadLock map[string]string) { + t.Helper() + meta := map[string]string{crypto.MetaSealedKeySSEC: "dummy-sealed-key"} + maps.Copy(meta, uploadLock) + res, err := obj.NewMultipartUpload(ctx, bucketName, object, ObjectOptions{VersionSuspended: true, UserDefined: meta}) + if err != nil { + t.Fatal(err) + } + part, err := obj.PutObjectPart(ctx, bucketName, object, res.UploadID, 1, + mustGetPutObjReader(t, bytes.NewReader([]byte("data")), 4, "", ""), ObjectOptions{}) + if err != nil { + t.Fatal(err) + } + if _, err := obj.CompleteMultipartUpload(ctx, bucketName, object, res.UploadID, + []CompletePart{{PartNumber: 1, ETag: part.ETag}}, + ObjectOptions{VersionSuspended: true, ReplicaLockReconcile: true}); err != nil { + t.Fatal(err) + } + } + + t.Run("null-and-uuid-present-reconciles-the-null-version", func(t *testing.T) { + object := "reconcile/null-vs-uuid" + // The null version holds a newer legal hold ON@10:00. + if _, err := obj.PutObject(ctx, bucketName, object, + mustGetPutObjReader(t, bytes.NewReader([]byte("data")), 4, "", ""), + ObjectOptions{VersionSuspended: true, MTime: time.Date(2026, 9, 3, 10, 0, 0, 0, time.UTC), UserDefined: map[string]string{ + strings.ToLower(xhttp.AmzObjectLockLegalHold): "ON", + ReservedMetadataPrefixLower + ObjectLockLegalHoldTimestamp: objectLockTestStamp1000, + }}); err != nil { + t.Fatal(err) + } + // A later UUID version holds an unrelated OFF@11:00 and is the latest. + uuidInfo, err := obj.PutObject(ctx, bucketName, object, + mustGetPutObjReader(t, bytes.NewReader([]byte("data")), 4, "", ""), + ObjectOptions{Versioned: true, MTime: time.Date(2026, 9, 3, 11, 0, 0, 0, time.UTC), UserDefined: map[string]string{ + strings.ToLower(xhttp.AmzObjectLockLegalHold): "OFF", + ReservedMetadataPrefixLower + ObjectLockLegalHoldTimestamp: objectLockTestStamp1100, + }}) + if err != nil { + t.Fatal(err) + } + + // A null-version SSE-C retransmit carrying an older OFF@09:00. + completeNullMPU(t, object, map[string]string{ + strings.ToLower(xhttp.AmzObjectLockLegalHold): "OFF", + ReservedMetadataPrefixLower + ObjectLockLegalHoldTimestamp: objectLockTestStamp0900, + }) + + // The null version keeps its own newer ON@10:00; the UUID is untouched. + wantNull := objectLockFields{legalHold: "ON", legalHoldStamp: objectLockTestStamp1000} + if got := readObjectLockFields(t, obj, bucketName, object, nullVersionID); got != wantNull { + t.Fatalf("%s: null-version completion reconciled against the wrong version: got %+v, want %+v", instanceType, got, wantNull) + } + wantUUID := objectLockFields{legalHold: "OFF", legalHoldStamp: objectLockTestStamp1100} + if got := readObjectLockFields(t, obj, bucketName, object, uuidInfo.VersionID); got != wantUUID { + t.Fatalf("%s: the UUID version was changed by a null-version completion: got %+v, want %+v", instanceType, got, wantUUID) + } + }) + + t.Run("absent-null-with-uuid-keeps-accepted-lock", func(t *testing.T) { + object := "reconcile/null-absent" + // Only a UUID version exists; there is no null version. + if _, err := obj.PutObject(ctx, bucketName, object, + mustGetPutObjReader(t, bytes.NewReader([]byte("data")), 4, "", ""), + ObjectOptions{Versioned: true, MTime: time.Date(2026, 9, 3, 11, 0, 0, 0, time.UTC), UserDefined: map[string]string{ + strings.ToLower(xhttp.AmzObjectLockLegalHold): "OFF", + ReservedMetadataPrefixLower + ObjectLockLegalHoldTimestamp: objectLockTestStamp1100, + }}); err != nil { + t.Fatal(err) + } + + // A null-version retransmit with its own validated legal hold ON. The null + // version does not exist, so the write must keep its accepted lock rather + // than order against the unrelated UUID version. + completeNullMPU(t, object, map[string]string{ + strings.ToLower(xhttp.AmzObjectLockLegalHold): "ON", + }) + if got := readObjectLockFields(t, obj, bucketName, object, nullVersionID); got.legalHold != "ON" { + t.Fatalf("%s: an absent null version was reconciled against the UUID version: got %+v", instanceType, got) + } + }) +}