snapshots/proxy: add optional default_timeout for proxy snapshotter calls

A proxy snapshotter is a separate process, and none of its 10 gRPC
methods carry a deadline unless the caller already set one. Snapshot GC
calls Walk and Remove under the snapshotter-level lock, so one wedged
call blocks every later mutating call on that snapshotter.

Add default_timeout to the proxy_plugins config section. When set, each
proxy snapshotter call gets that deadline if the incoming context has
none; a context that already carries its own deadline is left alone.
Empty or unset keeps today's behavior, so nothing changes for existing
configs.

NewSnapshotter keeps its signature and wraps the new
NewSnapshotterWithOpts, so existing callers of the exported constructor
are unaffected.

Assisted-by: Claude Code
Signed-off-by: Nahum Litvin <nahuml@wix.com>

cmd/containerd/server/config/config.go

@@ -404,6 +404,9 @@ type ProxyPlugin struct {
 	Platform     string            `toml:"platform"`
 	Exports      map[string]string `toml:"exports"`
 	Capabilities []string          `toml:"capabilities"`
+	// DefaultTimeout bounds each call to the plugin when the caller's context
+	// carries no deadline of its own. Empty leaves calls unbounded.
+	DefaultTimeout string `toml:"default_timeout"`
 }
 
 // Decode unmarshals a plugin specific configuration by plugin id

cmd/containerd/server/server.go

@@ -327,8 +327,16 @@ func LoadPlugins(ctx context.Context, config *srvconfig.Config) ([]plugin.Regist
 		case string(plugins.SnapshotPlugin), "snapshot":
 			t = plugins.SnapshotPlugin
 			ssname := name
+			var ssopts []ssproxy.Opt
+			if pp.DefaultTimeout != "" {
+				d, perr := time.ParseDuration(pp.DefaultTimeout)
+				if perr != nil {
+					return nil, fmt.Errorf("proxy plugin %q: unable to parse default_timeout %q into a time duration: %w", name, pp.DefaultTimeout, perr)
+				}
+				ssopts = append(ssopts, ssproxy.WithDefaultTimeout(d))
+			}
 			f = func(conn *grpc.ClientConn) any {
-				return ssproxy.NewSnapshotter(ssapi.NewSnapshotsClient(conn), ssname)
+				return ssproxy.NewSnapshotterWithOpts(ssapi.NewSnapshotsClient(conn), ssname, ssopts...)
 			}
 
 		case string(plugins.ContentPlugin), "content":

cmd/containerd/server/server_test.go

@@ -22,6 +22,7 @@ import (
 	"os"
 	"runtime"
 	"slices"
+	"strings"
 	"testing"
 
 	srvconfig "github.com/containerd/containerd/v2/cmd/containerd/server/config"
@@ -195,3 +196,25 @@ func TestSetTempDirEnv(t *testing.T) {
 		}
 	}
 }
+
+// A bad default_timeout must fail daemon startup rather than be ignored. Only
+// the error path is exercised here: LoadPlugins returns before it reaches the
+// global plugin registry, which panics if a test registers the same id twice.
+func TestLoadPluginsRejectsMalformedProxyDefaultTimeout(t *testing.T) {
+	cfg := &srvconfig.Config{
+		ProxyPlugins: map[string]srvconfig.ProxyPlugin{
+			"customsnapshot": {
+				Type:           "snapshot",
+				Address:        "/run/mysnapshotter.sock",
+				DefaultTimeout: "banana",
+			},
+		},
+	}
+	_, err := LoadPlugins(context.Background(), cfg)
+	if err == nil {
+		t.Fatal("LoadPlugins accepted a malformed default_timeout")
+	}
+	if !strings.Contains(err.Error(), "default_timeout") {
+		t.Fatalf("error does not name the offending key: %v", err)
+	}
+}

core/snapshots/proxy/proxy.go

@@ -19,6 +19,7 @@ package proxy
 import (
 	"context"
 	"io"
+	"time"
 
 	snapshotsapi "github.com/containerd/containerd/api/services/snapshots/v1"
 	"github.com/containerd/errdefs/pkg/errgrpc"
@@ -28,21 +29,56 @@ import (
 	protobuftypes "github.com/containerd/containerd/v2/pkg/protobuf/types"
 )
 
+// Opt configures a proxy snapshotter.
+type Opt func(*proxySnapshotter)
+
+// WithDefaultTimeout bounds each call to the proxy snapshotter when the
+// caller's context carries no deadline of its own. A non-positive duration
+// leaves calls unbounded.
+func WithDefaultTimeout(d time.Duration) Opt {
+	return func(p *proxySnapshotter) {
+		p.defaultTimeout = d
+	}
+}
+
 // NewSnapshotter returns a new Snapshotter which communicates over a GRPC
 // connection using the containerd snapshot GRPC API.
 func NewSnapshotter(client snapshotsapi.SnapshotsClient, snapshotterName string) snapshots.Snapshotter {
-	return &proxySnapshotter{
+	return NewSnapshotterWithOpts(client, snapshotterName)
+}
+
+// NewSnapshotterWithOpts is NewSnapshotter with options applied.
+func NewSnapshotterWithOpts(client snapshotsapi.SnapshotsClient, snapshotterName string, opts ...Opt) snapshots.Snapshotter {
+	p := &proxySnapshotter{
 		client:          client,
 		snapshotterName: snapshotterName,
 	}
+	for _, o := range opts {
+		o(p)
+	}
+	return p
 }
 
 type proxySnapshotter struct {
 	client          snapshotsapi.SnapshotsClient
 	snapshotterName string
+	defaultTimeout  time.Duration
+}
+
+func (p *proxySnapshotter) withTimeout(ctx context.Context) (context.Context, context.CancelFunc) {
+	if p.defaultTimeout <= 0 {
+		return ctx, func() {}
+	}
+	if _, ok := ctx.Deadline(); ok {
+		return ctx, func() {}
+	}
+	return context.WithTimeout(ctx, p.defaultTimeout)
 }
 
 func (p *proxySnapshotter) Stat(ctx context.Context, key string) (snapshots.Info, error) {
+	ctx, cancel := p.withTimeout(ctx)
+	defer cancel()
+
 	resp, err := p.client.Stat(ctx,
 		&snapshotsapi.StatSnapshotRequest{
 			Snapshotter: p.snapshotterName,
@@ -55,6 +91,9 @@ func (p *proxySnapshotter) Stat(ctx context.Context, key string) (snapshots.Info
 }
 
 func (p *proxySnapshotter) Update(ctx context.Context, info snapshots.Info, fieldpaths ...string) (snapshots.Info, error) {
+	ctx, cancel := p.withTimeout(ctx)
+	defer cancel()
+
 	resp, err := p.client.Update(ctx,
 		&snapshotsapi.UpdateSnapshotRequest{
 			Snapshotter: p.snapshotterName,
@@ -70,6 +109,9 @@ func (p *proxySnapshotter) Update(ctx context.Context, info snapshots.Info, fiel
 }
 
 func (p *proxySnapshotter) Usage(ctx context.Context, key string) (snapshots.Usage, error) {
+	ctx, cancel := p.withTimeout(ctx)
+	defer cancel()
+
 	resp, err := p.client.Usage(ctx, &snapshotsapi.UsageRequest{
 		Snapshotter: p.snapshotterName,
 		Key:         key,
@@ -81,6 +123,9 @@ func (p *proxySnapshotter) Usage(ctx context.Context, key string) (snapshots.Usa
 }
 
 func (p *proxySnapshotter) Mounts(ctx context.Context, key string) ([]mount.Mount, error) {
+	ctx, cancel := p.withTimeout(ctx)
+	defer cancel()
+
 	resp, err := p.client.Mounts(ctx, &snapshotsapi.MountsRequest{
 		Snapshotter: p.snapshotterName,
 		Key:         key,
@@ -92,6 +137,9 @@ func (p *proxySnapshotter) Mounts(ctx context.Context, key string) ([]mount.Moun
 }
 
 func (p *proxySnapshotter) Prepare(ctx context.Context, key, parent string, opts ...snapshots.Opt) ([]mount.Mount, error) {
+	ctx, cancel := p.withTimeout(ctx)
+	defer cancel()
+
 	var local snapshots.Info
 	for _, opt := range opts {
 		if err := opt(&local); err != nil {
@@ -111,6 +159,9 @@ func (p *proxySnapshotter) Prepare(ctx context.Context, key, parent string, opts
 }
 
 func (p *proxySnapshotter) View(ctx context.Context, key, parent string, opts ...snapshots.Opt) ([]mount.Mount, error) {
+	ctx, cancel := p.withTimeout(ctx)
+	defer cancel()
+
 	var local snapshots.Info
 	for _, opt := range opts {
 		if err := opt(&local); err != nil {
@@ -130,6 +181,9 @@ func (p *proxySnapshotter) View(ctx context.Context, key, parent string, opts ..
 }
 
 func (p *proxySnapshotter) Commit(ctx context.Context, name, key string, opts ...snapshots.Opt) error {
+	ctx, cancel := p.withTimeout(ctx)
+	defer cancel()
+
 	var local snapshots.Info
 	for _, opt := range opts {
 		if err := opt(&local); err != nil {
@@ -147,6 +201,9 @@ func (p *proxySnapshotter) Commit(ctx context.Context, name, key string, opts ..
 }
 
 func (p *proxySnapshotter) Remove(ctx context.Context, key string) error {
+	ctx, cancel := p.withTimeout(ctx)
+	defer cancel()
+
 	_, err := p.client.Remove(ctx, &snapshotsapi.RemoveSnapshotRequest{
 		Snapshotter: p.snapshotterName,
 		Key:         key,
@@ -155,6 +212,9 @@ func (p *proxySnapshotter) Remove(ctx context.Context, key string) error {
 }
 
 func (p *proxySnapshotter) Walk(ctx context.Context, fn snapshots.WalkFunc, fs ...string) error {
+	ctx, cancel := p.withTimeout(ctx)
+	defer cancel()
+
 	sc, err := p.client.List(ctx, &snapshotsapi.ListSnapshotsRequest{
 		Snapshotter: p.snapshotterName,
 		Filters:     fs,
@@ -186,6 +246,9 @@ func (p *proxySnapshotter) Close() error {
 }
 
 func (p *proxySnapshotter) Cleanup(ctx context.Context) error {
+	ctx, cancel := p.withTimeout(ctx)
+	defer cancel()
+
 	_, err := p.client.Cleanup(ctx, &snapshotsapi.CleanupRequest{
 		Snapshotter: p.snapshotterName,
 	})

core/snapshots/proxy/proxy_test.go

@@ -0,0 +1,109 @@
+/*
+   Copyright The containerd Authors.
+
+   Licensed under the Apache License, Version 2.0 (the "License");
+   you may not use this file except in compliance with the License.
+   You may obtain a copy of the License at
+
+       http://www.apache.org/licenses/LICENSE-2.0
+
+   Unless required by applicable law or agreed to in writing, software
+   distributed under the License is distributed on an "AS IS" BASIS,
+   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+   See the License for the specific language governing permissions and
+   limitations under the License.
+*/
+
+package proxy
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	snapshotsapi "github.com/containerd/containerd/api/services/snapshots/v1"
+	"google.golang.org/grpc"
+	"google.golang.org/protobuf/types/known/emptypb"
+)
+
+func TestWithTimeout(t *testing.T) {
+	t.Run("disabled", func(t *testing.T) {
+		p := &proxySnapshotter{}
+		ctx, cancel := p.withTimeout(context.Background())
+		defer cancel()
+		if _, ok := ctx.Deadline(); ok {
+			t.Fatal("expected no deadline when defaultTimeout is unset")
+		}
+	})
+
+	t.Run("applied", func(t *testing.T) {
+		p := &proxySnapshotter{defaultTimeout: time.Minute}
+		ctx, cancel := p.withTimeout(context.Background())
+		defer cancel()
+		dl, ok := ctx.Deadline()
+		if !ok {
+			t.Fatal("expected a deadline")
+		}
+		if d := time.Until(dl); d <= 0 || d > time.Minute {
+			t.Fatalf("deadline %v is not within the default timeout", d)
+		}
+	})
+
+	t.Run("caller deadline kept", func(t *testing.T) {
+		p := &proxySnapshotter{defaultTimeout: time.Minute}
+		want := time.Now().Add(2 * time.Hour)
+		parent, cancelParent := context.WithDeadline(context.Background(), want)
+		defer cancelParent()
+
+		ctx, cancel := p.withTimeout(parent)
+		defer cancel()
+		got, ok := ctx.Deadline()
+		if !ok {
+			t.Fatal("expected a deadline")
+		}
+		if !got.Equal(want) {
+			t.Fatalf("deadline %v was shortened, want %v", got, want)
+		}
+	})
+}
+
+// blockedClient blocks on Remove until its context is done. Other methods are
+// unimplemented and panic if called.
+type blockedClient struct {
+	snapshotsapi.SnapshotsClient
+}
+
+func (blockedClient) Remove(ctx context.Context, _ *snapshotsapi.RemoveSnapshotRequest, _ ...grpc.CallOption) (*emptypb.Empty, error) {
+	<-ctx.Done()
+	return nil, ctx.Err()
+}
+
+func TestRemoveAppliesDefaultTimeout(t *testing.T) {
+	s := NewSnapshotterWithOpts(blockedClient{}, "test", WithDefaultTimeout(50*time.Millisecond))
+
+	type result struct {
+		err     error
+		elapsed time.Duration
+	}
+	done := make(chan result, 1)
+	go func() {
+		start := time.Now()
+		err := s.Remove(context.Background(), "key")
+		done <- result{err, time.Since(start)}
+	}()
+
+	select {
+	case got := <-done:
+		if got.err == nil {
+			t.Fatal("Remove returned no error against a blocked client")
+		}
+		// A generous bound still catches a timeout that ignores the configured
+		// 50ms and uses some other value. errgrpc.ToNative flattens the error
+		// to an opaque errdefs value, so the duration is the signal here.
+		if got.elapsed > time.Second {
+			t.Fatalf("Remove returned after %v, want close to the 50ms default_timeout", got.elapsed)
+		}
+	case <-time.After(5 * time.Second):
+		t.Fatal("Remove did not return, so the default timeout was not applied")
+	}
+}

docs/PLUGINS.md

@@ -69,8 +69,15 @@ version = 2
   [proxy_plugins.customsnapshot]
     type = "snapshot"
     address = "/var/run/mysnapshotter.sock"
+    default_timeout = ""
 ```
 
+`default_timeout` is optional and honored by `snapshot` proxy plugins today. When
+set to a duration != "", for example to "30m", each call to the plugin gets that deadline if the caller has not set one of
+its own, so a stuck plugin cannot hold its lock for good. A caller that already
+has a deadline keeps it. Leaving the key out, or giving it a zero or negative
+duration, keeps calls unbounded.
+
 #### Implementation
 
 Implementing a proxy plugin is as easy as implementing the gRPC API for a

docs/man/containerd-config.toml.5.md

@@ -170,6 +170,9 @@ documentation.
 
 - **type** (Default: "")
 - **address** (Default: "")
+- **default_timeout** (Default: "") a duration bounding each call to the plugin
+  when the caller's context carries no deadline of its own, for example "30m".
+  Empty leaves calls unbounded unless the caller's context carries a deadline of its own. Currently honored by snapshot proxy plugins.
 
 **timeouts**
 : Timeouts specified as a duration