From 1c2840863b06a5790a8735dd45faffae5c5d52b6 Mon Sep 17 00:00:00 2001 From: Julien Goux Date: Wed, 10 Jun 2026 16:37:29 +0200 Subject: [PATCH] fix(cli): retry rate-limited function deployments --- apps/cli-e2e/src/tests/functions.e2e.test.ts | 7 +- apps/cli-go/pkg/function/deploy.go | 84 ++++++++++++++++++-- apps/cli-go/pkg/function/deploy_test.go | 67 ++++++++++++++++ 3 files changed, 148 insertions(+), 10 deletions(-) diff --git a/apps/cli-e2e/src/tests/functions.e2e.test.ts b/apps/cli-e2e/src/tests/functions.e2e.test.ts index cd4af63c0a..9f05ae0bfe 100644 --- a/apps/cli-e2e/src/tests/functions.e2e.test.ts +++ b/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", diff --git a/apps/cli-go/pkg/function/deploy.go b/apps/cli-go/pkg/function/deploy.go index 88e2f464ae..79de707d6f 100644 --- a/apps/cli-go/pkg/function/deploy.go +++ b/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 { diff --git a/apps/cli-go/pkg/function/deploy_test.go b/apps/cli-go/pkg/function/deploy_test.go index 50e2e859e5..fc4d5fcf91 100644 --- a/apps/cli-go/pkg/function/deploy_test.go +++ b/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}}