stub: bound Configure() wait in Start() and break connClosed() deadlock

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() {

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

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 {