update nri to latest

Signed-off-by: Michael Zappa <michael.zappa@gmail.com>

docs/NRI.md

@@ -44,6 +44,15 @@ The containerd CRI plugin registers itself as an above mentioned NRI
 Domain for the "k8s.io" namespace, to allow container configuration to be customized by external
 NRI plugins.
 
+CRI pod and container teardown continues when an NRI stop or removal
+notification fails; the failure is logged and the notification is not retried.
+Since NRI dispatches an event to plugins in order and stops dispatching when a
+plugin's `fail_request` policy makes the request fail, a timed-out plugin can
+prevent later plugins from receiving that teardown event even though the CRI
+teardown succeeds. Plugins that require teardown notifications should account
+for this behavior and should avoid `fail_request` on teardown events unless
+failing the NRI request is more important than notifying subsequent plugins.
+
 ### NRI Support for Other Container 'Domains'
 
 The main reason for this split of functionality is to allow

go.mod

@@ -23,7 +23,7 @@ require (
 	github.com/containerd/imgcrypt/v2 v2.0.3
 	github.com/containerd/log v0.2.0
 	github.com/containerd/log/otel v0.1.0
-	github.com/containerd/nri v0.12.3
+	github.com/containerd/nri v0.12.5
 	github.com/containerd/otelttrpc v0.1.0
 	github.com/containerd/platforms v1.0.0-rc.5
 	github.com/containerd/plugin v1.1.0

go.sum

@@ -60,8 +60,8 @@ github.com/containerd/log v0.2.0 h1:BewD/umNgVnoczglOpX8eRMyEy5t5iPlu5AIpnWDONc=
 github.com/containerd/log v0.2.0/go.mod h1:/M7L7CXKcPTfNC74XzaK+5H5KbO5+4lJVpuVI6vRLoM=
 github.com/containerd/log/otel v0.1.0 h1:Az5rFFo0+c4v2yC8ROR63sWsZpKl/vhtUnUhqLSpE6U=
 github.com/containerd/log/otel v0.1.0/go.mod h1:65C5iYF2xIQByCyxzoO2KKKK1+/tGxD4XqhdlrTtvzE=
-github.com/containerd/nri v0.12.3 h1:pVZwZCe0Z+d2WmScp51VTO3OmeuOZFoDGInX2n04e70=
-github.com/containerd/nri v0.12.3/go.mod h1:MrNpHL2YtNnf09t+D6NacWeN1kaHmy/YCm0+w4VHKL0=
+github.com/containerd/nri v0.12.5 h1:sPYUCT4fRbshF0ewiIaFXv2n6+tLIwY0kKII9ycMszY=
+github.com/containerd/nri v0.12.5/go.mod h1:KPI0bbRZeqYq4rRWNWQztAR1150JF937MGe5b45Zgrk=
 github.com/containerd/otelttrpc v0.1.0 h1:UOX68eVTE8H/T45JveIg+I22Ev2aFj4qPITCmXsskjw=
 github.com/containerd/otelttrpc v0.1.0/go.mod h1:XhoA2VvaGPW1clB2ULwrBZfXVuEWuyOd2NUD1IM0yTg=
 github.com/containerd/platforms v1.0.0-rc.5 h1:vXd569rDrz8LeMXzAnBsy6LADV5YtsD8oyaRarxdmSU=

integration/nri_test.go

@@ -98,6 +98,258 @@ func TestNriPluginSetup(t *testing.T) {
 	tc.setup()
 }
 
+// Test that NRI request timeouts follow the plugin's failure policy.
+func TestNriPluginFailurePolicy(t *testing.T) {
+	skipNriTestIfNecessary(t)
+
+	for _, tc := range []struct {
+		name         string
+		policy       *api.FailurePolicy
+		wantRunError bool
+		wantKept     bool
+	}{
+		{
+			name: "default policy drops plugin and proceeds",
+		},
+		{
+			name:         "fail request and drop plugin",
+			policy:       &api.FailurePolicy{FailRequest: true},
+			wantRunError: true,
+		},
+		{
+			name:     "keep plugin and proceed",
+			policy:   &api.FailurePolicy{KeepConnected: true},
+			wantKept: true,
+		},
+		{
+			name:         "fail request and keep plugin connected",
+			policy:       &api.FailurePolicy{FailRequest: true, KeepConnected: true},
+			wantRunError: true,
+			wantKept:     true,
+		},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			var (
+				timedOut = make(chan struct{})
+				healthy  = make(chan struct{})
+				once     sync.Once
+				plugin   = &mockPlugin{
+					runPodSandbox: func(mp *mockPlugin, pod *api.PodSandbox) error {
+						switch pod.Name {
+						case "timeout-target":
+							once.Do(func() {
+								time.Sleep(mp.stub.RequestTimeout() + 250*time.Millisecond)
+								close(timedOut)
+							})
+						case "healthy-target":
+							close(healthy)
+						}
+						return nil
+					},
+				}
+				nt = &nriTest{t: t, plugins: []*mockPlugin{plugin}}
+			)
+
+			if tc.policy != nil {
+				plugin.opts = append(plugin.opts, stub.WithFailurePolicy(tc.policy))
+			}
+			nt.setup()
+			t.Cleanup(plugin.Stop)
+
+			runPod := func(name string) (string, error) {
+				return nt.runtime.RunPodSandbox(PodSandboxConfig(name, nt.namespace), *runtimeHandler)
+			}
+			cleanupPod := func(id string) {
+				t.Cleanup(func() {
+					assert.NoError(t, nt.runtime.StopPodSandbox(id))
+					assert.NoError(t, nt.runtime.RemovePodSandbox(id))
+				})
+			}
+
+			id, err := runPod("timeout-target")
+			if id != "" {
+				cleanupPod(id)
+			}
+			if tc.wantRunError {
+				require.Error(t, err, "timed-out NRI hook should fail the pod request")
+			} else {
+				require.NoError(t, err, "policy should allow the pod request")
+			}
+
+			select {
+			case <-timedOut:
+			case <-time.After(plugin.stub.RequestTimeout() + pluginSyncTimeout):
+				t.Fatal("timed-out plugin callback did not return")
+			}
+
+			if tc.wantKept {
+				id, err = runPod("healthy-target")
+				require.NoError(t, err, "kept-connected plugin should handle the next pod")
+				cleanupPod(id)
+				select {
+				case <-healthy:
+				case <-time.After(pluginSyncTimeout):
+					t.Fatal("kept-connected plugin did not receive the next request")
+				}
+			} else {
+				select {
+				case <-plugin.closedCh:
+				case <-time.After(pluginSyncTimeout):
+					t.Fatal("policy should disconnect a timed-out plugin")
+				}
+			}
+		})
+	}
+}
+
+// Test that an NRI plugin which disconnects while idle does not fail the next
+// CRI lifecycle request or prevent later plugins from receiving it.
+func TestNriPluginIdleDisconnect(t *testing.T) {
+	skipNriTestIfNecessary(t)
+
+	var (
+		nextPodEvent = make(chan struct{}, 1)
+		failing      = &mockPlugin{
+			name: "idle-disconnect-plugin",
+			opts: []stub.Option{stub.WithFailurePolicy(&api.FailurePolicy{FailRequest: true, KeepConnected: true})},
+		}
+		observer = &mockPlugin{
+			name: "observer-plugin",
+			runPodSandbox: func(_ *mockPlugin, pod *api.PodSandbox) error {
+				if pod.Name == "after-disconnect" {
+					nextPodEvent <- struct{}{}
+				}
+				return nil
+			},
+		}
+		nt = &nriTest{t: t, plugins: []*mockPlugin{failing, observer}}
+	)
+
+	nt.setup()
+	t.Cleanup(failing.Stop)
+	t.Cleanup(observer.Stop)
+
+	nt.runPod("before-disconnect")
+
+	failing.Stop()
+	select {
+	case <-failing.closedCh:
+	case <-time.After(pluginSyncTimeout):
+		t.Fatal("disconnected plugin did not stop")
+	}
+
+	// Allow the runtime-side ttrpc connection watcher to observe the idle
+	// disconnect before sending the next request.
+	time.Sleep(250 * time.Millisecond)
+
+	nt.runPod("after-disconnect")
+	select {
+	case <-nextPodEvent:
+	case <-time.After(pluginSyncTimeout):
+		t.Fatal("later plugin did not receive the request after idle disconnect")
+	}
+}
+
+// Test that fail-request timeouts stop NRI fanout during pod teardown, while
+// CRI continues the teardown operation without retrying the notification.
+func TestNriPluginFailurePolicyPodTeardown(t *testing.T) {
+	skipNriTestIfNecessary(t)
+
+	for _, target := range []string{StopPodSandbox, RemovePodSandbox} {
+		t.Run(target, func(t *testing.T) {
+			var (
+				timedOut = make(chan struct{})
+				notified = make(chan string, 4)
+				once     sync.Once
+				failing  = &mockPlugin{
+					name: "timeout-plugin",
+					opts: []stub.Option{stub.WithFailurePolicy(&api.FailurePolicy{FailRequest: true, KeepConnected: true})},
+				}
+				observer = &mockPlugin{
+					name: "observer-plugin",
+					stopPodSandbox: func(_ *mockPlugin, pod *api.PodSandbox) error {
+						notified <- StopPodSandbox
+						return nil
+					},
+					removePodSandbox: func(_ *mockPlugin, pod *api.PodSandbox) error {
+						notified <- RemovePodSandbox
+						return nil
+					},
+				}
+				nt = &nriTest{t: t, plugins: []*mockPlugin{failing, observer}}
+			)
+
+			block := func(event string) {
+				if event != target {
+					return
+				}
+				once.Do(func() {
+					time.Sleep(failing.stub.RequestTimeout() + 250*time.Millisecond)
+					close(timedOut)
+				})
+			}
+			failing.stopPodSandbox = func(_ *mockPlugin, _ *api.PodSandbox) error {
+				block(StopPodSandbox)
+				return nil
+			}
+			failing.removePodSandbox = func(_ *mockPlugin, _ *api.PodSandbox) error {
+				block(RemovePodSandbox)
+				return nil
+			}
+
+			nt.setup()
+			t.Cleanup(failing.Stop)
+			t.Cleanup(observer.Stop)
+
+			id, err := nt.runtime.RunPodSandbox(PodSandboxConfig("pod0", nt.namespace), *runtimeHandler)
+			require.NoError(t, err, "run pod sandbox")
+			removed := false
+			t.Cleanup(func() {
+				if !removed {
+					assert.NoError(t, nt.runtime.StopPodSandbox(id))
+					assert.NoError(t, nt.runtime.RemovePodSandbox(id))
+				}
+			})
+
+			if target == RemovePodSandbox {
+				require.NoError(t, nt.runtime.StopPodSandbox(id), "stop pod sandbox before removal")
+				select {
+				case event := <-notified:
+					require.Equal(t, StopPodSandbox, event, "observer should receive successful stop notification")
+				case <-time.After(pluginSyncTimeout):
+					t.Fatal("observer did not receive stop notification")
+				}
+			}
+
+			switch target {
+			case StopPodSandbox:
+				require.NoError(t, nt.runtime.StopPodSandbox(id), "CRI should continue after NRI stop timeout")
+			case RemovePodSandbox:
+				err = nt.runtime.RemovePodSandbox(id)
+				require.NoError(t, err, "CRI should continue after NRI remove timeout")
+				removed = true
+			}
+
+			select {
+			case <-timedOut:
+			case <-time.After(failing.stub.RequestTimeout() + pluginSyncTimeout):
+				t.Fatal("timed-out plugin callback did not return")
+			}
+
+			select {
+			case event := <-notified:
+				t.Fatalf("later plugin unexpectedly received %s after %s failure", event, target)
+			case <-time.After(250 * time.Millisecond):
+			}
+
+			if target == StopPodSandbox {
+				require.NoError(t, nt.runtime.RemovePodSandbox(id), "remove stopped pod sandbox")
+				removed = true
+			}
+		})
+	}
+}
+
 // Test NRI runtime/plugin state synchronization.
 func TestNriPluginSynchronization(t *testing.T) {
 	skipNriTestIfNecessary(t)
@@ -802,12 +1054,15 @@ type mockPlugin struct {
 	idx  string
 	stub stub.Stub
 	mask stub.EventMask
+	opts []stub.Option
 
 	q    *eventQ
 	pods map[string]*api.PodSandbox
 	ctrs map[string]*api.Container
 
 	closed               bool
+	closedCh             chan struct{}
+	closedOnce           sync.Once
 	namespace            string
 	logf                 func(string, ...any)
 	synchronize          func(*mockPlugin, []*api.PodSandbox, []*api.Container) ([]*api.ContainerUpdate, error)
@@ -851,12 +1106,14 @@ func (m *mockPlugin) Start() error {
 		)
 	}
 
-	m.stub, err = stub.New(m,
+	opts := []stub.Option{
 		stub.WithPluginName(m.name),
 		stub.WithPluginIdx(m.idx),
 		stub.WithSocketPath(nriTestSocket),
 		stub.WithOnClose(m.onClose),
-	)
+	}
+	opts = append(opts, m.opts...)
+	m.stub, err = stub.New(m, opts...)
 	if err != nil {
 		return err
 	}
@@ -904,6 +1161,7 @@ func (m *mockPlugin) Start() error {
 	m.q = &eventQ{}
 	m.pods = make(map[string]*api.PodSandbox)
 	m.ctrs = make(map[string]*api.Container)
+	m.closedCh = make(chan struct{})
 
 	err = m.stub.Start(context.Background())
 	if err != nil {
@@ -922,6 +1180,7 @@ func (m *mockPlugin) Stop() {
 
 func (m *mockPlugin) onClose() {
 	m.closed = true
+	m.closedOnce.Do(func() { close(m.closedCh) })
 	if m.stub != nil {
 		m.stub.Stop()
 		m.stub.Wait()

vendor/github.com/containerd/nri/README.md

@@ -135,6 +135,58 @@ Once the handshake sequence is over and the plugin has registered with NRI,
 it will start receiving pod and container lifecycle events according to its
 subscription.
 
+### Request Errors, Timeouts and Failure Policy
+
+NRI calls plugins in index order for each lifecycle event. If a plugin
+returns an error for a request, NRI does not call the remaining plugins.
+It returns the error to the runtime, and the runtime fails the pod or
+container operation.
+
+Each request has a timeout. The timeout is configured in the runtime and
+applies to all plugins. A plugin fails to process a request when it exceeds
+this timeout, loses its connection, or causes a protocol error. The failure
+policy of the plugin tells NRI what to do in that case. The plugin declares
+the policy when it subscribes to events. The policy has two independent
+fields, one for the request and one for the plugin:
+
+  - `fail_request`: if set, the request fails with an error that names the
+    plugin and the request, like an error returned by the plugin. NRI does
+    not call the remaining plugins. By default, the request proceeds with
+    the remaining plugins and the runtime is not notified.
+  - `keep_connected`: if set, a plugin that exceeds the request timeout
+    stays registered and receives the next requests. The timeout only means
+    that the plugin did not handle this request in time. By default, NRI
+    unregisters the plugin. This field only applies to timeouts. NRI always
+    unregisters a plugin that loses its connection.
+
+The default of each field is the behavior of NRI before failure policies
+existed. A plugin that does not declare a policy is treated as before: NRI
+unregisters it and the request proceeds without it. A runtime with an NRI
+version without failure policy support ignores the policy and behaves the
+same way.
+
+Set both fields for a plugin that must process every request it subscribes
+to, for example a plugin that sets up networking or injects devices. With
+the stub library:
+
+```go
+stub.WithFailurePolicy(&api.FailurePolicy{FailRequest: true, KeepConnected: true})
+```
+
+Set `keep_connected` alone for a best-effort plugin that must not be lost
+when it is slow once, for example a logger or a metrics collector.
+
+Two notes. A plugin that is kept connected but stays unresponsive delays
+each request it receives by the full timeout. `fail_request` applies to all
+subscribed events, including `StopContainer` and `RemoveContainer`. A plugin
+must subscribe only to the events it must process.
+
+The failure policy only applies to connected plugins. A plugin that loses
+its connection, also between requests, does not receive the next requests
+and does not fail them. Use the required plugins check of the
+[default validator](#default-validation) to require that a plugin is
+present before containers are created.
+
 ### Pod Data and Available Lifecycle Events
 
 <details>

vendor/github.com/containerd/nri/pkg/adaptation/adaptation.go

@@ -248,6 +248,9 @@ func (r *Adaptation) RunPodSandbox(ctx context.Context, req *RunPodSandboxReques
 	defer r.removeClosedPlugins()
 
 	for _, plugin := range r.plugins {
+		if plugin.isClosed() {
+			continue
+		}
 		if _, err := plugin.runPodSandbox(ctx, req); err != nil {
 			return err
 		}
@@ -263,6 +266,9 @@ func (r *Adaptation) UpdatePodSandbox(ctx context.Context, req *UpdatePodSandbox
 	defer r.removeClosedPlugins()
 
 	for _, plugin := range r.plugins {
+		if plugin.isClosed() {
+			continue
+		}
 		_, err := plugin.updatePodSandbox(ctx, req)
 		if err != nil {
 			return nil, err
@@ -279,6 +285,9 @@ func (r *Adaptation) PostUpdatePodSandbox(ctx context.Context, req *PostUpdatePo
 	defer r.removeClosedPlugins()
 
 	for _, plugin := range r.plugins {
+		if plugin.isClosed() {
+			continue
+		}
 		_, err := plugin.postUpdatePodSandbox(ctx, req)
 		if err != nil {
 			return err
@@ -295,6 +304,9 @@ func (r *Adaptation) StopPodSandbox(ctx context.Context, req *StopPodSandboxRequ
 	defer r.removeClosedPlugins()
 
 	for _, plugin := range r.plugins {
+		if plugin.isClosed() {
+			continue
+		}
 		_, err := plugin.stopPodSandbox(ctx, req)
 		if err != nil {
 			return err
@@ -311,6 +323,9 @@ func (r *Adaptation) RemovePodSandbox(ctx context.Context, req *RemovePodSandbox
 	defer r.removeClosedPlugins()
 
 	for _, plugin := range r.plugins {
+		if plugin.isClosed() {
+			continue
+		}
 		_, err := plugin.removePodSandbox(ctx, req)
 		if err != nil {
 			return err
@@ -339,6 +354,9 @@ func (r *Adaptation) CreateContainer(ctx context.Context, req *CreateContainerRe
 	}
 
 	for _, plugin := range r.plugins {
+		if plugin.isClosed() {
+			continue
+		}
 		if validate != nil {
 			validate.AddPlugin(plugin.base, plugin.idx)
 		}
@@ -362,6 +380,9 @@ func (r *Adaptation) PostCreateContainer(ctx context.Context, req *PostCreateCon
 	defer r.removeClosedPlugins()
 
 	for _, plugin := range r.plugins {
+		if plugin.isClosed() {
+			continue
+		}
 		_, err := plugin.postCreateContainer(ctx, req)
 		if err != nil {
 			return err
@@ -378,6 +399,9 @@ func (r *Adaptation) StartContainer(ctx context.Context, req *StartContainerRequ
 	defer r.removeClosedPlugins()
 
 	for _, plugin := range r.plugins {
+		if plugin.isClosed() {
+			continue
+		}
 		_, err := plugin.startContainer(ctx, req)
 		if err != nil {
 			return err
@@ -394,6 +418,9 @@ func (r *Adaptation) PostStartContainer(ctx context.Context, req *PostStartConta
 	defer r.removeClosedPlugins()
 
 	for _, plugin := range r.plugins {
+		if plugin.isClosed() {
+			continue
+		}
 		_, err := plugin.postStartContainer(ctx, req)
 		if err != nil {
 			return err
@@ -411,6 +438,9 @@ func (r *Adaptation) UpdateContainer(ctx context.Context, req *UpdateContainerRe
 
 	result := collectUpdateContainerResult(req)
 	for _, plugin := range r.plugins {
+		if plugin.isClosed() {
+			continue
+		}
 		rpl, err := plugin.updateContainer(ctx, req)
 		if err != nil {
 			return nil, err
@@ -431,6 +461,9 @@ func (r *Adaptation) PostUpdateContainer(ctx context.Context, req *PostUpdateCon
 	defer r.removeClosedPlugins()
 
 	for _, plugin := range r.plugins {
+		if plugin.isClosed() {
+			continue
+		}
 		_, err := plugin.postUpdateContainer(ctx, req)
 		if err != nil {
 			return err
@@ -448,6 +481,9 @@ func (r *Adaptation) StopContainer(ctx context.Context, req *StopContainerReques
 
 	result := collectStopContainerResult()
 	for _, plugin := range r.plugins {
+		if plugin.isClosed() {
+			continue
+		}
 		rpl, err := plugin.stopContainer(ctx, req)
 		if err != nil {
 			return nil, err
@@ -468,6 +504,9 @@ func (r *Adaptation) RemoveContainer(ctx context.Context, req *RemoveContainerRe
 	defer r.removeClosedPlugins()
 
 	for _, plugin := range r.plugins {
+		if plugin.isClosed() {
+			continue
+		}
 		_, err := plugin.removeContainer(ctx, req)
 		if err != nil {
 			return err
@@ -497,6 +536,9 @@ func (r *Adaptation) validateContainerAdjustment(ctx context.Context, req *Valid
 	wg := sync.WaitGroup{}
 
 	for _, p := range r.validators {
+		if p.isClosed() {
+			continue
+		}
 		wg.Add(1)
 		go func(p *plugin) {
 			defer wg.Done()
@@ -605,6 +647,12 @@ func (r *Adaptation) stopPlugins() {
 	r.plugins = nil
 }
 
+// removeClosedPlugins removes closed plugins from the registry. The caller
+// must hold the lock. A plugin is closed when a request drops it, or when
+// its connection closes while no request is in progress. It is removed at
+// the end of the next request. Until then, requests skip it: a request
+// relayed to it fails with ttrpc.ErrClosed, and a fail_request policy
+// would fail the request for a plugin that is already gone.
 func (r *Adaptation) removeClosedPlugins() {
 	var active, closed, validators []*plugin
 	for _, p := range r.plugins {

vendor/github.com/containerd/nri/pkg/adaptation/api.go

@@ -112,6 +112,7 @@ type (
 
 	Event         = api.Event
 	EventMask     = api.EventMask
+	FailurePolicy = api.FailurePolicy
 )
 
 // Aliased consts for api/api.proto.

vendor/github.com/containerd/nri/pkg/adaptation/plugin.go

@@ -63,6 +63,7 @@ type plugin struct {
 	rpcl   stdnet.Listener
 	rpcs   *ttrpc.Server
 	events EventMask
+	policy *FailurePolicy
 	closed bool
 	regC   chan error
 	closeC chan struct{}
@@ -497,6 +498,9 @@ func (p *plugin) configure(ctx context.Context, name, version, config string) (e
 	}
 	p.events = events
 
+	// A nil policy means all defaults. The proto getters accept a nil receiver.
+	p.policy = rpl.FailurePolicy
+
 	return nil
 }
 
@@ -614,20 +618,14 @@ func (p *plugin) runPodSandbox(ctx context.Context, req *RunPodSandboxRequest) (
 		return nil, nil
 	}
 
-	ctx, cancel := context.WithTimeout(ctx, getPluginRequestTimeout())
+	rctx, cancel := context.WithTimeout(ctx, getPluginRequestTimeout())
 	defer cancel()
 
 	start := time.Now()
-	rpl, err := p.impl.RunPodSandbox(ctx, req)
+	rpl, err := p.impl.RunPodSandbox(rctx, req)
 	p.recordInvocation(event.PrettyName(), start, err)
 	if err != nil {
-		if isFatalError(err) {
-			log.Errorf(ctx, "closing plugin %s, failed to handle %s request: %v",
-				p.name(), event.PrettyName(), err)
-			p.close()
-			return nil, nil
-		}
-		return nil, err
+		return nil, p.handleRequestError(ctx, event, err)
 	}
 	p.warnDeprecatedEvent(ctx, event)
 
@@ -641,20 +639,14 @@ func (p *plugin) updatePodSandbox(ctx context.Context, req *UpdatePodSandboxRequ
 		return nil, nil
 	}
 
-	ctx, cancel := context.WithTimeout(ctx, getPluginRequestTimeout())
+	rctx, cancel := context.WithTimeout(ctx, getPluginRequestTimeout())
 	defer cancel()
 
 	start := time.Now()
-	_, err := p.impl.UpdatePodSandbox(ctx, req)
+	_, err := p.impl.UpdatePodSandbox(rctx, req)
 	p.recordInvocation(event.PrettyName(), start, err)
 	if err != nil {
-		if isFatalError(err) {
-			log.Errorf(ctx, "closing plugin %s, failed to handle %s request: %v",
-				p.name(), event.PrettyName(), err)
-			p.close()
-			return nil, nil
-		}
-		return nil, err
+		return nil, p.handleRequestError(ctx, event, err)
 	}
 
 	return &UpdatePodSandboxResponse{}, nil
@@ -667,20 +659,14 @@ func (p *plugin) postUpdatePodSandbox(ctx context.Context, req *PostUpdatePodSan
 		return nil, nil
 	}
 
-	ctx, cancel := context.WithTimeout(ctx, getPluginRequestTimeout())
+	rctx, cancel := context.WithTimeout(ctx, getPluginRequestTimeout())
 	defer cancel()
 
 	start := time.Now()
-	_, err := p.impl.PostUpdatePodSandbox(ctx, req)
+	_, err := p.impl.PostUpdatePodSandbox(rctx, req)
 	p.recordInvocation(event.PrettyName(), start, err)
 	if err != nil {
-		if isFatalError(err) {
-			log.Errorf(ctx, "closing plugin %s, failed to handle %s request: %v",
-				p.name(), event.PrettyName(), err)
-			p.close()
-			return nil, nil
-		}
-		return nil, err
+		return nil, p.handleRequestError(ctx, event, err)
 	}
 	p.warnDeprecatedEvent(ctx, event)
 
@@ -694,20 +680,14 @@ func (p *plugin) stopPodSandbox(ctx context.Context, req *StopPodSandboxRequest)
 		return nil, nil
 	}
 
-	ctx, cancel := context.WithTimeout(ctx, getPluginRequestTimeout())
+	rctx, cancel := context.WithTimeout(ctx, getPluginRequestTimeout())
 	defer cancel()
 
 	start := time.Now()
-	_, err := p.impl.StopPodSandbox(ctx, req)
+	_, err := p.impl.StopPodSandbox(rctx, req)
 	p.recordInvocation(event.PrettyName(), start, err)
 	if err != nil {
-		if isFatalError(err) {
-			log.Errorf(ctx, "closing plugin %s, failed to handle %s request: %v",
-				p.name(), event.PrettyName(), err)
-			p.close()
-			return nil, nil
-		}
-		return nil, err
+		return nil, p.handleRequestError(ctx, event, err)
 	}
 	p.warnDeprecatedEvent(ctx, event)
 
@@ -721,20 +701,14 @@ func (p *plugin) removePodSandbox(ctx context.Context, req *RemovePodSandboxRequ
 		return nil, nil
 	}
 
-	ctx, cancel := context.WithTimeout(ctx, getPluginRequestTimeout())
+	rctx, cancel := context.WithTimeout(ctx, getPluginRequestTimeout())
 	defer cancel()
 
 	start := time.Now()
-	_, err := p.impl.RemovePodSandbox(ctx, req)
+	_, err := p.impl.RemovePodSandbox(rctx, req)
 	p.recordInvocation(event.PrettyName(), start, err)
 	if err != nil {
-		if isFatalError(err) {
-			log.Errorf(ctx, "closing plugin %s, failed to handle %s request: %v",
-				p.name(), event.PrettyName(), err)
-			p.close()
-			return nil, nil
-		}
-		return nil, err
+		return nil, p.handleRequestError(ctx, event, err)
 	}
 	p.warnDeprecatedEvent(ctx, event)
 
@@ -748,20 +722,14 @@ func (p *plugin) createContainer(ctx context.Context, req *CreateContainerReques
 		return nil, nil
 	}
 
-	ctx, cancel := context.WithTimeout(ctx, getPluginRequestTimeout())
+	rctx, cancel := context.WithTimeout(ctx, getPluginRequestTimeout())
 	defer cancel()
 
 	start := time.Now()
-	rpl, err := p.impl.CreateContainer(ctx, req)
+	rpl, err := p.impl.CreateContainer(rctx, req)
 	p.recordInvocation(event.PrettyName(), start, err)
 	if err != nil {
-		if isFatalError(err) {
-			log.Errorf(ctx, "closing plugin %s, failed to handle %s request: %v",
-				p.name(), event.PrettyName(), err)
-			p.close()
-			return nil, nil
-		}
-		return nil, err
+		return nil, p.handleRequestError(ctx, event, err)
 	}
 	if rpl != nil {
 		p.recordAdjustments(event.PrettyName(), rpl.Adjust, len(rpl.Update), len(rpl.Evict))
@@ -777,20 +745,14 @@ func (p *plugin) postCreateContainer(ctx context.Context, req *PostCreateContain
 		return nil, nil
 	}
 
-	ctx, cancel := context.WithTimeout(ctx, getPluginRequestTimeout())
+	rctx, cancel := context.WithTimeout(ctx, getPluginRequestTimeout())
 	defer cancel()
 
 	start := time.Now()
-	_, err := p.impl.PostCreateContainer(ctx, req)
+	_, err := p.impl.PostCreateContainer(rctx, req)
 	p.recordInvocation(event.PrettyName(), start, err)
 	if err != nil {
-		if isFatalError(err) {
-			log.Errorf(ctx, "closing plugin %s, failed to handle %s request: %v",
-				p.name(), event.PrettyName(), err)
-			p.close()
-			return nil, nil
-		}
-		return nil, err
+		return nil, p.handleRequestError(ctx, event, err)
 	}
 	p.warnDeprecatedEvent(ctx, event)
 
@@ -804,20 +766,14 @@ func (p *plugin) startContainer(ctx context.Context, req *StartContainerRequest)
 		return nil, nil
 	}
 
-	ctx, cancel := context.WithTimeout(ctx, getPluginRequestTimeout())
+	rctx, cancel := context.WithTimeout(ctx, getPluginRequestTimeout())
 	defer cancel()
 
 	start := time.Now()
-	_, err := p.impl.StartContainer(ctx, req)
+	_, err := p.impl.StartContainer(rctx, req)
 	p.recordInvocation(event.PrettyName(), start, err)
 	if err != nil {
-		if isFatalError(err) {
-			log.Errorf(ctx, "closing plugin %s, failed to handle %s request: %v",
-				p.name(), event.PrettyName(), err)
-			p.close()
-			return nil, nil
-		}
-		return nil, err
+		return nil, p.handleRequestError(ctx, event, err)
 	}
 	p.warnDeprecatedEvent(ctx, event)
 
@@ -831,20 +787,14 @@ func (p *plugin) postStartContainer(ctx context.Context, req *PostStartContainer
 		return nil, nil
 	}
 
-	ctx, cancel := context.WithTimeout(ctx, getPluginRequestTimeout())
+	rctx, cancel := context.WithTimeout(ctx, getPluginRequestTimeout())
 	defer cancel()
 
 	start := time.Now()
-	_, err := p.impl.PostStartContainer(ctx, req)
+	_, err := p.impl.PostStartContainer(rctx, req)
 	p.recordInvocation(event.PrettyName(), start, err)
 	if err != nil {
-		if isFatalError(err) {
-			log.Errorf(ctx, "closing plugin %s, failed to handle %s request: %v",
-				p.name(), event.PrettyName(), err)
-			p.close()
-			return nil, nil
-		}
-		return nil, err
+		return nil, p.handleRequestError(ctx, event, err)
 	}
 	p.warnDeprecatedEvent(ctx, event)
 
@@ -858,20 +808,14 @@ func (p *plugin) updateContainer(ctx context.Context, req *UpdateContainerReques
 		return nil, nil
 	}
 
-	ctx, cancel := context.WithTimeout(ctx, getPluginRequestTimeout())
+	rctx, cancel := context.WithTimeout(ctx, getPluginRequestTimeout())
 	defer cancel()
 
 	start := time.Now()
-	rpl, err := p.impl.UpdateContainer(ctx, req)
+	rpl, err := p.impl.UpdateContainer(rctx, req)
 	p.recordInvocation(event.PrettyName(), start, err)
 	if err != nil {
-		if isFatalError(err) {
-			log.Errorf(ctx, "closing plugin %s, failed to handle %s request: %v",
-				p.name(), event.PrettyName(), err)
-			p.close()
-			return nil, nil
-		}
-		return nil, err
+		return nil, p.handleRequestError(ctx, event, err)
 	}
 	if rpl != nil {
 		p.recordAdjustments(event.PrettyName(), nil, len(rpl.Update), len(rpl.Evict))
@@ -887,20 +831,14 @@ func (p *plugin) postUpdateContainer(ctx context.Context, req *PostUpdateContain
 		return nil, nil
 	}
 
-	ctx, cancel := context.WithTimeout(ctx, getPluginRequestTimeout())
+	rctx, cancel := context.WithTimeout(ctx, getPluginRequestTimeout())
 	defer cancel()
 
 	start := time.Now()
-	_, err := p.impl.PostUpdateContainer(ctx, req)
+	_, err := p.impl.PostUpdateContainer(rctx, req)
 	p.recordInvocation(event.PrettyName(), start, err)
 	if err != nil {
-		if isFatalError(err) {
-			log.Errorf(ctx, "closing plugin %s, failed to handle %s request: %v",
-				p.name(), event.PrettyName(), err)
-			p.close()
-			return nil, nil
-		}
-		return nil, err
+		return nil, p.handleRequestError(ctx, event, err)
 	}
 	p.warnDeprecatedEvent(ctx, event)
 
@@ -914,20 +852,14 @@ func (p *plugin) stopContainer(ctx context.Context, req *StopContainerRequest) (
 		return nil, nil
 	}
 
-	ctx, cancel := context.WithTimeout(ctx, getPluginRequestTimeout())
+	rctx, cancel := context.WithTimeout(ctx, getPluginRequestTimeout())
 	defer cancel()
 
 	start := time.Now()
-	rpl, err = p.impl.StopContainer(ctx, req)
+	rpl, err = p.impl.StopContainer(rctx, req)
 	p.recordInvocation(event.PrettyName(), start, err)
 	if err != nil {
-		if isFatalError(err) {
-			log.Errorf(ctx, "closing plugin %s, failed to handle %s request: %v",
-				p.name(), event.PrettyName(), err)
-			p.close()
-			return nil, nil
-		}
-		return nil, err
+		return nil, p.handleRequestError(ctx, event, err)
 	}
 	if rpl != nil {
 		p.recordAdjustments(event.PrettyName(), nil, len(rpl.Update), 0)
@@ -943,20 +875,14 @@ func (p *plugin) removeContainer(ctx context.Context, req *RemoveContainerReques
 		return nil, nil
 	}
 
-	ctx, cancel := context.WithTimeout(ctx, getPluginRequestTimeout())
+	rctx, cancel := context.WithTimeout(ctx, getPluginRequestTimeout())
 	defer cancel()
 
 	start := time.Now()
-	_, err := p.impl.RemoveContainer(ctx, req)
+	_, err := p.impl.RemoveContainer(rctx, req)
 	p.recordInvocation(event.PrettyName(), start, err)
 	if err != nil {
-		if isFatalError(err) {
-			log.Errorf(ctx, "closing plugin %s, failed to handle %s request: %v",
-				p.name(), event.PrettyName(), err)
-			p.close()
-			return nil, nil
-		}
-		return nil, err
+		return nil, p.handleRequestError(ctx, event, err)
 	}
 	p.warnDeprecatedEvent(ctx, event)
 
@@ -970,19 +896,19 @@ func (p *plugin) ValidateContainerAdjustment(ctx context.Context, req *ValidateC
 		return nil
 	}
 
-	ctx, cancel := context.WithTimeout(ctx, getPluginRequestTimeout())
+	rctx, cancel := context.WithTimeout(ctx, getPluginRequestTimeout())
 	defer cancel()
 
 	start := time.Now()
-	rpl, err := p.impl.ValidateContainerAdjustment(ctx, req)
+	rpl, err := p.impl.ValidateContainerAdjustment(rctx, req)
 	p.recordInvocation(event.PrettyName(), start, err)
 	if err != nil {
-		if isFatalError(err) {
+		if ctx.Err() == nil && p.dropsPlugin(err) {
 			log.Errorf(ctx, "closing plugin %s, failed to handle %s request: %v",
 				p.name(), event.PrettyName(), err)
 			p.close()
 		}
-		return fmt.Errorf("validator plugin %s failed: %v", p.name(), err)
+		return fmt.Errorf("validator plugin %s failed: %w", p.name(), err)
 	}
 
 	return rpl.ValidationResult(p.name())
@@ -1018,6 +944,56 @@ func (p *plugin) warnDeprecatedEvent(ctx context.Context, event Event) {
 	p.impl.warned[event] = true
 }
 
+// handleRequestError handles an error from a relayed request. ctx is the
+// context of the runtime. It is not the context with the request timeout.
+//
+// Two kinds of errors are not plugin failures. The function returns them
+// unchanged and the request fails:
+//
+//   - an error returned by the plugin,
+//   - any error after the runtime cancelled ctx or its deadline expired.
+//     The plugin is not responsible for that, even if the request timeout
+//     expired too.
+//
+// A fatal error is a plugin failure: the plugin did not answer the request.
+// The failure policy of the plugin applies. The function takes two
+// independent decisions. It drops the plugin unless the policy keeps it
+// (see dropsPlugin). It fails the request only if the policy says so.
+// Otherwise the request proceeds without the plugin. A returned error
+// names the plugin and the request, because the runtime reports it.
+func (p *plugin) handleRequestError(ctx context.Context, event Event, err error) error {
+	if ctx.Err() != nil || !isFatalError(err) {
+		return err
+	}
+
+	if p.dropsPlugin(err) {
+		log.Errorf(ctx, "closing plugin %s, failed to handle %s request: %v",
+			p.name(), event.PrettyName(), err)
+		p.close()
+	}
+
+	if !p.policy.GetFailRequest() {
+		log.Warnf(ctx, "plugin %s failed to handle %s request, proceeding without it: %v",
+			p.name(), event.PrettyName(), err)
+		return nil
+	}
+
+	return fmt.Errorf("plugin %s failed to handle %s request: %w",
+		p.name(), event.PrettyName(), err)
+}
+
+// dropsPlugin returns true if the plugin must be closed and unregistered
+// for the error. A fatal error drops the plugin, with one exception: a
+// request timeout does not drop a plugin with KeepConnected set. The timeout
+// only means that the plugin did not handle this request in time. A lost
+// connection or a protocol error always drops the plugin.
+func (p *plugin) dropsPlugin(err error) bool {
+	if !isFatalError(err) {
+		return false
+	}
+	return !errors.Is(err, context.DeadlineExceeded) || !p.policy.GetKeepConnected()
+}
+
 // isFatalError returns true if the error is fatal and the plugin connection should be closed.
 func isFatalError(err error) bool {
 	switch {

vendor/github.com/containerd/nri/pkg/api/api.proto

@@ -183,6 +183,40 @@ message ConfigureResponse {
   // Events to subscribe the plugin for. Each bit set corresponds to an
   // enumerated Event.
   int32 events = 2;
+  // How the runtime handles a failure of the plugin to receive and respond
+  // to a request within a timeout period. If unset, all fields have their
+  // default value. The defaults match the behavior of NRI before this field
+  // existed.
+  FailurePolicy failure_policy = 3;
+}
+
+// FailurePolicy tells the runtime what to do when a plugin fails to receive
+// and respond to a request within a timeout period.
+// Failure types covered by the FailurePolicy include request timeout, lost
+// connection, and protocol error. Non-transport/timeout errors, such as a
+// plugin returning a rejection to the request, are not covered by the
+// FailurePolicy.
+//
+// The runtime takes two independent decisions for each failure: one about
+// the request and one about the plugin. Each field controls one decision.
+// The default of each field is the behavior of NRI before this field
+// existed. A plugin that does not set a policy is treated as before: the
+// runtime disconnects it and the request proceeds without it.
+message FailurePolicy {
+  // When fail_request is set to true, fail the request is the policy. The
+  // error names the plugin and the request, like an error returned by the
+  // plugin. The remaining plugins do not receive the request. The
+  // fail_request policy defaults to false when unset. False results in the
+  // request proceeding on to the remaining plugins, as if the plugin were
+  // not subscribed to it.
+  bool fail_request = 1;
+  // Keep the plugin registered when it exceeds the request timeout. The
+  // plugin continues receiving requests. A plugin that stays unresponsive
+  // delays each request it receives by the full timeout. Default: the
+  // runtime disconnects and unregisters the plugin. This field only applies
+  // to timeouts. A plugin that loses its connection or causes a protocol
+  // error is always disconnected.
+  bool keep_connected = 2;
 }
 
 message SynchronizeRequest {

vendor/github.com/containerd/nri/pkg/api/api_vtproto.pb.go

@@ -337,6 +337,16 @@ func (m *ConfigureResponse) MarshalToSizedBufferVT(dAtA []byte) (int, error) {
 		i -= len(m.unknownFields)
 		copy(dAtA[i:], m.unknownFields)
 	}
+	if m.FailurePolicy != nil {
+		size, err := m.FailurePolicy.MarshalToSizedBufferVT(dAtA[:i])
+		if err != nil {
+			return 0, err
+		}
+		i -= size
+		i = encodeVarint(dAtA, i, uint64(size))
+		i--
+		dAtA[i] = 0x1a
+	}
 	if m.Events != 0 {
 		i = encodeVarint(dAtA, i, uint64(m.Events))
 		i--
@@ -345,6 +355,59 @@ func (m *ConfigureResponse) MarshalToSizedBufferVT(dAtA []byte) (int, error) {
 	return len(dAtA) - i, nil
 }
 
+func (m *FailurePolicy) MarshalVT() (dAtA []byte, err error) {
+	if m == nil {
+		return nil, nil
+	}
+	size := m.SizeVT()
+	dAtA = make([]byte, size)
+	n, err := m.MarshalToSizedBufferVT(dAtA[:size])
+	if err != nil {
+		return nil, err
+	}
+	return dAtA[:n], nil
+}
+
+func (m *FailurePolicy) MarshalToVT(dAtA []byte) (int, error) {
+	size := m.SizeVT()
+	return m.MarshalToSizedBufferVT(dAtA[:size])
+}
+
+func (m *FailurePolicy) MarshalToSizedBufferVT(dAtA []byte) (int, error) {
+	if m == nil {
+		return 0, nil
+	}
+	i := len(dAtA)
+	_ = i
+	var l int
+	_ = l
+	if m.unknownFields != nil {
+		i -= len(m.unknownFields)
+		copy(dAtA[i:], m.unknownFields)
+	}
+	if m.KeepConnected {
+		i--
+		if m.KeepConnected {
+			dAtA[i] = 1
+		} else {
+			dAtA[i] = 0
+		}
+		i--
+		dAtA[i] = 0x10
+	}
+	if m.FailRequest {
+		i--
+		if m.FailRequest {
+			dAtA[i] = 1
+		} else {
+			dAtA[i] = 0
+		}
+		i--
+		dAtA[i] = 0x8
+	}
+	return len(dAtA) - i, nil
+}
+
 func (m *SynchronizeRequest) MarshalVT() (dAtA []byte, err error) {
 	if m == nil {
 		return nil, nil
@@ -5362,6 +5425,26 @@ func (m *ConfigureResponse) SizeVT() (n int) {
 	if m.Events != 0 {
 		n += 1 + sov(uint64(m.Events))
 	}
+	if m.FailurePolicy != nil {
+		l = m.FailurePolicy.SizeVT()
+		n += 1 + l + sov(uint64(l))
+	}
+	n += len(m.unknownFields)
+	return n
+}
+
+func (m *FailurePolicy) SizeVT() (n int) {
+	if m == nil {
+		return 0
+	}
+	var l int
+	_ = l
+	if m.FailRequest {
+		n += 2
+	}
+	if m.KeepConnected {
+		n += 2
+	}
 	n += len(m.unknownFields)
 	return n
 }
@@ -7957,6 +8040,133 @@ func (m *ConfigureResponse) UnmarshalVT(dAtA []byte) error {
 					break
 				}
 			}
+		case 3:
+			if wireType != 2 {
+				return fmt.Errorf("proto: wrong wireType = %d for field FailurePolicy", wireType)
+			}
+			var msglen int
+			for shift := uint(0); ; shift += 7 {
+				if shift >= 64 {
+					return ErrIntOverflow
+				}
+				if iNdEx >= l {
+					return io.ErrUnexpectedEOF
+				}
+				b := dAtA[iNdEx]
+				iNdEx++
+				msglen |= int(b&0x7F) << shift
+				if b < 0x80 {
+					break
+				}
+			}
+			if msglen < 0 {
+				return ErrInvalidLength
+			}
+			postIndex := iNdEx + msglen
+			if postIndex < 0 {
+				return ErrInvalidLength
+			}
+			if postIndex > l {
+				return io.ErrUnexpectedEOF
+			}
+			if m.FailurePolicy == nil {
+				m.FailurePolicy = &FailurePolicy{}
+			}
+			if err := m.FailurePolicy.UnmarshalVT(dAtA[iNdEx:postIndex]); err != nil {
+				return err
+			}
+			iNdEx = postIndex
+		default:
+			iNdEx = preIndex
+			skippy, err := skip(dAtA[iNdEx:])
+			if err != nil {
+				return err
+			}
+			if (skippy < 0) || (iNdEx+skippy) < 0 {
+				return ErrInvalidLength
+			}
+			if (iNdEx + skippy) > l {
+				return io.ErrUnexpectedEOF
+			}
+			m.unknownFields = append(m.unknownFields, dAtA[iNdEx:iNdEx+skippy]...)
+			iNdEx += skippy
+		}
+	}
+
+	if iNdEx > l {
+		return io.ErrUnexpectedEOF
+	}
+	return nil
+}
+func (m *FailurePolicy) UnmarshalVT(dAtA []byte) error {
+	l := len(dAtA)
+	iNdEx := 0
+	for iNdEx < l {
+		preIndex := iNdEx
+		var wire uint64
+		for shift := uint(0); ; shift += 7 {
+			if shift >= 64 {
+				return ErrIntOverflow
+			}
+			if iNdEx >= l {
+				return io.ErrUnexpectedEOF
+			}
+			b := dAtA[iNdEx]
+			iNdEx++
+			wire |= uint64(b&0x7F) << shift
+			if b < 0x80 {
+				break
+			}
+		}
+		fieldNum := int32(wire >> 3)
+		wireType := int(wire & 0x7)
+		if wireType == 4 {
+			return fmt.Errorf("proto: FailurePolicy: wiretype end group for non-group")
+		}
+		if fieldNum <= 0 {
+			return fmt.Errorf("proto: FailurePolicy: illegal tag %d (wire type %d)", fieldNum, wire)
+		}
+		switch fieldNum {
+		case 1:
+			if wireType != 0 {
+				return fmt.Errorf("proto: wrong wireType = %d for field FailRequest", wireType)
+			}
+			var v int
+			for shift := uint(0); ; shift += 7 {
+				if shift >= 64 {
+					return ErrIntOverflow
+				}
+				if iNdEx >= l {
+					return io.ErrUnexpectedEOF
+				}
+				b := dAtA[iNdEx]
+				iNdEx++
+				v |= int(b&0x7F) << shift
+				if b < 0x80 {
+					break
+				}
+			}
+			m.FailRequest = bool(v != 0)
+		case 2:
+			if wireType != 0 {
+				return fmt.Errorf("proto: wrong wireType = %d for field KeepConnected", wireType)
+			}
+			var v int
+			for shift := uint(0); ; shift += 7 {
+				if shift >= 64 {
+					return ErrIntOverflow
+				}
+				if iNdEx >= l {
+					return io.ErrUnexpectedEOF
+				}
+				b := dAtA[iNdEx]
+				iNdEx++
+				v |= int(b&0x7F) << shift
+				if b < 0x80 {
+					break
+				}
+			}
+			m.KeepConnected = bool(v != 0)
 		default:
 			iNdEx = preIndex
 			skippy, err := skip(dAtA[iNdEx:])

vendor/github.com/containerd/nri/pkg/net/multiplex/mux.go

@@ -322,6 +322,18 @@ func (m *mux) reader() {
 
 		cid = binary.BigEndian.Uint32(hdr[0:4])
 		cnt = binary.BigEndian.Uint32(hdr[4:8])
+		if cnt > maxPayloadSize {
+			// The write path never produces frames this large, so the
+			// stream is malformed: we cannot skip the frame (the framing
+			// has no resync marker) and we cannot allocate the claimed
+			// size (up to 4 GiB, and int(cnt) overflows on 32-bit
+			// architectures). Tear the mux down, like we do for queue
+			// overflow.
+			err = fmt.Errorf("conn %d: frame payload %d bytes, maximum is %d", cid, cnt, maxPayloadSize)
+			m.setError(err)
+			m.Close()
+			return
+		}
 		buf = make([]byte, int(cnt))
 
 		_, err = io.ReadFull(m.trunk, buf)

vendor/github.com/containerd/nri/pkg/stub/stub.go

@@ -283,6 +283,29 @@ func WithLogger(logger nrilog.Logger) Option {
 	}
 }
 
+// WithFailurePolicy sets how the runtime handles a failure of the plugin to
+// process a request. A failure is a request timeout, a lost connection or a
+// protocol error. The policy has two independent fields. The default of each
+// field is the behavior of NRI before failure policies existed:
+//
+//   - FailRequest: fail the request with an error. Default: the request
+//     proceeds without the plugin.
+//   - KeepConnected: keep the plugin registered when it exceeds the request
+//     timeout. Default: the runtime disconnects the plugin. A plugin that
+//     loses its connection is always disconnected.
+//
+// An error returned by the plugin always fails the request. A runtime with
+// an NRI version without failure policy support ignores this option.
+func WithFailurePolicy(policy *api.FailurePolicy) Option {
+	return func(s *stub) error {
+		if policy == nil {
+			return fmt.Errorf("nil failure policy")
+		}
+		s.failurePolicy = policy
+		return nil
+	}
+}
+
 // stub implements Stub.
 type stub struct {
 	sync.Mutex
@@ -310,6 +333,7 @@ type stub struct {
 
 	registrationTimeout time.Duration
 	requestTimeout      time.Duration
+	failurePolicy       *api.FailurePolicy
 	runtimeNRIVersion   string
 	pluginNRIVersion    string
 	logger              nrilog.Logger
@@ -710,6 +734,7 @@ func (stub *stub) Configure(ctx context.Context, req *api.ConfigureRequest) (rpl
 
 	return &api.ConfigureResponse{
 		Events:        int32(events),
+		FailurePolicy: stub.failurePolicy,
 	}, nil
 }
 

vendor/github.com/containerd/nri/pkg/version/version.go

@@ -17,11 +17,11 @@
 package version
 
 import (
+	"fmt"
 	"runtime/debug"
+	"slices"
 	"strconv"
 	"strings"
-
-	"golang.org/x/mod/semver"
 )
 
 const (
@@ -34,6 +34,205 @@ const (
 	nriModulePath = "github.com/containerd/nri"
 )
 
+// version represents a struct type that holds relevant data
+// that constitute a Semantic Version
+type version struct {
+	major string
+	minor string
+	patch string
+	pre   string
+}
+
+func (v version) String() string {
+	if v.pre != "" {
+		return fmt.Sprintf("v%s.%s.%s-%s", v.major, v.minor, v.patch, v.pre)
+	}
+	return fmt.Sprintf("v%s.%s.%s", v.major, v.minor, v.patch)
+}
+
+// compareVersion parses two semver strings into the "version" struct type and compares them.
+// NOTE: It ignores the build metadata when making the comparison
+func compareVersion(a, b string) int {
+	aVer, errA := parseVersion(a)
+	bVer, errB := parseVersion(b)
+	// Like semver.Compare, order invalid versions before valid ones
+	// and consider any two invalid versions equal.
+	switch {
+	case errA != nil && errB != nil:
+		return 0
+	case errA != nil:
+		return -1
+	case errB != nil:
+		return 1
+	}
+
+	if c := compareInt(aVer.major, bVer.major); c != 0 {
+		return c
+	}
+	if c := compareInt(aVer.minor, bVer.minor); c != 0 {
+		return c
+	}
+	if c := compareInt(aVer.patch, bVer.patch); c != 0 {
+		return c
+	}
+	return comparePrerelease(aVer.pre, bVer.pre)
+}
+
+func compareInt(x, y string) int {
+	if x == y {
+		return 0
+	}
+	if len(x) < len(y) {
+		return -1
+	}
+	if len(x) > len(y) {
+		return 1
+	}
+	if x < y {
+		return -1
+	}
+	return 1
+}
+
+func comparePrerelease(x, y string) int {
+	if x == y {
+		return 0
+	}
+	if x == "" {
+		return 1
+	}
+	if y == "" {
+		return -1
+	}
+	for x != "" && y != "" {
+		var dx, dy string
+		dx, x, _ = strings.Cut(x, ".")
+		dy, y, _ = strings.Cut(y, ".")
+		if dx != dy {
+			ix := isNum(dx)
+			iy := isNum(dy)
+			if ix != iy {
+				if ix {
+					return -1
+				}
+				return 1
+			}
+			if ix {
+				if len(dx) < len(dy) {
+					return -1
+				}
+				if len(dx) > len(dy) {
+					return 1
+				}
+			}
+			if dx < dy {
+				return -1
+			}
+			return 1
+		}
+	}
+	if x == "" {
+		return -1
+	}
+	return 1
+}
+
+func isNum(v string) bool {
+	if len(v) == 0 {
+		return false
+	}
+	i := 0
+	for i < len(v) && '0' <= v[i] && v[i] <= '9' {
+		i++
+	}
+	return i == len(v)
+}
+
+// parseVersion parses a semantic version string. Like the semver package it replaces, it requires a leading "v"
+// and accepts the shorthand forms v<Major> and v<Major>.<Minor> (pre-release and build metadata are only allowed
+// on a full v<Major>.<Minor>.<Patch> version). Build metadata is validated but discarded.
+// parseVersion is inspired by the internal parse written in the upstream "golang.org/x/mod/semver" package
+// https://cs.opensource.google/go/x/mod/+/master:semver/semver.go;l=178
+func parseVersion(s string) (version, error) {
+	rest, ok := strings.CutPrefix(s, "v")
+	if !ok {
+		return version{}, fmt.Errorf("invalid version %q: missing 'v' prefix", s)
+	}
+
+	rest, build, hasBuild := strings.Cut(rest, "+")
+	if hasBuild {
+		if err := validateIdentifiers(build, false); err != nil {
+			return version{}, fmt.Errorf("invalid build metadata in %q: %w", s, err)
+		}
+	}
+
+	core, pre, hasPre := strings.Cut(rest, "-")
+	if hasPre {
+		if err := validateIdentifiers(pre, true); err != nil {
+			return version{}, fmt.Errorf("invalid pre-release in %q: %w", s, err)
+		}
+	}
+
+	parts := strings.Split(core, ".")
+	if len(parts) > 3 {
+		return version{}, fmt.Errorf("invalid version %q: too many components", s)
+	}
+	if len(parts) < 3 && (hasPre || hasBuild) {
+		return version{}, fmt.Errorf("invalid version %q: pre-release or build metadata requires major.minor.patch", s)
+	}
+
+	var nums [3]string
+	for i, part := range parts {
+		n, err := parseNumber(part)
+		if err != nil {
+			return version{}, fmt.Errorf("invalid version %q: %w", s, err)
+		}
+		nums[i] = n
+	}
+	for i := range nums {
+		if nums[i] == "" {
+			nums[i] = "0"
+		}
+	}
+
+	return version{major: nums[0], minor: nums[1], patch: nums[2], pre: pre}, nil
+}
+
+// parseNumber parses a non-negative decimal number without leading zeros.
+func parseNumber(s string) (string, error) {
+	if s == "" || !isNum(s) {
+		return "", fmt.Errorf("invalid number %q", s)
+	}
+	if len(s) > 1 && s[0] == '0' {
+		return "", fmt.Errorf("number %q has a leading zero", s)
+	}
+	return s, nil
+}
+
+// validateIdentifiers checks a dot-separated list of pre-release or build identifiers:
+// each must be non-empty and consist of [0-9A-Za-z-]. Numeric pre-release identifiers must not have leading zeros.
+func validateIdentifiers(s string, pre bool) error {
+	for id := range strings.SplitSeq(s, ".") {
+		if id == "" {
+			return fmt.Errorf("empty identifier in %q", s)
+		}
+		for i := 0; i < len(id); i++ {
+			c := id[i]
+			if !isValidIdentifierChar(c) {
+				return fmt.Errorf("invalid character %q in identifier %q", c, id)
+			}
+		}
+		if pre && isNum(id) && len(id) > 1 && id[0] == '0' {
+			return fmt.Errorf("numeric identifier %q has a leading zero", id)
+		}
+	}
+	return nil
+}
+
+func isValidIdentifierChar(c byte) bool {
+	return (c >= '0' && c <= '9') || (c >= 'a' && c <= 'z') || (c >= 'A' && c <= 'Z') || c == '-'
+}
+
 // GetFromBuildInfo returns the locally used NRI version. This
 // is taken either from the debug/build info provided by the
 // golang runtime, or for plugins hosted in the NRI repository
@@ -63,8 +262,8 @@ func GetFromBuildInfo() string {
 }
 
 // majorMinorPatch returns the major.minor.patch prefix of the semantic version v.
-func majorMinorPatch(v string) string {
-	return strings.TrimSuffix(strings.TrimSuffix(v, semver.Build(v)), semver.Prerelease(v))
+func majorMinorPatch(v version) string {
+	return fmt.Sprintf("v%s.%s.%s", v.major, v.minor, v.patch)
 }
 
 // FindClosestMatch returns the largest version smaller or equal to a given one.
@@ -76,11 +275,12 @@ func FindClosestMatch(v string, versions []string) string {
 	// obviously not the case. In lack of a better choice, we strip any such
 	// suffix from v before comparison.
 	v = stripGitSuffix(v)
-	semver.Sort(versions)
+
+	slices.SortFunc(versions, compareVersion)
 
 	latest := ""
 	for _, ver := range versions {
-		if semver.Compare(ver, v) > 0 {
+		if compareVersion(ver, v) > 0 {
 			break
 		}
 		latest = ver
@@ -92,26 +292,27 @@ func FindClosestMatch(v string, versions []string) string {
 // We expect a valid git suffix to be of the form "-N-gSHA1[.m], where
 // N is an decimal integer and SHA1 is a hexadecimal integer.
 func stripGitSuffix(version string) string {
-	mmp := majorMinorPatch(version)
-	pre := semver.Prerelease(version)
-	if mmp+pre != version {
+	pv, _ := parseVersion(version)
+	mmp := majorMinorPatch(pv)
+	if pv.String() != version {
 		return version
 	}
 
-	if len(pre) == 0 || pre[0] != '-' {
+	pre := pv.pre
+	if len(pre) == 0 {
 		return version
 	}
 
-	commits, gsha1, ok := strings.Cut(pre[1:], "-")
+	commits, gsha1, ok := strings.Cut(pre, "-")
 	if !ok || len(gsha1) == 0 || gsha1[0] != 'g' {
 		return version
 	}
-	if _, err := strconv.ParseInt(commits, 10, 64); err != nil {
+	if _, err := strconv.ParseUint(commits, 10, 64); err != nil {
 		return version
 	}
 
 	sha1, _, _ := strings.Cut(gsha1[1:], ".")
-	if _, err := strconv.ParseInt(sha1, 16, 64); err != nil {
+	if _, err := strconv.ParseUint(sha1, 16, 64); err != nil {
 		return version
 	}
 

vendor/modules.txt

@@ -235,7 +235,7 @@ github.com/containerd/log/logtest
 # github.com/containerd/log/otel v0.1.0
 ## explicit; go 1.23
 github.com/containerd/log/otel
-# github.com/containerd/nri v0.12.3
+# github.com/containerd/nri v0.12.5
 ## explicit; go 1.24.0
 github.com/containerd/nri
 github.com/containerd/nri/pkg/adaptation