fix(cli): retry rate-limited function deployments

apps/cli-e2e/src/tests/functions.e2e.test.ts

@@ -124,10 +124,13 @@ describe("functions", () => {
 
     testBehaviour("exits non-zero on 429", async ({ run, apiUrl }) => {
       await run(["functions", "new", FUNCTION_NAME]);
-      await fetch(`${apiUrl}/_ctrl/error-all`, {
+      await fetch(`${apiUrl}/_ctrl/rate-limit`, {
         method: "POST",
         headers: { "Content-Type": "application/json" },
-        body: JSON.stringify({ status: 429, body: { message: "Too Many Requests" } }),
+        body: JSON.stringify({
+          path: `/v1/projects/${PROJECT_REF}/functions/deploy`,
+          retryAfterSeconds: 0,
+        }),
       });
       const result = await run([
         "functions",

apps/cli-go/pkg/function/deploy.go

@@ -7,8 +7,11 @@ import (
 	"io"
 	"io/fs"
 	"mime/multipart"
+	"net/http"
 	"os"
 	"path/filepath"
+	"strconv"
+	"time"
 
 	"github.com/go-errors/errors"
 	"github.com/supabase/cli/pkg/api"
@@ -19,6 +22,8 @@ import (
 
 var ErrNoDeploy = errors.New("All Functions are up to date.")
 
+const deployRateLimitMaxRetries = 8
+
 func (s *EdgeRuntimeAPI) Deploy(ctx context.Context, functionConfig config.FunctionConfig, fsys fs.FS) error {
 	if s.eszip != nil {
 		return s.UpsertFunctions(ctx, functionConfig)
@@ -97,15 +102,36 @@ func (s *EdgeRuntimeAPI) bulkUpload(ctx context.Context, toDeploy []FunctionDepl
 	if err := jq.Collect(); err != nil {
 		return err
 	}
-	if resp, err := s.client.V1BulkUpdateFunctionsWithResponse(ctx, s.project, toUpdate); err != nil {
-		return errors.Errorf("failed to bulk update: %w", err)
-	} else if resp.JSON200 == nil {
-		return errors.Errorf("unexpected bulk update status %d: %s", resp.StatusCode(), string(resp.Body))
+	for attempt := 0; ; attempt++ {
+		resp, err := s.client.V1BulkUpdateFunctionsWithResponse(ctx, s.project, toUpdate)
+		if err != nil {
+			return errors.Errorf("failed to bulk update: %w", err)
+		} else if resp.JSON200 != nil {
+			return nil
+		} else if resp.StatusCode() != http.StatusTooManyRequests || attempt >= deployRateLimitMaxRetries {
+			return errors.Errorf("unexpected bulk update status %d: %s", resp.StatusCode(), string(resp.Body))
+		} else if err := waitForRateLimit(ctx, responseHeaders(resp.HTTPResponse), attempt, "bulk updating functions"); err != nil {
+			return err
+		}
 	}
-	return nil
 }
 
 func (s *EdgeRuntimeAPI) upload(ctx context.Context, param api.V1DeployAFunctionParams, meta FunctionDeployMetadata, fsys fs.FS) (*api.DeployFunctionResponse, error) {
+	for attempt := 0; ; attempt++ {
+		resp, err := s.uploadOnce(ctx, param, meta, fsys)
+		if resp != nil && resp.JSON201 != nil {
+			return resp.JSON201, nil
+		} else if err != nil {
+			return nil, err
+		} else if resp.StatusCode() != http.StatusTooManyRequests || attempt >= deployRateLimitMaxRetries {
+			return nil, errors.Errorf("unexpected deploy status %d: %s", resp.StatusCode(), string(resp.Body))
+		} else if err := waitForRateLimit(ctx, responseHeaders(resp.HTTPResponse), attempt, "deploying function "+cast.Val(meta.Name, "")); err != nil {
+			return nil, err
+		}
+	}
+}
+
+func (s *EdgeRuntimeAPI) uploadOnce(ctx context.Context, param api.V1DeployAFunctionParams, meta FunctionDeployMetadata, fsys fs.FS) (*api.V1DeployAFunctionResponse, error) {
 	body, w := io.Pipe()
 	form := multipart.NewWriter(w)
 	ctx, cancel := context.WithCancelCause(ctx)
@@ -123,10 +149,52 @@ func (s *EdgeRuntimeAPI) upload(ctx context.Context, param api.V1DeployAFunction
 		return nil, cause
 	} else if err != nil {
 		return nil, errors.Errorf("failed to deploy function: %w", err)
-	} else if resp.JSON201 == nil {
-		return nil, errors.Errorf("unexpected deploy status %d: %s", resp.StatusCode(), string(resp.Body))
 	}
-	return resp.JSON201, nil
+	return resp, nil
+}
+
+func waitForRateLimit(ctx context.Context, headers http.Header, attempt int, action string) error {
+	delay := rateLimitDelay(headers, attempt)
+	fmt.Fprintf(os.Stderr, "Rate limit exceeded while %s. Retrying in %s.\n", action, delay.Round(time.Second))
+	timer := time.NewTimer(delay)
+	defer timer.Stop()
+	select {
+	case <-ctx.Done():
+		return ctx.Err()
+	case <-timer.C:
+		return nil
+	}
+}
+
+func responseHeaders(resp *http.Response) http.Header {
+	if resp == nil {
+		return nil
+	}
+	return resp.Header
+}
+
+func rateLimitDelay(headers http.Header, attempt int) time.Duration {
+	if delay, ok := parseRateLimitDelay(headers.Get("Retry-After")); ok {
+		return delay
+	}
+	if delay, ok := parseRateLimitDelay(headers.Get("X-RateLimit-Reset")); ok {
+		return delay
+	}
+	delay := time.Second << min(attempt, 5)
+	return delay
+}
+
+func parseRateLimitDelay(value string) (time.Duration, bool) {
+	if len(value) == 0 {
+		return 0, false
+	}
+	if seconds, err := strconv.Atoi(value); err == nil {
+		return max(time.Duration(seconds)*time.Second, 0), true
+	}
+	if retryAt, err := http.ParseTime(value); err == nil {
+		return max(time.Until(retryAt), 0), true
+	}
+	return 0, false
 }
 
 func writeForm(form *multipart.Writer, meta FunctionDeployMetadata, fsys fs.FS) error {

apps/cli-go/pkg/function/deploy_test.go

@@ -104,6 +104,34 @@ func TestDeployAll(t *testing.T) {
 		assert.Empty(t, gock.GetUnmatchedRequests())
 	})
 
+	t.Run("retries single slug after rate limit", func(t *testing.T) {
+		c := config.FunctionConfig{"demo": {
+			Enabled:    true,
+			Entrypoint: "testdata/shared/whatever.ts",
+		}}
+		// Setup in-memory fs
+		fsys := testImports
+		// Setup mock api
+		defer gock.OffAll()
+		gock.New(mockApiHost).
+			Post("/v1/projects/"+mockProject+"/functions/deploy").
+			MatchParam("slug", "demo").
+			Reply(http.StatusTooManyRequests).
+			SetHeader("Retry-After", "0").
+			JSON(map[string]string{"message": "Too Many Requests"})
+		gock.New(mockApiHost).
+			Post("/v1/projects/"+mockProject+"/functions/deploy").
+			MatchParam("slug", "demo").
+			Reply(http.StatusCreated).
+			JSON(api.DeployFunctionResponse{})
+		// Run test
+		err := client.Deploy(context.Background(), c, fsys)
+		// Check error
+		assert.NoError(t, err)
+		assert.Empty(t, gock.Pending())
+		assert.Empty(t, gock.GetUnmatchedRequests())
+	})
+
 	t.Run("deploys multiple slugs", func(t *testing.T) {
 		c := config.FunctionConfig{
 			"test-ts": {
@@ -138,6 +166,45 @@ func TestDeployAll(t *testing.T) {
 		assert.Empty(t, gock.GetUnmatchedRequests())
 	})
 
+	t.Run("retries bulk update after rate limit reset", func(t *testing.T) {
+		c := config.FunctionConfig{
+			"test-ts": {
+				Enabled:    true,
+				Entrypoint: "testdata/shared/whatever.ts",
+			},
+			"test-js": {
+				Enabled:    true,
+				Entrypoint: "testdata/geometries/Geometries.js",
+			},
+		}
+		// Setup in-memory fs
+		fsys := testImports
+		// Setup mock api
+		defer gock.OffAll()
+		for slug := range c {
+			gock.New(mockApiHost).
+				Post("/v1/projects/"+mockProject+"/functions/deploy").
+				MatchParam("slug", slug).
+				Reply(http.StatusCreated).
+				JSON(api.DeployFunctionResponse{Id: slug})
+		}
+		gock.New(mockApiHost).
+			Put("/v1/projects/"+mockProject+"/functions").
+			Reply(http.StatusTooManyRequests).
+			SetHeader("X-RateLimit-Reset", "0").
+			JSON(map[string]string{"message": "Too Many Requests"})
+		gock.New(mockApiHost).
+			Put("/v1/projects/" + mockProject + "/functions").
+			Reply(http.StatusOK).
+			JSON(api.BulkUpdateFunctionResponse{})
+		// Run test
+		err := client.Deploy(context.Background(), c, fsys)
+		// Check error
+		assert.NoError(t, err)
+		assert.Empty(t, gock.Pending())
+		assert.Empty(t, gock.GetUnmatchedRequests())
+	})
+
 	t.Run("throws error on network failure", func(t *testing.T) {
 		errNetwork := errors.New("network")
 		c := config.FunctionConfig{"demo": {Enabled: true}}