From a8824b341fce071cdd48356caaed37d76580c39e Mon Sep 17 00:00:00 2001 From: Feng Ruohang Date: Wed, 11 Nov 2026 12:29:28 +0800 Subject: [PATCH] fix(copy): plan historical version placement before transforms Locate exact versions for in-place updates and rewrite cross-pool references through the encryption and compression pipeline selected before copying. Fixes #343 Signed-off-by: Feng Ruohang --- cmd/erasure-server-pool.go | 76 ++++++++++---- cmd/object-api-interface.go | 3 + cmd/object-copy-pools_test.go | 190 ++++++++++++++++++++++++++++++++++ cmd/object-handlers.go | 18 ++-- 4 files changed, 258 insertions(+), 29 deletions(-) create mode 100644 cmd/object-copy-pools_test.go diff --git a/cmd/erasure-server-pool.go b/cmd/erasure-server-pool.go index 0bff3529d..b2a5a2719 100644 --- a/cmd/erasure-server-pool.go +++ b/cmd/erasure-server-pool.go @@ -1689,6 +1689,48 @@ func (z *erasureServerPools) DeleteObjects(ctx context.Context, bucket string, o return dobjects, derrs } +// copyObjectPlan fixes placement before a caller prepares encrypted or compressed +// bytes. A reference can only be made in the pool holding the source version. +type copyObjectPlan struct { + poolIdx int + rewrite bool +} + +// prepareCopyObject requires the same source/destination locks as CopyObject. +// In-place changes address the exact version; new versions retain the normal +// write placement and rewrite the data when it lives in a different pool. +func (z *erasureServerPools) prepareCopyObject(ctx context.Context, srcBucket, srcObject, dstBucket, dstObject string, srcInfo ObjectInfo, srcOpts, dstOpts ObjectOptions) (copyObjectPlan, error) { + if err := checkCopyObjArgs(ctx, dstBucket, dstObject); err != nil { + return copyObjectPlan{}, err + } + srcObject, dstObject = encodeDirObject(srcObject), encodeDirObject(dstObject) + sameObject := isStringEqual(pathJoin(srcBucket, srcObject), pathJoin(dstBucket, dstObject)) + 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, + }) + return copyObjectPlan{poolIdx: idx}, err + } + idx, err := z.getPoolIdxNoLock(ctx, dstBucket, dstObject, srcInfo.Size) + if err != nil { + return copyObjectPlan{}, err + } + plan := copyObjectPlan{poolIdx: idx} + if reference && !z.SinglePool() { + sourceOpts := srcOpts + sourceOpts.NoLock = true + _, sourcePool, err := z.getLatestObjectInfoWithIdx(ctx, srcBucket, srcObject, sourceOpts) + if err != nil { + return copyObjectPlan{}, err + } + plan.rewrite = sourcePool != idx + } + return plan, nil +} + func (z *erasureServerPools) CopyObject(ctx context.Context, srcBucket, srcObject, dstBucket, dstObject string, srcInfo ObjectInfo, srcOpts, dstOpts ObjectOptions) (objInfo ObjectInfo, err error) { if err := checkCopyObjArgs(ctx, srcBucket, srcObject); err != nil { return ObjectInfo{}, err @@ -1715,6 +1757,15 @@ func (z *erasureServerPools) CopyObject(ctx context.Context, srcBucket, srcObjec dstOpts.NoLock = true } + if dstOpts.copyPlan == nil { + plan, err := z.prepareCopyObject(ctx, srcBucket, srcObject, dstBucket, dstObject, srcInfo, srcOpts, dstOpts) + if err != nil { + return ObjectInfo{}, err + } + dstOpts.copyPlan = &plan + } + poolIdx := dstOpts.copyPlan.poolIdx + if !z.SinglePool() && cpSrcDstSame && srcInfo.metadataOnly && dstOpts.ReplicaLockReconcile && srcOpts.VersionID == dstOpts.VersionID { copies, err := z.metadataPoolInfos(ctx, dstBucket, dstObject, dstOpts) if err != nil { @@ -1724,7 +1775,7 @@ func (z *erasureServerPools) CopyObject(ctx context.Context, srcBucket, srcObjec reconcileStoredObjectLock(srcInfo.UserDefined, storedObjectLockState(stored.UserDefined)) reconcileStoredObjectTags(srcInfo.UserDefined, stored.UserTags, stored.UserDefined[ReservedMetadataPrefixLower+TaggingTimestamp]) dstOpts.replicaObjectInfo = z.replicaObjectInfo - idx := copies[0].Index + idx := poolIdx oi, err := z.serverPools[idx].CopyObject(ctx, srcBucket, srcObject, dstBucket, dstObject, srcInfo, srcOpts, dstOpts) if err == nil { err = z.retireReplicaCopies(ctx, dstBucket, dstObject, idx, oi) @@ -1732,25 +1783,6 @@ func (z *erasureServerPools) CopyObject(ctx context.Context, srcBucket, srcObjec return oi, err } - var poolIdx int - if dstOpts.MetadataChg && cpSrcDstSame && srcInfo.metadataOnly && srcOpts.VersionID == dstOpts.VersionID { - // Restore status updates address an existing version. Keep ordinary - // copy/write allocation unchanged, including its latest-version lookup. - poolIdx, err = z.getPoolIdxExistingWithOpts(ctx, dstBucket, dstObject, ObjectOptions{ - VersionID: dstOpts.VersionID, - MetadataChg: true, - NoLock: true, - Transition: dstOpts.Transition, - SkipDecommissioned: true, - SkipRebalancing: true, - }) - } else { - poolIdx, err = z.getPoolIdxNoLock(ctx, dstBucket, dstObject, srcInfo.Size) - } - if err != nil { - return objInfo, err - } - if !z.SinglePool() { dstOpts.replicaContentGuard = z.guardReplicaContent } @@ -1765,9 +1797,7 @@ func (z *erasureServerPools) CopyObject(ctx context.Context, srcBucket, srcObjec } } - // CopyObjectHandler predicts the outcome of this decision in - // copyRewritesObjectData(); keep the two in sync. - if cpSrcDstSame && srcInfo.metadataOnly { + if cpSrcDstSame && !copyRewritesObjectData(srcInfo.metadataOnly && !srcInfo.Legacy, srcOpts, dstOpts) { // Version ID is set for the destination and source == destination version ID. if dstOpts.VersionID != "" && srcOpts.VersionID == dstOpts.VersionID { return z.serverPools[poolIdx].CopyObject(ctx, srcBucket, srcObject, dstBucket, dstObject, srcInfo, srcOpts, dstOpts) diff --git a/cmd/object-api-interface.go b/cmd/object-api-interface.go index ffd21230d..454ec8383 100644 --- a/cmd/object-api-interface.go +++ b/cmd/object-api-interface.go @@ -124,6 +124,8 @@ type ObjectOptions struct { replicaObjectInfo GetObjectInfoFn // Verifies protected content in every pool before a replica commits. replicaContentGuard replicaContentGuardFn + // Selected under the object locks, before copy encryption/compression. + copyPlan *copyObjectPlan // RefuseLockedOverwrite refuses, under the destination's write lock, a // write that would replace a stored version Object Lock protects: the @@ -299,6 +301,7 @@ type ObjectLayer interface { NewObjectNSLock(bucket, object string) RWLocker // lockObjects locks several objects together, see erasureServerPools. lockObjects(ctx context.Context, reqs ...objectLockRequest) (LockContext, func(), error) + prepareCopyObject(ctx context.Context, srcBucket, srcObject, dstBucket, dstObject string, srcInfo ObjectInfo, srcOpts, dstOpts ObjectOptions) (copyObjectPlan, error) // Storage operations. Shutdown(context.Context) error diff --git a/cmd/object-copy-pools_test.go b/cmd/object-copy-pools_test.go new file mode 100644 index 000000000..73aaa0cac --- /dev/null +++ b/cmd/object-copy-pools_test.go @@ -0,0 +1,190 @@ +// Copyright (c) 2026 PGSTY +// SPDX-License-Identifier: AGPL-3.0-or-later + +package cmd + +import ( + "bytes" + "context" + "crypto/md5" + "encoding/base64" + "fmt" + "maps" + "net/http" + "net/http/httptest" + "testing" + "time" + + "github.com/minio/minio/internal/hash" + xhttp "github.com/minio/minio/internal/http" + "github.com/minio/minio/internal/kms" +) + +// Place uploaded versions in specific pools, while retaining the real HTTP +// encryption/compression pipeline and the normal pools-layer copy operation. +type copyUploadPool struct { + *erasureServerPools + uploadPool int +} + +func (z *copyUploadPool) PutObject(ctx context.Context, bucket, object string, r *PutObjReader, opts ObjectOptions) (ObjectInfo, error) { + return z.serverPools[z.uploadPool].PutObject(ctx, bucket, object, r, opts) +} + +func TestAPICopyHistoricalVersionAcrossPools(t *testing.T) { + z, _ := consistencyPools(t) + placed := ©UploadPool{erasureServerPools: z} + bucket, router, err := initAPIHandlerTest(t.Context(), placed, nil, MakeBucketOptions{VersioningEnabled: true}) + if err != nil { + t.Fatal(err) + } + setObjectLayer(placed) + if _, err = globalBucketMetadataSys.Update(t.Context(), bucket, bucketVersioningConfig, enabledBucketVersioningConfig); err != nil { + t.Fatal(err) + } + previousKMS, previousTLS := GlobalKMS, globalIsTLS + GlobalKMS = kms.NewStub("copy-key") + globalIsTLS = true + defer func() { GlobalKMS, globalIsTLS = previousKMS, previousTLS }() + customer := func(value byte) map[string]string { + key := bytes.Repeat([]byte{value}, 32) + sum := md5.Sum(key) + return map[string]string{xhttp.AmzServerSideEncryptionCustomerAlgorithm: xhttp.AmzEncryptionAES, xhttp.AmzServerSideEncryptionCustomerKey: base64.StdEncoding.EncodeToString(key), xhttp.AmzServerSideEncryptionCustomerKeyMD5: base64.StdEncoding.EncodeToString(sum[:])} + } + for _, mode := range []string{"plain", "compressed", "sse-s3", "sse-c"} { + for _, sourcePool := range []int{0, 1} { + t.Run(fmt.Sprintf("%s/source=%d", mode, sourcePool), func(t *testing.T) { + compressed := mode == "compressed" + restore := setCopyChecksumCompression(true) + globalCompressConfigMu.Lock() + globalCompressConfig.Enabled = compressed + globalCompressConfigMu.Unlock() + defer restore() + sourceHeaders := map[string]string{xhttp.AmzChecksumCRC32: ""} + if mode == "sse-s3" { + sourceHeaders[xhttp.AmzServerSideEncryption] = xhttp.AmzEncryptionAES + } + if mode == "sse-c" { + maps.Copy(sourceHeaders, customer(0x11)) + } + data := bytes.Repeat([]byte("historical encrypted compressed content-"), 8192) + sourceHeaders[xhttp.AmzChecksumCRC32] = mustChecksum(t, hash.ChecksumCRC32, data) + key := fmt.Sprintf("copy-%s-%d.txt", mode, sourcePool) + placed.uploadPool = sourcePool + putCopyChecksumSource(t, router, globalActiveCred, bucket, key, data, sourceHeaders) + original, err := z.GetObjectInfo(t.Context(), bucket, key, ObjectOptions{}) + if err != nil { + t.Fatal(err) + } + if original.VersionID == "" { + t.Fatal("fixture has no version ID") + } + if _, err := z.serverPools[sourcePool].GetObjectInfo(t.Context(), bucket, key, ObjectOptions{VersionID: original.VersionID}); err != nil { + t.Fatal(err) + } + if original.IsCompressed() != compressed { + t.Fatalf("fixture compressed=%v", original.IsCompressed()) + } + placed.uploadPool = 1 - sourcePool + putCopyChecksumSource(t, router, globalActiveCred, bucket, key, []byte("newer version"), nil) + newer, err := z.GetObjectInfo(t.Context(), bucket, key, ObjectOptions{}) + if err != nil { + t.Fatal(err) + } + // Toggle compression so metadata has to describe the bytes written now. + restore() + restoreTarget := setCopyChecksumCompression(true) + globalCompressConfigMu.Lock() + globalCompressConfig.Enabled = !compressed + globalCompressConfigMu.Unlock() + defer restoreTarget() + headers := map[string]string{xhttp.AmzMetadataDirective: "REPLACE"} + getHeaders := map[string]string{} + if mode == "sse-s3" { + headers[xhttp.AmzServerSideEncryption] = xhttp.AmzEncryptionAES + } + if mode == "sse-c" { + maps.Copy(headers, customer(0x22)) + maps.Copy(getHeaders, customer(0x22)) + old := customer(0x11) + headers[xhttp.AmzServerSideEncryptionCopyCustomerAlgorithm] = old[xhttp.AmzServerSideEncryptionCustomerAlgorithm] + headers[xhttp.AmzServerSideEncryptionCopyCustomerKey] = old[xhttp.AmzServerSideEncryptionCustomerKey] + headers[xhttp.AmzServerSideEncryptionCopyCustomerKeyMD5] = old[xhttp.AmzServerSideEncryptionCustomerKeyMD5] + } + rec := copyChecksumRequest(t, router, globalActiveCred, bucket, key+"?versionId="+original.VersionID, key, headers) + if rec.Code != http.StatusOK { + t.Fatalf("copy: %d %s", rec.Code, rec.Body.String()) + } + copied, err := z.GetObjectInfo(t.Context(), bucket, key, ObjectOptions{}) + if err != nil || copied.VersionID == "" || copied.VersionID == original.VersionID || copied.VersionID == newer.VersionID { + t.Fatalf("copy did not create a version: %+v %v", copied, err) + } + if copied.IsCompressed() != (!compressed && mode != "sse-c") { + t.Fatalf("copy did not use destination compression setting: %v", copied.IsCompressed()) + } + if _, err = z.serverPools[1-sourcePool].GetObjectInfo(t.Context(), bucket, key, ObjectOptions{VersionID: copied.VersionID}); err != nil { + t.Fatalf("copy was not placed with destination: %v", err) + } + read := func(version string, headers map[string]string, want []byte) { + t.Helper() + req, err := newTestSignedRequestV4(http.MethodGet, getGetObjectURL("", bucket, key)+"?versionId="+version, 0, nil, globalActiveCred.AccessKey, globalActiveCred.SecretKey, headers) + if err != nil { + t.Fatal(err) + } + rec := httptest.NewRecorder() + router.ServeHTTP(rec, req) + if rec.Code != http.StatusOK || !bytes.Equal(rec.Body.Bytes(), want) { + t.Fatalf("GET version %s: %d body length=%d, want %d", version, rec.Code, rec.Body.Len(), len(want)) + } + } + read(copied.VersionID, getHeaders, data) + oldHeaders := map[string]string{} + if mode == "sse-c" { + oldHeaders = customer(0x11) + } + read(original.VersionID, oldHeaders, data) + read(newer.VersionID, nil, []byte("newer version")) + }) + } + } +} + +// Internal metadata updates do not all set MetadataChg. They must still +// address the requested historical version, and must never create a new one. +func TestPoolsCopyExactVersionMetadata(t *testing.T) { + z, bucket := consistencyPools(t) + for sourcePool := range 2 { + key := fmt.Sprintf("exact-%d", sourcePool) + older := putConsistencyObject(t, z, bucket, key, sourcePool, "old data", ObjectOptions{Versioned: true, MTime: UTCNow().Add(-time.Hour)}) + newer := putConsistencyObject(t, z, bucket, key, 1-sourcePool, "new data", ObjectOptions{Versioned: true, MTime: UTCNow()}) + gr, err := z.GetObjectNInfo(t.Context(), bucket, key, nil, nil, ObjectOptions{VersionID: older.VersionID, NoLock: true}) + if err != nil { + t.Fatal(err) + } + info := gr.ObjInfo + info.Reader, err = hash.NewReader(t.Context(), gr, info.Size, "", "", info.Size) + if err != nil { + gr.Close() + t.Fatal(err) + } + info.PutObjReader = NewPutObjReader(info.Reader) + info.metadataOnly = true + info.UserDefined["updated"] = "yes" + _, err = z.CopyObject(t.Context(), bucket, key, bucket, key, info, ObjectOptions{VersionID: older.VersionID}, ObjectOptions{Versioned: true, VersionID: older.VersionID, MTime: older.ModTime}) + gr.Close() + if err != nil { + t.Fatal(err) + } + got, err := z.serverPools[sourcePool].GetObjectInfo(t.Context(), bucket, key, ObjectOptions{VersionID: older.VersionID}) + if err != nil || got.UserDefined["updated"] != "yes" { + t.Fatalf("historical metadata not updated: %v %v", got.UserDefined, err) + } + got, err = z.GetObjectInfo(t.Context(), bucket, key, ObjectOptions{}) + if err != nil || got.VersionID != newer.VersionID || got.UserDefined["updated"] != "" { + t.Fatalf("latest changed: %+v %v", got, err) + } + if got := readCopyChecksumObject(t, z, bucket, key, ObjectOptions{VersionID: older.VersionID}); string(got) != "old data" { + t.Fatalf("historical body changed: %q", got) + } + } +} diff --git a/cmd/object-handlers.go b/cmd/object-handlers.go index b9464c641..b50751e16 100644 --- a/cmd/object-handlers.go +++ b/cmd/object-handlers.go @@ -1317,13 +1317,11 @@ func isRemoteCallRequired(ctx context.Context, bucket string, objAPI ObjectLayer // copyRewritesObjectData reports whether the object layer stores new object data // for this copy instead of updating metadata in place or adding a -// self-referential version. It mirrors the metadata-only decision taken by -// erasureServerPools.CopyObject and erasureSets.CopyObject. CopyObjectHandler -// has to predict that decision because the compression metadata it records must -// describe whichever bytes are finally stored. metadataOnly already excludes -// legacy sources, which the object layer always rewrites. +// self-referential version. The selected pool must contain the source data to +// add a reference. metadataOnly already excludes legacy sources, which the +// object layer always rewrites. func copyRewritesObjectData(metadataOnly bool, srcOpts, dstOpts ObjectOptions) bool { - if !metadataOnly { + if !metadataOnly || dstOpts.copyPlan != nil && dstOpts.copyPlan.rewrite { return true } switch { @@ -1682,6 +1680,14 @@ 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