chore: add streamer unit tests

Author: sweatybridge

internal/functions/serve/streamer.go

@@ -3,44 +3,54 @@ package serve
 import (
 	"context"
 	"os"
-	"strings"
+	"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
-	cancel context.CancelFunc
-	ErrCh  chan error
+	ctx   context.Context
+	clock backoff.Clock
+	Close context.CancelFunc
+	ErrCh chan error
 }
 
-func NewLogStreamer(ctx context.Context) logStreamer {
-	cancelCtx, cancel := context.WithCancel(ctx)
-	return logStreamer{
-		ctx:    cancelCtx,
-		cancel: cancel,
-		ErrCh:  make(chan error, 1),
+func NewLogStreamer(ctx context.Context, opts ...func(*logStreamer)) logStreamer {
+	s := logStreamer{
+		clock: backoff.SystemClock,
+		ErrCh: make(chan error, 1),
 	}
+	s.ctx, s.Close = context.WithCancel(ctx)
+	for _, apply := range opts {
+		apply(&s)
+	}
+	return s
 }
 
+const (
+	initialInterval = time.Millisecond * 50
+	maxElapsedTime  = time.Second * 20
+)
+
 func (s *logStreamer) Start(containerID string) {
-	for {
+	policy := backoff.WithContext(backoff.NewExponentialBackOff(
+		backoff.WithInitialInterval(initialInterval),
+		backoff.WithMaxElapsedTime(maxElapsedTime),
+		backoff.WithClockProvider(s.clock),
+	), s.ctx)
+	fetch := func() error {
 		if err := utils.DockerStreamLogs(s.ctx, containerID, os.Stdout, os.Stderr, func(lo *container.LogsOptions) {
 			lo.Timestamps = true
-		}); err != nil &&
-			!errdefs.IsNotFound(err) &&
-			!strings.HasSuffix(err.Error(), "exit 137") &&
-			!strings.HasSuffix(err.Error(), "can not get logs from container which is dead or marked for removal") {
-			s.ErrCh <- err
-			break
+		}); 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)
 	}
-}
-
-func (s *logStreamer) Close() {
-	if s.cancel != nil {
-		s.cancel()
-	}
+	s.ErrCh <- backoff.Retry(fetch, policy)
 }

internal/functions/serve/watcher_test.go

@@ -84,7 +84,7 @@ func (s *WatcherIntegrationSetup) CreateFileWatcher() (*debounceFileWatcher, err
 	return watcher, nil
 }
 
-func TestFileWatcherIntegration(t *testing.T) {
+func TestFileWatcher(t *testing.T) {
 	t.Run("detects TypeScript function changes and triggers restart", func(t *testing.T) {
 		setup := NewWatcherIntegrationSetup(t)
 		defer setup.Cleanup()

internal/testing/apitest/docker.go

@@ -3,7 +3,9 @@ package apitest
 import (
 	"bytes"
 	"fmt"
+	"io"
 	"net/http"
+	"strings"
 
 	"github.com/docker/docker/api"
 	"github.com/docker/docker/api/types/container"
@@ -84,9 +86,17 @@ func MockDockerStop(docker *client.Client) {
 
 // Ref: internal/utils/docker.go::DockerRunOnce
 func setupDockerLogs(docker *client.Client, containerID, stdout string, exitCode int) error {
+	err := MockDockerLogsStream(docker, containerID, exitCode, strings.NewReader(stdout))
+	gock.New(docker.DaemonHost()).
+		Delete("/v" + docker.ClientVersion() + "/containers/" + containerID).
+		Reply(http.StatusOK)
+	return err
+}
+
+func MockDockerLogsStream(docker *client.Client, containerID string, exitCode int, r io.Reader) error {
 	var body bytes.Buffer
 	writer := stdcopy.NewStdWriter(&body, stdcopy.Stdout)
-	_, err := writer.Write([]byte(stdout))
+	_, err := io.Copy(writer, r)
 	gock.New(docker.DaemonHost()).
 		Get("/v"+docker.ClientVersion()+"/containers/"+containerID+"/logs").
 		Reply(http.StatusOK).
@@ -99,9 +109,6 @@ func setupDockerLogs(docker *client.Client, containerID, stdout string, exitCode
 			State: &container.State{
 				ExitCode: exitCode,
 			}}})
-	gock.New(docker.DaemonHost()).
-		Delete("/v" + docker.ClientVersion() + "/containers/" + containerID).
-		Reply(http.StatusOK)
 	return err
 }
 

internal/utils/docker.go

@@ -376,6 +376,8 @@ func DockerRunOnceWithConfig(ctx context.Context, config container.Config, hostC
 	return DockerStreamLogs(ctx, container, stdout, stderr)
 }
 
+var ErrContainerKilled = errors.New("exit 137")
+
 func DockerStreamLogs(ctx context.Context, containerId string, stdout, stderr io.Writer, opts ...func(*container.LogsOptions)) error {
 	logsOptions := container.LogsOptions{
 		ShowStdout: true,
@@ -399,10 +401,15 @@ func DockerStreamLogs(ctx context.Context, containerId string, stdout, stderr io
 	if err != nil {
 		return errors.Errorf("failed to inspect docker container: %w", err)
 	}
-	if resp.State.ExitCode > 0 {
-		return errors.Errorf("error running container: exit %d", resp.State.ExitCode)
+	switch resp.State.ExitCode {
+	case 0:
+		return nil
+	case 137:
+		err = ErrContainerKilled
+	default:
+		err = errors.Errorf("exit %d", resp.State.ExitCode)
 	}
-	return nil
+	return errors.Errorf("error running container: %w", err)
 }
 
 func DockerStreamLogsOnce(ctx context.Context, containerId string, stdout, stderr io.Writer) error {