Bucket exists

.github/CODEOWNERS

@@ -1,4 +1,4 @@
 # These owners will be the default owners for everything in
 # the repo. Unless a later match takes precedence,
 
-* @schucly @arttor @fa-at-pulsit
+* @schucly @arttor @fa-at-pulsit @aiivashchenko

service/worker/handler/bucket_handlers.go

@@ -69,30 +69,9 @@ func (s *svc) HandleBucketCreate(ctx context.Context, t *asynq.Task) (err error)
 	}
 
 	// 1. create bucket
-	_, err = toClient.AWS().CreateBucketWithContext(ctx, &s3.CreateBucketInput{
-		Bucket: &p.Bucket,
-		CreateBucketConfiguration: &s3.CreateBucketConfiguration{
-			LocationConstraint: &p.Location,
-		},
-	})
-	if err != nil && (dom.ErrContains(err, "region", "location")) {
-		defaultRegion := toClient.Config().DefaultRegion
-		if defaultRegion == "" {
-			defaultRegion = s.clients.DefaultRegion()
-		}
-		logger.Warn().Msgf("unable to create bucket: invalid region %q: retry with default region %q", p.Location, defaultRegion)
-		_, err = toClient.AWS().CreateBucketWithContext(ctx, &s3.CreateBucketInput{
-			Bucket: &p.Bucket,
-			CreateBucketConfiguration: &s3.CreateBucketConfiguration{
-				LocationConstraint: &defaultRegion,
-			},
-		})
-	}
+	err = s.createBucketIfNotExists(ctx, toClient, p)
 	if err != nil {
-		if !dom.ErrContains(err, "BucketAlreadyExists") {
-			return err
-		}
-		// else BucketAlreadyExists - ok
+		return err
 	}
 
 	// 2. copy tags
@@ -145,6 +124,47 @@ func (s *svc) HandleBucketCreate(ctx context.Context, t *asynq.Task) (err error)
 	return nil
 }
 
+func (s *svc) createBucketIfNotExists(ctx context.Context, toClient s3client.Client, p tasks.BucketCreatePayload) error {
+	ctx = log.WithBucket(ctx, p.Bucket)
+	logger := zerolog.Ctx(ctx)
+	// check if bucket already exists:
+	_, err := toClient.AWS().HeadBucketWithContext(ctx, &s3.HeadBucketInput{
+		Bucket: &p.Bucket,
+	})
+	if err == nil {
+		// already exists
+		return nil
+	}
+
+	// create bucket
+	_, err = toClient.AWS().CreateBucketWithContext(ctx, &s3.CreateBucketInput{
+		Bucket: &p.Bucket,
+		CreateBucketConfiguration: &s3.CreateBucketConfiguration{
+			LocationConstraint: &p.Location,
+		},
+	})
+	if err != nil && (dom.ErrContains(err, "region", "location")) {
+		defaultRegion := toClient.Config().DefaultRegion
+		if defaultRegion == "" {
+			defaultRegion = s.clients.DefaultRegion()
+		}
+		logger.Warn().Msgf("unable to create bucket: invalid region %q: retry with default region %q", p.Location, defaultRegion)
+		_, err = toClient.AWS().CreateBucketWithContext(ctx, &s3.CreateBucketInput{
+			Bucket: &p.Bucket,
+			CreateBucketConfiguration: &s3.CreateBucketConfiguration{
+				LocationConstraint: &defaultRegion,
+			},
+		})
+	}
+	if err != nil {
+		if !dom.ErrContains(err, "BucketAlreadyExists") {
+			return err
+		}
+		// else BucketAlreadyExists - ok
+	}
+	return nil
+}
+
 func (s *svc) bucketCopyLC(ctx context.Context, fromClient, toClient s3client.Client, p tasks.BucketCreatePayload) error {
 	logger := zerolog.Ctx(ctx)
 	if !features.Lifecycle(ctx) {

test/migration/init_test.go

@@ -203,7 +203,7 @@ func TestMain(m *testing.M) {
 	go func() {
 		app := dom.AppInfo{
 			Version: "test",
-			App:     "proxy",
+			App:     "worker",
 			AppID:   xid.New().String(),
 		}
 		workerCtx, cancelFn := context.WithCancel(ctx)