14692275893

Committed 27 Apr 2025 12:54PM UTC coverage: 51.186% (-0.02%) from 51.205%

Build # 14692275893

Build Type

push

github

Committed by

web-flow

Commit Message

fix: retry bulk update endpoint on error (#3492)

Run Details

0 of 9 new or added lines in 1 file covered. (0.0%)

1 existing line in 1 file now uncovered.

6993 of 13662 relevant lines covered (51.19%)

184.84 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

25.66

/pkg/function/batch.go

package function

import (
        "bytes"
        "context"
        "fmt"
        "io"
        "net/http"
        "os"
        "strings"
        "time"

        "github.com/cenkalti/backoff/v4"
        "github.com/docker/go-units"
        "github.com/go-errors/errors"
        "github.com/supabase/cli/pkg/api"
        "github.com/supabase/cli/pkg/config"
)

const (
        eszipContentType = "application/vnd.denoland.eszip"
        maxRetries       = 3
)

func (s *EdgeRuntimeAPI) UpsertFunctions(ctx context.Context, functionConfig config.FunctionConfig, filter ...func(string) bool) error {
        policy := backoff.WithContext(backoff.WithMaxRetries(backoff.NewExponentialBackOff(), maxRetries), ctx)
        result, err := backoff.RetryWithData(func() ([]api.FunctionResponse, error) {
                resp, err := s.client.V1ListAllFunctionsWithResponse(ctx, s.project)
                if err != nil {
                        return nil, errors.Errorf("failed to list functions: %w", err)
                } else if resp.JSON200 == nil {
                        err = errors.Errorf("unexpected list functions status %d: %s", resp.StatusCode(), string(resp.Body))
                        if resp.StatusCode() < http.StatusInternalServerError {
                                err = &backoff.PermanentError{Err: err}
                        }
                        return nil, err
                }
                return *resp.JSON200, nil
        }, policy)
        if err != nil {
                return err
        }
        policy.Reset()
        exists := make(map[string]struct{}, len(result))
        for _, f := range result {
                exists[f.Slug] = struct{}{}
        }
        var toUpdate []api.BulkUpdateFunctionBody
OUTER:
        for slug, function := range functionConfig {
                if !function.Enabled {
                        fmt.Fprintln(os.Stderr, "Skipped deploying Function:", slug)
                        continue
                }
                for _, keep := range filter {
                        if !keep(slug) {
                                continue OUTER
                        }
                }
                var body bytes.Buffer
                meta, err := s.eszip.Bundle(ctx, slug, function.Entrypoint, function.ImportMap, function.StaticFiles, &body)
                if err != nil {
                        return err
                }
                meta.VerifyJwt = &function.VerifyJWT
                // Update if function already exists
                upsert := func() (api.BulkUpdateFunctionBody, error) {
                        if _, ok := exists[slug]; ok {
                                return s.updateFunction(ctx, slug, meta, bytes.NewReader(body.Bytes()))
                        }
                        return s.createFunction(ctx, slug, meta, bytes.NewReader(body.Bytes()))
                }
                functionSize := units.HumanSize(float64(body.Len()))
                fmt.Fprintf(os.Stderr, "Deploying Function: %s (script size: %s)\n", slug, functionSize)
                result, err := backoff.RetryNotifyWithData(upsert, policy, func(err error, d time.Duration) {
                        if strings.Contains(err.Error(), "Duplicated function slug") {
                                exists[slug] = struct{}{}
                        }
                })
                if err != nil {
                        return err
                }
                toUpdate = append(toUpdate, result)
                policy.Reset()
        }
        if len(toUpdate) > 1 {
                if err := backoff.Retry(func() error {
                        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))
                        }
                        return nil
                }, policy); err != nil {
                        return err
                }
        }
        return nil
}

func (s *EdgeRuntimeAPI) updateFunction(ctx context.Context, slug string, meta api.FunctionDeployMetadata, body io.Reader) (api.BulkUpdateFunctionBody, error) {
        resp, err := s.client.V1UpdateAFunctionWithBodyWithResponse(ctx, s.project, slug, &api.V1UpdateAFunctionParams{
                VerifyJwt:      meta.VerifyJwt,
                ImportMapPath:  meta.ImportMapPath,
                EntrypointPath: &meta.EntrypointPath,
        }, eszipContentType, body)
        if err != nil {
                return api.BulkUpdateFunctionBody{}, errors.Errorf("failed to update function: %w", err)
        } else if resp.JSON200 == nil {
                return api.BulkUpdateFunctionBody{}, errors.Errorf("unexpected update function status %d: %s", resp.StatusCode(), string(resp.Body))
        }
        return api.BulkUpdateFunctionBody{
                Id:             resp.JSON200.Id,
                Name:           resp.JSON200.Name,
                Slug:           resp.JSON200.Slug,
                Version:        resp.JSON200.Version,
                EntrypointPath: resp.JSON200.EntrypointPath,
                ImportMap:      resp.JSON200.ImportMap,
                ImportMapPath:  resp.JSON200.ImportMapPath,
                VerifyJwt:      resp.JSON200.VerifyJwt,
                Status:         api.BulkUpdateFunctionBodyStatus(resp.JSON200.Status),
                CreatedAt:      &resp.JSON200.CreatedAt,
        }, nil
}

func (s *EdgeRuntimeAPI) createFunction(ctx context.Context, slug string, meta api.FunctionDeployMetadata, body io.Reader) (api.BulkUpdateFunctionBody, error) {
        resp, err := s.client.V1CreateAFunctionWithBodyWithResponse(ctx, s.project, &api.V1CreateAFunctionParams{
                Slug:           &slug,
                Name:           &slug,
                VerifyJwt:      meta.VerifyJwt,
                ImportMapPath:  meta.ImportMapPath,
                EntrypointPath: &meta.EntrypointPath,
        }, eszipContentType, body)
        if err != nil {
                return api.BulkUpdateFunctionBody{}, errors.Errorf("failed to create function: %w", err)
        } else if resp.JSON201 == nil {
                return api.BulkUpdateFunctionBody{}, errors.Errorf("unexpected create function status %d: %s", resp.StatusCode(), string(resp.Body))
        }
        return api.BulkUpdateFunctionBody{
                Id:             resp.JSON201.Id,
                Name:           resp.JSON201.Name,
                Slug:           resp.JSON201.Slug,
                Version:        resp.JSON201.Version,
                EntrypointPath: resp.JSON201.EntrypointPath,
                ImportMap:      resp.JSON201.ImportMap,
                ImportMapPath:  resp.JSON201.ImportMapPath,
                VerifyJwt:      resp.JSON201.VerifyJwt,
                Status:         api.BulkUpdateFunctionBodyStatus(resp.JSON201.Status),
                CreatedAt:      &resp.JSON201.CreatedAt,
        }, nil
}

1	package function
2
3	import (
4	"bytes"
5	"context"
6	"fmt"
7	"io"
8	"net/http"
9	"os"
10	"strings"
11	"time"
12
13	"github.com/cenkalti/backoff/v4"
14	"github.com/docker/go-units"
15	"github.com/go-errors/errors"
16	"github.com/supabase/cli/pkg/api"
17	"github.com/supabase/cli/pkg/config"
18	)
19
20	const (
21	eszipContentType = "application/vnd.denoland.eszip"
22	maxRetries = 3
23	)
24
25	func (s *EdgeRuntimeAPI) UpsertFunctions(ctx context.Context, functionConfig config.FunctionConfig, filter ...func(string) bool) error {	5✔
26	policy := backoff.WithContext(backoff.WithMaxRetries(backoff.NewExponentialBackOff(), maxRetries), ctx)	5✔
27	result, err := backoff.RetryWithData(func() ([]api.FunctionResponse, error) {	12✔
28	resp, err := s.client.V1ListAllFunctionsWithResponse(ctx, s.project)	7✔
29	if err != nil {	8✔
30	return nil, errors.Errorf("failed to list functions: %w", err)	1✔
31	} else if resp.JSON200 == nil {	9✔
32	err = errors.Errorf("unexpected list functions status %d: %s", resp.StatusCode(), string(resp.Body))	2✔
33	if resp.StatusCode() < http.StatusInternalServerError {	3✔
34	err = &backoff.PermanentError{Err: err}	1✔
35	}	1✔
36	return nil, err	2✔
37	}
38	return *resp.JSON200, nil	4✔
39	}, policy)
40	if err != nil {	6✔
41	return err	1✔
42	}	1✔
43	policy.Reset()	4✔
44	exists := make(map[string]struct{}, len(result))	4✔
45	for _, f := range result {	6✔
46	exists[f.Slug] = struct{}{}	2✔
47	}	2✔
48	var toUpdate []api.BulkUpdateFunctionBody	4✔
49	OUTER:	4✔
50	for slug, function := range functionConfig {	9✔
51	if !function.Enabled {	10✔
52	fmt.Fprintln(os.Stderr, "Skipped deploying Function:", slug)	5✔
53	continue	5✔
54	}
55	for _, keep := range filter {	×
56	if !keep(slug) {	×
57	continue OUTER	×
58	}
59	}
60	var body bytes.Buffer	×
61	meta, err := s.eszip.Bundle(ctx, slug, function.Entrypoint, function.ImportMap, function.StaticFiles, &body)	×
62	if err != nil {	×
63	return err	×
64	}	×
65	meta.VerifyJwt = &function.VerifyJWT	×
66	// Update if function already exists	×
67	upsert := func() (api.BulkUpdateFunctionBody, error) {	×
68	if _, ok := exists[slug]; ok {	×
69	return s.updateFunction(ctx, slug, meta, bytes.NewReader(body.Bytes()))	×
70	}	×
71	return s.createFunction(ctx, slug, meta, bytes.NewReader(body.Bytes()))	×
72	}
73	functionSize := units.HumanSize(float64(body.Len()))	×
74	fmt.Fprintf(os.Stderr, "Deploying Function: %s (script size: %s)\n", slug, functionSize)	×
75	result, err := backoff.RetryNotifyWithData(upsert, policy, func(err error, d time.Duration) {	×
76	if strings.Contains(err.Error(), "Duplicated function slug") {	×
77	exists[slug] = struct{}{}	×
78	}	×
79	})
80	if err != nil {	×
81	return err	×
82	}	×
83	toUpdate = append(toUpdate, result)	×
84	policy.Reset()	×
85	}
86	if len(toUpdate) > 1 {	4✔
NEW 87	if err := backoff.Retry(func() error {	×
NEW 88	if resp, err := s.client.V1BulkUpdateFunctionsWithResponse(ctx, s.project, toUpdate); err != nil {	×
NEW 89	return errors.Errorf("failed to bulk update: %w", err)	×
NEW 90	} else if resp.JSON200 == nil {	×
NEW 91	return errors.Errorf("unexpected bulk update status %d: %s", resp.StatusCode(), string(resp.Body))	×
NEW 92	}	×
NEW 93	return nil	×
NEW 94	}, policy); err != nil {	×
NEW 95	return err	×
UNCOV 96	}	×
97	}
98	return nil	4✔
99	}
100
101	func (s *EdgeRuntimeAPI) updateFunction(ctx context.Context, slug string, meta api.FunctionDeployMetadata, body io.Reader) (api.BulkUpdateFunctionBody, error) {	×
102	resp, err := s.client.V1UpdateAFunctionWithBodyWithResponse(ctx, s.project, slug, &api.V1UpdateAFunctionParams{	×
103	VerifyJwt: meta.VerifyJwt,	×
104	ImportMapPath: meta.ImportMapPath,	×
105	EntrypointPath: &meta.EntrypointPath,	×
106	}, eszipContentType, body)	×
107	if err != nil {	×
108	return api.BulkUpdateFunctionBody{}, errors.Errorf("failed to update function: %w", err)	×
109	} else if resp.JSON200 == nil {	×
110	return api.BulkUpdateFunctionBody{}, errors.Errorf("unexpected update function status %d: %s", resp.StatusCode(), string(resp.Body))	×
111	}	×
112	return api.BulkUpdateFunctionBody{	×
113	Id: resp.JSON200.Id,	×
114	Name: resp.JSON200.Name,	×
115	Slug: resp.JSON200.Slug,	×
116	Version: resp.JSON200.Version,	×
117	EntrypointPath: resp.JSON200.EntrypointPath,	×
118	ImportMap: resp.JSON200.ImportMap,	×
119	ImportMapPath: resp.JSON200.ImportMapPath,	×
120	VerifyJwt: resp.JSON200.VerifyJwt,	×
121	Status: api.BulkUpdateFunctionBodyStatus(resp.JSON200.Status),	×
122	CreatedAt: &resp.JSON200.CreatedAt,	×
123	}, nil	×
124	}
125
126	func (s *EdgeRuntimeAPI) createFunction(ctx context.Context, slug string, meta api.FunctionDeployMetadata, body io.Reader) (api.BulkUpdateFunctionBody, error) {	×
127	resp, err := s.client.V1CreateAFunctionWithBodyWithResponse(ctx, s.project, &api.V1CreateAFunctionParams{	×
128	Slug: &slug,	×
129	Name: &slug,	×
130	VerifyJwt: meta.VerifyJwt,	×
131	ImportMapPath: meta.ImportMapPath,	×
132	EntrypointPath: &meta.EntrypointPath,	×
133	}, eszipContentType, body)	×
134	if err != nil {	×
135	return api.BulkUpdateFunctionBody{}, errors.Errorf("failed to create function: %w", err)	×
136	} else if resp.JSON201 == nil {	×
137	return api.BulkUpdateFunctionBody{}, errors.Errorf("unexpected create function status %d: %s", resp.StatusCode(), string(resp.Body))	×
138	}	×
139	return api.BulkUpdateFunctionBody{	×
140	Id: resp.JSON201.Id,	×
141	Name: resp.JSON201.Name,	×
142	Slug: resp.JSON201.Slug,	×
143	Version: resp.JSON201.Version,	×
144	EntrypointPath: resp.JSON201.EntrypointPath,	×
145	ImportMap: resp.JSON201.ImportMap,	×
146	ImportMapPath: resp.JSON201.ImportMapPath,	×
147	VerifyJwt: resp.JSON201.VerifyJwt,	×
148	Status: api.BulkUpdateFunctionBodyStatus(resp.JSON201.Status),	×
149	CreatedAt: &resp.JSON201.CreatedAt,	×
150	}, nil	×
151	}

supabase / cli / 14692275893

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous