fix(knative): fix goroutine leak in Deploy

pkg/knative/deployer.go

@@ -46,6 +46,8 @@ const (
 	KnativeDeployerName = deployers.Knative
 )
 
+var errPrivateRegistry = stdErrors.New("your function image is unreachable. It is possible that your docker registry is private. If so, make sure you have set up pull secrets https://knative.dev/docs/developer/serving/deploying-from-private-registry")
+
 type DeployerOpt func(*Deployer)
 
 type Deployer struct {
@@ -212,15 +214,24 @@ consider using the --image-pull-secret flag, or setting up pull secrets manually
 		out = os.Stderr
 	}
 	since := time.Now()
+	deployCtx, cancelDeploy := context.WithCancel(ctx)
+	logsDone := make(chan struct{})
 	go func() {
-		_ = GetKServiceLogs(ctx, LogsOptions{
+		defer close(logsDone)
+		_ = GetKServiceLogs(deployCtx, LogsOptions{
 			Name:      f.Name,
 			Namespace: namespace,
 			Image:     f.Deploy.Image,
 			Since:     &since,
 			Follow:    true,
 		}, out)
 	}()
+	// Stop the log follower before Deploy returns. Cancel first: Follow blocks
+	// until the context is done, and waiting without cancelling would deadlock.
+	defer func() {
+		cancelDeploy()
+		<-logsDone
+	}()
 
 	previousService, err := client.GetService(ctx, f.Name)
 	if err != nil {
@@ -254,44 +265,21 @@ consider using the --image-pull-secret flag, or setting up pull secrets manually
 			if d.verbose {
 				fmt.Println("Waiting for Knative Service to become ready")
 			}
-			chprivate := make(chan bool)
-			cherr := make(chan error)
-			go func() {
-				private := false
-				for !private {
-					time.Sleep(5 * time.Second)
-					private = d.isImageInPrivateRegistry(ctx, client, f)
-					chprivate <- private
-				}
-				close(chprivate)
-			}()
-			go func() {
+			waitForService := func(ctx context.Context) error {
 				err, _ := client.WaitForService(ctx, f.Name,
 					clientservingv1.WaitConfig{Timeout: k8s.DefaultWaitingTimeout, ErrorWindow: k8s.DefaultErrorWindowTimeout},
 					wait.NoopMessageCallback())
-				cherr <- err
-				close(cherr)
-			}()
-
-			presumePrivate := false
-		main:
-			// Wait for either a timeout or a container condition signaling the image is unreachable
-			for {
-				select {
-				case private := <-chprivate:
-					if private {
-						presumePrivate = true
-						break main
-					}
-				case err = <-cherr:
-					break main
-				}
+				return err
 			}
-			if presumePrivate {
-				err := fmt.Errorf("your function image is unreachable. It is possible that your docker registry is private. If so, make sure you have set up pull secrets https://knative.dev/docs/developer/serving/deploying-from-private-registry")
-				return fn.DeploymentResult{}, err
+			isPrivateRegistry := func(ctx context.Context) bool {
+				return d.isImageInPrivateRegistry(ctx, client, f)
 			}
+
+			err = waitForReadyOrPrivateRegistry(deployCtx, waitForService, isPrivateRegistry, 5*time.Second)
 			if err != nil {
+				if stdErrors.Is(err, errPrivateRegistry) {
+					return fn.DeploymentResult{}, err
+				}
 				err = fmt.Errorf("knative deployer failed to wait for the Knative Service to become ready: %v", err)
 				if !d.verbose {
 					fmt.Fprintln(os.Stderr, "\nService output:")
@@ -890,3 +878,47 @@ func wrapK8sConnectionError(err error) error {
 
 	return nil
 }
+
+// waitForReadyOrPrivateRegistry handles polling for a private registry image pull error
+// or successfully waiting for the service to become ready.
+func waitForReadyOrPrivateRegistry(
+	ctx context.Context,
+	waitForService func(context.Context) error,
+	isPrivateRegistry func(context.Context) bool,
+	pollInterval time.Duration,
+) error {
+	waitCtx, cancel := context.WithCancel(ctx)
+	defer cancel()
+
+	ticker := time.NewTicker(pollInterval)
+	defer ticker.Stop()
+
+	cherr := make(chan error, 1)
+	waitDone := make(chan struct{})
+	go func() {
+		defer close(waitDone)
+		cherr <- waitForService(waitCtx)
+	}()
+
+	var err error
+loop:
+	for {
+		select {
+		case <-ctx.Done():
+			err = ctx.Err()
+			break loop
+		case err = <-cherr:
+			break loop
+		case <-ticker.C:
+			if isPrivateRegistry(waitCtx) {
+				err = errPrivateRegistry
+				break loop
+			}
+		}
+	}
+	// Cancel before joining. Otherwise a WaitForService that blocks on
+	// waitCtx never closes waitDone, and this wait deadlocks.
+	cancel()
+	<-waitDone
+	return err
+}

pkg/knative/deployer_test.go

@@ -13,6 +13,7 @@ import (
 	"strings"
 	"sync/atomic"
 	"testing"
+	"time"
 
 	"github.com/docker/cli/cli/config/configfile"
 	"github.com/docker/cli/cli/config/types"
@@ -529,3 +530,119 @@ func assertAuth(uname, pwd string, w http.ResponseWriter, r *http.Request) bool
 	_, _ = fmt.Fprintln(w, "Unauthorised.")
 	return false
 }
+
+func TestWaitForReadyOrPrivateRegistry_Success(t *testing.T) {
+	waitForService := func(context.Context) error {
+		return nil // returns immediately with success
+	}
+	isPrivateRegistry := func(context.Context) bool {
+		return false
+	}
+	err := waitForReadyOrPrivateRegistry(t.Context(), waitForService, isPrivateRegistry, 10*time.Millisecond)
+	if err != nil {
+		t.Fatalf("expected nil error, got %v", err)
+	}
+}
+
+func TestWaitForReadyOrPrivateRegistry_WaitError(t *testing.T) {
+	expectedErr := errors.New("timeout waiting for service")
+	waitForService := func(context.Context) error {
+		return expectedErr // returns immediately with error
+	}
+	isPrivateRegistry := func(context.Context) bool {
+		return false
+	}
+	err := waitForReadyOrPrivateRegistry(t.Context(), waitForService, isPrivateRegistry, 10*time.Millisecond)
+	if !errors.Is(err, expectedErr) {
+		t.Fatalf("expected error %v, got %v", expectedErr, err)
+	}
+}
+
+func TestWaitForReadyOrPrivateRegistry_PrivateRegistry(t *testing.T) {
+	waitCanceled := make(chan struct{})
+	waitForService := func(ctx context.Context) error {
+		<-ctx.Done()
+		close(waitCanceled)
+		return ctx.Err()
+	}
+	isPrivate := false
+	isPrivateRegistry := func(context.Context) bool {
+		isPrivate = true
+		return true // instantly return true
+	}
+	err := waitForReadyOrPrivateRegistry(t.Context(), waitForService, isPrivateRegistry, 10*time.Millisecond)
+	if !errors.Is(err, errPrivateRegistry) {
+		t.Fatalf("expected errPrivateRegistry, got %v", err)
+	}
+	if !isPrivate {
+		t.Fatalf("expected isPrivateRegistry to be called")
+	}
+	select {
+	case <-waitCanceled:
+	case <-time.After(time.Second):
+		t.Fatal("expected waitForService context to be canceled")
+	}
+}
+
+func TestWaitForReadyOrPrivateRegistry_DelayedSuccess(t *testing.T) {
+	waitForService := func(context.Context) error {
+		time.Sleep(50 * time.Millisecond)
+		return nil
+	}
+	isPrivateRegistry := func(context.Context) bool {
+		return false // never private
+	}
+	err := waitForReadyOrPrivateRegistry(t.Context(), waitForService, isPrivateRegistry, 10*time.Millisecond)
+	if err != nil {
+		t.Fatalf("expected nil error, got %v", err)
+	}
+}
+
+// The waiter must have left waitForService before we return. Cancel alone is
+// not enough: a goroutine can still be unwinding after Deploy has returned.
+func TestWaitForReadyOrPrivateRegistry_PrivateRegistryJoinsWaiter(t *testing.T) {
+	var exited atomic.Bool
+	waitForService := func(ctx context.Context) error {
+		<-ctx.Done()
+		time.Sleep(150 * time.Millisecond)
+		exited.Store(true)
+		return ctx.Err()
+	}
+	isPrivateRegistry := func(context.Context) bool {
+		return true
+	}
+	err := waitForReadyOrPrivateRegistry(t.Context(), waitForService, isPrivateRegistry, time.Millisecond)
+	if !errors.Is(err, errPrivateRegistry) {
+		t.Fatalf("expected errPrivateRegistry, got %v", err)
+	}
+	if !exited.Load() {
+		t.Fatal("waitForService still running after return")
+	}
+}
+
+func TestWaitForReadyOrPrivateRegistry_CallerCancelJoinsWaiter(t *testing.T) {
+	ctx, cancel := context.WithCancel(t.Context())
+	started := make(chan struct{})
+	var exited atomic.Bool
+	waitForService := func(ctx context.Context) error {
+		close(started)
+		<-ctx.Done()
+		time.Sleep(150 * time.Millisecond)
+		exited.Store(true)
+		return ctx.Err()
+	}
+	isPrivateRegistry := func(context.Context) bool {
+		return false
+	}
+	go func() {
+		<-started
+		cancel()
+	}()
+	err := waitForReadyOrPrivateRegistry(ctx, waitForService, isPrivateRegistry, time.Hour)
+	if !errors.Is(err, context.Canceled) {
+		t.Fatalf("expected context.Canceled, got %v", err)
+	}
+	if !exited.Load() {
+		t.Fatal("waitForService still running after caller cancel")
+	}
+}