diff --git a/pkg/backend/internal/patchcopy/copy.go b/pkg/backend/internal/patchcopy/copy.go new file mode 100644 index 00000000..cf2043ad --- /dev/null +++ b/pkg/backend/internal/patchcopy/copy.go @@ -0,0 +1,154 @@ +// Package patchcopy runs retained multipart copies with bounded concurrency. +package patchcopy + +import ( + "context" + "errors" + "fmt" + "sync" + "time" +) + +// Copier copies an object range into one multipart part. +type Copier interface { + UploadPartCopy(ctx context.Context, destKey, uploadID string, partNumber int, sourceKey string, startByte, endByte int64) (string, error) +} + +// Aborter aborts an incomplete multipart upload. +type Aborter interface { + AbortMultipartUpload(ctx context.Context, key, uploadID string) error +} + +// Client supports both retained-part copies and multipart cleanup. +type Client interface { + Copier + Aborter +} + +// Task describes one retained multipart range. +type Task struct { + PartNumber int + StartByte int64 + EndByte int64 +} + +// PartError identifies the retained part whose copy failed. +type PartError struct { + PartNumber int + Err error +} + +func (e *PartError) Error() string { + return fmt.Sprintf("copy part %d: %v", e.PartNumber, e.Err) +} + +func (e *PartError) Unwrap() error { + return e.Err +} + +// Copy runs every task with at most maxConcurrency in-flight requests. +func Copy( + ctx context.Context, + copier Copier, + destKey string, + uploadID string, + sourceKey string, + tasks []Task, + maxConcurrency int, +) error { + if maxConcurrency <= 0 { + return fmt.Errorf("patch copy concurrency must be positive") + } + if err := ctx.Err(); err != nil { + return err + } + if len(tasks) == 0 { + return nil + } + + copyCtx, cancel := context.WithCancel(ctx) + defer cancel() + + workerCount := min(maxConcurrency, len(tasks)) + jobs := make(chan Task) + + var workers sync.WaitGroup + var failureOnce sync.Once + var failure *PartError + workers.Add(workerCount) + for range workerCount { + go func() { + defer workers.Done() + for task := range jobs { + if copyCtx.Err() != nil { + return + } + if _, err := copier.UploadPartCopy( + copyCtx, + destKey, + uploadID, + task.PartNumber, + sourceKey, + task.StartByte, + task.EndByte, + ); err != nil { + failureOnce.Do(func() { + failure = &PartError{PartNumber: task.PartNumber, Err: err} + cancel() + }) + return + } + } + }() + } + +feed: + for _, task := range tasks { + select { + case jobs <- task: + case <-copyCtx.Done(): + break feed + } + } + close(jobs) + workers.Wait() + + if err := ctx.Err(); err != nil { + return err + } + if failure != nil { + return failure + } + return nil +} + +// CopyOrAbort waits for all workers to stop, then aborts once after failure. +func CopyOrAbort( + ctx context.Context, + client Client, + destKey string, + uploadID string, + sourceKey string, + tasks []Task, + maxConcurrency int, + abortTimeout time.Duration, +) error { + copyErr := Copy(ctx, client, destKey, uploadID, sourceKey, tasks, maxConcurrency) + if copyErr == nil { + return nil + } + if abortErr := Abort(ctx, client, destKey, uploadID, abortTimeout); abortErr != nil { + return errors.Join(copyErr, fmt.Errorf("abort patch multipart upload: %w", abortErr)) + } + return copyErr +} + +// Abort uses a detached bounded context so caller cancellation cannot skip cleanup. +func Abort(ctx context.Context, aborter Aborter, key string, uploadID string, timeout time.Duration) error { + if timeout <= 0 { + return fmt.Errorf("patch abort timeout must be positive") + } + abortCtx, cancel := context.WithTimeout(context.WithoutCancel(ctx), timeout) + defer cancel() + return aborter.AbortMultipartUpload(abortCtx, key, uploadID) +} diff --git a/pkg/backend/internal/patchcopy/copy_test.go b/pkg/backend/internal/patchcopy/copy_test.go new file mode 100644 index 00000000..864b0983 --- /dev/null +++ b/pkg/backend/internal/patchcopy/copy_test.go @@ -0,0 +1,473 @@ +package patchcopy + +import ( + "context" + "errors" + "fmt" + "sort" + "sync" + "testing" + "time" +) + +type testClient struct { + mu sync.Mutex + + active int + maxActive int + calls []int + ranges map[int][2]int64 + + started chan int + release chan struct{} + + failPart int + failErr error + + abortCalls int + activeAtAbort int + abortCtxErr error + abortErr error +} + +type latencyClient struct { + delay time.Duration +} + +func (c latencyClient) UploadPartCopy( + ctx context.Context, + _ string, + _ string, + _ int, + _ string, + _ int64, + _ int64, +) (string, error) { + timer := time.NewTimer(c.delay) + defer timer.Stop() + select { + case <-timer.C: + return "etag", nil + case <-ctx.Done(): + return "", ctx.Err() + } +} + +func (c *testClient) UploadPartCopy( + ctx context.Context, + _ string, + _ string, + partNumber int, + _ string, + startByte int64, + endByte int64, +) (string, error) { + c.mu.Lock() + c.active++ + if c.active > c.maxActive { + c.maxActive = c.active + } + c.calls = append(c.calls, partNumber) + if c.ranges == nil { + c.ranges = make(map[int][2]int64) + } + c.ranges[partNumber] = [2]int64{startByte, endByte} + c.mu.Unlock() + + defer func() { + c.mu.Lock() + c.active-- + c.mu.Unlock() + }() + + if c.started != nil { + c.started <- partNumber + } + if partNumber == c.failPart { + return "", c.failErr + } + if c.release != nil { + select { + case <-c.release: + case <-ctx.Done(): + return "", ctx.Err() + } + } + return "etag", nil +} + +func (c *testClient) AbortMultipartUpload(ctx context.Context, _ string, _ string) error { + c.mu.Lock() + defer c.mu.Unlock() + c.abortCalls++ + c.activeAtAbort = c.active + c.abortCtxErr = ctx.Err() + return c.abortErr +} + +func tasks(count int) []Task { + out := make([]Task, count) + for i := range out { + out[i] = Task{ + PartNumber: i + 1, + StartByte: int64(i * 10), + EndByte: int64(i*10 + 9), + } + } + return out +} + +func waitForStarts(t *testing.T, started <-chan int, count int) { + t.Helper() + for i := 0; i < count; i++ { + select { + case <-started: + case <-time.After(5 * time.Second): + t.Fatalf("started %d copies, want %d", i, count) + } + } +} + +func TestCopyRunsWithBoundedConcurrency(t *testing.T) { + const concurrency = 3 + client := &testClient{ + started: make(chan int, 10), + release: make(chan struct{}), + } + done := make(chan error, 1) + go func() { + done <- Copy( + context.Background(), + client, + "dest", + "upload", + "source", + tasks(10), + concurrency, + ) + }() + + waitForStarts(t, client.started, concurrency) + select { + case part := <-client.started: + t.Fatalf("copy part %d started above concurrency limit %d", part, concurrency) + default: + } + + client.mu.Lock() + maxActive := client.maxActive + client.mu.Unlock() + if maxActive != concurrency { + t.Fatalf("max active copies = %d, want %d", maxActive, concurrency) + } + + close(client.release) + if err := <-done; err != nil { + t.Fatalf("Copy: %v", err) + } + + client.mu.Lock() + calls := append([]int(nil), client.calls...) + ranges := make(map[int][2]int64, len(client.ranges)) + for part, byteRange := range client.ranges { + ranges[part] = byteRange + } + client.mu.Unlock() + sort.Ints(calls) + if len(calls) != 10 { + t.Fatalf("copy calls = %v, want 10 parts", calls) + } + for i, part := range calls { + if part != i+1 { + t.Fatalf("copied parts = %v, want 1..10", calls) + } + wantRange := [2]int64{int64(i * 10), int64(i*10 + 9)} + if ranges[part] != wantRange { + t.Fatalf("part %d range = %v, want %v", part, ranges[part], wantRange) + } + } +} + +func TestCopyCancelsWorkersOnFirstFailure(t *testing.T) { + copyErr := errors.New("copy failed") + client := &testClient{ + started: make(chan int, 4), + release: make(chan struct{}), + failPart: 2, + failErr: copyErr, + } + + err := Copy( + context.Background(), + client, + "dest", + "upload", + "source", + tasks(4), + 2, + ) + var partErr *PartError + if !errors.As(err, &partErr) { + t.Fatalf("Copy error = %v, want PartError", err) + } + if partErr.PartNumber != 2 { + t.Fatalf("failed part = %d, want 2", partErr.PartNumber) + } + if !errors.Is(err, copyErr) { + t.Fatalf("Copy error = %v, want wrapped copy error", err) + } + + client.mu.Lock() + calls := append([]int(nil), client.calls...) + active := client.active + client.mu.Unlock() + sort.Ints(calls) + if len(calls) > 2 { + t.Fatalf("copy calls = %v, want no parts after the first worker batch", calls) + } + if len(calls) == 0 || calls[len(calls)-1] != 2 { + t.Fatalf("copy calls = %v, want failed part 2", calls) + } + if active != 0 { + t.Fatalf("active copies after return = %d, want 0", active) + } +} + +func TestCopyReturnsParentCancellation(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + client := &testClient{ + started: make(chan int, 4), + release: make(chan struct{}), + } + done := make(chan error, 1) + go func() { + done <- Copy( + ctx, + client, + "dest", + "upload", + "source", + tasks(4), + 2, + ) + }() + + waitForStarts(t, client.started, 2) + cancel() + if err := <-done; !errors.Is(err, context.Canceled) { + t.Fatalf("Copy error = %v, want context.Canceled", err) + } + + client.mu.Lock() + calls := append([]int(nil), client.calls...) + active := client.active + client.mu.Unlock() + if len(calls) != 2 { + t.Fatalf("copy calls = %v, want two in-flight copies only", calls) + } + if active != 0 { + t.Fatalf("active copies after return = %d, want 0", active) + } +} + +func TestCopyRejectsInvalidConcurrency(t *testing.T) { + err := Copy( + context.Background(), + &testClient{}, + "dest", + "upload", + "source", + tasks(1), + 0, + ) + if err == nil { + t.Fatal("Copy error = nil, want invalid concurrency error") + } +} + +func TestCopyReturnsCancellationWithoutTasks(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + cancel() + + err := Copy(ctx, &testClient{}, "dest", "upload", "source", nil, 8) + if !errors.Is(err, context.Canceled) { + t.Fatalf("Copy error = %v, want context.Canceled", err) + } +} + +func TestCopyOrAbortWaitsAndAbortsExactlyOnce(t *testing.T) { + copyErr := errors.New("copy failed") + client := &testClient{ + release: make(chan struct{}), + failPart: 2, + failErr: copyErr, + } + + err := CopyOrAbort( + context.Background(), + client, + "dest", + "upload", + "source", + tasks(4), + 2, + time.Second, + ) + if !errors.Is(err, copyErr) { + t.Fatalf("CopyOrAbort error = %v, want copy error", err) + } + + client.mu.Lock() + abortCalls := client.abortCalls + activeAtAbort := client.activeAtAbort + client.mu.Unlock() + if abortCalls != 1 { + t.Fatalf("abort calls = %d, want 1", abortCalls) + } + if activeAtAbort != 0 { + t.Fatalf("active copies when abort started = %d, want 0", activeAtAbort) + } +} + +func TestCopyOrAbortDoesNotAbortAfterSuccess(t *testing.T) { + client := &testClient{} + + if err := CopyOrAbort( + context.Background(), + client, + "dest", + "upload", + "source", + tasks(4), + 2, + time.Second, + ); err != nil { + t.Fatalf("CopyOrAbort: %v", err) + } + + client.mu.Lock() + abortCalls := client.abortCalls + client.mu.Unlock() + if abortCalls != 0 { + t.Fatalf("abort calls = %d, want 0", abortCalls) + } +} + +func TestCopyOrAbortUsesDetachedContextAfterParentCancellation(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + client := &testClient{ + started: make(chan int, 2), + release: make(chan struct{}), + } + done := make(chan error, 1) + go func() { + done <- CopyOrAbort( + ctx, + client, + "dest", + "upload", + "source", + tasks(2), + 2, + time.Second, + ) + }() + + waitForStarts(t, client.started, 2) + cancel() + if err := <-done; !errors.Is(err, context.Canceled) { + t.Fatalf("CopyOrAbort error = %v, want context.Canceled", err) + } + + client.mu.Lock() + abortCalls := client.abortCalls + activeAtAbort := client.activeAtAbort + abortCtxErr := client.abortCtxErr + client.mu.Unlock() + if abortCalls != 1 { + t.Fatalf("abort calls = %d, want 1", abortCalls) + } + if activeAtAbort != 0 { + t.Fatalf("active copies when abort started = %d, want 0", activeAtAbort) + } + if abortCtxErr != nil { + t.Fatalf("abort context error = %v, want nil", abortCtxErr) + } +} + +func TestCopyOrAbortReturnsAbortFailure(t *testing.T) { + copyErr := errors.New("copy failed") + abortErr := errors.New("abort failed") + client := &testClient{ + failPart: 1, + failErr: copyErr, + abortErr: abortErr, + } + + err := CopyOrAbort( + context.Background(), + client, + "dest", + "upload", + "source", + tasks(1), + 1, + time.Second, + ) + if !errors.Is(err, copyErr) { + t.Fatalf("CopyOrAbort error = %v, want copy error", err) + } + if !errors.Is(err, abortErr) { + t.Fatalf("CopyOrAbort error = %v, want abort error", err) + } +} + +func TestAbortUsesDetachedContext(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + cancel() + client := &testClient{} + + if err := Abort(ctx, client, "dest", "upload", time.Second); err != nil { + t.Fatalf("Abort: %v", err) + } + client.mu.Lock() + abortCalls := client.abortCalls + abortCtxErr := client.abortCtxErr + client.mu.Unlock() + if abortCalls != 1 { + t.Fatalf("abort calls = %d, want 1", abortCalls) + } + if abortCtxErr != nil { + t.Fatalf("abort context error = %v, want nil", abortCtxErr) + } +} + +func TestAbortRejectsInvalidTimeout(t *testing.T) { + err := Abort(context.Background(), &testClient{}, "dest", "upload", 0) + if err == nil { + t.Fatal("Abort error = nil, want invalid timeout error") + } +} + +func BenchmarkCopy(b *testing.B) { + for _, concurrency := range []int{1, 8} { + b.Run(fmt.Sprintf("concurrency_%d", concurrency), func(b *testing.B) { + client := latencyClient{delay: time.Millisecond} + copyTasks := tasks(36) + b.ResetTimer() + for i := 0; i < b.N; i++ { + if err := Copy( + context.Background(), + client, + "dest", + "upload", + "source", + copyTasks, + concurrency, + ); err != nil { + b.Fatalf("Copy: %v", err) + } + } + }) + } +} diff --git a/pkg/backend/patch.go b/pkg/backend/patch.go index daaa9f5b..06e9e22c 100644 --- a/pkg/backend/patch.go +++ b/pkg/backend/patch.go @@ -3,9 +3,11 @@ package backend import ( "context" "database/sql" + "errors" "fmt" "time" + "github.com/mem9-ai/drive9/pkg/backend/internal/patchcopy" "github.com/mem9-ai/drive9/pkg/datastore" "github.com/mem9-ai/drive9/pkg/logger" "github.com/mem9-ai/drive9/pkg/pathutil" @@ -43,6 +45,8 @@ type PatchUploadPart struct { ReadHeaders map[string]string `json:"read_headers,omitempty"` // required headers for the GET (e.g. Range, signed into the presigned URL) } +const patchPartCopyConcurrency = 8 + // InitiatePatchUpload creates a multipart upload for modifying an existing // large file. Only the dirty parts are uploaded by the client; unchanged parts // are copied server-side within S3 using UploadPartCopy. @@ -240,6 +244,7 @@ func (b *Dat9Backend) InitiatePatchUploadIfRevision(ctx context.Context, path st UploadID: "", // set below after DB insert PartSize: partSize, } + copyTasks := make([]patchcopy.Task, 0, len(newParts)) // Process each part for _, p := range newParts { @@ -251,20 +256,16 @@ func (b *Dat9Backend) InitiatePatchUploadIfRevision(ctx context.Context, path st if partEnd >= origSize { partEnd = origSize - 1 } - - _, err := b.s3.UploadPartCopy(ctx, newS3Key, mpu.UploadID, p.Number, sourceKey, partStart, partEnd) - if err != nil { - _ = b.s3.AbortMultipartUpload(ctx, newS3Key, mpu.UploadID) - logger.Error(ctx, "backend_patch_upload_copy_failed", zap.String("tenant_id", b.tenantID), zap.String("path", path), zap.Int("part", p.Number), zap.Error(err)) - b.recordTenantOperation("backend", "patch_upload", "error", time.Since(start)) - return nil, fmt.Errorf("copy part %d: %w", p.Number, err) - } - plan.CopiedParts = append(plan.CopiedParts, p.Number) + copyTasks = append(copyTasks, patchcopy.Task{ + PartNumber: p.Number, + StartByte: partStart, + EndByte: partEnd, + }) } else { // Dirty part or new part beyond original → client must upload u, err := b.s3.PresignUploadPart(ctx, newS3Key, mpu.UploadID, p.Number, p.Size, s3client.ChecksumAlgoNone, "", s3client.UploadTTL) if err != nil { - _ = b.s3.AbortMultipartUpload(ctx, newS3Key, mpu.UploadID) + _ = patchcopy.Abort(ctx, b.s3, newS3Key, mpu.UploadID, postS3UploadFinalizeTimeout) logger.Error(ctx, "backend_patch_upload_presign_failed", zap.String("tenant_id", b.tenantID), zap.String("path", path), zap.Int("part", p.Number), zap.Error(err)) b.recordTenantOperation("backend", "patch_upload", "error", time.Since(start)) return nil, fmt.Errorf("presign part %d: %w", p.Number, err) @@ -303,6 +304,28 @@ func (b *Dat9Backend) InitiatePatchUploadIfRevision(ctx context.Context, path st plan.UploadParts = append(plan.UploadParts, pup) } } + if err := patchcopy.CopyOrAbort( + ctx, + b.s3, + newS3Key, + mpu.UploadID, + sourceKey, + copyTasks, + patchPartCopyConcurrency, + postS3UploadFinalizeTimeout, + ); err != nil { + var partErr *patchcopy.PartError + if errors.As(err, &partErr) { + logger.Error(ctx, "backend_patch_upload_copy_failed", zap.String("tenant_id", b.tenantID), zap.String("path", path), zap.Int("part", partErr.PartNumber), zap.Error(err)) + } else { + logger.Error(ctx, "backend_patch_upload_copy_failed", zap.String("tenant_id", b.tenantID), zap.String("path", path), zap.Error(err)) + } + b.recordTenantOperation("backend", "patch_upload", "error", time.Since(start)) + return nil, fmt.Errorf("copy retained parts: %w", err) + } + for _, task := range copyTasks { + plan.CopiedParts = append(plan.CopiedParts, task.PartNumber) + } // Insert DB records (same pattern as InitiateUploadWithChecksums) now := time.Now() @@ -312,7 +335,7 @@ func (b *Dat9Backend) InitiatePatchUploadIfRevision(ctx context.Context, path st // Server-reserve-first saga (same as upload initiate). reserved, err := b.reserveUploadOnServer(ctx, uploadID, path, newSize, 0) if err != nil { - _ = b.s3.AbortMultipartUpload(ctx, newS3Key, mpu.UploadID) + _ = patchcopy.Abort(ctx, b.s3, newS3Key, mpu.UploadID, postS3UploadFinalizeTimeout) b.recordTenantOperation("backend", "patch_upload", "quota_exceeded", time.Since(start)) return nil, err } @@ -326,7 +349,7 @@ func (b *Dat9Backend) InitiatePatchUploadIfRevision(ctx context.Context, path st if reserved { b.abortUploadReservation(ctx, uploadID, newSize) } - _ = b.s3.AbortMultipartUpload(ctx, newS3Key, mpu.UploadID) + _ = patchcopy.Abort(ctx, b.s3, newS3Key, mpu.UploadID, postS3UploadFinalizeTimeout) b.recordTenantOperation("backend", "patch_upload", "error", time.Since(start)) return nil, fmt.Errorf("lookup active upload for %s: %w", path, err) } @@ -335,7 +358,7 @@ func (b *Dat9Backend) InitiatePatchUploadIfRevision(ctx context.Context, path st if reserved { b.abortUploadReservation(ctx, uploadID, newSize) } - _ = b.s3.AbortMultipartUpload(ctx, newS3Key, mpu.UploadID) + _ = patchcopy.Abort(ctx, b.s3, newS3Key, mpu.UploadID, postS3UploadFinalizeTimeout) b.recordTenantOperation("backend", "patch_upload", "error", time.Since(start)) return nil, fmt.Errorf("supersede active upload for %s: %w", path, err) } @@ -381,7 +404,7 @@ func (b *Dat9Backend) InitiatePatchUploadIfRevision(ctx context.Context, path st if reserved { b.abortUploadReservation(ctx, uploadID, newSize) } - _ = b.s3.AbortMultipartUpload(ctx, newS3Key, mpu.UploadID) + _ = patchcopy.Abort(ctx, b.s3, newS3Key, mpu.UploadID, postS3UploadFinalizeTimeout) logger.Error(ctx, "backend_patch_upload_insert_upload_failed", zap.String("tenant_id", b.tenantID), zap.String("path", path), zap.Error(err)) b.recordTenantOperation("backend", "patch_upload", "error", time.Since(start)) return nil, err diff --git a/pkg/backend/patch_test.go b/pkg/backend/patch_test.go index bb50a7af..f2aac28e 100644 --- a/pkg/backend/patch_test.go +++ b/pkg/backend/patch_test.go @@ -5,6 +5,7 @@ import ( "context" "errors" "os" + "sync" "testing" "time" @@ -15,6 +16,238 @@ import ( "github.com/mem9-ai/drive9/pkg/s3client" ) +type patchCopyRecordingS3Client struct { + s3client.S3Client + + mu sync.Mutex + + active int + maxActive int + started chan int + release chan struct{} + failPart int + failErr error + abortCalls int + activeAtAbort int + abortedKey string + abortedID string +} + +func (c *patchCopyRecordingS3Client) UploadPartCopy( + ctx context.Context, + destKey string, + uploadID string, + partNumber int, + sourceKey string, + startByte int64, + endByte int64, +) (string, error) { + c.mu.Lock() + c.active++ + if c.active > c.maxActive { + c.maxActive = c.active + } + c.mu.Unlock() + defer func() { + c.mu.Lock() + c.active-- + c.mu.Unlock() + }() + + if c.started != nil { + c.started <- partNumber + } + if partNumber == c.failPart { + return "", c.failErr + } + if c.release != nil { + select { + case <-c.release: + case <-ctx.Done(): + return "", ctx.Err() + } + } + return c.S3Client.UploadPartCopy(ctx, destKey, uploadID, partNumber, sourceKey, startByte, endByte) +} + +func (c *patchCopyRecordingS3Client) AbortMultipartUpload(ctx context.Context, key string, uploadID string) error { + c.mu.Lock() + c.abortCalls++ + c.activeAtAbort = c.active + c.abortedKey = key + c.abortedID = uploadID + c.mu.Unlock() + return c.S3Client.AbortMultipartUpload(ctx, key, uploadID) +} + +func uploadPatchTestFile(t *testing.T, b *Dat9Backend, localS3 *s3client.LocalS3Client, path string, partCount int) { + t.Helper() + ctx := context.Background() + totalSize := int64(partCount) * s3client.PartSize + plan, err := b.InitiateUpload(ctx, path, totalSize) + if err != nil { + t.Fatalf("InitiateUpload: %v", err) + } + upload, err := b.GetUpload(ctx, plan.UploadID) + if err != nil { + t.Fatalf("GetUpload: %v", err) + } + for _, part := range plan.Parts { + data := bytes.Repeat([]byte{byte(part.Number)}, int(part.Size)) + if _, err := localS3.UploadPart(ctx, upload.S3UploadID, part.Number, bytes.NewReader(data)); err != nil { + t.Fatalf("upload part %d: %v", part.Number, err) + } + } + if err := b.ConfirmUpload(ctx, plan.UploadID); err != nil { + t.Fatalf("ConfirmUpload: %v", err) + } +} + +func waitForPatchCopyStarts(t *testing.T, started <-chan int, count int) { + t.Helper() + for i := 0; i < count; i++ { + select { + case <-started: + case <-time.After(5 * time.Second): + t.Fatalf("started %d copies, want %d", i, count) + } + } +} + +func TestPatchUploadCopiesRetainedPartsConcurrentlyAndCompletesInPartOrder(t *testing.T) { + b := newTestBackendWithS3(t) + localS3, ok := b.s3.(*s3client.LocalS3Client) + if !ok { + t.Fatalf("S3 client = %T, want *LocalS3Client", b.s3) + } + const path = "/patch-copy-concurrency.bin" + uploadPatchTestFile(t, b, localS3, path, 4) + + recorder := &patchCopyRecordingS3Client{ + S3Client: localS3, + started: make(chan int, 4), + release: make(chan struct{}), + } + b.s3 = recorder + + ctx := context.Background() + planDone := make(chan *PatchPlan, 1) + errDone := make(chan error, 1) + go func() { + plan, err := b.InitiatePatchUploadIfRevision( + ctx, + path, + 4*s3client.PartSize, + []int{4}, + s3client.PartSize, + -1, + ) + planDone <- plan + errDone <- err + }() + + waitForPatchCopyStarts(t, recorder.started, 3) + recorder.mu.Lock() + maxActive := recorder.maxActive + recorder.mu.Unlock() + close(recorder.release) + if maxActive != 3 { + t.Fatalf("max active copies = %d, want 3", maxActive) + } + + plan := <-planDone + if err := <-errDone; err != nil { + t.Fatalf("InitiatePatchUploadIfRevision: %v", err) + } + if len(plan.CopiedParts) != 3 || plan.CopiedParts[0] != 1 || plan.CopiedParts[1] != 2 || plan.CopiedParts[2] != 3 { + t.Fatalf("copied parts = %v, want [1 2 3]", plan.CopiedParts) + } + + upload, err := b.GetUpload(ctx, plan.UploadID) + if err != nil { + t.Fatalf("GetUpload: %v", err) + } + dirty := bytes.Repeat([]byte{9}, s3client.PartSize) + if _, err := localS3.UploadPart(ctx, upload.S3UploadID, 4, bytes.NewReader(dirty)); err != nil { + t.Fatalf("upload dirty part: %v", err) + } + if err := b.ConfirmUpload(ctx, plan.UploadID); err != nil { + t.Fatalf("ConfirmUpload: %v", err) + } + + for partNumber := 1; partNumber <= 4; partNumber++ { + got, err := b.ReadCtx(ctx, path, int64(partNumber-1)*s3client.PartSize, 1) + if err != nil { + t.Fatalf("ReadCtx part %d: %v", partNumber, err) + } + want := byte(partNumber) + if partNumber == 4 { + want = 9 + } + if len(got) != 1 || got[0] != want { + t.Fatalf("part %d first byte = %v, want [%d]", partNumber, got, want) + } + } + + recorder.mu.Lock() + abortCalls := recorder.abortCalls + recorder.mu.Unlock() + if abortCalls != 0 { + t.Fatalf("abort calls = %d, want 0", abortCalls) + } +} + +func TestPatchUploadCopyFailureWaitsThenAbortsWithoutMetadata(t *testing.T) { + b := newTestBackendWithS3(t) + localS3, ok := b.s3.(*s3client.LocalS3Client) + if !ok { + t.Fatalf("S3 client = %T, want *LocalS3Client", b.s3) + } + const path = "/patch-copy-failure.bin" + uploadPatchTestFile(t, b, localS3, path, 3) + + copyErr := errors.New("copy failed") + recorder := &patchCopyRecordingS3Client{ + S3Client: localS3, + failPart: 2, + failErr: copyErr, + } + b.s3 = recorder + + _, err := b.InitiatePatchUploadIfRevision( + context.Background(), + path, + 3*s3client.PartSize, + []int{3}, + s3client.PartSize, + -1, + ) + if !errors.Is(err, copyErr) { + t.Fatalf("InitiatePatchUploadIfRevision error = %v, want copy error", err) + } + + recorder.mu.Lock() + abortCalls := recorder.abortCalls + activeAtAbort := recorder.activeAtAbort + abortedKey := recorder.abortedKey + abortedID := recorder.abortedID + recorder.mu.Unlock() + if abortCalls != 1 { + t.Fatalf("abort calls = %d, want 1", abortCalls) + } + if activeAtAbort != 0 { + t.Fatalf("active copies when abort started = %d, want 0", activeAtAbort) + } + if _, err := localS3.ListParts(context.Background(), abortedKey, abortedID); err == nil { + t.Fatal("ListParts after abort = nil error, want missing multipart upload") + } + if upload, err := b.activeUploadByPath(context.Background(), path); err != nil { + t.Fatalf("activeUploadByPath: %v", err) + } else if upload != nil { + t.Fatalf("active upload after copy failure = %+v, want nil", upload) + } +} + func TestPatchAndAppendRejectDBBackedFilesWithSentinel(t *testing.T) { b := newTestBackendWithS3(t) ctx := context.Background() diff --git a/pkg/server/patch_copy_e2e_test.go b/pkg/server/patch_copy_e2e_test.go new file mode 100644 index 00000000..54a9d8a6 --- /dev/null +++ b/pkg/server/patch_copy_e2e_test.go @@ -0,0 +1,624 @@ +package server + +import ( + "bytes" + "context" + "crypto/sha256" + "encoding/json" + "errors" + "io" + "net/http" + "net/http/httptest" + "os" + "sync" + "testing" + "time" + + "github.com/mem9-ai/drive9/internal/testmysql" + "github.com/mem9-ai/drive9/pkg/backend" + "github.com/mem9-ai/drive9/pkg/datastore" + "github.com/mem9-ai/drive9/pkg/s3client" +) + +type patchCopyE2ES3 struct { + s3client.S3Client + + mu sync.Mutex + + active int + maxActive int + calls []int + ranges map[int][2]int64 + completedOrder []int + gates map[int]chan struct{} + started chan int + completed chan int + failPart int + failErr error + abortCalls int + activeAtAbort int + abortedKey string + abortedUploadID string + aborted chan struct{} +} + +func (s *patchCopyE2ES3) UploadPartCopy( + ctx context.Context, + destKey string, + uploadID string, + partNumber int, + sourceKey string, + startByte int64, + endByte int64, +) (etag string, err error) { + s.mu.Lock() + s.active++ + if s.active > s.maxActive { + s.maxActive = s.active + } + s.calls = append(s.calls, partNumber) + s.ranges[partNumber] = [2]int64{startByte, endByte} + gate := s.gates[partNumber] + s.mu.Unlock() + + defer func() { + s.mu.Lock() + s.active-- + if err == nil { + s.completedOrder = append(s.completedOrder, partNumber) + } + s.mu.Unlock() + if err == nil { + s.completed <- partNumber + } + }() + + s.started <- partNumber + if gate != nil { + select { + case <-gate: + case <-ctx.Done(): + return "", ctx.Err() + } + } + if partNumber == s.failPart { + return "", s.failErr + } + return s.S3Client.UploadPartCopy(ctx, destKey, uploadID, partNumber, sourceKey, startByte, endByte) +} + +func (s *patchCopyE2ES3) AbortMultipartUpload(ctx context.Context, key string, uploadID string) error { + s.mu.Lock() + s.abortCalls++ + s.activeAtAbort = s.active + s.abortedKey = key + s.abortedUploadID = uploadID + s.mu.Unlock() + + err := s.S3Client.AbortMultipartUpload(ctx, key, uploadID) + s.aborted <- struct{}{} + return err +} + +func newPatchCopyE2EServer( + t *testing.T, + configure func(*patchCopyE2ES3), +) (*Server, *s3client.LocalS3Client, *patchCopyE2ES3) { + t.Helper() + + s3Dir, err := os.MkdirTemp("", "dat9-patch-copy-e2e-*") + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = os.RemoveAll(s3Dir) }) + + initServerTenantSchema(t, testDSN) + store, err := datastore.Open(testDSN) + if err != nil { + t.Fatal(err) + } + testmysql.ResetDB(t, store.DB()) + t.Cleanup(func() { _ = store.Close() }) + + localS3, err := s3client.NewLocal(s3Dir, "/s3") + if err != nil { + t.Fatal(err) + } + recorder := &patchCopyE2ES3{ + S3Client: localS3, + ranges: make(map[int][2]int64), + gates: make(map[int]chan struct{}), + started: make(chan int, 32), + completed: make(chan int, 32), + aborted: make(chan struct{}, 4), + } + if configure != nil { + configure(recorder) + } + + b, err := backend.NewWithS3ModeAndOptions(store, recorder, true, backend.Options{}) + if err != nil { + t.Fatal(err) + } + t.Cleanup(b.Close) + + server := NewWithConfig(Config{Backend: b, LocalS3: localS3}) + t.Cleanup(server.Close) + return server, localS3, recorder +} + +func patchCopyE2EPartBounds(totalSize int64, partSize int64, partNumber int) (int64, int64) { + start := int64(partNumber-1) * partSize + end := start + partSize + if end > totalSize { + end = totalSize + } + return start, end +} + +func patchCopyE2EBody(totalSize int64, partSize int64) []byte { + body := make([]byte, int(totalSize)) + partCount := int((totalSize + partSize - 1) / partSize) + for partNumber := 1; partNumber <= partCount; partNumber++ { + start, end := patchCopyE2EPartBounds(totalSize, partSize, partNumber) + for offset := start; offset < end; offset++ { + body[offset] = byte((int64(partNumber)*37 + (offset-start)%251) % 256) + } + } + return body +} + +func patchCopyE2EExpected(original []byte, partSize int64, dirtyParts []int) []byte { + expected := bytes.Clone(original) + totalSize := int64(len(expected)) + for _, partNumber := range dirtyParts { + start, end := patchCopyE2EPartBounds(totalSize, partSize, partNumber) + for offset := start; offset < end; offset++ { + expected[offset] = byte((int64(partNumber)*83 + (offset-start)%239) % 256) + } + } + return expected +} + +func waitPatchCopyE2EPart(t *testing.T, events <-chan int, want int) { + t.Helper() + select { + case got := <-events: + if got != want { + t.Fatalf("copy event = part %d, want part %d", got, want) + } + case <-time.After(10 * time.Second): + t.Fatalf("timed out waiting for copy event for part %d", want) + } +} + +func waitPatchCopyE2EStarts(t *testing.T, started <-chan int, count int) []int { + t.Helper() + parts := make([]int, 0, count) + for len(parts) < count { + select { + case partNumber := <-started: + parts = append(parts, partNumber) + case <-time.After(10 * time.Second): + t.Fatalf("started %d copies, want %d", len(parts), count) + } + } + return parts +} + +func equalPatchCopyE2EParts(got []int, want []int) bool { + if len(got) != len(want) { + return false + } + for i := range got { + if got[i] != want[i] { + return false + } + } + return true +} + +func assertPatchCopyE2EPartSet(t *testing.T, got []int, want []int) { + t.Helper() + counts := make(map[int]int, len(got)) + for _, partNumber := range got { + counts[partNumber]++ + } + if len(counts) != len(want) { + t.Fatalf("part set = %v, want %v", got, want) + } + for _, partNumber := range want { + if counts[partNumber] != 1 { + t.Fatalf("part %d count = %d in %v, want 1", partNumber, counts[partNumber], got) + } + } +} + +func assertPatchCopyE2EPlanCoverage(t *testing.T, plan *backend.PatchPlan, partCount int) { + t.Helper() + counts := make(map[int]int, partCount) + for _, partNumber := range plan.CopiedParts { + counts[partNumber]++ + } + for _, part := range plan.UploadParts { + counts[part.Number]++ + } + if len(counts) != partCount { + t.Fatalf("patch plan covers %d distinct parts, want %d", len(counts), partCount) + } + for partNumber := 1; partNumber <= partCount; partNumber++ { + if counts[partNumber] != 1 { + t.Fatalf("patch plan part %d count = %d, want 1", partNumber, counts[partNumber]) + } + } +} + +func readPatchCopyE2EFile(t *testing.T, client *http.Client, baseURL string, path string) []byte { + t.Helper() + resp, err := client.Get(baseURL + "/v1/fs" + path) + if err != nil { + t.Fatalf("GET %s: %v", path, err) + } + defer func() { _ = resp.Body.Close() }() + body, err := io.ReadAll(resp.Body) + if err != nil { + t.Fatalf("read %s: %v", path, err) + } + if resp.StatusCode != http.StatusOK { + t.Fatalf("GET %s = %d, want 200: %s", path, resp.StatusCode, body) + } + return body +} + +func assertNoPatchCopyE2EUploadMetadata(t *testing.T, s *Server, path string) { + t.Helper() + var totalCount int + if err := s.fallback.Store().DB().QueryRowContext( + context.Background(), + `SELECT COUNT(*) FROM uploads WHERE target_path = ?`, + path, + ).Scan(&totalCount); err != nil { + t.Fatalf("count uploads for %s: %v", path, err) + } + if totalCount != 1 { + t.Fatalf("upload metadata for %s = %d rows, want only the 1 completed seed upload", path, totalCount) + } + + var activeCount int + err := s.fallback.Store().DB().QueryRowContext( + context.Background(), + `SELECT COUNT(*) FROM uploads + WHERE target_path = ? AND status IN (?, ?)`, + path, + datastore.UploadInitiated, + datastore.UploadUploading, + ).Scan(&activeCount) + if err != nil { + t.Fatalf("count active uploads for %s: %v", path, err) + } + if activeCount != 0 { + t.Fatalf("active upload metadata for %s = %d rows, want 0", path, activeCount) + } +} + +func assertPatchCopyE2EAbort(t *testing.T, localS3 *s3client.LocalS3Client, recorder *patchCopyE2ES3) { + t.Helper() + recorder.mu.Lock() + abortCalls := recorder.abortCalls + activeAtAbort := recorder.activeAtAbort + abortedKey := recorder.abortedKey + abortedUploadID := recorder.abortedUploadID + recorder.mu.Unlock() + + if abortCalls != 1 { + t.Fatalf("abort calls = %d, want 1", abortCalls) + } + if activeAtAbort != 0 { + t.Fatalf("active copies when abort started = %d, want 0", activeAtAbort) + } + if _, err := localS3.ListParts(context.Background(), abortedKey, abortedUploadID); err == nil { + t.Fatal("ListParts after abort = nil error, want missing multipart upload") + } + if err := localS3.CompleteMultipartUpload(context.Background(), abortedKey, abortedUploadID, nil); err == nil { + t.Fatal("CompleteMultipartUpload after abort = nil error, want missing multipart upload") + } +} + +type patchCopyE2EHTTPResult struct { + resp *http.Response + err error +} + +func startPatchCopyE2ERequest(client *http.Client, req *http.Request) <-chan patchCopyE2EHTTPResult { + done := make(chan patchCopyE2EHTTPResult, 1) + go func() { + resp, err := client.Do(req) + done <- patchCopyE2EHTTPResult{resp: resp, err: err} + }() + return done +} + +func waitPatchCopyE2EHTTPResult(t *testing.T, done <-chan patchCopyE2EHTTPResult) patchCopyE2EHTTPResult { + t.Helper() + select { + case result := <-done: + return result + case <-time.After(20 * time.Second): + t.Fatal("timed out waiting for PATCH response") + return patchCopyE2EHTTPResult{} + } +} + +func TestPatchCopyE2EConcurrentRetainedPartsPreserveWholeFile(t *testing.T) { + const ( + path = "/patch-copy-e2e.bin" + partCount = 12 + ) + partSize := int64(s3client.MinPartSize) + totalSize := int64(partCount-1)*partSize + 257 + dirtyParts := []int{2, 5, 9} + retainedParts := []int{1, 3, 4, 6, 7, 8, 10, 11, 12} + firstWave := []int{1, 3, 4, 6, 7, 8, 10, 11} + completionOrder := []int{11, 12, 10, 8, 7, 6, 4, 3, 1} + + gates := make(map[int]chan struct{}, len(retainedParts)) + for _, partNumber := range retainedParts { + gates[partNumber] = make(chan struct{}) + } + s, localS3, recorder := newPatchCopyE2EServer(t, func(recorder *patchCopyE2ES3) { + recorder.gates = gates + }) + ts := httptest.NewServer(s) + defer ts.Close() + client := &http.Client{Timeout: 30 * time.Second} + + original := patchCopyE2EBody(totalSize, partSize) + expected := patchCopyE2EExpected(original, partSize, dirtyParts) + mustUploadLargeServerFile(t, ts, s, localS3, path, original) + + requestBody, err := json.Marshal(map[string]any{ + "new_size": totalSize, + "dirty_parts": dirtyParts, + "part_size": partSize, + }) + if err != nil { + t.Fatal(err) + } + req, err := http.NewRequest(http.MethodPatch, ts.URL+"/v1/fs"+path, bytes.NewReader(requestBody)) + if err != nil { + t.Fatal(err) + } + req.Header.Set("Content-Type", "application/json") + patchDone := startPatchCopyE2ERequest(client, req) + + started := waitPatchCopyE2EStarts(t, recorder.started, 8) + assertPatchCopyE2EPartSet(t, started, firstWave) + recorder.mu.Lock() + callCountBeforeRelease := len(recorder.calls) + maxActiveBeforeRelease := recorder.maxActive + recorder.mu.Unlock() + if callCountBeforeRelease != 8 { + t.Fatalf("copy calls before release = %d, want 8", callCountBeforeRelease) + } + if maxActiveBeforeRelease != 8 { + t.Fatalf("max active copies before release = %d, want 8", maxActiveBeforeRelease) + } + + close(gates[11]) + waitPatchCopyE2EPart(t, recorder.completed, 11) + waitPatchCopyE2EPart(t, recorder.started, 12) + for _, partNumber := range completionOrder[1:] { + close(gates[partNumber]) + waitPatchCopyE2EPart(t, recorder.completed, partNumber) + } + + result := waitPatchCopyE2EHTTPResult(t, patchDone) + if result.err != nil { + t.Fatalf("PATCH %s: %v", path, result.err) + } + defer func() { _ = result.resp.Body.Close() }() + responseBody, err := io.ReadAll(result.resp.Body) + if err != nil { + t.Fatal(err) + } + if result.resp.StatusCode != http.StatusAccepted { + t.Fatalf("PATCH %s = %d, want 202: %s", path, result.resp.StatusCode, responseBody) + } + var plan backend.PatchPlan + if err := json.Unmarshal(responseBody, &plan); err != nil { + t.Fatalf("decode patch plan: %v", err) + } + if !equalPatchCopyE2EParts(plan.CopiedParts, retainedParts) { + t.Fatalf("copied parts = %v, want %v", plan.CopiedParts, retainedParts) + } + gotDirtyParts := make([]int, 0, len(plan.UploadParts)) + for _, part := range plan.UploadParts { + gotDirtyParts = append(gotDirtyParts, part.Number) + } + if !equalPatchCopyE2EParts(gotDirtyParts, dirtyParts) { + t.Fatalf("dirty upload parts = %v, want %v", gotDirtyParts, dirtyParts) + } + assertPatchCopyE2EPlanCoverage(t, &plan, partCount) + + recorder.mu.Lock() + calls := append([]int(nil), recorder.calls...) + ranges := make(map[int][2]int64, len(recorder.ranges)) + for partNumber, byteRange := range recorder.ranges { + ranges[partNumber] = byteRange + } + gotCompletionOrder := append([]int(nil), recorder.completedOrder...) + maxActive := recorder.maxActive + abortCalls := recorder.abortCalls + recorder.mu.Unlock() + assertPatchCopyE2EPartSet(t, calls, retainedParts) + if maxActive != 8 { + t.Fatalf("max active copies = %d, want 8", maxActive) + } + if !equalPatchCopyE2EParts(gotCompletionOrder, completionOrder) { + t.Fatalf("copy completion order = %v, want %v", gotCompletionOrder, completionOrder) + } + if abortCalls != 0 { + t.Fatalf("abort calls = %d, want 0", abortCalls) + } + for _, partNumber := range retainedParts { + start, end := patchCopyE2EPartBounds(totalSize, partSize, partNumber) + wantRange := [2]int64{start, end - 1} + if got := ranges[partNumber]; got != wantRange { + t.Fatalf("part %d copied range = %v, want %v", partNumber, got, wantRange) + } + } + + upload, err := s.fallback.GetUpload(context.Background(), plan.UploadID) + if err != nil { + t.Fatalf("GetUpload(%q): %v", plan.UploadID, err) + } + for _, part := range plan.UploadParts { + start, end := patchCopyE2EPartBounds(totalSize, plan.PartSize, part.Number) + if _, err := localS3.UploadPart( + context.Background(), + upload.S3UploadID, + part.Number, + bytes.NewReader(expected[start:end]), + ); err != nil { + t.Fatalf("upload dirty part %d: %v", part.Number, err) + } + } + + completeReq, err := http.NewRequest(http.MethodPost, ts.URL+"/v1/uploads/"+plan.UploadID+"/complete", nil) + if err != nil { + t.Fatal(err) + } + completeResp, err := client.Do(completeReq) + if err != nil { + t.Fatalf("complete patch upload: %v", err) + } + defer func() { _ = completeResp.Body.Close() }() + completeBody, err := io.ReadAll(completeResp.Body) + if err != nil { + t.Fatal(err) + } + if completeResp.StatusCode != http.StatusOK { + t.Fatalf("complete patch upload = %d, want 200: %s", completeResp.StatusCode, completeBody) + } + + got := readPatchCopyE2EFile(t, client, ts.URL, path) + if len(got) != len(expected) { + t.Fatalf("patched file size = %d, want %d", len(got), len(expected)) + } + gotSHA256 := sha256.Sum256(got) + wantSHA256 := sha256.Sum256(expected) + if gotSHA256 != wantSHA256 || !bytes.Equal(got, expected) { + t.Fatalf("patched file bytes differ: sha256 got %x, want %x", gotSHA256, wantSHA256) + } +} + +func TestPatchCopyE2ECopyFailureLeavesOriginalAndNoUpload(t *testing.T) { + const path = "/patch-copy-failure-e2e.bin" + partSize := int64(s3client.MinPartSize) + totalSize := 3*partSize + 193 + copyErr := errors.New("injected retained-part copy failure") + + s, localS3, recorder := newPatchCopyE2EServer(t, func(recorder *patchCopyE2ES3) { + recorder.failPart = 2 + recorder.failErr = copyErr + }) + ts := httptest.NewServer(s) + defer ts.Close() + client := &http.Client{Timeout: 30 * time.Second} + + original := patchCopyE2EBody(totalSize, partSize) + mustUploadLargeServerFile(t, ts, s, localS3, path, original) + + requestBody, err := json.Marshal(map[string]any{ + "new_size": totalSize, + "dirty_parts": []int{4}, + "part_size": partSize, + }) + if err != nil { + t.Fatal(err) + } + req, err := http.NewRequest(http.MethodPatch, ts.URL+"/v1/fs"+path, bytes.NewReader(requestBody)) + if err != nil { + t.Fatal(err) + } + req.Header.Set("Content-Type", "application/json") + resp, err := client.Do(req) + if err != nil { + t.Fatalf("PATCH %s: %v", path, err) + } + defer func() { _ = resp.Body.Close() }() + responseBody, err := io.ReadAll(resp.Body) + if err != nil { + t.Fatal(err) + } + if resp.StatusCode != http.StatusInternalServerError { + t.Fatalf("PATCH %s = %d, want 500: %s", path, resp.StatusCode, responseBody) + } + + assertPatchCopyE2EAbort(t, localS3, recorder) + assertNoPatchCopyE2EUploadMetadata(t, s, path) + got := readPatchCopyE2EFile(t, client, ts.URL, path) + if gotSHA256, wantSHA256 := sha256.Sum256(got), sha256.Sum256(original); gotSHA256 != wantSHA256 || !bytes.Equal(got, original) { + t.Fatalf("original file changed after copy failure: sha256 got %x, want %x", gotSHA256, wantSHA256) + } +} + +func TestPatchCopyE2ERequestCancellationAbortsAfterCopiesExit(t *testing.T) { + const path = "/patch-copy-cancel-e2e.bin" + partSize := int64(s3client.MinPartSize) + totalSize := 2*partSize + 211 + gates := map[int]chan struct{}{ + 1: make(chan struct{}), + 2: make(chan struct{}), + } + + s, localS3, recorder := newPatchCopyE2EServer(t, func(recorder *patchCopyE2ES3) { + recorder.gates = gates + }) + ts := httptest.NewServer(s) + defer ts.Close() + client := &http.Client{Timeout: 30 * time.Second} + + original := patchCopyE2EBody(totalSize, partSize) + mustUploadLargeServerFile(t, ts, s, localS3, path, original) + + requestBody, err := json.Marshal(map[string]any{ + "new_size": totalSize, + "dirty_parts": []int{3}, + "part_size": partSize, + }) + if err != nil { + t.Fatal(err) + } + ctx, cancel := context.WithCancel(context.Background()) + req, err := http.NewRequestWithContext(ctx, http.MethodPatch, ts.URL+"/v1/fs"+path, bytes.NewReader(requestBody)) + if err != nil { + t.Fatal(err) + } + req.Header.Set("Content-Type", "application/json") + patchDone := startPatchCopyE2ERequest(client, req) + + started := waitPatchCopyE2EStarts(t, recorder.started, 2) + assertPatchCopyE2EPartSet(t, started, []int{1, 2}) + cancel() + + result := waitPatchCopyE2EHTTPResult(t, patchDone) + if result.resp != nil { + _ = result.resp.Body.Close() + } + if !errors.Is(result.err, context.Canceled) { + t.Fatalf("PATCH cancellation error = %v, want context.Canceled", result.err) + } + select { + case <-recorder.aborted: + case <-time.After(10 * time.Second): + t.Fatal("timed out waiting for detached multipart abort") + } + + assertPatchCopyE2EAbort(t, localS3, recorder) + assertNoPatchCopyE2EUploadMetadata(t, s, path) + got := readPatchCopyE2EFile(t, client, ts.URL, path) + if gotSHA256, wantSHA256 := sha256.Sum256(got), sha256.Sum256(original); gotSHA256 != wantSHA256 || !bytes.Equal(got, original) { + t.Fatalf("original file changed after request cancellation: sha256 got %x, want %x", gotSHA256, wantSHA256) + } +}