From 6225e0a7057a54a3563b06e59fbb46e048c1e49a Mon Sep 17 00:00:00 2001 From: Jay Hemnani <193022578+jayhemnani9910@users.noreply.github.com> Date: Thu, 29 Oct 2026 04:19:37 +0530 Subject: [PATCH] s3: return x-amz-copy-source-version-id from CopyObject and UploadPartCopy (#12181) * s3: return x-amz-copy-source-version-id from CopyObject and UploadPartCopy * s3: cover x-amz-copy-source-version-id on latest copies and part copies The explicit-version case was already covered; add a .versions pointer fixture so the unversioned request resolves the latest version and asserts the header on CopyObject, plus the same assertion on UploadPartCopy. * s3: carry the resolved id on a version file that lacks it A .versions child written before the version-id attribute resolves to an empty source version even though the pointer or the rescan already knows it. Stamp the id on the returned entry so copy and get readers report it; the version copy tests now drop the attribute to cover that shape. * s3: report the null version in x-amz-copy-source-version-id A regular-path source in a versioning-configured bucket is the null version; echo it like every other version id, matching the GET path and AWS. Unversioned sources still report no header. * s3: report the version a self-copy retry actually read --------- Co-authored-by: user.email --- weed/s3api/s3api_object_handlers_copy.go | 31 ++++ ...bject_handlers_copy_version_header_test.go | 173 ++++++++++++++++++ weed/s3api/s3api_object_versioning.go | 20 +- 3 files changed, 223 insertions(+), 1 deletion(-) create mode 100644 weed/s3api/s3api_object_handlers_copy_version_header_test.go diff --git a/weed/s3api/s3api_object_handlers_copy.go b/weed/s3api/s3api_object_handlers_copy.go index 69ae416f4e..1b716126d9 100644 --- a/weed/s3api/s3api_object_handlers_copy.go +++ b/weed/s3api/s3api_object_handlers_copy.go @@ -259,6 +259,7 @@ func (s3a *S3ApiServer) CopyObjectHandler(w http.ResponseWriter, r *http.Request s3err.WriteErrorResponse(w, r, errCode) return } + copySourceVersionId := resolveCopySourceVersionId(srcVersionId, srcVersioningState, entry) if errCode := s3a.recheckCopySourcePolicy(r, srcBucket, srcObject, srcVersionId, entry.Extended, "CopyObjectHandler"); errCode != s3err.ErrNone { s3err.WriteErrorResponse(w, r, errCode) @@ -346,6 +347,9 @@ func (s3a *S3ApiServer) CopyObjectHandler(w http.ResponseWriter, r *http.Request if errCode := classifyCopySourceError(currentEntry, currentErr); errCode != s3err.ErrNone { return errCode } + // Report the snapshot this attempt actually read, not the first + // resolution — a retry may observe a different source entry. + copySourceVersionId = resolveCopySourceVersionId(srcVersionId, srcVersioningState, currentEntry) if errCode := s3a.validateConditionalCopyHeaders(r, currentEntry); errCode != s3err.ErrNone { return errCode } @@ -439,6 +443,7 @@ func (s3a *S3ApiServer) CopyObjectHandler(w http.ResponseWriter, r *http.Request if dstVersionId != "" { w.Header().Set("x-amz-version-id", dstVersionId) } + setCopySourceVersionId(w, copySourceVersionId) setEtag(w, etag) writeSuccessResponseXML(w, r, CopyObjectResult{ ETag: quoteETag(etag), @@ -620,6 +625,7 @@ func (s3a *S3ApiServer) CopyObjectHandler(w http.ResponseWriter, r *http.Request if dstVersionId != "" { w.Header().Set("x-amz-version-id", dstVersionId) } + setCopySourceVersionId(w, copySourceVersionId) setEtag(w, etag) @@ -1109,6 +1115,7 @@ func (s3a *S3ApiServer) CopyObjectPartHandler(w http.ResponseWriter, r *http.Req s3err.WriteErrorResponse(w, r, errCode) return } + copySourceVersionId := resolveCopySourceVersionId(srcVersionId, srcVersioningState, entry) // Cache remote-only sources before copying; the part-copy paths below // iterate entry.GetChunks() and would otherwise produce an empty part. @@ -1201,6 +1208,7 @@ func (s3a *S3ApiServer) CopyObjectPartHandler(w http.ResponseWriter, r *http.Req if !s3a.checkUploadStillOpen(w, r, dstBucket, dstObject, uploadID) { return } + setCopySourceVersionId(w, copySourceVersionId) setEtag(w, etag) // Mirror PutObjectPartHandler: write x-amz-server-side-encryption / // x-amz-server-side-encryption-aws-kms-key-id headers on the response @@ -1293,11 +1301,34 @@ func (s3a *S3ApiServer) CopyObjectPartHandler(w http.ResponseWriter, r *http.Req // Calculate ETag for the part etag := copyEntryETag(dstEntry) + setCopySourceVersionId(w, copySourceVersionId) setEtag(w, etag) writeSuccessResponseXML(w, r, buildCopyPartResult(etag, t, SSEResponseMetadata{})) } +// resolveCopySourceVersionId reports which source version a copy read: an +// explicit versionId or the entry's recorded id, defaulting to "null" when the +// source bucket is versioning-configured — a regular-path object there is the +// null version, mirroring the GET path. Unversioned sources report no version. +func resolveCopySourceVersionId(requested, versioningState string, entry *filer_pb.Entry) string { + if v := resolvedSourceVersionId(requested, entry); v != "" { + return v + } + if versioningState != "" { + return "null" + } + return "" +} + +// setCopySourceVersionId reports which source version a copy read, as AWS does — +// the null version is a real version id and is echoed like any other. +func setCopySourceVersionId(w http.ResponseWriter, versionId string) { + if versionId != "" { + w.Header().Set("x-amz-copy-source-version-id", versionId) + } +} + func replaceDirective(reqHeader http.Header) (replaceMeta, replaceTagging bool) { return reqHeader.Get(s3_constants.AmzUserMetaDirective) == DirectiveReplace, reqHeader.Get(s3_constants.AmzObjectTaggingDirective) == DirectiveReplace } diff --git a/weed/s3api/s3api_object_handlers_copy_version_header_test.go b/weed/s3api/s3api_object_handlers_copy_version_header_test.go new file mode 100644 index 0000000000..38c28052e0 --- /dev/null +++ b/weed/s3api/s3api_object_handlers_copy_version_header_test.go @@ -0,0 +1,173 @@ +package s3api + +import ( + "net/http" + "net/http/httptest" + "testing" + "time" + + "github.com/gorilla/mux" + "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" + "github.com/seaweedfs/seaweedfs/weed/s3api/s3_constants" + util_http "github.com/seaweedfs/seaweedfs/weed/util/http" +) + +// CopyObject reports the source version it read in x-amz-copy-source-version-id. +func TestCopyObjectReportsSourceVersionId(t *testing.T) { + const src, dst, object, versionId = "versioned-src", "plain-dst", "object", "3a8f2c1d9e7b4a6f" + filer := ©ACLFiler{ambiguousPutFiler: &ambiguousPutFiler{apply: true, entries: map[string]*filer_pb.Entry{}}} + s3a := newPutTestServer(t, startFakeFiler(t, filer)) + s3a.iam = &IdentityAccessManagement{} + s3a.bucketConfigCache = NewBucketConfigCache(time.Minute) + s3a.bucketConfigCache.Set(src, &BucketConfig{Name: src, Versioning: s3_constants.VersioningEnabled, Ownership: s3_constants.OwnershipObjectWriter}) + s3a.bucketConfigCache.Set(dst, &BucketConfig{Name: dst, Ownership: s3_constants.OwnershipObjectWriter}) + filer.entries["/buckets/"+src+"/"+object+s3_constants.VersionsFolder+"/v_"+versionId] = &filer_pb.Entry{ + Name: "v_" + versionId, Content: []byte("data"), + Attributes: &filer_pb.FuseAttributes{FileMode: defaultFileMode, FileSize: 4}, + Extended: map[string][]byte{s3_constants.ExtVersionIdKey: []byte(versionId)}, + } + + req := httptest.NewRequest(http.MethodPut, "/"+dst+"/"+object, nil) + req = mux.SetURLVars(req, map[string]string{"bucket": dst, "object": object}) + req.Header.Set("X-Amz-Copy-Source", "/"+src+"/"+object+"?versionId="+versionId) + rr := httptest.NewRecorder() + s3a.CopyObjectHandler(rr, req) + if rr.Code != http.StatusOK { + t.Fatalf("status = %d; body: %s", rr.Code, rr.Body.String()) + } + if got := rr.Header().Get("x-amz-copy-source-version-id"); got != versionId { + t.Fatalf("x-amz-copy-source-version-id = %q, want %q", got, versionId) + } +} + +func versionedCopyFixture(src, object, versionId string, filer *copyACLFiler) { + filer.entries["/buckets/"+src+"/"+object+s3_constants.VersionsFolder] = &filer_pb.Entry{ + Name: object + s3_constants.VersionsFolder, IsDirectory: true, + Extended: map[string][]byte{ + s3_constants.ExtLatestVersionIdKey: []byte(versionId), + s3_constants.ExtLatestVersionFileNameKey: []byte("v_" + versionId), + }, + } + filer.entries["/buckets/"+src+"/"+object+s3_constants.VersionsFolder+"/v_"+versionId] = &filer_pb.Entry{ + Name: "v_" + versionId, Content: []byte("data"), + Attributes: &filer_pb.FuseAttributes{FileMode: defaultFileMode, FileSize: 4}, + Extended: map[string][]byte{s3_constants.ExtVersionIdKey: []byte(versionId)}, + } +} + +// A copy naming no version reads the latest one, resolved from the .versions +// pointer, and still reports it in x-amz-copy-source-version-id — even when the +// version file predates the version-id attribute. +func TestCopyObjectReportsLatestSourceVersionId(t *testing.T) { + const src, dst, object, versionId = "versioned-src", "plain-dst", "object", "3a8f2c1d9e7b4a6f" + filer := ©ACLFiler{ambiguousPutFiler: &ambiguousPutFiler{apply: true, entries: map[string]*filer_pb.Entry{}}} + s3a := newPutTestServer(t, startFakeFiler(t, filer)) + s3a.iam = &IdentityAccessManagement{} + s3a.bucketConfigCache = NewBucketConfigCache(time.Minute) + s3a.bucketConfigCache.Set(src, &BucketConfig{Name: src, Versioning: s3_constants.VersioningEnabled, Ownership: s3_constants.OwnershipObjectWriter}) + s3a.bucketConfigCache.Set(dst, &BucketConfig{Name: dst, Ownership: s3_constants.OwnershipObjectWriter}) + versionedCopyFixture(src, object, versionId, filer) + delete(filer.entries["/buckets/"+src+"/"+object+s3_constants.VersionsFolder+"/v_"+versionId].Extended, s3_constants.ExtVersionIdKey) + + req := httptest.NewRequest(http.MethodPut, "/"+dst+"/"+object, nil) + req = mux.SetURLVars(req, map[string]string{"bucket": dst, "object": object}) + req.Header.Set("X-Amz-Copy-Source", "/"+src+"/"+object) + rr := httptest.NewRecorder() + s3a.CopyObjectHandler(rr, req) + if rr.Code != http.StatusOK { + t.Fatalf("status = %d; body: %s", rr.Code, rr.Body.String()) + } + if got := rr.Header().Get("x-amz-copy-source-version-id"); got != versionId { + t.Fatalf("x-amz-copy-source-version-id = %q, want %q", got, versionId) + } +} + +// UploadPartCopy answers with the same header for a versioned source. +func TestCopyObjectPartReportsSourceVersionId(t *testing.T) { + const src, dst, object, versionId = "versioned-src", "plain-dst", "object", "3a8f2c1d9e7b4a6f" + filer := ©ACLFiler{ambiguousPutFiler: &ambiguousPutFiler{volume: startFakeVolumeServer(t), apply: true, entries: map[string]*filer_pb.Entry{}}} + s3a := newPutTestServer(t, startFakeFiler(t, filer)) + s3a.iam = &IdentityAccessManagement{} + s3a.bucketConfigCache = NewBucketConfigCache(time.Minute) + s3a.bucketConfigCache.Set(src, &BucketConfig{Name: src, Versioning: s3_constants.VersioningEnabled, Ownership: s3_constants.OwnershipObjectWriter}) + s3a.bucketConfigCache.Set(dst, &BucketConfig{Name: dst, Ownership: s3_constants.OwnershipObjectWriter}) + util_http.InitGlobalHttpClient() + versionedCopyFixture(src, object, versionId, filer) + delete(filer.entries["/buckets/"+src+"/"+object+s3_constants.VersionsFolder+"/v_"+versionId].Extended, s3_constants.ExtVersionIdKey) + + uploadID := s3a.generateUploadID(object) + filer.entries["/buckets/"+dst+"/.uploads/"+uploadID] = &filer_pb.Entry{ + Name: uploadID, + IsDirectory: true, + Extended: map[string][]byte{s3_constants.ExtMultipartObjectKey: []byte(object)}, + } + + req := httptest.NewRequest(http.MethodPut, "/"+dst+"/"+object+"?partNumber=1&uploadId="+uploadID, nil) + req = mux.SetURLVars(req, map[string]string{"bucket": dst, "object": object}) + req.Header.Set("X-Amz-Copy-Source", "/"+src+"/"+object) + rr := httptest.NewRecorder() + s3a.CopyObjectPartHandler(rr, req) + if rr.Code != http.StatusOK { + t.Fatalf("status = %d; body: %s", rr.Code, rr.Body.String()) + } + if got := rr.Header().Get("x-amz-copy-source-version-id"); got != versionId { + t.Fatalf("x-amz-copy-source-version-id = %q, want %q", got, versionId) + } +} + +// A copy whose source is the null version — a regular-path object in a +// versioning-configured bucket — reports it as AWS does: "null". +func TestCopyObjectReportsNullSourceVersionId(t *testing.T) { + const src, dst, object = "suspended-src", "plain-dst", "object" + filer := ©ACLFiler{ambiguousPutFiler: &ambiguousPutFiler{apply: true, entries: map[string]*filer_pb.Entry{}}} + s3a := newPutTestServer(t, startFakeFiler(t, filer)) + s3a.iam = &IdentityAccessManagement{} + s3a.bucketConfigCache = NewBucketConfigCache(time.Minute) + s3a.bucketConfigCache.Set(src, &BucketConfig{Name: src, Versioning: s3_constants.VersioningSuspended, Ownership: s3_constants.OwnershipObjectWriter}) + s3a.bucketConfigCache.Set(dst, &BucketConfig{Name: dst, Ownership: s3_constants.OwnershipObjectWriter}) + filer.entries["/buckets/"+src+"/"+object] = &filer_pb.Entry{ + Name: object, Content: []byte("data"), + Attributes: &filer_pb.FuseAttributes{FileMode: defaultFileMode, FileSize: 4}, + } + + for _, copySource := range []string{"/" + src + "/" + object, "/" + src + "/" + object + "?versionId=null"} { + rr := httptest.NewRecorder() + req := httptest.NewRequest(http.MethodPut, "/"+dst+"/"+object, nil) + req = mux.SetURLVars(req, map[string]string{"bucket": dst, "object": object}) + req.Header.Set("X-Amz-Copy-Source", copySource) + s3a.CopyObjectHandler(rr, req) + if rr.Code != http.StatusOK { + t.Fatalf("copySource %q: status = %d; body: %s", copySource, rr.Code, rr.Body.String()) + } + if got := rr.Header().Get("x-amz-copy-source-version-id"); got != "null" { + t.Fatalf("copySource %q: x-amz-copy-source-version-id = %q, want %q", copySource, got, "null") + } + } +} + +// An unversioned source reports no copy-source version header at all. +func TestCopyObjectOmitsSourceVersionIdWhenUnversioned(t *testing.T) { + const src, dst, object = "plain-src", "plain-dst", "object" + filer := ©ACLFiler{ambiguousPutFiler: &ambiguousPutFiler{apply: true, entries: map[string]*filer_pb.Entry{}}} + s3a := newPutTestServer(t, startFakeFiler(t, filer)) + s3a.iam = &IdentityAccessManagement{} + s3a.bucketConfigCache = NewBucketConfigCache(time.Minute) + s3a.bucketConfigCache.Set(src, &BucketConfig{Name: src, Ownership: s3_constants.OwnershipObjectWriter}) + s3a.bucketConfigCache.Set(dst, &BucketConfig{Name: dst, Ownership: s3_constants.OwnershipObjectWriter}) + filer.entries["/buckets/"+src+"/"+object] = &filer_pb.Entry{ + Name: object, Content: []byte("data"), + Attributes: &filer_pb.FuseAttributes{FileMode: defaultFileMode, FileSize: 4}, + } + + rr := httptest.NewRecorder() + req := httptest.NewRequest(http.MethodPut, "/"+dst+"/"+object, nil) + req = mux.SetURLVars(req, map[string]string{"bucket": dst, "object": object}) + req.Header.Set("X-Amz-Copy-Source", "/"+src+"/"+object) + s3a.CopyObjectHandler(rr, req) + if rr.Code != http.StatusOK { + t.Fatalf("status = %d; body: %s", rr.Code, rr.Body.String()) + } + if got := rr.Header().Get("x-amz-copy-source-version-id"); got != "" { + t.Fatalf("x-amz-copy-source-version-id = %q, want empty", got) + } +} diff --git a/weed/s3api/s3api_object_versioning.go b/weed/s3api/s3api_object_versioning.go index d2b10c2016..f7ec672d29 100644 --- a/weed/s3api/s3api_object_versioning.go +++ b/weed/s3api/s3api_object_versioning.go @@ -2078,7 +2078,23 @@ func (s3a *S3ApiServer) doGetLatestObjectVersion(bucket, object string, maxRetri return nil, fmt.Errorf("failed to get latest version file %s: %v", latestVersionPath, err) } - return latestVersionEntry, nil + return attachResolvedVersionId(latestVersionEntry, latestVersionId), nil +} + +// attachResolvedVersionId stamps the resolved id on a versioned child that +// predates the version-id attribute, so readers see the id the pointer or the +// rescan already recovered without re-deriving it from the file name. +func attachResolvedVersionId(entry *filer_pb.Entry, versionId string) *filer_pb.Entry { + if entry == nil || versionId == "" { + return entry + } + if entry.Extended == nil { + entry.Extended = make(map[string][]byte) + } + if len(entry.Extended[s3_constants.ExtVersionIdKey]) == 0 { + entry.Extended[s3_constants.ExtVersionIdKey] = []byte(versionId) + } + return entry } // recoverLatestVersionWithoutPointer handles a .versions directory that exists @@ -2249,6 +2265,8 @@ func (s3a *S3ApiServer) healStaleLatestVersionPointer(bucket, normalizedObject s return nil, fmt.Errorf("%w: no remaining version in %s", filer_pb.ErrNotFound, versionsDir) } + latestEntry = attachResolvedVersionId(latestEntry, latestVersionId) + // The persist is CAS-style (mirroring clearStaleLatestVersionPointer) so a // stale repair cannot supersede a concurrent writer: the rescan takes time, // during which a PUT or delete may have atomically advanced the pointer on