s3: give a versioned metadata-only copy its own chunks (#10594)

* s3: give a versioned metadata-only copy its own chunks

A self-copy that only rewrites metadata clones the source entry, chunk fids
and all, and writes the clone back. With no versioning that is exactly right:
the clone replaces the entry it came from, so one entry owns the needles the
whole time. Under versioning the clone lands in a new .versions/ file and the
source stays live, and nothing refcounts a plain shared chunk list -- deleting
either version (a NoncurrentVersionExpiration rule, say) frees needles the
other still points at, and the next vacuum makes that permanent. rclone hits
this on every upload, since it stamps mtime with exactly this copy.

Take the metadata-only path only where the write replaces the entry it read:
the bare key of a bucket without versioning. Versioned, suspended, and
versionId-pinned copies fall through to the regular copy path, which gives the
destination its own chunks.

* s3: reencrypt a versioned SSE-KMS key rotation instead of reusing the chunks

A same-object copy that changes the KMS key id hands the source chunks straight
back, on the assumption that the copy overwrites the entry they came from. A
versioned bucket writes a new version beside the source instead, so the two end
up sharing needles that nothing refcounts, and deleting either one frees the
other's data. Reuse the chunks only when the destination really is the source
entry; otherwise fall through to the reencrypt path, which also gives the new
version the key it asked for rather than leaving it on the old one.

* s3: make one predicate decide whether a copy replaces its source

The metadata-only branch and the key-rotation strategy both answer the same
question -- does this copy write back to the entry it read -- so let them share
one predicate instead of pairing a same-destination check with it separately at
each site.

* test(s3): fail the copy regression tests when the vacuum does not run

The helper swallowed a failed or non-200 request to the master, so a vacuum
that never ran turned both chunk-ownership assertions into no-ops: the
tombstoned needles were still readable and the surviving version looked fine
either way. Require the endpoint, the request, and a 200.

* ci(s3): run every versioning test in the regression gate

The gate named the tests it wanted, so a new regression test sat there
uncovered until someone remembered this file -- it fooled me into thinking two
tests added in this PR never ran anywhere, when the comprehensive job had them
all along. Invert it: run everything, and name a test only to keep it out. The
delete job beside this one already works that way, and the suite costs about
two minutes.

Only the pagination stress tests are excluded; they build 1500+ versions, skip
themselves without ENABLE_STRESS_TESTS, and have their own make target.

Go's regexp has no negation, so the pattern is still assembled from a listing,
the way the volume-server integration workflow does it. Note the trailing $$:
make eats a lone trailing $ and takes the anchor with it.

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()