diff --git a/pkg/blob/s3.go b/pkg/blob/s3.go index ab45bb19f..22f0bd270 100644 --- a/pkg/blob/s3.go +++ b/pkg/blob/s3.go @@ -19,7 +19,9 @@ import ( // Writes go to a hidden staging prefix (.staging/) so that // in-progress multipart uploads can't collide with the final // content-addressed key. Complete renames staging -> the configured -// key via CopyObject + DeleteObject. +// key via a server-side Copy (single CopyObject, or a multipart +// UploadPartCopy for objects past S3's 5 GiB single-copy cap) + +// DeleteObject. type S3Store struct { client *awss3.Client bucket string @@ -84,12 +86,10 @@ func (s *S3Store) NewWriter(ctx context.Context, key, contentType string) (Write } func (s *S3Store) Move(ctx context.Context, srcKey, dstKey string) error { - _, err := s.client.CopyObject(ctx, &awss3.CopyObjectInput{ - Bucket: aws.String(s.bucket), - Key: aws.String(dstKey), - CopySource: aws.String(s.bucket + "/" + srcKey), - }) - if err != nil { + // Copy handles the >5 GiB VODs that a single CopyObject can't (it falls + // back to a multipart server-side copy), and HEADs the source first so a + // missing key surfaces as a NotFound we can treat idempotently below. + if err := s3pkg.Copy(ctx, s.client, s.bucket, srcKey, dstKey); err != nil { if isS3NotFound(err) { // Idempotency: maybe a previous Move already renamed // source -> dest. If dest exists, we're done. diff --git a/pkg/blob/store.go b/pkg/blob/store.go index c54e1df5c..fb5a4dfee 100644 --- a/pkg/blob/store.go +++ b/pkg/blob/store.go @@ -51,7 +51,7 @@ type Store interface { // Move relocates the blob from srcKey to dstKey atomically (where // the underlying storage permits — POSIX rename on FileStore, - // CopyObject+DeleteObject on S3Store). If dstKey already exists, + // server-side Copy+DeleteObject on S3Store). If dstKey already exists, // it is overwritten. Returns nil if srcKey does not exist after a // successful Move (idempotency for retried renames). Move(ctx context.Context, srcKey, dstKey string) error diff --git a/pkg/s3/copy.go b/pkg/s3/copy.go new file mode 100644 index 000000000..b7f8e4dbe --- /dev/null +++ b/pkg/s3/copy.go @@ -0,0 +1,188 @@ +package s3 + +import ( + "context" + "fmt" + + "github.com/aws/aws-sdk-go-v2/aws" + "github.com/aws/aws-sdk-go-v2/service/s3" + "github.com/aws/aws-sdk-go-v2/service/s3/types" + "go.opentelemetry.io/otel/attribute" + "go.opentelemetry.io/otel/codes" + "go.opentelemetry.io/otel/trace" + "golang.org/x/sync/errgroup" + "stream.place/streamplace/pkg/log" +) + +// maxCopyObjectSize is the largest object S3 will copy in a single +// CopyObject call (5 GiB). Anything larger fails with EntityTooLarge and +// must be copied part-by-part with a multipart upload driven by +// UploadPartCopy. +const maxCopyObjectSize = 5 * 1024 * 1024 * 1024 + +// copyPartSize is the byte range each UploadPartCopy transfers. The copy +// happens server-side inside S3, so — unlike MultipartPartSize, which sizes +// an in-process buffer — this never costs us memory; it's sized large to +// keep the part (and request) count low while staying under S3's 5 GiB +// per-part ceiling. At 1 GiB/part the 10000-part limit allows objects up to +// ~10 TiB, far beyond any VOD. +const copyPartSize = 1024 * 1024 * 1024 + +// copyConcurrency bounds how many UploadPartCopy requests run at once. +// Mirrors multipartUploadConcurrency: server-side range copies serialize +// badly otherwise, dragging a large VOD's finalize out for minutes. +const copyConcurrency = 8 + +// copyAPI is the subset of *s3.Client that Copy uses. Pulled out so tests +// can inject a fake; *s3.Client satisfies it. +type copyAPI interface { + HeadObject(context.Context, *s3.HeadObjectInput, ...func(*s3.Options)) (*s3.HeadObjectOutput, error) + CopyObject(context.Context, *s3.CopyObjectInput, ...func(*s3.Options)) (*s3.CopyObjectOutput, error) + CreateMultipartUpload(context.Context, *s3.CreateMultipartUploadInput, ...func(*s3.Options)) (*s3.CreateMultipartUploadOutput, error) + UploadPartCopy(context.Context, *s3.UploadPartCopyInput, ...func(*s3.Options)) (*s3.UploadPartCopyOutput, error) + CompleteMultipartUpload(context.Context, *s3.CompleteMultipartUploadInput, ...func(*s3.Options)) (*s3.CompleteMultipartUploadOutput, error) + AbortMultipartUpload(context.Context, *s3.AbortMultipartUploadInput, ...func(*s3.Options)) (*s3.AbortMultipartUploadOutput, error) +} + +// Copy copies an object within bucket from srcKey to dstKey, preserving the +// source's content type. Objects at or below maxCopyObjectSize use a single +// CopyObject; larger objects exceed S3's single-copy limit, so they're +// copied with a multipart upload whose parts are server-side UploadPartCopy +// range copies. A HeadObject against the source missing key surfaces as a +// NotFound error the caller can sniff for idempotency. +func Copy(ctx context.Context, client *s3.Client, bucket, srcKey, dstKey string) error { + return copyObject(ctx, client, bucket, srcKey, dstKey) +} + +func copyObject(ctx context.Context, client copyAPI, bucket, srcKey, dstKey string) error { + ctx = log.WithLogValues(ctx, "func", "s3.Copy") + ctx, span := s3Tracer.Start(ctx, "s3.Copy", trace.WithAttributes( + attribute.String("bucket", bucket), + attribute.String("src_key", srcKey), + attribute.String("dst_key", dstKey), + )) + defer span.End() + + head, err := client.HeadObject(ctx, &s3.HeadObjectInput{ + Bucket: aws.String(bucket), + Key: aws.String(srcKey), + }) + if err != nil { + span.RecordError(err) + return fmt.Errorf("head s3://%s/%s: %w", bucket, srcKey, err) + } + size := aws.ToInt64(head.ContentLength) + contentType := aws.ToString(head.ContentType) + span.SetAttributes(attribute.Int64("size_bytes", size)) + + // CopySource is "bucket/key"; keep it unescaped to match what the rest + // of the codebase already sends (our keys are content hashes, UUIDs and + // DIDs that the endpoint accepts raw). + copySource := bucket + "/" + srcKey + + if size <= maxCopyObjectSize { + if _, err := client.CopyObject(ctx, &s3.CopyObjectInput{ + Bucket: aws.String(bucket), + Key: aws.String(dstKey), + CopySource: aws.String(copySource), + }); err != nil { + span.RecordError(err) + span.SetStatus(codes.Error, "copy_object") + return fmt.Errorf("copy s3://%s/%s -> %s: %w", bucket, srcKey, dstKey, err) + } + return nil + } + + if err := multipartCopy(ctx, client, bucket, dstKey, copySource, contentType, size); err != nil { + span.RecordError(err) + span.SetStatus(codes.Error, "multipart_copy") + return err + } + return nil +} + +// multipartCopy copies an over-5-GiB object by opening a multipart upload at +// dstKey and filling it with server-side UploadPartCopy range copies of the +// source. Parts run concurrently (bounded by copyConcurrency) but land at +// their fixed part numbers, so the completed-parts list is already ordered. +// Any failure aborts the upload so S3 doesn't retain orphaned parts. +func multipartCopy(ctx context.Context, client copyAPI, bucket, dstKey, copySource, contentType string, size int64) error { + create := &s3.CreateMultipartUploadInput{ + Bucket: aws.String(bucket), + Key: aws.String(dstKey), + } + if contentType != "" { + create.ContentType = aws.String(contentType) + } + resp, err := client.CreateMultipartUpload(ctx, create) + if err != nil { + return fmt.Errorf("create multipart copy s3://%s/%s: %w", bucket, dstKey, err) + } + uploadID := aws.ToString(resp.UploadId) + + // Pre-compute the (start,end) byte range of every part. Part numbers are + // 1-based; CopySourceRange's end offset is inclusive. + type byteRange struct{ start, end int64 } + var ranges []byteRange + for off := int64(0); off < size; off += copyPartSize { + end := off + copyPartSize - 1 + if end > size-1 { + end = size - 1 + } + ranges = append(ranges, byteRange{start: off, end: end}) + } + + parts := make([]types.CompletedPart, len(ranges)) + g, gctx := errgroup.WithContext(ctx) + g.SetLimit(copyConcurrency) + for i, r := range ranges { + partNum := int32(i + 1) + g.Go(func() error { + res, err := client.UploadPartCopy(gctx, &s3.UploadPartCopyInput{ + Bucket: aws.String(bucket), + Key: aws.String(dstKey), + UploadId: aws.String(uploadID), + PartNumber: aws.Int32(partNum), + CopySource: aws.String(copySource), + CopySourceRange: aws.String(fmt.Sprintf("bytes=%d-%d", r.start, r.end)), + }) + if err != nil { + return fmt.Errorf("upload part copy %d (bytes %d-%d) s3://%s/%s: %w", partNum, r.start, r.end, bucket, dstKey, err) + } + if res.CopyPartResult == nil { + return fmt.Errorf("upload part copy %d s3://%s/%s: missing CopyPartResult", partNum, bucket, dstKey) + } + parts[i] = types.CompletedPart{ + ETag: res.CopyPartResult.ETag, + PartNumber: aws.Int32(partNum), + } + return nil + }) + } + if err := g.Wait(); err != nil { + // Best-effort cleanup so a failed copy doesn't leave parts S3 keeps + // billing for. Use ctx (not the cancelled gctx) so the abort runs. + _, _ = client.AbortMultipartUpload(ctx, &s3.AbortMultipartUploadInput{ + Bucket: aws.String(bucket), + Key: aws.String(dstKey), + UploadId: aws.String(uploadID), + }) + return err + } + + if _, err := client.CompleteMultipartUpload(ctx, &s3.CompleteMultipartUploadInput{ + Bucket: aws.String(bucket), + Key: aws.String(dstKey), + UploadId: aws.String(uploadID), + MultipartUpload: &types.CompletedMultipartUpload{Parts: parts}, + }); err != nil { + _, _ = client.AbortMultipartUpload(ctx, &s3.AbortMultipartUploadInput{ + Bucket: aws.String(bucket), + Key: aws.String(dstKey), + UploadId: aws.String(uploadID), + }) + return fmt.Errorf("complete multipart copy s3://%s/%s: %w", bucket, dstKey, err) + } + log.Log(ctx, "completed S3 multipart copy", "bucket", bucket, "key", dstKey, "parts", len(parts), "size", size) + return nil +} diff --git a/pkg/s3/copy_test.go b/pkg/s3/copy_test.go new file mode 100644 index 000000000..e98bc524c --- /dev/null +++ b/pkg/s3/copy_test.go @@ -0,0 +1,214 @@ +package s3 + +import ( + "context" + "fmt" + "sync" + "testing" + "time" + + "github.com/aws/aws-sdk-go-v2/aws" + "github.com/aws/aws-sdk-go-v2/service/s3" + "github.com/aws/aws-sdk-go-v2/service/s3/types" + "github.com/stretchr/testify/require" +) + +// fakeCopyClient is an in-memory stand-in for the subset of *s3.Client that +// Copy uses. HeadObject reports a configurable size/content-type (so the +// >5 GiB multipart path can be exercised without allocating gigabytes), and +// the multipart ops record the ranges they were asked to copy. +type fakeCopyClient struct { + headSize int64 + headContentType string + headErr error + + copyDelay time.Duration + failPart int32 // if >0, UploadPartCopy for this part number returns an error + + srcBytes []byte // for the small-object CopyObject path + dst map[string][]byte // CopyObject writes here when non-nil + + mu sync.Mutex + copyObjectCalls int + createCalls int + createdType string + partRanges map[int32][2]int64 + completedParts []types.CompletedPart + aborted bool + inFlight int + maxInFlight int +} + +func (f *fakeCopyClient) HeadObject(context.Context, *s3.HeadObjectInput, ...func(*s3.Options)) (*s3.HeadObjectOutput, error) { + if f.headErr != nil { + return nil, f.headErr + } + out := &s3.HeadObjectOutput{ContentLength: aws.Int64(f.headSize)} + if f.headContentType != "" { + out.ContentType = aws.String(f.headContentType) + } + return out, nil +} + +func (f *fakeCopyClient) CopyObject(_ context.Context, in *s3.CopyObjectInput, _ ...func(*s3.Options)) (*s3.CopyObjectOutput, error) { + f.mu.Lock() + defer f.mu.Unlock() + f.copyObjectCalls++ + if f.dst != nil { + f.dst[aws.ToString(in.Key)] = f.srcBytes + } + return &s3.CopyObjectOutput{}, nil +} + +func (f *fakeCopyClient) CreateMultipartUpload(_ context.Context, in *s3.CreateMultipartUploadInput, _ ...func(*s3.Options)) (*s3.CreateMultipartUploadOutput, error) { + f.mu.Lock() + defer f.mu.Unlock() + f.createCalls++ + f.createdType = aws.ToString(in.ContentType) + return &s3.CreateMultipartUploadOutput{UploadId: aws.String("test-upload-id")}, nil +} + +func (f *fakeCopyClient) UploadPartCopy(_ context.Context, in *s3.UploadPartCopyInput, _ ...func(*s3.Options)) (*s3.UploadPartCopyOutput, error) { + f.mu.Lock() + f.inFlight++ + if f.inFlight > f.maxInFlight { + f.maxInFlight = f.inFlight + } + f.mu.Unlock() + + if f.copyDelay > 0 { + time.Sleep(f.copyDelay) + } + + num := aws.ToInt32(in.PartNumber) + var start, end int64 + if _, err := fmt.Sscanf(aws.ToString(in.CopySourceRange), "bytes=%d-%d", &start, &end); err != nil { + return nil, fmt.Errorf("bad CopySourceRange %q: %w", aws.ToString(in.CopySourceRange), err) + } + + f.mu.Lock() + defer f.mu.Unlock() + f.inFlight-- + if f.failPart > 0 && num == f.failPart { + return nil, fmt.Errorf("simulated failure on part %d", num) + } + f.partRanges[num] = [2]int64{start, end} + return &s3.UploadPartCopyOutput{ + CopyPartResult: &types.CopyPartResult{ETag: aws.String(fmt.Sprintf("etag-%d", num))}, + }, nil +} + +func (f *fakeCopyClient) CompleteMultipartUpload(_ context.Context, in *s3.CompleteMultipartUploadInput, _ ...func(*s3.Options)) (*s3.CompleteMultipartUploadOutput, error) { + f.mu.Lock() + defer f.mu.Unlock() + f.completedParts = in.MultipartUpload.Parts + return &s3.CompleteMultipartUploadOutput{}, nil +} + +func (f *fakeCopyClient) AbortMultipartUpload(context.Context, *s3.AbortMultipartUploadInput, ...func(*s3.Options)) (*s3.AbortMultipartUploadOutput, error) { + f.mu.Lock() + defer f.mu.Unlock() + f.aborted = true + return &s3.AbortMultipartUploadOutput{}, nil +} + +// TestCopySmallObjectUsesCopyObject verifies sub-5-GiB objects take the +// single-call CopyObject path and land byte-for-byte at the destination. +func TestCopySmallObjectUsesCopyObject(t *testing.T) { + src := []byte("hello world, a small object") + fake := &fakeCopyClient{ + headSize: int64(len(src)), + headContentType: "video/mp4", + srcBytes: src, + dst: map[string][]byte{}, + partRanges: map[int32][2]int64{}, + } + require.NoError(t, copyObject(context.Background(), fake, "bucket", "src", "dst")) + require.Equal(t, 1, fake.copyObjectCalls) + require.Equal(t, 0, fake.createCalls, "small object must not open a multipart upload") + require.Equal(t, src, fake.dst["dst"]) +} + +// TestCopyAtThresholdUsesCopyObject pins the boundary: an object exactly at +// maxCopyObjectSize still copies in one shot. +func TestCopyAtThresholdUsesCopyObject(t *testing.T) { + fake := &fakeCopyClient{headSize: maxCopyObjectSize, partRanges: map[int32][2]int64{}} + require.NoError(t, copyObject(context.Background(), fake, "bucket", "src", "dst")) + require.Equal(t, 1, fake.copyObjectCalls) + require.Equal(t, 0, fake.createCalls) +} + +// TestCopyLargeObjectUsesMultipart verifies an over-5-GiB object is copied +// with a multipart UploadPartCopy whose parts tile the source exactly, +// preserve content type, finish in ascending order, and run concurrently. +func TestCopyLargeObjectUsesMultipart(t *testing.T) { + const size = int64(12) * 1024 * 1024 * 1024 // 12 GiB -> spans many parts + fake := &fakeCopyClient{ + headSize: size, + headContentType: "video/mp4", + copyDelay: 5 * time.Millisecond, // force overlap so concurrency is observable + partRanges: map[int32][2]int64{}, + } + require.NoError(t, copyObject(context.Background(), fake, "bucket", "src", "dst")) + + require.Equal(t, 0, fake.copyObjectCalls, "large object must not use single CopyObject") + require.Equal(t, 1, fake.createCalls) + require.Equal(t, "video/mp4", fake.createdType, "content type must be preserved") + + wantParts := int((size + copyPartSize - 1) / copyPartSize) + require.Len(t, fake.partRanges, wantParts) + require.Len(t, fake.completedParts, wantParts) + + // Parts must tile [0,size) contiguously with no gaps or overlaps, and + // every part except the last is exactly copyPartSize. + var covered int64 + for n := int32(1); n <= int32(wantParts); n++ { + r, ok := fake.partRanges[n] + require.Truef(t, ok, "missing part %d", n) + require.Equalf(t, covered, r[0], "part %d start should continue from previous end", n) + require.GreaterOrEqual(t, r[1], r[0]) + if int(n) < wantParts { + require.Equalf(t, int64(copyPartSize), r[1]-r[0]+1, "non-final part %d should be a full part", n) + } + covered = r[1] + 1 + } + require.Equal(t, size, covered, "parts must cover the whole object") + + // CompleteMultipartUpload requires ascending part numbers. + for i := 1; i < len(fake.completedParts); i++ { + require.Less(t, + aws.ToInt32(fake.completedParts[i-1].PartNumber), + aws.ToInt32(fake.completedParts[i].PartNumber)) + } + require.Greater(t, fake.maxInFlight, 1, "expected part copies to run concurrently") + require.False(t, fake.aborted) +} + +// TestCopyLargeObjectAbortsOnPartError verifies a failed part copy aborts +// the multipart upload (so S3 doesn't retain orphaned parts) and never +// completes it. +func TestCopyLargeObjectAbortsOnPartError(t *testing.T) { + const size = int64(12) * 1024 * 1024 * 1024 + fake := &fakeCopyClient{ + headSize: size, + partRanges: map[int32][2]int64{}, + failPart: 3, + } + err := copyObject(context.Background(), fake, "bucket", "src", "dst") + require.Error(t, err) + require.Contains(t, err.Error(), "upload part copy 3") + require.True(t, fake.aborted, "a failed part must abort the upload") + require.Empty(t, fake.completedParts, "must not complete after a part failure") +} + +// TestCopyHeadErrorPropagates verifies a missing source surfaces the +// HeadObject error (which blob.Move sniffs as NotFound for idempotency) +// before any copy is attempted. +func TestCopyHeadErrorPropagates(t *testing.T) { + fake := &fakeCopyClient{headErr: fmt.Errorf("api error NotFound: object missing")} + err := copyObject(context.Background(), fake, "bucket", "missing", "dst") + require.Error(t, err) + require.Contains(t, err.Error(), "NotFound") + require.Equal(t, 0, fake.copyObjectCalls) + require.Equal(t, 0, fake.createCalls) +}