🐛 Fix goroutine leaks in cache Start() methods

pkg/cache/defaulting_test.go

@@ -17,14 +17,18 @@ limitations under the License.
 package cache
 
 import (
+	"context"
+	"errors"
 	"reflect"
 	"sync"
 	"testing"
+	"testing/synctest"
 	"time"
 
 	"github.com/google/go-cmp/cmp"
 	"github.com/google/go-cmp/cmp/cmpopts"
 	fuzz "github.com/google/gofuzz"
+	. "github.com/onsi/gomega"
 	corev1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/api/meta"
 	"k8s.io/apimachinery/pkg/fields"
@@ -517,3 +521,126 @@ func TestDefaultConfigConsidersAllFields(t *testing.T) {
 		}
 	}
 }
+
+// mockCache is a mock implementation of Cache for testing Start() behavior.
+type mockCache struct {
+	startFunc func(ctx context.Context) error
+	Cache
+}
+
+func (m *mockCache) Start(ctx context.Context) error {
+	if m.startFunc != nil {
+		return m.startFunc(ctx)
+	}
+	<-ctx.Done()
+	return nil
+}
+
+func TestMultiNamespaceCacheStart_MultipleErrors(t *testing.T) {
+	t.Parallel()
+	synctest.Test(t, func(t *testing.T) {
+		g := NewWithT(t)
+
+		errTest := errors.New("test error")
+		namespaceToCache := map[string]Cache{
+			"ns1": &mockCache{startFunc: func(ctx context.Context) error { return errTest }},
+			"ns2": &mockCache{startFunc: func(ctx context.Context) error { return errTest }},
+			"ns3": &mockCache{startFunc: func(ctx context.Context) error { return errTest }},
+		}
+
+		c := &multiNamespaceCache{
+			namespaceToCache: namespaceToCache,
+		}
+
+		ctx, cancel := context.WithTimeout(t.Context(), 100*time.Millisecond)
+		defer cancel()
+
+		err := c.Start(ctx)
+		g.Expect(err).To(HaveOccurred())
+		g.Expect(err).To(MatchError(errTest))
+	})
+}
+
+func TestMultiNamespaceCacheStart_ContextCancel(t *testing.T) {
+	t.Parallel()
+	synctest.Test(t, func(t *testing.T) {
+		g := NewWithT(t)
+
+		normalCache := &mockCache{
+			startFunc: func(ctx context.Context) error {
+				<-ctx.Done()
+				return nil
+			},
+		}
+
+		namespaceToCache := map[string]Cache{
+			"ns1": normalCache,
+			"ns2": normalCache,
+		}
+
+		c := &multiNamespaceCache{
+			namespaceToCache: namespaceToCache,
+		}
+
+		ctx, cancel := context.WithCancel(t.Context())
+		cancel() // Cancel immediately
+
+		err := c.Start(ctx)
+		g.Expect(err).NotTo(HaveOccurred())
+	})
+}
+
+func TestDelegatingByGVKCacheStart_MultipleErrors(t *testing.T) {
+	t.Parallel()
+	synctest.Test(t, func(t *testing.T) {
+		g := NewWithT(t)
+
+		errTest := errors.New("test error")
+		caches := map[schema.GroupVersionKind]Cache{
+			{Group: "apps", Version: "v1", Kind: "Deployment"}:  &mockCache{startFunc: func(ctx context.Context) error { return errTest }},
+			{Group: "apps", Version: "v1", Kind: "StatefulSet"}: &mockCache{startFunc: func(ctx context.Context) error { return errTest }},
+			{Group: "apps", Version: "v1", Kind: "DaemonSet"}:   &mockCache{startFunc: func(ctx context.Context) error { return errTest }},
+		}
+
+		c := &delegatingByGVKCache{
+			caches:       caches,
+			defaultCache: &mockCache{startFunc: func(ctx context.Context) error { return errTest }},
+		}
+
+		ctx, cancel := context.WithTimeout(t.Context(), 100*time.Millisecond)
+		defer cancel()
+
+		err := c.Start(ctx)
+		g.Expect(err).To(HaveOccurred())
+		g.Expect(err).To(MatchError(errTest))
+	})
+}
+
+func TestDelegatingByGVKCacheStart_ContextCancel(t *testing.T) {
+	t.Parallel()
+	synctest.Test(t, func(t *testing.T) {
+		g := NewWithT(t)
+
+		normalCache := &mockCache{
+			startFunc: func(ctx context.Context) error {
+				<-ctx.Done()
+				return nil
+			},
+		}
+
+		caches := map[schema.GroupVersionKind]Cache{
+			{Group: "apps", Version: "v1", Kind: "Deployment"}: normalCache,
+		}
+
+		c := &delegatingByGVKCache{
+			caches:       caches,
+			defaultCache: normalCache,
+		}
+
+		ctx, cancel := context.WithCancel(t.Context())
+		cancel() // Cancel immediately
+
+		err := c.Start(ctx)
+		g.Expect(err).NotTo(HaveOccurred())
+	})
+}

pkg/cache/delegating_by_gvk_cache.go

@@ -21,8 +21,8 @@ import (
 	"maps"
 	"slices"
 	"strings"
-	"sync"
 
+	"golang.org/x/sync/errgroup"
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/runtime/schema"
 	"sigs.k8s.io/controller-runtime/pkg/client"
@@ -77,24 +77,15 @@ func (dbt *delegatingByGVKCache) Start(ctx context.Context) error {
 	allCaches := slices.Collect(maps.Values(dbt.caches))
 	allCaches = append(allCaches, dbt.defaultCache)
 
-	wg := &sync.WaitGroup{}
-	errs := make(chan error)
+	group, childCtx := errgroup.WithContext(ctx)
 	for idx := range allCaches {
 		cache := allCaches[idx]
-		wg.Go(func() {
-			if err := cache.Start(ctx); err != nil {
-				errs <- err
-			}
+		group.Go(func() error {
+			return cache.Start(childCtx)
 		})
 	}
 
-	select {
-	case err := <-errs:
-		return err
-	case <-ctx.Done():
-		wg.Wait()
-		return nil
-	}
+	return ignoreContextCanceled(group.Wait())
 }
 
 func (dbt *delegatingByGVKCache) WaitForCacheSync(ctx context.Context) bool {

pkg/cache/multi_namespace_cache.go

@@ -18,10 +18,12 @@ package cache
 
 import (
 	"context"
+	"errors"
 	"fmt"
 	"strings"
 	"time"
 
+	"golang.org/x/sync/errgroup"
 	corev1 "k8s.io/api/core/v1"
 	apimeta "k8s.io/apimachinery/pkg/api/meta"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
@@ -160,31 +162,34 @@ func (c *multiNamespaceCache) GetInformerForKind(ctx context.Context, gvk schema
 }
 
 func (c *multiNamespaceCache) Start(ctx context.Context) error {
-	errs := make(chan error)
-	// start global cache
+	group, childCtx := errgroup.WithContext(ctx)
+
 	if c.clusterCache != nil {
-		go func() {
-			err := c.clusterCache.Start(ctx)
-			if err != nil {
-				errs <- fmt.Errorf("failed to start cluster-scoped cache: %w", err)
-			}
-		}()
+		clusterCache := c.clusterCache
+		group.Go(func() error {
+			return clusterCache.Start(childCtx)
+		})
 	}
 
-	// start namespaced caches
 	for ns, cache := range c.namespaceToCache {
-		go func(ns string, cache Cache) {
-			if err := cache.Start(ctx); err != nil {
-				errs <- fmt.Errorf("failed to start cache for namespace %s: %w", ns, err)
+		group.Go(func() error {
+			if err := cache.Start(childCtx); err != nil {
+				return fmt.Errorf("failed to start cache for namespace %s: %w", ns, err)
 			}
-		}(ns, cache)
+			return nil
+		})
 	}
-	select {
-	case <-ctx.Done():
+
+	return ignoreContextCanceled(group.Wait())
+}
+
+// ignoreContextCanceled returns nil if the error is a context.Canceled error,
+// otherwise returns the error unchanged.
+func ignoreContextCanceled(err error) error {
+	if errors.Is(err, context.Canceled) {
 		return nil
-	case err := <-errs:
-		return err
 	}
+	return err
 }
 
 func (c *multiNamespaceCache) WaitForCacheSync(ctx context.Context) bool {