runtime: make task.Delete API retriable

To make task.Delete retriable, this patch caches the delete result in
memory after the shim successfully deletes the task. If a later shutdown
attempt fails, a subsequent Delete retry can complete cleanup and return the
cached exit information.

If containerd restarts after deleting the task, the in-memory cache is lost,
and containerd will publish TaskExit and TaskDelete events again to
subscribers. Currently, the exit status returned by shim-delete is always
137. In a follow-up, containerd-shim-runc-v2 should store the actual exit
status in the state directory, so that shim-delete can reuse it instead of
returning a hard-coded value.

Implements: #8981

Signed-off-by: Wei Fu <fuweid89@gmail.com>

core/runtime/v2/shim.go

@@ -27,6 +27,7 @@ import (
 	"os"
 	"path/filepath"
 	"strings"
+	"sync/atomic"
 	"time"
 
 	"go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc"
@@ -152,9 +153,16 @@ func cleanupAfterDeadShim(ctx context.Context, id string, rt *runtime.NSMap[Shim
 		log.G(ctx).WithError(err).WithField("id", id).Warn("failed to clean up after shim disconnected")
 	}
 
-	if _, err := rt.Get(ctx, id); err != nil {
-		// Task was never started or was already successfully deleted
-		// No need to publish events
+	s, err := rt.Get(ctx, id)
+	if err != nil {
+		// Task was never started, or its record has already been removed.
+		// No need to publish events.
+		return
+	}
+
+	// If the task delete already succeeded, the shim itself has delivered the
+	// exit and delete events. No need to publish duplicates.
+	if s.(taskDeleteState).taskDeleteResult() != nil {
 		return
 	}
 
@@ -274,6 +282,14 @@ type clientVersionDowngrader interface {
 	Downgrade() error
 }
 
+// taskDeleteState caches the result of a successful task delete on the shim
+// instance, so that a Delete retried after a failed shutdown can still return
+// the original exit.
+type taskDeleteState interface {
+	recordTaskDeleteResult(*runtime.Exit)
+	taskDeleteResult() *runtime.Exit
+}
+
 func parseStartResponse(response []byte) (*bootapi.BootstrapResult, error) {
 	var result bootapi.BootstrapResult
 
@@ -469,11 +485,14 @@ type shim struct {
 	// rather than decoded into fields here so that a new capability needs no
 	// further plumbing through the shim instance.
 	bootstrap *bootapi.BootstrapResult
+
+	taskDeleteExit atomic.Pointer[runtime.Exit]
 }
 
 var _ ShimInstance = (*shim)(nil)
 var _ clientVersionDowngrader = (*shim)(nil)
 var _ shimCapabilities = (*shim)(nil)
+var _ taskDeleteState = (*shim)(nil)
 
 // BootstrapResult returns what the shim advertised when it started.
 func (s *shim) BootstrapResult() *bootapi.BootstrapResult {
@@ -554,6 +573,25 @@ func (s *shim) Delete(ctx context.Context) error {
 	return errors.Join(result...)
 }
 
+// recordTaskDeleteResult caches the result of a successful task delete. The
+// value is copied both in and out so that neither the caller that recorded it
+// nor a later retry can mutate the cached result.
+func (s *shim) recordTaskDeleteResult(exit *runtime.Exit) {
+	cached := *exit
+	s.taskDeleteExit.Store(&cached)
+}
+
+// taskDeleteResult returns a copy of the cached delete result, or nil if this
+// containerd process has never deleted the task successfully.
+func (s *shim) taskDeleteResult() *runtime.Exit {
+	cached := s.taskDeleteExit.Load()
+	if cached == nil {
+		return nil
+	}
+	exit := *cached
+	return &exit
+}
+
 var _ runtime.Task = &shimTask{}
 
 // shimTask wraps shim process and adds task service client for compatibility with existing shim manager.
@@ -617,6 +655,8 @@ func (s *shimTask) delete(ctx context.Context, removeTask func(ctx context.Conte
 		}
 	}
 
+	deleteState := s.ShimInstance.(taskDeleteState)
+
 	// NOTE: If the shim has been killed and ttrpc connection has been
 	// closed, the shimErr will not be nil. For this case, the event
 	// subscriber, like moby/moby, might have received the exit or delete
@@ -624,8 +664,8 @@ func (s *shimTask) delete(ctx context.Context, removeTask func(ctx context.Conte
 	// send the exit and delete events again. And the exit status will
 	// depend on result of shimV2.Delete.
 	//
-	// If not, the shim has been delivered the exit and delete events.
-	// So we should remove the record and prevent duplicate events from
+	// If not, the shim has delivered the exit and delete events. Cache the
+	// delete result so a retry can return it and prevent duplicate events from
 	// ttrpc-callback-on-close.
 	//
 	// TODO: It's hard to guarantee that the event is unique and sent only
@@ -633,16 +673,22 @@ func (s *shimTask) delete(ctx context.Context, removeTask func(ctx context.Conte
 	// only one exit event. The moby/moby should handle the duplicate events.
 	//
 	// REF: https://github.com/containerd/containerd/issues/4769
+	var exit *runtime.Exit
 	if shimErr == nil {
-		removeTask(ctx, s.ID())
+		exit = &runtime.Exit{
+			Status:    response.ExitStatus,
+			Timestamp: protobuf.FromTimestamp(response.ExitedAt),
+			Pid:       response.Pid,
+		}
+		deleteState.recordTaskDeleteResult(exit)
 	}
 
+	// NOTE: Returning here deliberately leaves the shim record, the ttrpc
+	// client and the bundle in place, so that the caller can retry Delete.
+	// The result cached above lets that retry return the original exit, even
+	// though the shim reports NotFound for the task by then.
 	if err := s.waitShutdown(ctx); err != nil {
-		// FIXME(fuweid):
-		//
-		// If the error is context canceled, should we use context.TODO()
-		// to wait for it?
-		log.G(ctx).WithField("id", s.ID()).WithError(err).Error("failed to shutdown shim task and the shim might be leaked")
+		return nil, fmt.Errorf("failed to invoke shutdown: %w", err)
 	}
 
 	if err := s.ShimInstance.Delete(ctx); err != nil {
@@ -653,15 +699,15 @@ func (s *shimTask) delete(ctx context.Context, removeTask func(ctx context.Conte
 	// this seems dirty but it cleans up the API across runtimes, tasks, and the service
 	removeTask(ctx, s.ID())
 
-	if shimErr != nil {
+	if exit == nil {
+		// An earlier attempt already deleted the task in the shim, so shimErr
+		// is NotFound. Return the exit recorded by that attempt instead.
+		exit = deleteState.taskDeleteResult()
+	}
+	if exit == nil {
 		return nil, shimErr
 	}
-
-	return &runtime.Exit{
-		Status:    response.ExitStatus,
-		Timestamp: protobuf.FromTimestamp(response.ExitedAt),
-		Pid:       response.Pid,
-	}, nil
+	return exit, nil
 }
 
 func (s *shimTask) Create(ctx context.Context, opts runtime.CreateOpts) (runtime.Task, error) {

core/runtime/v2/task_manager.go

@@ -320,14 +320,25 @@ func (m *TaskManager) Delete(ctx context.Context, taskID string) (*runtime.Exit,
 		m.manager.shims.Delete(ctx, id)
 	})
 
-	if err != nil {
+	// An ErrNotFound here means the shim has no record of the task and there
+	// was no cached delete result to fall back to. For example, the task was
+	// never created successfully in the shim, or a previous containerd process
+	// deleted it. The runtime side has still been cleaned up, so we should
+	// deactivate the mounts before returning the error.
+	if err != nil && !errdefs.IsNotFound(err) {
 		return nil, fmt.Errorf("failed to delete task: %w", err)
 	}
 
-	if err := m.taskMounts.Deactivate(ctx, taskID); err != nil && !errdefs.IsNotFound(err) {
-		log.G(ctx).WithError(err).WithField("task", taskID).Errorf("failed to deactivate mounts")
+	// FIXME(fuweid): It seems that cleaning this up is best-effort because
+	// GC can guarantee that the mount is deleted when the container is deleted.
+	// What if we reuse the container and restart the task?
+	if merr := m.taskMounts.Deactivate(ctx, taskID); merr != nil && !errdefs.IsNotFound(merr) {
+		log.G(ctx).WithError(merr).WithField("task", taskID).Errorf("failed to deactivate mounts")
 	}
 
+	if err != nil {
+		return nil, fmt.Errorf("failed to delete task: %w", err)
+	}
 	return exit, nil
 }
 

integration/issue7496_shutdown_linux_test.go

@@ -17,36 +17,37 @@
 package integration
 
 import (
-	"context"
+	"fmt"
 	"syscall"
 	"testing"
 	"time"
 
 	"github.com/containerd/errdefs"
+	"github.com/containerd/typeurl/v2"
 	"github.com/stretchr/testify/require"
 
+	eventtypes "github.com/containerd/containerd/api/events"
 	apitask "github.com/containerd/containerd/api/runtime/task/v3"
+	"github.com/containerd/containerd/v2/core/events"
+	ctrdruntime "github.com/containerd/containerd/v2/core/runtime"
 	"github.com/containerd/containerd/v2/integration/images"
 	"github.com/containerd/containerd/v2/pkg/namespaces"
+	ptypes "github.com/containerd/containerd/v2/pkg/protobuf/types"
 	runtime "k8s.io/cri-api/pkg/apis/runtime/v1"
 )
 
 // TestIssue7496_ShouldRetryShutdown is based on https://github.com/containerd/containerd/issues/7496.
 //
-// Assume that the shim.Delete takes almost 10 seconds and returns successfully
-// and there is no container in shim. However, the context is close to be
-// canceled. It will fail to call Shutdown. If we ignores the Canceled error,
-// the shim will be leaked. In order to reproduce this, this case will use
-// failpoint to inject error into Shutdown API, and then check whether the shim
-// is leaked.
+// The first task Delete succeeds, but an injected Shutdown failure prevents the
+// shim from exiting. The CRI retry then sees the task as NotFound and must still
+// shut down the shim without publishing duplicate exit or delete events.
 func TestIssue7496_ShouldRetryShutdown(t *testing.T) {
-	// TODO: re-enable if we can retry Shutdown API.
-	t.Skipf("Please re-enable me if we can retry Shutdown API")
+	const eventBarrierTopic = "/tests/issue7496/event-barrier"
 
-	ctx := namespaces.WithNamespace(context.Background(), "k8s.io")
+	ctx := namespaces.WithNamespace(t.Context(), "k8s.io")
 
 	t.Logf("Create a pod config with shutdown failpoint")
-	sbConfig := PodSandboxConfig("sandbox", "issue7496_shouldretryshutdown")
+	sbConfig := PodSandboxConfig("sandbox", "issue7496_shouldretryshutdown", WithHostNetwork)
 	injectShimFailpoint(t, sbConfig, map[string]string{
 		"Shutdown": "1*error(please retry)",
 	})
@@ -60,13 +61,135 @@ func TestIssue7496_ShouldRetryShutdown(t *testing.T) {
 
 	t.Logf("Log shim %s's pid: %d", sbID, shimPid(ctx, t, shimCli))
 
+	t.Logf("Subscribe task exit/delete events")
+	eventCh, eventErrCh := containerdClient.EventService().Subscribe(ctx,
+		fmt.Sprintf(`topic=="%s",event.container_id=="%s"`, ctrdruntime.TaskExitEventTopic, sbID),
+		fmt.Sprintf(`topic=="%s",event.container_id=="%s"`, ctrdruntime.TaskDeleteEventTopic, sbID),
+		fmt.Sprintf(`topic=="%s"`, eventBarrierTopic),
+	)
+
 	t.Logf("StopPodSandbox and RemovePodSandbox")
 	require.NoError(t, runtimeService.StopPodSandbox(sbID))
 	require.NoError(t, runtimeService.RemovePodSandbox(sbID))
 
 	t.Logf("Check the shim connection")
 	_, err = shimCli.Connect(ctx, &apitask.ConnectRequest{})
-	require.Error(t, err, "should failed to call shim connect API")
+	require.Error(t, err, "should fail to call the shim Connect API")
+	require.ErrorContains(t, err, "ttrpc: closed")
+
+	// The task delete succeeded on the first attempt, so the shim itself has
+	// already delivered the exit and delete events. When the retry finally
+	// shuts the shim down, ttrpc-callback-on-close must not publish them a
+	// second time.
+	//
+	// REF: https://github.com/containerd/containerd/issues/4769
+	t.Logf("Check that task exit/delete events are not duplicated")
+	require.NoError(t, containerdClient.EventService().Publish(ctx,
+		eventBarrierTopic, &ptypes.Empty{},
+	))
+	exits, deletes := countTaskExitDeleteEventsUntilBarrier(
+		t, sbID, eventBarrierTopic, eventCh, eventErrCh, 10*time.Second,
+	)
+	require.Equal(t, 1, exits, "task exit event should be published only once")
+	require.Equal(t, 1, deletes, "task delete event should be published only once")
+}
+
+func TestKillShimPublishesTaskExitAndDeleteEvents(t *testing.T) {
+	ctx := namespaces.WithNamespace(t.Context(), "k8s.io")
+
+	sbConfig := PodSandboxConfig("sandbox", t.Name(), WithHostNetwork)
+	sbID, err := runtimeService.RunPodSandbox(sbConfig, "")
+	require.NoError(t, err)
+	t.Cleanup(func() {
+		require.NoError(t, runtimeService.StopPodSandbox(sbID))
+		require.NoError(t, runtimeService.RemovePodSandbox(sbID))
+	})
+
+	shimCli := connectToShim(ctx, t, containerdEndpoint, 3, sbID)
+	shimPID := shimPid(ctx, t, shimCli)
+
+	eventCh, eventErrCh := containerdClient.EventService().Subscribe(ctx,
+		fmt.Sprintf(`topic=="%s",event.container_id=="%s"`, ctrdruntime.TaskExitEventTopic, sbID),
+		fmt.Sprintf(`topic=="%s",event.container_id=="%s"`, ctrdruntime.TaskDeleteEventTopic, sbID),
+	)
+
+	require.NoError(t, syscall.Kill(int(shimPID), syscall.SIGKILL))
+	waitForTaskExitDeleteEvents(t, sbID, eventCh, eventErrCh, 10*time.Second)
+}
+
+func waitForTaskExitDeleteEvents(t *testing.T, id string,
+	ch <-chan *events.Envelope, errCh <-chan error, timeout time.Duration,
+) {
+	t.Helper()
+
+	pending := map[string]struct{}{
+		ctrdruntime.TaskExitEventTopic:   {},
+		ctrdruntime.TaskDeleteEventTopic: {},
+	}
+	timer := time.NewTimer(timeout)
+	defer timer.Stop()
+
+	for len(pending) != 0 {
+		select {
+		case <-timer.C:
+			t.Fatalf("timed out waiting for task exit/delete events for %q", id)
+		case err, ok := <-errCh:
+			if !ok {
+				t.Fatalf("event subscription closed while waiting for %q", id)
+			}
+			require.NoError(t, err, "event subscription failed")
+		case env, ok := <-ch:
+			require.True(t, ok, "event subscription closed while waiting for %q", id)
+			t.Logf("received event %q for %q", env.Topic, id)
+			delete(pending, env.Topic)
+		}
+	}
+}
+
+// countTaskExitDeleteEventsUntilBarrier counts task exit and delete events for
+// the given container ID until the event stream reaches barrierTopic. The
+// barrier is published after shim deletion has waited for its close callback,
+// so any duplicate events from cleanupAfterDeadShim are ordered before it.
+func countTaskExitDeleteEventsUntilBarrier(t *testing.T, id, barrierTopic string,
+	ch <-chan *events.Envelope, errCh <-chan error, timeout time.Duration,
+) (exits, deletes int) {
+	t.Helper()
+
+	timer := time.NewTimer(timeout)
+	defer timer.Stop()
+
+	for {
+		select {
+		case <-timer.C:
+			t.Fatalf("timed out waiting for event barrier %q", barrierTopic)
+		case err, ok := <-errCh:
+			if !ok {
+				t.Fatalf("event subscription closed before barrier %q", barrierTopic)
+			}
+			require.NoError(t, err, "event subscription failed")
+		case env, ok := <-ch:
+			require.True(t, ok, "event subscription closed before barrier %q", barrierTopic)
+			if env.Topic == barrierTopic {
+				return exits, deletes
+			}
+
+			evt, err := typeurl.UnmarshalAny(env.Event)
+			require.NoError(t, err, "failed to unmarshal event")
+
+			switch e := evt.(type) {
+			case *eventtypes.TaskExit:
+				if e.ContainerID == id {
+					t.Logf("received task exit event: %+v", e)
+					exits++
+				}
+			case *eventtypes.TaskDelete:
+				if e.ContainerID == id {
+					t.Logf("received task delete event: %+v", e)
+					deletes++
+				}
+			}
+		}
+	}
 }
 
 func TestShutdownShimWhenPauseExitsBeforeWorkload(t *testing.T) {