feat: hot reload eszip bundle when function source changes (#3710)

* fix: file watcher skeleton for serving functions

* chore: add debounce file watcher

* chore: remove unnecessary log

* chore: impose default watch limit

* chore: add watcher tests and main.ts (#3717)

* chore: add watcher tests and main.ts

* chore: add serve tests

* chore: fix lints

* chore: remove sleep in tests

* chore: simulate fs event when possible

* chore: avoid flaky tests

* chore: undo debounce test

---------

Co-authored-by: Qiao Han <qiao@supabase.io>

* chore: simplify event ignore logic

* chore: add streamer unit tests

* chore: simplify streamer test

* chore: update serve tests

* chore: update unit tests

* chore: update watcher tests

---------

Co-authored-by: Andrew Valleteau <avallete@users.noreply.github.com>

Author: sweatybridge

internal/functions/serve/serve.go

@@ -10,7 +10,9 @@ import (
 	"strconv"
 	"strings"
 
+	"github.com/docker/cli/cli/compose/loader"
 	"github.com/docker/docker/api/types/container"
+	"github.com/docker/docker/api/types/mount"
 	"github.com/docker/docker/api/types/network"
 	"github.com/docker/go-connections/nat"
 	"github.com/go-errors/errors"
@@ -46,6 +48,7 @@ func (mode InspectMode) toFlag() string {
 type RuntimeOption struct {
 	InspectMode *InspectMode
 	InspectMain bool
+	fileWatcher *debounceFileWatcher
 }
 
 func (i *RuntimeOption) toArgs() []string {
@@ -68,6 +71,36 @@ const (
 var mainFuncEmbed string
 
 func Run(ctx context.Context, envFilePath string, noVerifyJWT *bool, importMapPath string, runtimeOption RuntimeOption, fsys afero.Fs) error {
+	watcher, err := NewDebounceFileWatcher()
+	if err != nil {
+		return err
+	}
+	go watcher.Start()
+	defer watcher.Close()
+	// TODO: refactor this to edge runtime service
+	runtimeOption.fileWatcher = watcher
+	if err := restartEdgeRuntime(ctx, envFilePath, noVerifyJWT, importMapPath, runtimeOption, fsys); err != nil {
+		return err
+	}
+	streamer := NewLogStreamer(ctx)
+	go streamer.Start(utils.EdgeRuntimeId)
+	defer streamer.Close()
+	for {
+		select {
+		case <-ctx.Done():
+			fmt.Println("Stopped serving " + utils.Bold(utils.FunctionsDir))
+			return ctx.Err()
+		case <-watcher.RestartCh:
+			if err := restartEdgeRuntime(ctx, envFilePath, noVerifyJWT, importMapPath, runtimeOption, fsys); err != nil {
+				return err
+			}
+		case err := <-streamer.ErrCh:
+			return err
+		}
+	}
+}
+
+func restartEdgeRuntime(ctx context.Context, envFilePath string, noVerifyJWT *bool, importMapPath string, runtimeOption RuntimeOption, fsys afero.Fs) error {
 	// 1. Sanity checks.
 	if err := flags.LoadConfig(fsys); err != nil {
 		return err
@@ -84,14 +117,7 @@ func Run(ctx context.Context, envFilePath string, noVerifyJWT *bool, importMapPa
 	dbUrl := fmt.Sprintf("postgresql://postgres:postgres@%s:5432/postgres", utils.DbAliases[0])
 	// 3. Serve and log to console
 	fmt.Fprintln(os.Stderr, "Setting up Edge Functions runtime...")
-	if err := ServeFunctions(ctx, envFilePath, noVerifyJWT, importMapPath, dbUrl, runtimeOption, fsys); err != nil {
-		return err
-	}
-	if err := utils.DockerStreamLogs(ctx, utils.EdgeRuntimeId, os.Stdout, os.Stderr); err != nil {
-		return err
-	}
-	fmt.Println("Stopped serving " + utils.Bold(utils.FunctionsDir))
-	return nil
+	return ServeFunctions(ctx, envFilePath, noVerifyJWT, importMapPath, dbUrl, runtimeOption, fsys)
 }
 
 func ServeFunctions(ctx context.Context, envFilePath string, noVerifyJWT *bool, importMapPath string, dbUrl string, runtimeOption RuntimeOption, fsys afero.Fs) error {
@@ -131,6 +157,19 @@ func ServeFunctions(ctx context.Context, envFilePath string, noVerifyJWT *bool,
 	if err != nil {
 		return err
 	}
+	if watcher := runtimeOption.fileWatcher; watcher != nil {
+		var watchPaths []string
+		for _, b := range binds {
+			if spec, err := loader.ParseVolume(b); err != nil {
+				return errors.Errorf("failed to parse docker volume: %w", err)
+			} else if spec.Type == string(mount.TypeBind) {
+				watchPaths = append(watchPaths, spec.Source)
+			}
+		}
+		if err := watcher.SetWatchPaths(watchPaths, fsys); err != nil {
+			return err
+		}
+	}
 	env = append(env, "SUPABASE_INTERNAL_FUNCTIONS_CONFIG="+functionsConfigString)
 	// 3. Parse entrypoint script
 	cmd := append([]string{
@@ -215,6 +254,7 @@ func populatePerFunctionConfigs(cwd, importMapPath string, noVerifyJWT *bool, fs
 	for slug, fc := range functionsConfig {
 		if !fc.Enabled {
 			fmt.Fprintln(os.Stderr, "Skipped serving Function:", slug)
+			delete(functionsConfig, slug)
 			continue
 		}
 		modules, err := deploy.GetBindMounts(cwd, utils.FunctionsDir, "", fc.Entrypoint, fc.ImportMap, fsys)

internal/functions/serve/serve_test.go

@@ -2,9 +2,10 @@ package serve
 
 import (
 	"context"
+	"embed"
 	"net/http"
-	"os"
 	"path/filepath"
+	"strings"
 	"testing"
 
 	"github.com/docker/docker/api/types/container"
@@ -17,13 +18,20 @@ import (
 	"github.com/supabase/cli/pkg/cast"
 )
 
+var (
+	//go:embed testdata/config.toml
+	testConfig []byte
+	//go:embed testdata/*
+	testdata embed.FS
+)
+
 func TestServeCommand(t *testing.T) {
 	t.Run("serves all functions", func(t *testing.T) {
 		// Setup in-memory fs
 		fsys := afero.NewMemMapFs()
-		require.NoError(t, utils.InitConfig(utils.InitParams{ProjectId: "test"}, fsys))
+		require.NoError(t, afero.WriteFile(fsys, utils.ConfigPath, testConfig, 0644))
 		require.NoError(t, afero.WriteFile(fsys, utils.FallbackEnvFilePath, []byte{}, 0644))
-		require.NoError(t, afero.WriteFile(fsys, utils.FallbackImportMapPath, []byte{}, 0644))
+		require.NoError(t, afero.WriteFile(fsys, utils.FallbackImportMapPath, []byte("{}"), 0644))
 		// Setup mock docker
 		require.NoError(t, apitest.MockDocker(utils.Docker))
 		defer gock.OffAll()
@@ -36,11 +44,11 @@ func TestServeCommand(t *testing.T) {
 			Delete("/v" + utils.Docker.ClientVersion() + "/containers/" + containerId).
 			Reply(http.StatusOK)
 		apitest.MockDockerStart(utils.Docker, utils.GetRegistryImageUrl(utils.Config.EdgeRuntime.Image), containerId)
-		require.NoError(t, apitest.MockDockerLogs(utils.Docker, containerId, "success"))
-		// Run test
+		require.NoError(t, apitest.MockDockerLogsStream(utils.Docker, containerId, 1, strings.NewReader("failed")))
+		// Run test with timeout context
 		err := Run(context.Background(), "", nil, "", RuntimeOption{}, fsys)
 		// Check error
-		assert.NoError(t, err)
+		assert.ErrorContains(t, err, "error running container: exit 1")
 		assert.Empty(t, apitest.ListUnmatchedRequests())
 	})
 
@@ -88,7 +96,6 @@ func TestServeCommand(t *testing.T) {
 	})
 
 	t.Run("throws error on missing import map", func(t *testing.T) {
-		utils.CurrentDirAbs = "/"
 		// Setup in-memory fs
 		fsys := afero.NewMemMapFs()
 		require.NoError(t, utils.InitConfig(utils.InitParams{ProjectId: "test"}, fsys))
@@ -105,6 +112,62 @@ func TestServeCommand(t *testing.T) {
 		// Run test
 		err := Run(context.Background(), ".env", cast.Ptr(true), "import_map.json", RuntimeOption{}, fsys)
 		// Check error
-		assert.ErrorIs(t, err, os.ErrNotExist)
+		assert.ErrorContains(t, err, "failed to resolve relative path:")
+	})
+}
+
+func TestServeFunctions(t *testing.T) {
+	require.NoError(t, utils.Config.Load("testdata/config.toml", testdata))
+	utils.UpdateDockerIds()
+
+	t.Run("runs inspect mode", func(t *testing.T) {
+		// Setup in-memory fs
+		fsys := afero.FromIOFS{FS: testdata}
+		// Setup mock docker
+		require.NoError(t, apitest.MockDocker(utils.Docker))
+		defer gock.OffAll()
+		apitest.MockDockerStart(utils.Docker, utils.GetRegistryImageUrl(utils.Config.EdgeRuntime.Image), utils.EdgeRuntimeId)
+		// Run test
+		err := ServeFunctions(context.Background(), "", nil, "", "", RuntimeOption{
+			InspectMode: cast.Ptr(InspectModeRun),
+			InspectMain: true,
+		}, fsys)
+		// Check error
+		assert.NoError(t, err)
+		assert.Empty(t, apitest.ListUnmatchedRequests())
+	})
+
+	t.Run("parses env file", func(t *testing.T) {
+		envPath := "/project/.env"
+		// Setup in-memory fs
+		fsys := afero.NewMemMapFs()
+		require.NoError(t, utils.WriteFile(envPath, []byte(`
+			DATABASE_URL=postgresql://localhost:5432/test
+			API_KEY=secret123
+			DEBUG=true
+		`), fsys))
+		// Run test
+		env, err := parseEnvFile(envPath, fsys)
+		// Check error
+		assert.NoError(t, err)
+		assert.ElementsMatch(t, []string{
+			"DATABASE_URL=postgresql://localhost:5432/test",
+			"API_KEY=secret123",
+			"DEBUG=true",
+		}, env)
+	})
+
+	t.Run("parses function config", func(t *testing.T) {
+		// Setup in-memory fs
+		fsys := afero.FromIOFS{FS: testdata}
+		// Run test
+		binds, configString, err := populatePerFunctionConfigs("/", "", nil, fsys)
+		// Check error
+		assert.NoError(t, err)
+		assert.ElementsMatch(t, []string{
+			"supabase_edge_runtime_test:/root/.cache/deno:rw",
+			"/supabase/functions/:/supabase/functions/:ro",
+		}, binds)
+		assert.Equal(t, `{"hello":{"verifyJWT":true,"entrypointPath":"testdata/functions/hello/index.ts","staticFiles":["testdata/image.png"]}}`, configString)
 	})
 }

internal/functions/serve/streamer.go

@@ -0,0 +1,47 @@
+package serve
+
+import (
+	"context"
+	"os"
+	"time"
+
+	"github.com/cenkalti/backoff/v4"
+	"github.com/containerd/errdefs"
+	"github.com/docker/docker/api/types/container"
+	"github.com/go-errors/errors"
+	"github.com/supabase/cli/internal/utils"
+)
+
+type logStreamer struct {
+	ctx   context.Context
+	Close context.CancelFunc
+	ErrCh chan error
+}
+
+func NewLogStreamer(ctx context.Context) logStreamer {
+	cancelCtx, cancel := context.WithCancel(ctx)
+	return logStreamer{
+		ctx:   cancelCtx,
+		Close: cancel,
+		ErrCh: make(chan error, 1),
+	}
+}
+
+// Used by unit tests
+var retryInterval = time.Millisecond * 400
+
+func (s *logStreamer) Start(containerID string) {
+	// Retry indefinitely until stream is closed
+	policy := backoff.WithContext(backoff.NewConstantBackOff(retryInterval), s.ctx)
+	fetch := func() error {
+		if err := utils.DockerStreamLogs(s.ctx, containerID, os.Stdout, os.Stderr, func(lo *container.LogsOptions) {
+			lo.Timestamps = true
+		}); errdefs.IsNotFound(err) || errdefs.IsConflict(err) || errors.Is(err, utils.ErrContainerKilled) {
+			return err
+		} else if err != nil {
+			return &backoff.PermanentError{Err: err}
+		}
+		return errors.Errorf("container exited gracefully: %s", containerID)
+	}
+	s.ErrCh <- backoff.Retry(fetch, policy)
+}

internal/functions/serve/streamer_test.go

@@ -0,0 +1,91 @@
+package serve
+
+import (
+	"context"
+	"net/http"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/h2non/gock"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"github.com/supabase/cli/internal/testing/apitest"
+	"github.com/supabase/cli/internal/utils"
+)
+
+func TestLogStreamer(t *testing.T) {
+	containerID := "test-container"
+	retryInterval = 0
+
+	t.Run("streams logs from container", func(t *testing.T) {
+		// Setup mock docker
+		require.NoError(t, apitest.MockDocker(utils.Docker))
+		defer gock.OffAll()
+		require.NoError(t, apitest.MockDockerLogsStream(utils.Docker, containerID, 1, strings.NewReader("")))
+		// Run test
+		streamer := NewLogStreamer(context.Background())
+		streamer.Start(containerID)
+		// Check error
+		select {
+		case err := <-streamer.ErrCh:
+			assert.ErrorContains(t, err, "error running container: exit 1")
+		case <-time.After(2 * time.Second):
+			assert.Fail(t, "missing error signal from closing")
+		}
+		assert.Empty(t, apitest.ListUnmatchedRequests())
+	})
+
+	t.Run("retries on container exit", func(t *testing.T) {
+		// Setup mock docker
+		require.NoError(t, apitest.MockDocker(utils.Docker))
+		defer gock.OffAll()
+		require.NoError(t, apitest.MockDockerLogsStream(utils.Docker, containerID, 0, strings.NewReader("")))
+		require.NoError(t, apitest.MockDockerLogsStream(utils.Docker, containerID, 137, strings.NewReader("")))
+		require.NoError(t, apitest.MockDockerLogsStream(utils.Docker, containerID, 1, strings.NewReader("")))
+		// Run test
+		streamer := NewLogStreamer(context.Background())
+		streamer.Start(containerID)
+		// Check error
+		select {
+		case err := <-streamer.ErrCh:
+			assert.ErrorContains(t, err, "error running container: exit 1")
+		case <-time.After(2 * time.Second):
+			assert.Fail(t, "missing error signal from closing")
+		}
+		assert.Empty(t, apitest.ListUnmatchedRequests())
+	})
+
+	t.Run("retries on missing container", func(t *testing.T) {
+		// Setup mock docker
+		require.NoError(t, apitest.MockDocker(utils.Docker))
+		defer gock.OffAll()
+		gock.New(utils.Docker.DaemonHost()).
+			Get("/v" + utils.Docker.ClientVersion() + "/containers/" + containerID + "/logs").
+			Reply(http.StatusNotFound).
+			BodyString("No such container")
+		gock.New(utils.Docker.DaemonHost()).
+			Get("/v" + utils.Docker.ClientVersion() + "/containers/" + containerID + "/logs").
+			Reply(http.StatusConflict).
+			BodyString("can not get logs from container which is dead or marked for removal")
+		gock.New(utils.Docker.DaemonHost()).
+			Get("/v" + utils.Docker.ClientVersion() + "/containers/" + containerID + "/logs").
+			Reply(http.StatusOK)
+		gock.New(utils.Docker.DaemonHost()).
+			Get("/v" + utils.Docker.ClientVersion() + "/containers/" + containerID + "/json").
+			Reply(http.StatusNotFound).
+			BodyString("No such object")
+		require.NoError(t, apitest.MockDockerLogsStream(utils.Docker, containerID, 1, strings.NewReader("")))
+		// Run test
+		streamer := NewLogStreamer(context.Background())
+		streamer.Start(containerID)
+		// Check error
+		select {
+		case err := <-streamer.ErrCh:
+			assert.ErrorContains(t, err, "error running container: exit 1")
+		case <-time.After(2 * time.Second):
+			assert.Fail(t, "missing error signal from closing")
+		}
+		assert.Empty(t, apitest.ListUnmatchedRequests())
+	})
+}