diff --git a/.github/workflows/s3-mutation-regression-tests.yml b/.github/workflows/s3-mutation-regression-tests.yml index 16df18c8b..acf09a8f2 100644 --- a/.github/workflows/s3-mutation-regression-tests.yml +++ b/.github/workflows/s3-mutation-regression-tests.yml @@ -38,7 +38,22 @@ jobs: working-directory: test/s3/versioning run: | set -x - make test-with-server TEST_PATTERN="TestVersioningCompleteMultipartUploadIsIdempotent|TestVersioningSelfCopyMetadataReplaceCreatesNewVersion|TestVersioningSelfCopyMetadataReplaceSuspendedKeepsNullVersion|TestSuspendedDeleteCreatesDeleteMarker" + # Run every versioning test, so a regression test lands covered instead + # of waiting for someone to remember this file. Name a test in EXCLUDE, + # with the reason, to keep it out. + # + # TestVersioningPagination*: opt-in stress tests that build 1500+ + # versions. They self-skip without ENABLE_STRESS_TESTS and have their + # own make target, so this gate should not carry them. + EXCLUDE='TestVersioningPagination.*' + tests=$(go test . -list '.*' | grep '^Test' | sort -u) + # An empty list would make -run match nothing and pass this job vacuously. + [ -n "$tests" ] || { echo "listed no versioning tests"; exit 1; } + selected=$(echo "$tests" | grep -vE "^($EXCLUDE)$") + [ -n "$selected" ] || { echo "EXCLUDE matched every test"; exit 1; } + echo "running $(echo "$selected" | wc -l) of $(echo "$tests" | wc -l) versioning tests" + # make swallows a lone trailing $, taking the anchor with it, so escape it. + make test-with-server TEST_PATTERN="^($(echo "$selected" | paste -sd'|' -))"'$$' - name: Show server logs on failure if: failure() diff --git a/test/s3/versioning/s3_copy_versioning_regression_test.go b/test/s3/versioning/s3_copy_versioning_regression_test.go index 074aa283a..92bbc7864 100644 --- a/test/s3/versioning/s3_copy_versioning_regression_test.go +++ b/test/s3/versioning/s3_copy_versioning_regression_test.go @@ -6,8 +6,11 @@ import ( "errors" "fmt" "io" + "net/http" "net/url" + "strings" "testing" + "time" "github.com/aws/aws-sdk-go-v2/aws" "github.com/aws/aws-sdk-go-v2/service/s3" @@ -31,6 +34,152 @@ func suspendVersioning(t *testing.T, client *s3.Client, bucketName string) { require.NoError(t, err) } +// vacuumVolumes asks the master to compact away the needles a delete tombstoned. +// Tests that assert a surviving object still has its data need this: deleting an +// entry only tombstones the needles it points at, so a shared chunk list reads +// fine right up until the vacuum makes the loss permanent. A vacuum that does not +// run leaves those tests asserting nothing, so treat every failure as fatal. +func vacuumVolumes(t *testing.T) { + t.Helper() + require.NotEmpty(t, defaultConfig.MasterEndpoint, "vacuum needs a master endpoint; set MASTER_ENDPOINT") + endpoint := strings.TrimRight(defaultConfig.MasterEndpoint, "/") + "/vol/vacuum?garbageThreshold=0.001" + httpClient := &http.Client{Timeout: 30 * time.Second} + resp, err := httpClient.Get(endpoint) + require.NoError(t, err, "vacuum request to %s", endpoint) + defer resp.Body.Close() + _, err = io.Copy(io.Discard, resp.Body) + require.NoError(t, err, "reading the vacuum response") + require.Equal(t, http.StatusOK, resp.StatusCode, "vacuum request to %s", endpoint) +} + +func requireVersionBody(t *testing.T, client *s3.Client, bucketName, objectKey, versionId string, want []byte, msg string) { + t.Helper() + getResp, err := client.GetObject(context.TODO(), &s3.GetObjectInput{ + Bucket: aws.String(bucketName), + Key: aws.String(objectKey), + VersionId: aws.String(versionId), + }) + require.NoError(t, err, msg) + defer getResp.Body.Close() + body, err := io.ReadAll(getResp.Body) + require.NoError(t, err, msg) + require.Equal(t, len(want), len(body), msg) + require.True(t, bytes.Equal(want, body), msg) +} + +// chunkedTestContent returns a body large enough to land in volume needles rather +// than inline in the filer entry, so a copy that reuses the source fids is visible +// once those needles are freed. +func chunkedTestContent(size int) []byte { + content := make([]byte, size) + for i := range content { + content[i] = byte(i * 31 % 251) + } + return content +} + +// TestVersioningSelfCopyMetadataReplaceKeepsChunksIndependent covers the copy that +// only rewrites metadata: it used to hand the source version's chunk fids to the +// new version, so nothing owned those needles and deleting either version freed +// the survivor's data (silently, once a vacuum ran). +func TestVersioningSelfCopyMetadataReplaceKeepsChunksIndependent(t *testing.T) { + client := getS3Client(t) + bucketName := getNewBucketName() + + createBucket(t, client, bucketName) + defer deleteBucket(t, client, bucketName) + + enableVersioning(t, client, bucketName) + + objectKey := "self-copy-chunk-ownership.bin" + content := chunkedTestContent(6 << 20) + + putResp, err := client.PutObject(context.TODO(), &s3.PutObjectInput{ + Bucket: aws.String(bucketName), + Key: aws.String(objectKey), + Body: bytes.NewReader(content), + }) + require.NoError(t, err) + require.NotNil(t, putResp.VersionId) + + copyResp, err := client.CopyObject(context.TODO(), &s3.CopyObjectInput{ + Bucket: aws.String(bucketName), + Key: aws.String(objectKey), + CopySource: aws.String(versioningCopySource(bucketName, objectKey)), + Metadata: map[string]string{"mtime": "1653465360"}, + MetadataDirective: types.MetadataDirectiveReplace, + }) + require.NoError(t, err) + require.NotNil(t, copyResp.VersionId) + require.NotEqual(t, *putResp.VersionId, *copyResp.VersionId) + + _, err = client.DeleteObject(context.TODO(), &s3.DeleteObjectInput{ + Bucket: aws.String(bucketName), + Key: aws.String(objectKey), + VersionId: putResp.VersionId, + }) + require.NoError(t, err) + + // The filer frees a deleted entry's chunks asynchronously, so re-check across + // a few vacuum rounds instead of racing a single one. + for round := 0; round < 4; round++ { + time.Sleep(time.Second) + vacuumVolumes(t) + requireVersionBody(t, client, bucketName, objectKey, *copyResp.VersionId, content, + "the surviving version must keep its own data after the other version is deleted") + } +} + +// TestSuspendedSelfCopyMetadataReplaceKeepsChunksIndependent is the same defect on +// a suspended bucket: the null version the copy writes sits beside a .versions/ +// entry that stays live, so the two must not share needles either. +func TestSuspendedSelfCopyMetadataReplaceKeepsChunksIndependent(t *testing.T) { + client := getS3Client(t) + bucketName := getNewBucketName() + + createBucket(t, client, bucketName) + defer deleteBucket(t, client, bucketName) + + enableVersioning(t, client, bucketName) + + objectKey := "suspended-self-copy-chunk-ownership.bin" + content := chunkedTestContent(6 << 20) + + putResp, err := client.PutObject(context.TODO(), &s3.PutObjectInput{ + Bucket: aws.String(bucketName), + Key: aws.String(objectKey), + Body: bytes.NewReader(content), + }) + require.NoError(t, err) + require.NotNil(t, putResp.VersionId) + + suspendVersioning(t, client, bucketName) + + _, err = client.CopyObject(context.TODO(), &s3.CopyObjectInput{ + Bucket: aws.String(bucketName), + Key: aws.String(objectKey), + CopySource: aws.String(versioningCopySource(bucketName, objectKey)), + Metadata: map[string]string{"mtime": "1653465360"}, + MetadataDirective: types.MetadataDirectiveReplace, + }) + require.NoError(t, err) + + // Drop the version the copy read from; the null version it wrote must survive. + _, err = client.DeleteObject(context.TODO(), &s3.DeleteObjectInput{ + Bucket: aws.String(bucketName), + Key: aws.String(objectKey), + VersionId: putResp.VersionId, + }) + require.NoError(t, err) + + for round := 0; round < 4; round++ { + time.Sleep(time.Second) + vacuumVolumes(t) + requireVersionBody(t, client, bucketName, objectKey, "null", content, + "the null version must keep its own data after the version it was copied from is deleted") + } +} + func TestVersioningSelfCopyMetadataReplaceCreatesNewVersion(t *testing.T) { client := getS3Client(t) bucketName := getNewBucketName() diff --git a/weed/s3api/s3api_object_handlers_copy.go b/weed/s3api/s3api_object_handlers_copy.go index 5ed64ec86..3ae75be90 100644 --- a/weed/s3api/s3api_object_handlers_copy.go +++ b/weed/s3api/s3api_object_handlers_copy.go @@ -232,12 +232,13 @@ func (s3a *S3ApiServer) CopyObjectHandler(w http.ResponseWriter, r *http.Request return } - if sameDestination && (replaceMeta || replaceTagging) && s3a.canUseMetadataOnlySelfCopy(entry, r, dstBucket, dstObject) { + replacesSource := copyReplacesSourceEntry(sameDestination, dstVersioningState, srcVersionId) + + if replacesSource && (replaceMeta || replaceTagging) && s3a.canUseMetadataOnlySelfCopy(entry, r, dstBucket, dstObject) { var dstVersionId string var etag string - // A non-versioned in-place metadata replace routes to the owner as a - // serialized PATCH (off the distributed lock); versioned/suspended (which - // create a new version) and the no-owner bootstrap keep the lock. + // An in-place metadata replace routes to the owner as a serialized PATCH + // (off the distributed lock); the no-owner bootstrap keeps the lock. // // REPLACE can also change Content-Type, which lives on Attributes.Mime, // not Extended. The routed PATCH only carries Extended keys, so when the @@ -246,7 +247,7 @@ func (s3a *S3ApiServer) CopyObjectHandler(w http.ResponseWriter, r *http.Request owner := s3a.objectWriteOwner(dstBucket, dstObject) sourceMime := entry.GetAttributes().GetMime() mimeChanged := resolveDestinationMime(r.Header, sourceMime, replaceMeta) != sourceMime - routeInPlace := owner != "" && dstVersioningState == "" && !mimeChanged + routeInPlace := owner != "" && !mimeChanged selfCopyBody := func() s3err.ErrorCode { currentEntry, currentErr := s3a.resolveCopySourceEntry(srcBucket, srcObject, srcVersionId, srcVersioningState) if errCode := classifyCopySourceError(currentEntry, currentErr); errCode != s3err.ErrNone { @@ -416,7 +417,7 @@ func (s3a *S3ApiServer) CopyObjectHandler(w http.ResponseWriter, r *http.Request } } else { // Use unified copy strategy approach - dstChunks, dstMetadata, copyErr := s3a.executeUnifiedCopyStrategy(entry, r, srcBucket, dstBucket, srcObject, dstObject) + dstChunks, dstMetadata, copyErr := s3a.executeUnifiedCopyStrategy(entry, r, srcBucket, dstBucket, srcObject, dstObject, replacesSource) if copyErr != nil { glog.Errorf("CopyObjectHandler unified copy error: %v", copyErr) // Map errors to appropriate S3 errors @@ -474,6 +475,18 @@ func (s3a *S3ApiServer) CopyObjectHandler(w http.ResponseWriter, r *http.Request } +// copyReplacesSourceEntry reports whether a copy writes back to the very entry it +// read, which is what lets a strategy hand the source's chunk fids to the +// destination instead of copying the data. Nothing refcounts a plain shared chunk +// list, so a second live entry on the same chunks loses its data as soon as either +// side is deleted. A versioned destination writes a new version file, a suspended +// one writes the null version next to a .versions/ entry that stays live, and a +// source pinned to a versionId reads a version file that outlives the copy — those +// all need the chunks copied for real, as does any copy to a different key. +func copyReplacesSourceEntry(sameDestination bool, dstVersioningState, srcVersionId string) bool { + return sameDestination && dstVersioningState == "" && srcVersionId == "" +} + func cloneProtoEntry(entry *filer_pb.Entry) *filer_pb.Entry { if entry == nil { return nil diff --git a/weed/s3api/s3api_object_handlers_copy_self_reuse_test.go b/weed/s3api/s3api_object_handlers_copy_self_reuse_test.go new file mode 100644 index 000000000..cf17af7d8 --- /dev/null +++ b/weed/s3api/s3api_object_handlers_copy_self_reuse_test.go @@ -0,0 +1,31 @@ +package s3api + +import ( + "testing" + + "github.com/seaweedfs/seaweedfs/weed/s3api/s3_constants" +) + +func TestCopyReplacesSourceEntry(t *testing.T) { + cases := []struct { + name string + sameDestination bool + versioningState string + srcVersionId string + want bool + }{ + {"no versioning replaces the bare key", true, "", "", true}, + {"a copy to another key writes its own entry", false, "", "", false}, + {"versioning enabled writes a new version file", true, s3_constants.VersioningEnabled, "", false}, + {"suspended writes the null version beside live versions", true, s3_constants.VersioningSuspended, "", false}, + {"pinned source version outlives the copy", true, "", "6736fb618f225b190c06e5b4fb63c83b", false}, + {"pinned null source version", true, "", "null", false}, + } + for _, c := range cases { + t.Run(c.name, func(t *testing.T) { + if got := copyReplacesSourceEntry(c.sameDestination, c.versioningState, c.srcVersionId); got != c.want { + t.Errorf("copyReplacesSourceEntry(%v, %q, %q) = %v, want %v", c.sameDestination, c.versioningState, c.srcVersionId, got, c.want) + } + }) + } +} diff --git a/weed/s3api/s3api_object_handlers_copy_unified.go b/weed/s3api/s3api_object_handlers_copy_unified.go index 3728ac7d9..613e74844 100644 --- a/weed/s3api/s3api_object_handlers_copy_unified.go +++ b/weed/s3api/s3api_object_handlers_copy_unified.go @@ -12,8 +12,10 @@ import ( ) // executeUnifiedCopyStrategy executes the appropriate copy strategy based on encryption state -// Returns chunks and destination metadata that should be applied to the destination entry -func (s3a *S3ApiServer) executeUnifiedCopyStrategy(entry *filer_pb.Entry, r *http.Request, srcBucket, dstBucket, srcObject, dstObject string) ([]*filer_pb.FileChunk, map[string][]byte, error) { +// Returns chunks and destination metadata that should be applied to the destination entry. +// replacesSource says the destination entry is the source entry, which is what lets the +// key-rotation strategy hand back the source chunks instead of copying them. +func (s3a *S3ApiServer) executeUnifiedCopyStrategy(entry *filer_pb.Entry, r *http.Request, srcBucket, dstBucket, srcObject, dstObject string, replacesSource bool) ([]*filer_pb.FileChunk, map[string][]byte, error) { // Per-chunk copy must see data chunks: a manifest chunk copied raw becomes // object data. Resolved manifests stay with the source. if _, err := s3a.flattenManifestChunks(r.Context(), entry); err != nil { @@ -51,7 +53,7 @@ func (s3a *S3ApiServer) executeUnifiedCopyStrategy(entry *filer_pb.Entry, r *htt return chunks, nil, err case CopyStrategyKeyRotation: - return s3a.executeKeyRotation(entry, r, state, dstBucket, dstPath) + return s3a.executeKeyRotation(entry, r, state, dstBucket, dstPath, replacesSource) case CopyStrategyEncrypt: return s3a.executeEncryptCopy(entry, r, state, dstBucket, dstPath) @@ -96,7 +98,7 @@ func (s3a *S3ApiServer) mapCopyErrorToS3Error(err error) s3err.ErrorCode { } // executeKeyRotation handles key rotation for same-object copies -func (s3a *S3ApiServer) executeKeyRotation(entry *filer_pb.Entry, r *http.Request, state *EncryptionState, dstBucket, dstPath string) ([]*filer_pb.FileChunk, map[string][]byte, error) { +func (s3a *S3ApiServer) executeKeyRotation(entry *filer_pb.Entry, r *http.Request, state *EncryptionState, dstBucket, dstPath string, replacesSource bool) ([]*filer_pb.FileChunk, map[string][]byte, error) { // For key rotation, we only need to update metadata, not re-copy chunks // This is a significant optimization for same-object key changes @@ -105,7 +107,10 @@ func (s3a *S3ApiServer) executeKeyRotation(entry *filer_pb.Entry, r *http.Reques return s3a.executeReencryptCopy(entry, r, state, dstBucket, dstPath) } - if state.SrcSSEKMS && state.DstSSEKMS { + // Handing back the source chunks leaves the destination sharing needles nothing + // refcounts, so it is only safe when the destination overwrites the source entry. + // A versioned rotation writes a new version beside the source and has to reencrypt. + if state.SrcSSEKMS && state.DstSSEKMS && replacesSource { // SSE-KMS key rotation - return existing chunks, metadata will be updated by caller return entry.GetChunks(), nil, nil }