diff --git a/pkg/docker/docker.go b/pkg/docker/docker.go index 7ae900c3..8d54d179 100644 --- a/pkg/docker/docker.go +++ b/pkg/docker/docker.go @@ -626,8 +626,36 @@ 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, @@ -635,12 +663,18 @@ func (m *manager) WaitForContainerExit( 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() } }() diff --git a/pkg/docker/wait_exit_test.go b/pkg/docker/wait_exit_test.go new file mode 100644 index 00000000..9b6b87a8 --- /dev/null +++ b/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) +} diff --git a/pkg/podman/connwithctx_test.go b/pkg/podman/connwithctx_test.go new file mode 100644 index 00000000..c425f17a --- /dev/null +++ b/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") + } +} diff --git a/pkg/podman/podman.go b/pkg/podman/podman.go index 3aa00f46..7aa96f52 100644 --- a/pkg/podman/podman.go +++ b/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() diff --git a/pkg/runner/lifecycle.go b/pkg/runner/lifecycle.go index e3613f48..1fc9f49f 100644 --- a/pkg/runner/lifecycle.go +++ b/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") }