fix(batch): let a write refuse to replace a locked version

Add ObjectOptions.RefuseLockedOverwrite. With it set, PutObject (before
the rename) and CompleteMultipartUpload check, under the destination's
write lock, the stored version the write would replace: the version it
addresses, or the null version. DeleteObject checks the stored version
that a delete marker it writes names; in a multi-pool deployment that
version is resolved across pools. A version under retention or legal
hold is refused with ObjectLocked. The check reads the version's own
metadata, whatever the bucket configuration, and returns read errors.

Nothing sets the option yet.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Signed-off-by: Feng Ruohang <rh@vonng.com>

cmd/bucket-object-lock.go

@@ -291,6 +291,53 @@ func checkObjectLockedForOverwrite(ctx context.Context, bucket, object string, g
 	return nil
 }
 
+// checkVersionLockedForOverwrite re-checks under the object write lock
+// whether a write would replace the stored version versionID (the null
+// version if empty) while Object Lock protects it. It reads the version's own
+// metadata, whatever the bucket configuration, and a failed read is an error.
+func checkVersionLockedForOverwrite(ctx context.Context, bucket, object, versionID string, getObjectInfoFn GetObjectInfoFn, transition TransitionOptions) error {
+	if isMinioMetaBucketName(bucket) {
+		return nil
+	}
+	if versionID == "" {
+		versionID = nullVersionID
+	}
+	objInfo, err := getObjectInfoFn(ctx, bucket, object, ObjectOptions{
+		VersionID:  versionID,
+		Transition: transition,
+		NoLock:     true,
+		NoAuditLog: true,
+	})
+	if err != nil {
+		// A delete marker is not protected.
+		if isErrObjectNotFound(err) || isErrVersionNotFound(err) || errors.As(err, &MethodNotAllowed{}) {
+			return nil
+		}
+		return err
+	}
+	if enforceRetentionForDeletion(ctx, objInfo) {
+		return ObjectLocked{Bucket: bucket, Object: decodeDirObject(object), VersionID: versionID}
+	}
+	return nil
+}
+
+// refuseLockedMarkerOverwrite refuses a delete marker write that would replace
+// the stored version it names while Object Lock protects it. goi and gerr are
+// that version as read under the write lock. A new marker, or one replacing a
+// marker, is allowed.
+func refuseLockedMarkerOverwrite(ctx context.Context, bucket, object, versionID string, goi ObjectInfo, gerr error) error {
+	if gerr != nil {
+		if isErrObjectNotFound(gerr) || isErrVersionNotFound(gerr) || errors.As(gerr, &MethodNotAllowed{}) {
+			return nil
+		}
+		return gerr
+	}
+	if enforceRetentionForDeletion(ctx, goi) {
+		return ObjectLocked{Bucket: bucket, Object: decodeDirObject(object), VersionID: versionID}
+	}
+	return nil
+}
+
 // checkPutObjectLockAllowed enforces object retention policy and legal hold policy
 // for requests with WORM headers
 // See https://docs.aws.amazon.com/AmazonS3/latest/dev/object-lock-managing.html for the spec.

cmd/erasure-multipart.go

@@ -1687,6 +1687,15 @@ func (er erasureObjects) CompleteMultipartUpload(ctx context.Context, bucket str
 		defer lk.Unlock(lkctx)
 	}
 
+	if opts.RefuseLockedOverwrite {
+		getObjectInfo := er.getObjectInfo
+		if opts.replicaObjectInfo != nil {
+			getObjectInfo = opts.replicaObjectInfo
+		}
+		if err := checkVersionLockedForOverwrite(ctx, bucket, object, fi.VersionID, getObjectInfo, opts.Transition); err != nil {
+			return oi, toObjectErr(err, bucket, object)
+		}
+	}
 	if !opts.ReplicationRequest && !opts.ReplicaLockReconcile && !opts.DataMovement {
 		getObjectInfo := er.getObjectInfo
 		if opts.replicaObjectInfo != nil {

cmd/erasure-object.go

@@ -1676,6 +1676,15 @@ func (er erasureObjects) putObject(ctx context.Context, bucket string, object st
 			return ObjectInfo{}, toObjectErr(err, bucket, object)
 		}
 	}
+	if opts.RefuseLockedOverwrite {
+		getObjectInfo := er.getObjectInfo
+		if opts.replicaObjectInfo != nil {
+			getObjectInfo = opts.replicaObjectInfo
+		}
+		if err := checkVersionLockedForOverwrite(ctx, bucket, object, fi.VersionID, getObjectInfo, opts.Transition); err != nil {
+			return ObjectInfo{}, toObjectErr(err, bucket, object)
+		}
+	}
 
 	// Rename the successfully written temporary object to final location.
 	onlineDisks, versions, oldDataDir, err := renameData(ctx, onlineDisks, minioMetaTmpBucket, tempObj, partsMetadata, bucket, object, writeQuorum)
@@ -2199,6 +2208,13 @@ func (er erasureObjects) DeleteObject(ctx context.Context, bucket, object string
 		}
 	}
 
+	// A delete marker that names a stored version replaces it.
+	if opts.RefuseLockedOverwrite && opts.DeleteMarker && opts.VersionID != "" {
+		if err := refuseLockedMarkerOverwrite(ctx, bucket, object, opts.VersionID, goi, gerr); err != nil {
+			return ObjectInfo{}, err
+		}
+	}
+
 	if opts.Expiration.Expire && !opts.Expiration.CheckedAllPools {
 		if gerr == nil {
 			evt := evalActionFromLifecycle(ctx, *lc, rcfg, replcfg, goi)

cmd/erasure-server-pool.go

@@ -1277,6 +1277,15 @@ func (z *erasureServerPools) DeleteObject(ctx context.Context, bucket string, ob
 		return z.deleteAllVersions(ctx, bucket, object, opts)
 	}
 
+	// A delete marker that names a version replaces that version in
+	// whichever pool holds it.
+	if opts.RefuseLockedOverwrite && opts.DeleteMarker && opts.VersionID != "" && !z.SinglePool() {
+		goi, gerr := z.replicaObjectInfo(ctx, bucket, object, ObjectOptions{VersionID: opts.VersionID, NoLock: true, NoAuditLog: true})
+		if err := refuseLockedMarkerOverwrite(ctx, bucket, object, opts.VersionID, goi, gerr); err != nil {
+			return ObjectInfo{Name: decodeDirObject(object), VersionID: opts.VersionID}, err
+		}
+	}
+
 	// An ILM expiry is decided on the versions in one pool. Under the lock,
 	// check it with the versions in all pools; the sets then take it as
 	// decided, without a recheck by the versions each one holds. They only

cmd/object-api-interface.go

@@ -117,6 +117,12 @@ type ObjectOptions struct {
 	// Pools-layer version resolver; the caller holds the shared object lock.
 	replicaObjectInfo GetObjectInfoFn
 
+	// RefuseLockedOverwrite refuses, under the destination's write lock, a
+	// write that would replace a stored version Object Lock protects: the
+	// version it writes, or the version a delete marker it writes names.
+	// It returns ObjectLocked. Set by batch replication pulls.
+	RefuseLockedOverwrite bool
+
 	Speedtest bool // object call specifically meant for SpeedTest code, set to 'true' when invoked by SpeedtestHandler.
 
 	// Use the maximum parity (N/2), used when saving server configuration files