Merge pull request #84 from authzed/cancelfix

fix: store cancel func before launching controller goroutine

Author: ecordell

bootstrap/crds.go

@@ -26,6 +26,7 @@ const (
 )
 
 // CRD installs the CRDs in the filesystem into the kube cluster configured by the rest config.
+//
 // Deprecated: Use CRDs instead.
 func CRD(restConfig *rest.Config, crdFS fs.ReadDirFS, dir string) error {
 	return CRDs(context.Background(), restConfig, crdFS, dir)

manager/manager.go

@@ -150,12 +150,16 @@ func (m *Manager) Go(controllers ...Controller) error {
 	for _, c := range controllers {
 		c := c
 		m.healthzHandler.AddHealthChecker(controllerhealthz.NamedHealthChecker(c.Name(), c.HealthChecker()))
+		// Create the cancel context and store it before launching the goroutine.
+		// This prevents a race condition where Cancel() is called between Go()
+		// returning and the goroutine storing the cancel function, which would
+		// cause the controller to run indefinitely with an uncancellable context.
+		cCtx, cancel := context.WithCancel(ctx) //nolint:gosec // cancel stored in cancelFuncs map for later use
+		m.lock.Lock()
+		m.cancelFuncs[c] = cancel
+		m.lock.Unlock()
 		errG.Go(func() error {
-			ctx, cancel := context.WithCancel(ctx)
-			m.lock.Lock()
-			m.cancelFuncs[c] = cancel
-			m.lock.Unlock()
-			c.Start(ctx, runtime.GOMAXPROCS(0))
+			c.Start(cCtx, runtime.GOMAXPROCS(0))
 			return nil
 		})
 		if ctx.Err() != nil {

manager/manager_test.go

@@ -85,6 +85,44 @@ func testController(t *testing.T, name string) Controller {
 	})
 }
 
+// TestGoThenImmediateCancelStopsController verifies that calling Cancel immediately
+// after Go correctly cancels the controller. There is a potential race condition
+// where if the cancel context is created inside the goroutine launched by Go,
+// a Cancel call between Go() returning and the goroutine executing will miss the
+// cancel function, leaving the controller running indefinitely.
+func TestGoThenImmediateCancelStopsController(t *testing.T) {
+	t.Parallel()
+	ctx := t.Context()
+
+	m := NewManager(&config.DebuggingConfiguration{}, ":"+getFreePort(t), nil, nil)
+	ready := make(chan struct{})
+	go func() { _ = m.Start(ctx, ready) }()
+	<-ready
+
+	ctrl := &blockingController{BasicController: *NewBasicController("blocking"), done: make(chan struct{})}
+	require.NoError(t, m.Go(ctrl))
+	// Cancel immediately, before the goroutine launched by Go has had a chance to run.
+	m.Cancel(ctrl)
+
+	select {
+	case <-ctrl.done:
+		// Controller was cancelled and shut down correctly.
+	case <-time.After(5 * time.Second):
+		t.Fatal("controller did not shut down after Cancel — likely race condition in Manager.Go")
+	}
+}
+
+// blockingController is a Controller whose Start blocks until the context is cancelled.
+type blockingController struct {
+	BasicController
+	done chan struct{}
+}
+
+func (c *blockingController) Start(ctx context.Context, _ int) {
+	<-ctx.Done()
+	close(c.done)
+}
+
 func requireCancelFnCount(t *testing.T, m *Manager, count int) {
 	t.Helper()
 	require.Eventually(t, func() bool {

state/state_test.go

@@ -2276,7 +2276,7 @@ func TestContinueWithCancellation(t *testing.T) {
 }
 
 func ExampleWithErrorHandler() {
-	ctx, cancel := context.WithCancel(context.Background())
+	ctx, cancel := context.WithCancel(context.Background()) //nolint:gosec // cancel called inside pipeline step
 
 	// Set up error handler
 	ctx = WithErrorHandler(ctx, func(_ error) Step {

typed/registry.go

@@ -94,6 +94,7 @@ func (r *Registry) NewFilteredDynamicSharedInformerFactory(key FactoryKey, clien
 }
 
 // ListerFor returns a typed Lister from a Registry
+//
 // Deprecated: Use MustListerForKey instead
 func ListerFor[K runtime.Object](r *Registry, key RegistryKey) *Lister[K] {
 	return MustListerForKey[K](r, key)
@@ -114,6 +115,7 @@ func ListerForKey[K runtime.Object](r *Registry, key RegistryKey) (*Lister[K], e
 }
 
 // IndexerFor returns a typed Indexer from a Registry
+//
 // Deprecated: Use MustIndexerForKey instead
 func IndexerFor[K runtime.Object](r *Registry, key RegistryKey) *Indexer[K] {
 	return MustIndexerForKey[K](r, key)
@@ -154,6 +156,7 @@ func (r *Registry) Remove(key FactoryKey) {
 }
 
 // InformerFactoryFor returns GVR-specific InformerFactory from the Registry.
+//
 // Deprecated: use MustInformerFactoryForKey instead.
 func (r *Registry) InformerFactoryFor(key RegistryKey) informers.GenericInformer {
 	return r.MustInformerFactoryForKey(key)
@@ -182,6 +185,7 @@ func (r *Registry) InformerFactoryForKey(key RegistryKey) (informers.GenericInfo
 }
 
 // ListerFor returns the GVR-specific Lister from the Registry
+//
 // Deprecated: use MustListerForKey instead.
 func (r *Registry) ListerFor(key RegistryKey) cache.GenericLister {
 	return r.MustInformerFactoryForKey(key).Lister()
@@ -204,6 +208,7 @@ func (r *Registry) ListerForKey(key RegistryKey) (cache.GenericLister, error) {
 }
 
 // InformerFor returns the GVR-specific Informer from the Registry
+//
 // Deprecated: use MustInformerForKey instead.
 func (r *Registry) InformerFor(key RegistryKey) cache.SharedIndexInformer {
 	return r.MustInformerFactoryForKey(key).Informer()
@@ -226,6 +231,7 @@ func (r *Registry) InformerForKey(key RegistryKey) (cache.SharedIndexInformer, e
 }
 
 // IndexerFor returns the GVR-specific Indexer from the Registry
+//
 // Deprecated: use MustIndexerForKey instead.
 func (r *Registry) IndexerFor(key RegistryKey) cache.Indexer {
 	return r.MustInformerForKey(key).GetIndexer()