fix(recovery): wait for concurrent disposable container removal

runners/dockerrun/disposable_recovery.go

@@ -5,11 +5,13 @@ import (
 	"errors"
 	"fmt"
 	"strconv"
+	"strings"
 	"time"
 
 	"github.com/containerd/errdefs"
 	"github.com/docker/docker/api/types/container"
 	"github.com/docker/docker/api/types/filters"
+	"github.com/docker/docker/client"
 
 	"github.com/codefly-dev/core/runners/base"
 	"github.com/codefly-dev/core/wool"
@@ -59,7 +61,7 @@ func ReapDisposableContainers(ctx context.Context, scope ContainerRecoveryScope)
 			continue
 		}
 		removeCtx, removeCancel := context.WithTimeout(ctx, 10*time.Second)
-		err := cli.ContainerRemove(removeCtx, c.ID, container.RemoveOptions{Force: true})
+		err := removeDisposableContainer(removeCtx, cli, c.ID)
 		removeCancel()
 		if err != nil && !errdefs.IsNotFound(err) {
 			failures = append(failures, fmt.Errorf("recover disposable container %s: %w", c.ID, err))
@@ -68,6 +70,34 @@ func ReapDisposableContainers(ctx context.Context, scope ContainerRecoveryScope)
 	return errors.Join(append(failures, ctx.Err())...)
 }
 
+func removeDisposableContainer(ctx context.Context, cli client.APIClient, id string) error {
+	for {
+		if err := ctx.Err(); err != nil {
+			return err
+		}
+		err := cli.ContainerRemove(ctx, id, container.RemoveOptions{Force: true})
+		if err == nil || errdefs.IsNotFound(err) {
+			return nil
+		}
+		// Docker reports this specific conflict while another remover owns the
+		// operation. A conflict alone does not prove removal is underway.
+		if !removalInProgress(err, id) {
+			return err
+		}
+		timer := time.NewTimer(100 * time.Millisecond)
+		select {
+		case <-ctx.Done():
+			timer.Stop()
+			return fmt.Errorf("waiting for concurrent container removal: %w", ctx.Err())
+		case <-timer.C:
+		}
+	}
+}
+
+func removalInProgress(err error, id string) bool {
+	return errdefs.IsConflict(err) && strings.HasSuffix(err.Error(), fmt.Sprintf("removal of container %s is already in progress", id))
+}
+
 // disposableContainerInNamespace states the cross-scope rules directly rather
 // than borrowing staleContainerInScope with a scope built from the container's
 // own labels, which made that predicate's exact-scope comparison tautological

runners/dockerrun/disposable_remove_integration_test.go

@@ -0,0 +1,46 @@
+//go:build !skip_infra
+
+package dockerrun
+
+import (
+	"context"
+	"sync"
+	"testing"
+
+	"github.com/containerd/errdefs"
+	"github.com/docker/docker/api/types/container"
+	"github.com/stretchr/testify/require"
+)
+
+// Only containers created by this test are removed; this never runs a sweep.
+func TestRemoveDisposableContainerConcurrentRemoval(t *testing.T) {
+	backend := newBackend(t)
+	for range 10 {
+		created, err := backend.client.ContainerCreate(t.Context(), &container.Config{Image: "alpine:latest", Cmd: []string{"sleep", "300"}}, &container.HostConfig{NetworkMode: "none"}, nil, nil, uniqueName(t))
+		require.NoError(t, err)
+		id := created.ID
+		t.Cleanup(func() {
+			_ = backend.client.ContainerRemove(context.Background(), id, container.RemoveOptions{Force: true})
+		})
+		require.NoError(t, backend.client.ContainerStart(t.Context(), id, container.StartOptions{}))
+		cancelled, cancel := context.WithCancel(t.Context())
+		cancel()
+		require.ErrorIs(t, removeDisposableContainer(cancelled, backend.client, id), context.Canceled)
+		_, err = backend.client.ContainerInspect(t.Context(), id)
+		require.NoError(t, err)
+		start := make(chan struct{})
+		errs := make(chan error, 2)
+		var ready sync.WaitGroup
+		ready.Add(2)
+		for range 2 {
+			go func() { ready.Done(); <-start; errs <- removeDisposableContainer(t.Context(), backend.client, id) }()
+		}
+		ready.Wait()
+		close(start)
+		first, second := <-errs, <-errs
+		require.NoError(t, first)
+		require.NoError(t, second)
+		_, err = backend.client.ContainerInspect(t.Context(), id)
+		require.True(t, errdefs.IsNotFound(err), "%v", err)
+	}
+}

runners/dockerrun/disposable_remove_test.go

@@ -0,0 +1,19 @@
+package dockerrun
+
+import (
+	"errors"
+	"fmt"
+	"testing"
+
+	dockererrors "github.com/docker/docker/errdefs"
+	"github.com/stretchr/testify/require"
+)
+
+func TestRemovalInProgressRequiresSpecificConflict(t *testing.T) {
+	const id = "abc123"
+	message := fmt.Sprintf("removal of container %s is already in progress", id)
+	require.True(t, removalInProgress(dockererrors.Conflict(errors.New("Error response from daemon: "+message)), id))
+	require.False(t, removalInProgress(errors.New(message), id))
+	require.False(t, removalInProgress(dockererrors.Conflict(errors.New("other conflict")), id))
+	require.False(t, removalInProgress(dockererrors.Conflict(errors.New("removal of container another is already in progress")), id))
+}