fix(runner,docker,podman): guard containerDied read and release container-wait goroutines on stop

pkg/docker/docker.go

@@ -626,21 +626,55 @@ func (m *manager) GetClient() *client.Client {
 	return m.client
 }
 
+// ctxWithDone derives a context that is canceled either when ctx is
+// canceled or when the manager is stopping (m.done closes). The watcher
+// goroutine is tracked on m.wg so Stop() waits for it.
+func (m *manager) ctxWithDone(ctx context.Context) (context.Context, context.CancelFunc) {
+	derived, cancel := context.WithCancel(ctx)
+
+	m.wg.Add(1)
+
+	go func() {
+		defer m.wg.Done()
+
+		select {
+		case <-ctx.Done():
+			cancel()
+		case <-m.done:
+			cancel()
+		case <-derived.Done():
+		}
+	}()
+
+	return derived, cancel
+}
+
 // WaitForContainerExit returns channels that signal when a container exits.
 // The statusCh receives exit info (code + OOM status), errCh receives any wait errors.
+//
+// The producer goroutine is tracked on m.wg and its ContainerWait call runs
+// against a context derived via ctxWithDone, so Stop() actually cancels
+// the in-flight call and waits for the goroutine to finish instead of
+// returning while it - and its underlying connection - are still running.
 func (m *manager) WaitForContainerExit(
 	ctx context.Context,
 	containerID string,
 ) (<-chan ContainerExitInfo, <-chan error) {
 	statusCh := make(chan ContainerExitInfo, 1)
 	errCh := make(chan error, 1)
 
+	waitCtx, cancel := m.ctxWithDone(ctx)
+
+	m.wg.Add(1)
+
 	go func() {
+		defer m.wg.Done()
 		defer close(statusCh)
 		defer close(errCh)
+		defer cancel()
 
 		waitStatusCh, waitErrCh := m.client.ContainerWait(
-			ctx, containerID, container.WaitConditionNotRunning,
+			waitCtx, containerID, container.WaitConditionNotRunning,
 		)
 
 		select {
@@ -674,8 +708,8 @@ func (m *manager) WaitForContainerExit(
 			statusCh <- info
 		case err := <-waitErrCh:
 			errCh <- err
-		case <-ctx.Done():
-			errCh <- ctx.Err()
+		case <-waitCtx.Done():
+			errCh <- waitCtx.Err()
 		}
 	}()
 

pkg/docker/wait_exit_test.go

@@ -0,0 +1,111 @@
+package docker
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	"github.com/sirupsen/logrus"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+// TestCtxWithDone_CancelsOnManagerDone is a regression test for the
+// WaitForContainerExit goroutine leak: the manager stopping (m.done
+// closing) must cancel any in-flight call derived via ctxWithDone, not
+// just leave it running until the caller's own ctx eventually cancels.
+func TestCtxWithDone_CancelsOnManagerDone(t *testing.T) {
+	m := &manager{done: make(chan struct{})}
+
+	derived, cancel := m.ctxWithDone(context.Background())
+	defer cancel()
+
+	select {
+	case <-derived.Done():
+		t.Fatal("derived context should not be done before m.done closes")
+	default:
+	}
+
+	close(m.done)
+
+	select {
+	case <-derived.Done():
+	case <-time.After(2 * time.Second):
+		t.Fatal("derived context was not canceled after m.done closed")
+	}
+
+	assert.ErrorIs(t, derived.Err(), context.Canceled)
+}
+
+func TestCtxWithDone_CancelsOnParentContext(t *testing.T) {
+	m := &manager{done: make(chan struct{})}
+	parentCtx, parentCancel := context.WithCancel(context.Background())
+
+	derived, cancel := m.ctxWithDone(parentCtx)
+	defer cancel()
+
+	parentCancel()
+
+	select {
+	case <-derived.Done():
+	case <-time.After(2 * time.Second):
+		t.Fatal("derived context was not canceled after the parent context was canceled")
+	}
+}
+
+func TestCtxWithDone_ExplicitCancelDoesNotLeakWatcher(t *testing.T) {
+	m := &manager{done: make(chan struct{})}
+
+	_, cancel := m.ctxWithDone(context.Background())
+	cancel()
+
+	waited := make(chan struct{})
+
+	go func() {
+		m.wg.Wait()
+		close(waited)
+	}()
+
+	select {
+	case <-waited:
+	case <-time.After(2 * time.Second):
+		t.Fatal("watcher goroutine did not exit after its own cancel func was called")
+	}
+}
+
+// TestManagerWgWait_ReturnsAfterDoneClosed is a regression test for the
+// core NM-08 claim: previously, closing m.done (what Stop() does first)
+// did not cause any tracked goroutine to release, so m.wg.Wait() could
+// block past what a caller would reasonably expect from Stop().
+func TestManagerWgWait_ReturnsAfterDoneClosed(t *testing.T) {
+	m := &manager{done: make(chan struct{})}
+
+	_, cancel := m.ctxWithDone(context.Background())
+	defer cancel()
+
+	stopped := make(chan struct{})
+
+	go func() {
+		close(m.done)
+		m.wg.Wait()
+		close(stopped)
+	}()
+
+	select {
+	case <-stopped:
+	case <-time.After(2 * time.Second):
+		t.Fatal("wg.Wait() did not return promptly after m.done closed")
+	}
+}
+
+func TestNewManager_WgAndDoneAreWired(t *testing.T) {
+	// Sanity check that the fields ctxWithDone depends on are actually
+	// initialized by the constructor used in production.
+	mgr, err := NewManager(logrus.New())
+	require.NoError(t, err)
+
+	m, ok := mgr.(*manager)
+	require.True(t, ok)
+
+	require.NotNil(t, m.done)
+}

pkg/podman/connwithctx_test.go

@@ -0,0 +1,98 @@
+package podman
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+)
+
+// TestConnWithCtx_CancelsOnManagerDone is a regression test for the
+// goroutine leak in every Podman call site that uses connWithCtx
+// (including WaitForContainerExit, the long-lived one): the manager
+// stopping (m.done closing) must cancel the derived connection context,
+// not just leave any in-flight call running until the caller's own ctx
+// eventually cancels.
+func TestConnWithCtx_CancelsOnManagerDone(t *testing.T) {
+	m := &manager{conn: context.Background(), done: make(chan struct{})}
+
+	derived, cancel := m.connWithCtx(context.Background())
+	defer cancel()
+
+	select {
+	case <-derived.Done():
+		t.Fatal("derived context should not be done before m.done closes")
+	default:
+	}
+
+	close(m.done)
+
+	select {
+	case <-derived.Done():
+	case <-time.After(2 * time.Second):
+		t.Fatal("derived context was not canceled after m.done closed")
+	}
+
+	assert.ErrorIs(t, derived.Err(), context.Canceled)
+}
+
+func TestConnWithCtx_CancelsOnParentContext(t *testing.T) {
+	m := &manager{conn: context.Background(), done: make(chan struct{})}
+	parentCtx, parentCancel := context.WithCancel(context.Background())
+
+	derived, cancel := m.connWithCtx(parentCtx)
+	defer cancel()
+
+	parentCancel()
+
+	select {
+	case <-derived.Done():
+	case <-time.After(2 * time.Second):
+		t.Fatal("derived context was not canceled after the parent context was canceled")
+	}
+}
+
+func TestConnWithCtx_ExplicitCancelDoesNotLeakWatcher(t *testing.T) {
+	m := &manager{conn: context.Background(), done: make(chan struct{})}
+
+	_, cancel := m.connWithCtx(context.Background())
+	cancel()
+
+	waited := make(chan struct{})
+
+	go func() {
+		m.wg.Wait()
+		close(waited)
+	}()
+
+	select {
+	case <-waited:
+	case <-time.After(2 * time.Second):
+		t.Fatal("watcher goroutine did not exit after its own cancel func was called")
+	}
+}
+
+// TestManagerWgWait_ReturnsAfterDoneClosed mirrors the docker package's
+// equivalent test: closing m.done (what Stop() does first) must cause
+// every connWithCtx watcher to release so wg.Wait() returns promptly.
+func TestManagerWgWait_ReturnsAfterDoneClosed(t *testing.T) {
+	m := &manager{conn: context.Background(), done: make(chan struct{})}
+
+	_, cancel := m.connWithCtx(context.Background())
+	defer cancel()
+
+	stopped := make(chan struct{})
+
+	go func() {
+		close(m.done)
+		m.wg.Wait()
+		close(stopped)
+	}()
+
+	select {
+	case <-stopped:
+	case <-time.After(2 * time.Second):
+		t.Fatal("wg.Wait() did not return promptly after m.done closed")
+	}
+}

pkg/podman/podman.go

@@ -53,10 +53,19 @@ type manager struct {
 func (m *manager) connWithCtx(ctx context.Context) (context.Context, context.CancelFunc) {
 	derived, cancel := context.WithCancel(m.conn)
 
+	m.wg.Add(1)
+
 	go func() {
+		defer m.wg.Done()
+
 		select {
 		case <-ctx.Done():
 			cancel()
+		case <-m.done:
+			// Manager is stopping - release any in-flight call derived
+			// from this context instead of leaving it running past
+			// Stop() returning.
+			cancel()
 		case <-derived.Done():
 		}
 	}()
@@ -703,7 +712,10 @@ func (m *manager) WaitForContainerExit(
 
 	waitConn, cancel := m.connWithCtx(ctx)
 
+	m.wg.Add(1)
+
 	go func() {
+		defer m.wg.Done()
 		defer close(statusCh)
 		defer close(errCh)
 		defer cancel()

pkg/runner/lifecycle.go

@@ -1353,8 +1353,14 @@ func (r *runner) runContainerLifecycle(
 	}
 
 	// Return an error if the container died so callers (e.g. multi-genesis
-	// loop) stop instead of continuing with the next group.
-	if containerDied {
+	// loop) stop instead of continuing with the next group. Read under mu:
+	// the death-monitor goroutine (still running until execCancel/r.done
+	// fires) writes containerDied under the same lock.
+	mu.Lock()
+	died := containerDied
+	mu.Unlock()
+
+	if died {
 		return fmt.Errorf("container died during execution")
 	}