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