Paginate S3 prefix deletes

packages/shared/pkg/storage/storage_aws.go

@@ -75,26 +75,54 @@ func newAWSStorage(ctx context.Context, bucketName string) (*awsStorage, error)
 }
 
 func (s *awsStorage) DeleteObjectsWithPrefix(ctx context.Context, prefix string) error {
-	ctx, cancel := context.WithTimeout(ctx, awsOperationTimeout)
-	defer cancel()
-
-	list, err := s.client.ListObjectsV2(ctx, &s3.ListObjectsV2Input{Bucket: &s.bucketName, Prefix: &prefix})
-	if err != nil {
-		return err
+	if prefix == "" {
+		return errors.New("refusing to delete objects with an empty prefix")
 	}
 
-	objects := make([]types.ObjectIdentifier, 0, len(list.Contents))
-	for _, obj := range list.Contents {
-		objects = append(objects, types.ObjectIdentifier{Key: obj.Key})
+	// A large prefix spans many pages, so scope the timeout per round-trip
+	// below instead of wrapping the whole walk.
+	paginator := s3.NewListObjectsV2Paginator(s.client, &s3.ListObjectsV2Input{
+		Bucket: aws.String(s.bucketName),
+		Prefix: aws.String(prefix),
+	})
+
+	deleted := false
+	for paginator.HasMorePages() {
+		pageCtx, cancel := context.WithTimeout(ctx, awsOperationTimeout)
+		list, err := paginator.NextPage(pageCtx)
+		cancel()
+		if err != nil {
+			return err
+		}
+
+		objects := make([]types.ObjectIdentifier, 0, len(list.Contents))
+		for _, obj := range list.Contents {
+			objects = append(objects, types.ObjectIdentifier{Key: obj.Key})
+		}
+
+		// AWS S3 delete operation requires at least one object to delete.
+		if len(objects) == 0 {
+			continue
+		}
+
+		if err := s.deleteObjects(ctx, objects); err != nil {
+			return err
+		}
+
+		deleted = true
 	}
 
-	// AWS S3 delete operation requires at least one object to delete.
-	if len(objects) == 0 {
+	if !deleted {
 		logger.L().Warn(ctx, "No objects found to delete with the given prefix", zap.String("prefix", prefix), zap.String("bucket", s.bucketName))
-
-		return nil
 	}
 
+	return nil
+}
+
+func (s *awsStorage) deleteObjects(ctx context.Context, objects []types.ObjectIdentifier) error {
+	ctx, cancel := context.WithTimeout(ctx, awsOperationTimeout)
+	defer cancel()
+
 	output, err := s.client.DeleteObjects(
 		ctx, &s3.DeleteObjectsInput{
 			Bucket: aws.String(s.bucketName),

packages/shared/pkg/storage/storage_aws_test.go

@@ -0,0 +1,18 @@
+package storage
+
+import (
+	"context"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+)
+
+func TestAWSDeleteObjectsWithPrefixRejectsEmptyPrefix(t *testing.T) {
+	t.Parallel()
+
+	s := &awsStorage{bucketName: "test-bucket"}
+
+	err := s.DeleteObjectsWithPrefix(context.Background(), "")
+	require.Error(t, err)
+	require.Contains(t, err.Error(), "empty prefix")
+}