perf(backend): parallelize retained PATCH copies

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)
+}