From e36ea51d9c3a28df7d60f1d105226b1d0049342e Mon Sep 17 00:00:00 2001 From: Feng Ruohang Date: Wed, 11 Nov 2026 14:39:30 +0800 Subject: [PATCH] fix(copy): preserve authoritative content when pools are moving Bind in-place metadata updates to the same version and pool as reads. Resolve encryption and checksum rewrites before write placement while retaining the SSE-C rewrap fast path and invalid self-copy checks. Add HTTP and object-layer regressions for moving pools, duplicate bodies, null versions, timestamp ties and active-pool rewrites. Refs #343. Signed-off-by: Feng Ruohang --- cmd/erasure-server-pool.go | 11 +- cmd/object-copy-pools_test.go | 188 +++++++++++++++++++++++++++++++--- cmd/object-handlers.go | 43 +++++--- 3 files changed, 206 insertions(+), 36 deletions(-) diff --git a/cmd/erasure-server-pool.go b/cmd/erasure-server-pool.go index b2a5a2719..4c7f4f17f 100644 --- a/cmd/erasure-server-pool.go +++ b/cmd/erasure-server-pool.go @@ -1708,10 +1708,13 @@ func (z *erasureServerPools) prepareCopyObject(ctx context.Context, srcBucket, s reference := sameObject && !copyRewritesObjectData(srcInfo.metadataOnly && !srcInfo.Legacy, srcOpts, dstOpts) inPlace := reference && (dstOpts.VersionID != "" && srcOpts.VersionID == dstOpts.VersionID || !dstOpts.Versioned && srcOpts.VersionID == "") if inPlace { - idx, err := z.getPoolIdxExistingWithOpts(ctx, dstBucket, dstObject, ObjectOptions{ - VersionID: dstOpts.VersionID, MetadataChg: true, NoLock: true, - Transition: dstOpts.Transition, SkipDecommissioned: true, SkipRebalancing: true, - }) + // Metadata must stay with the authoritative source bytes, including + // when that pool is draining. Write allocation could select a stale + // duplicate and later retire the only current copy. + sourceOpts := srcOpts + sourceOpts.NoLock = true + sourceOpts.Transition = dstOpts.Transition + _, idx, err := z.getLatestObjectInfoWithIdx(ctx, srcBucket, srcObject, sourceOpts) return copyObjectPlan{poolIdx: idx}, err } idx, err := z.getPoolIdxNoLock(ctx, dstBucket, dstObject, srcInfo.Size) diff --git a/cmd/object-copy-pools_test.go b/cmd/object-copy-pools_test.go index aa928fada..bbec8a0a0 100644 --- a/cmd/object-copy-pools_test.go +++ b/cmd/object-copy-pools_test.go @@ -108,23 +108,7 @@ func TestAPICopyHistoricalVersionAcrossPools(t *testing.T) { if err != nil { t.Fatal(err) } - switch tc.movement { - case "decommission": - z.poolMetaMutex.Lock() - z.poolMeta.Pools[sourcePool].Decommission = &PoolDecommissionInfo{} - z.poolMetaMutex.Unlock() - defer func() { - z.poolMetaMutex.Lock() - z.poolMeta.Pools[sourcePool].Decommission = nil - z.poolMetaMutex.Unlock() - }() - case "rebalance": - z.rebalMu.Lock() - z.rebalMeta = &rebalanceMeta{PoolStats: []*rebalanceStats{{}, {}}} - z.rebalMeta.PoolStats[sourcePool] = &rebalanceStats{Participating: true, Info: rebalanceInfo{Status: rebalStarted}} - z.rebalMu.Unlock() - defer func() { z.rebalMu.Lock(); z.rebalMeta = nil; z.rebalMu.Unlock() }() - } + setCopyMovingPool(t, z, sourcePool, tc.movement) // Toggle compression so metadata has to describe the bytes written now. restore() restoreTarget := setCopyChecksumCompression(true) @@ -231,3 +215,173 @@ func TestPoolsCopyExactVersionMetadata(t *testing.T) { } } } + +func setCopyMovingPool(t *testing.T, z *erasureServerPools, pool int, movement string) { + t.Helper() + switch movement { + case "decommission": + z.poolMetaMutex.Lock() + z.poolMeta.Pools[pool].Decommission = &PoolDecommissionInfo{} + z.poolMetaMutex.Unlock() + t.Cleanup(func() { + z.poolMetaMutex.Lock() + z.poolMeta.Pools[pool].Decommission = nil + z.poolMetaMutex.Unlock() + }) + case "rebalance": + z.rebalMu.Lock() + z.rebalMeta = &rebalanceMeta{PoolStats: []*rebalanceStats{{}, {}}} + z.rebalMeta.PoolStats[pool] = &rebalanceStats{Participating: true, Info: rebalanceInfo{Status: rebalStarted}} + z.rebalMu.Unlock() + t.Cleanup(func() { z.rebalMu.Lock(); z.rebalMeta = nil; z.rebalMu.Unlock() }) + } +} + +// The content that a metadata-only replica preserves must be the same content +// that reads selected, even when write allocation would skip its pool. +func TestCopyReplicaMetadataMovingPool(t *testing.T) { + z, _ := consistencyPools(t) + bucket, router, err := initAPIHandlerTest(t.Context(), z, nil, MakeBucketOptions{LockEnabled: true}) + mustIAM(t, err) + setObjectLayer(z) + _, err = globalBucketMetadataSys.Update(t.Context(), bucket, bucketVersioningConfig, enabledBucketVersioningConfig) + mustIAM(t, err) + const body = "current-body" + for _, api := range []bool{false, true} { + for sourcePool := range 2 { + for _, movement := range []string{"decommission", "rebalance"} { + for _, kind := range []string{"uuid", "null"} { + for _, duplicate := range []string{"absent", "older", "tie"} { + // The lower pool index wins equal ModTimes, just as on reads. + if duplicate == "tie" && sourcePool != 0 { + continue + } + t.Run(fmt.Sprintf("api=%t/source=%d/%s/%s/%s", api, sourcePool, movement, kind, duplicate), func(t *testing.T) { + key := fmt.Sprintf("metadata-%t-%d-%s-%s-%s", api, sourcePool, movement, kind, duplicate) + oi := putConsistencyObject(t, z, bucket, key, sourcePool, body, ObjectOptions{ + Versioned: kind == "uuid", MTime: UTCNow().Add(-time.Minute), + WantChecksum: hash.NewChecksumFromData(hash.ChecksumCRC32, []byte(body)), + }) + version := oi.VersionID + if version == "" { + version = nullVersionID + } + if duplicate != "absent" { + stamp := oi.ModTime + if duplicate == "older" { + stamp = stamp.Add(-time.Hour) + } + putConsistencyObject(t, z, bucket, key, 1-sourcePool, "stale---body", ObjectOptions{ + Versioned: kind == "uuid", VersionID: version, MTime: stamp, + UserDefined: poolLockMetadata("", "ON", "", objectLockTestStamp1000), + }) + } + if got := readCopyChecksumObject(t, z, bucket, key, ObjectOptions{VersionID: version}); string(got) != body { + t.Fatalf("fixture: %q", got) + } + setCopyMovingPool(t, z, sourcePool, movement) + if api { + headers := map[string]string{ + xhttp.AmzCopySource: "/" + bucket + "/" + key + "?versionId=" + version, + xhttp.AmzMetadataDirective: "REPLACE", "X-Amz-Meta-Updated": "yes", + xhttp.MinIOSourceReplicationRequest: "true", xhttp.AmzBucketReplicationStatus: "REPLICA", + } + rec := r5Request(t, router, globalActiveCred, http.MethodPut, "/"+bucket+"/"+key+"?versionId="+version, "", headers) + if rec.Code != http.StatusOK { + t.Fatalf("metadata copy: %d %s", rec.Code, rec.Body.String()) + } + assertCopyChecksumResponse(t, rec, hash.ChecksumCRC32, []byte(body)) + } else { + oi.metadataOnly = true + oi.UserDefined["X-Amz-Meta-Updated"] = "yes" + _, err = z.CopyObject(t.Context(), bucket, key, bucket, key, oi, + ObjectOptions{VersionID: version}, ObjectOptions{Versioned: kind == "uuid", VersionID: version, MTime: oi.ModTime, ReplicaLockReconcile: true}) + mustIAM(t, err) + } + if got := readCopyChecksumObject(t, z, bucket, key, ObjectOptions{VersionID: version}); string(got) != body { + t.Fatalf("metadata copy replaced content: %q", got) + } + stored, err := z.serverPools[sourcePool].GetObjectInfo(t.Context(), bucket, key, ObjectOptions{VersionID: version}) + mustIAM(t, err) + if stored.UserDefined["X-Amz-Meta-Updated"] != "yes" || !bytes.Equal(stored.Checksum, oi.Checksum) { + t.Fatalf("metadata/checksum: %v, %x; want checksum %x", stored.UserDefined, stored.Checksum, oi.Checksum) + } + if duplicate != "absent" && storedObjectLockState(stored.UserDefined).legalHold != "ON" { + t.Fatal("lost protection from the retired copy") + } + _, err = z.serverPools[1-sourcePool].GetObjectInfo(t.Context(), bucket, key, ObjectOptions{VersionID: version}) + if !isErrObjectNotFound(err) && !isErrVersionNotFound(err) { + t.Fatalf("stale copy not retired: %v", err) + } + }) + } + } + } + } + } +} + +func TestAPICopyRewriteMovingPool(t *testing.T) { + z, _ := consistencyPools(t) + placed := ©UploadPool{erasureServerPools: z} + bucket, router, err := initAPIHandlerTest(t.Context(), placed, nil, MakeBucketOptions{}) + mustIAM(t, err) + setObjectLayer(placed) + previousKMS := GlobalKMS + GlobalKMS, err = kms.NewBuiltin("copy-key", bytes.Repeat([]byte{0x33}, 32)) + mustIAM(t, err) + defer func() { GlobalKMS = previousKMS }() + for sourcePool := range 2 { + for _, movement := range []string{"decommission", "rebalance"} { + for _, mode := range []string{"sse-s3", "checksum", "default-checksum", "invalid-self-copy"} { + t.Run(fmt.Sprintf("source=%d/%s/%s", sourcePool, movement, mode), func(t *testing.T) { + key := fmt.Sprintf("rewrite-%d-%s-%s", sourcePool, movement, mode) + data := []byte("rewriting a source on a draining pool") + sourceHeaders := map[string]string{} + if mode != "default-checksum" { + sourceHeaders[xhttp.AmzChecksumCRC32] = mustChecksum(t, hash.ChecksumCRC32, data) + } + placed.uploadPool = sourcePool + putCopyChecksumSource(t, router, globalActiveCred, bucket, key, data, sourceHeaders) + setCopyMovingPool(t, z, sourcePool, movement) + headers := map[string]string{xhttp.AmzMetadataDirective: "REPLACE"} + wantChecksum := hash.ChecksumCRC32 + switch mode { + case "sse-s3": + headers[xhttp.AmzServerSideEncryption] = xhttp.AmzEncryptionAES + case "checksum", "invalid-self-copy": + headers[xhttp.AmzChecksumAlgo] = "SHA256" + wantChecksum = hash.ChecksumSHA256 + if mode == "invalid-self-copy" { + delete(headers, xhttp.AmzMetadataDirective) + } + case "default-checksum": + wantChecksum = hash.ChecksumCRC64NVME + } + rec := copyChecksumRequest(t, router, globalActiveCred, bucket, key, key, headers) + if mode == "invalid-self-copy" { + if rec.Code != http.StatusBadRequest { + t.Fatalf("invalid self-copy: %d %s", rec.Code, rec.Body.String()) + } + _, err := z.serverPools[1-sourcePool].GetObjectInfo(t.Context(), bucket, key, ObjectOptions{}) + if !isErrObjectNotFound(err) { + t.Fatalf("rejected copy wrote a destination: %v", err) + } + return + } + if rec.Code != http.StatusOK { + t.Fatalf("rewrite: %d %s", rec.Code, rec.Body.String()) + } + assertCopyChecksumResponse(t, rec, wantChecksum, data) + if got := readCopyChecksumObject(t, z, bucket, key, ObjectOptions{}); !bytes.Equal(got, data) { + t.Fatalf("rewritten body: %q", got) + } + if _, err := z.serverPools[1-sourcePool].GetObjectInfo(t.Context(), bucket, key, ObjectOptions{}); err != nil { + t.Fatalf("rewrite did not reach active pool: %v", err) + } + assertCopyChecksum(t, z, bucket, key, wantChecksum, data, false, nil) + }) + } + } + } +} diff --git a/cmd/object-handlers.go b/cmd/object-handlers.go index b50751e16..0a2b2412a 100644 --- a/cmd/object-handlers.go +++ b/cmd/object-handlers.go @@ -1680,14 +1680,6 @@ func (api objectAPIHandlers) CopyObjectHandler(w http.ResponseWriter, r *http.Re if dstOpts.Versioned && copySrcOpts.VersionID == "" { copySrcOpts.VersionID = srcInfo.VersionID } - if !remoteCallRequired { - plan, err := objectAPI.prepareCopyObject(lockCtx, srcBucket, srcObject, dstBucket, dstObject, srcInfo, copySrcOpts, dstOpts) - if err != nil { - writeErrorResponse(ctx, w, toAPIError(ctx, err), r.URL) - return - } - dstOpts.copyPlan = &plan - } // A key rotation rewraps the object key held in metadata; it never // re-encrypts the stored bytes. When the object layer stores new object @@ -1704,16 +1696,23 @@ func (api objectAPIHandlers) CopyObjectHandler(w http.ResponseWriter, r *http.Re // Without a requested destination checksum algorithm, an in-place SSE-C // rotation preserves the stored checksum state, including absence; it does // not add the default CRC64NVME. - canRotateKeyInPlace := !srcInfo.Legacy && + canRotateKeyInPlace := cpSrcDstSame && sseCopyC && sseC && !chStorageClass && !srcInfo.Legacy && !copyRewritesObjectData(srcInfo.metadataOnly, copySrcOpts, dstOpts) && (replicaTrusted || !hash.NewChecksumHeader(r.Header).IsSet()) - // If src == dst and either - // - the object is encrypted using SSE-C and two different SSE-C keys are present - // - the object is encrypted using SSE-S3 and the SSE-S3 header is present - // - the object storage class is not changing - // then execute a key rotation. - if cpSrcDstSame && (sseCopyC && sseC) && !chStorageClass && canRotateKeyInPlace { + if canRotateKeyInPlace { + // A new version may need another pool. Resolve that before rewrapping + // the key, since a cross-pool copy must re-encrypt the source stream. + plan, err := objectAPI.prepareCopyObject(lockCtx, srcBucket, srcObject, dstBucket, dstObject, srcInfo, copySrcOpts, dstOpts) + if err != nil { + writeErrorResponse(ctx, w, toAPIError(ctx, err), r.URL) + return + } + dstOpts.copyPlan = &plan + canRotateKeyInPlace = !plan.rewrite + } + + if canRotateKeyInPlace { oldKey, err = ParseSSECopyCustomerRequest(r.Header, srcInfo.UserDefined) if err != nil { writeErrorResponse(ctx, w, toAPIError(ctx, err), r.URL) @@ -1804,6 +1803,20 @@ func (api objectAPIHandlers) CopyObjectHandler(w http.ResponseWriter, r *http.Re } } + if !remoteCallRequired && dstOpts.copyPlan == nil { + // Encryption and checksum recomputation require new data. Plan + // with that final decision before starting either transform, but + // retain metadataOnly for the invalid self-copy check below. + planInfo := srcInfo + planInfo.metadataOnly = srcInfo.metadataOnly && !dstOpts.WantServerSideChecksumType.IsSet() + plan, err := objectAPI.prepareCopyObject(lockCtx, srcBucket, srcObject, dstBucket, dstObject, planInfo, copySrcOpts, dstOpts) + if err != nil { + writeErrorResponse(ctx, w, toAPIError(ctx, err), r.URL) + return + } + dstOpts.copyPlan = &plan + } + if isDstCompressed { checksumReader := srcInfo.Reader wantEncryption := crypto.Requested(r.Header)