14585959955

Committed 22 Apr 2025 03:20AM UTC coverage: 51.164% (-0.008%) from 51.172%

Build # 14585959955

Build Type

push

github

Committed by

web-flow

Commit Message

fix: race condition when deploying from concurrent jobs (#3469)

Run Details

1 of 6 new or added lines in 1 file covered. (16.67%)

6987 of 13656 relevant lines covered (51.16%)

184.92 hits per line

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

21.78

/pkg/function/batch.go

package function

import (
        "bytes"
        "context"
        "fmt"
        "io"
        "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 {
        var result []api.FunctionResponse
        if resp, err := s.client.V1ListAllFunctionsWithResponse(ctx, s.project); err != nil {
                return errors.Errorf("failed to list functions: %w", err)
        } else if resp.JSON200 == nil {
                return errors.Errorf("unexpected list functions status %d: %s", resp.StatusCode(), string(resp.Body))
        } else {
                result = *resp.JSON200
        }
        exists := make(map[string]struct{}, len(result))
        for _, f := range result {
                exists[f.Slug] = struct{}{}
        }
        policy := backoff.WithContext(backoff.WithMaxRetries(backoff.NewExponentialBackOff(), maxRetries), ctx)
        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 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
}

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

supabase / cli / 14585959955

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