From 78ea2748a133775f19e82200c41eba74fb2e9ab0 Mon Sep 17 00:00:00 2001 From: Chris Evans Date: Thu, 2 Jul 2026 19:11:36 -0500 Subject: [PATCH] stub: bound Configure() wait in Start() and break connClosed() deadlock Start() holds stub.Lock() for its entire body and blocks on <-cfgErrC waiting for the runtime's Configure() callback. If the runtime accepts RegisterPlugin but never sends Configure() (observed when a plugin and containerd race during node boot), Start() blocks forever holding the lock. When the connection later closes, the ttrpc OnClose handler connClosed() tries to take the same lock and deadlocks, so the plugin never observes the disconnect and cannot recover. Fix by: - replacing the naked <-cfgErrC receive with a select that also honours ctx cancellation and a registrationTimeout-bounded timer; - having connClosed() do a non-blocking send of a ttrpc.ErrClosed-wrapped error on cfgErrC before taking the lock, so a blocked Start() is woken and releases the lock; - making Configure()'s deferred cfgErrC send non-blocking so it cannot wedge if Start() has already given up or connClosed() has already filled the buffer. To keep this race-clean under -race: - snapshot stub.registrationTimeout into a local before any goroutines are running, since Configure() rewrites that field without the lock; - create srvErrC/cfgErrC before ttrpc.NewClient spawns the goroutine whose OnClose reads cfgErrC without the lock (this is the race that #238's test surfaced). Add two adaptation-suite regression specs: one that lets Register succeed and blocks the plugin's Configure handler so cfgErrC is never written, then has the runtime time out and close the connection (deterministic deadlock repro); and one that has the runtime drop the connection during registration (surfaces the cfgErrC field race under -race). Supersedes #238. Signed-off-by: Chris Evans --- pkg/adaptation/adaptation_suite_test.go | 96 +++++++++++++++++++++++++ pkg/adaptation/suite_test.go | 6 +- pkg/stub/stub.go | 46 ++++++++++-- 3 files changed, 141 insertions(+), 7 deletions(-) diff --git a/pkg/adaptation/adaptation_suite_test.go b/pkg/adaptation/adaptation_suite_test.go index b8ebde7c..b1403c34 100644 --- a/pkg/adaptation/adaptation_suite_test.go +++ b/pkg/adaptation/adaptation_suite_test.go @@ -39,6 +39,7 @@ import ( nri "github.com/containerd/nri/pkg/adaptation" "github.com/containerd/nri/pkg/api" "github.com/containerd/nri/pkg/plugin" + "github.com/containerd/nri/pkg/stub" validator "github.com/containerd/nri/plugins/default-validator/builtin" ) @@ -95,6 +96,101 @@ var _ = Describe("Configuration", func() { Expect(plugin.Start(s.dir)).ToNot(Succeed()) }) }) + + When("the connection is lost while Start() is waiting for Configure()", func() { + var ( + inConfigure chan struct{} + releaseConfigure chan struct{} + ) + + BeforeEach(func() { + inConfigure = make(chan struct{}) + releaseConfigure = make(chan struct{}) + // Short request timeout so the runtime gives up on the blocked + // Configure RPC quickly and closes the connection. + nri.SetPluginRequestTimeout(200 * time.Millisecond) + s.Prepare( + &mockRuntime{}, + &mockPlugin{ + idx: "00", + name: "test", + configure: func(_ *mockPlugin, _ context.Context, _, _, _ string) (stub.EventMask, error) { + close(inConfigure) + <-releaseConfigure + return 0, fmt.Errorf("test: configure released") + }, + }, + ) + }) + + AfterEach(func() { + close(releaseConfigure) + nri.SetPluginRequestTimeout(nri.DefaultPluginRequestTimeout) + }) + + It("should cause plugin Start() to fail instead of deadlocking", func() { + var ( + runtime = s.runtime + plugin = s.plugins[0] + errCh = make(chan error, 1) + ) + + Expect(runtime.Start(s.dir)).To(Succeed()) + + go func() { + errCh <- plugin.Start(s.dir) + }() + + // Once the plugin's Configure handler is running we know register() + // has succeeded and stub.Start() is blocked on cfgErrC holding the + // stub lock. + Eventually(inConfigure, 2*time.Second).Should(BeClosed()) + + // The runtime's Configure RPC now times out and closes the + // connection, firing the plugin's ttrpc OnClose -> connClosed(). + select { + case <-time.After(3 * time.Second): + Fail("plugin Start() did not return: stub deadlocked waiting for Configure()") + case err := <-errCh: + Expect(err).To(HaveOccurred()) + } + }) + }) + + When("the connection is lost during plugin registration", func() { + BeforeEach(func() { + // Make the runtime give up on registration essentially at accept + // time so the connection is torn down while the plugin's Start() + // is still setting up. + nri.SetPluginRegistrationTimeout(1 * time.Nanosecond) + s.Prepare(&mockRuntime{}, &mockPlugin{idx: "00", name: "test"}) + }) + + AfterEach(func() { + nri.SetPluginRegistrationTimeout(nri.DefaultPluginRegistrationTimeout) + }) + + It("should cause plugin Start() to fail instead of hanging", func() { + var ( + runtime = s.runtime + plugin = s.plugins[0] + errCh = make(chan error, 1) + ) + + Expect(runtime.Start(s.dir)).To(Succeed()) + + go func() { + errCh <- plugin.Start(s.dir) + }() + + select { + case <-time.After(3 * time.Second): + Fail("plugin Start() did not return within 3s") + case err := <-errCh: + Expect(err).To(HaveOccurred()) + } + }) + }) }) var _ = Describe("Adaptation", func() { diff --git a/pkg/adaptation/suite_test.go b/pkg/adaptation/suite_test.go index 36c0830e..736cc0f8 100644 --- a/pkg/adaptation/suite_test.go +++ b/pkg/adaptation/suite_test.go @@ -385,6 +385,7 @@ type mockPlugin struct { pods map[string]*api.PodSandbox ctrs map[string]*api.Container + configure func(*mockPlugin, context.Context, string, string, string) (stub.EventMask, error) runPodSandbox func(*mockPlugin, *api.PodSandbox, *api.Container) error updatePodSandbox func(*mockPlugin, *api.PodSandbox, *api.LinuxResources, *api.LinuxResources) error postUpdatePodSandbox func(*mockPlugin, *api.PodSandbox, *api.Container) error @@ -575,7 +576,10 @@ func (m *mockPlugin) onClose() { } } -func (m *mockPlugin) Configure(_ context.Context, _, runtime, version string) (stub.EventMask, error) { +func (m *mockPlugin) Configure(ctx context.Context, cfg, runtime, version string) (stub.EventMask, error) { + if m.configure != nil { + return m.configure(m, ctx, cfg, runtime, version) + } m.q.Add(PluginConfigured) m.runtime = runtime diff --git a/pkg/stub/stub.go b/pkg/stub/stub.go index b0288bd6..62a15358 100644 --- a/pkg/stub/stub.go +++ b/pkg/stub/stub.go @@ -380,6 +380,9 @@ func (stub *stub) Start(ctx context.Context) (retErr error) { return fmt.Errorf("stub already started") } stub.doneC = make(chan struct{}) + // Snapshot before any goroutines are running: Configure() rewrites this + // field without the stub lock once the ttrpc server is up. + cfgTimeout := stub.registrationTimeout err := stub.connect() if err != nil { @@ -423,6 +426,12 @@ func (stub *stub) Start(ctx context.Context) (retErr error) { return fmt.Errorf("failed to multiplex ttrpc client connection: %w", err) } + // Create the error channels before ttrpc.NewClient spawns the goroutine + // whose OnClose callback (connClosed) may send on cfgErrC without holding + // stub.Lock(). + stub.srvErrC = make(chan error, 1) + stub.cfgErrC = make(chan error, 1) + clientOpts := []ttrpc.ClientOpts{ ttrpc.WithOnClose(func() { stub.connClosed() @@ -436,9 +445,6 @@ func (stub *stub) Start(ctx context.Context) (retErr error) { } }() - stub.srvErrC = make(chan error, 1) - stub.cfgErrC = make(chan error, 1) - go func(l stdnet.Listener, doneC chan struct{}, srvErrC chan error) { srvErrC <- rpcs.Serve(ctx, l) close(doneC) @@ -456,8 +462,22 @@ func (stub *stub) Start(ctx context.Context) (retErr error) { return err } - if err = <-stub.cfgErrC; err != nil { - return err + // The runtime should Configure() us immediately after a successful register(). + // Bound the wait so Start() cannot block forever holding stub.Lock() if the + // runtime accepts registration but never issues Configure(), and let context + // cancellation and connection loss (via connClosed -> cfgErrC) break the wait. + cfgTimer := time.NewTimer(cfgTimeout) + defer cfgTimer.Stop() + + select { + case err = <-stub.cfgErrC: + if err != nil { + return err + } + case <-cfgTimer.C: + return fmt.Errorf("timed out waiting for Configure() from runtime after %s", cfgTimeout) + case <-ctx.Done(): + return fmt.Errorf("context cancelled while waiting for Configure() from runtime: %w", ctx.Err()) } stub.logger.Infof(ctx, "Started plugin %s...", stub.Name()) @@ -624,6 +644,14 @@ func (stub *stub) register(ctx context.Context) error { // Handle a lost connection. func (stub *stub) connClosed() { + // Start() may be blocked on cfgErrC while holding stub.Lock(). Signal it + // first (non-blocking; cfgErrC has cap 1) so it can return and release the + // lock before we take it below. + select { + case stub.cfgErrC <- fmt.Errorf("connection closed before Configure(): %w", ttrpc.ErrClosed): + default: + } + stub.Lock() stub.close() stub.Unlock() @@ -680,7 +708,13 @@ func (stub *stub) Configure(ctx context.Context, req *api.ConfigureRequest) (rpl stub.runtimeNRIVersion = req.NRIVersion defer func() { - stub.cfgErrC <- retErr + // Non-blocking: Start() may have already given up (timeout / ctx / conn + // loss) and is no longer receiving; the buffered slot may also already + // hold connClosed()'s error. + select { + case stub.cfgErrC <- retErr: + default: + } }() if handler := stub.handlers.Configure; handler == nil {