fix(run): a failed run removes the containers its own agents created

cmd/run/service.go

@@ -92,9 +92,15 @@ func runServiceCommand(cmd *cobra.Command, args []string) (returnErr error) {
 	// but `kill` sends SIGTERM, which fell through and orphaned every plugin.
 	ctx, stop := common.SignalContext(ctx)
 	defer stop()
+	// signalled tells a run stopped on purpose apart from one that failed.
+	signalled := ctx
+	started := time.Now()
 	ctx, cancelRun := context.WithCancel(ctx)
 	defer cancelRun()
 
+	// cleanupRun is the run once it exists, for the deferred cleanup below,
+	// which is registered before it.
+	var cleanupRun *serviceRun
 	cli.Init()
 	defer func() {
 		services.ClearAgents()
@@ -108,6 +114,9 @@ func runServiceCommand(cmd *cobra.Command, args []string) (returnErr error) {
 		} else if recovered := evidence.RecoveredPGIDs(); len(recovered) > 0 {
 			cli.Info("reaped %d managed process groups after run: %v", len(recovered), recovered)
 		}
+		if returnErr != nil && signalled.Err() == nil && cleanupRun != nil {
+			returnErr = errors.Join(returnErr, removeFailedRunContainers(context.WithoutCancel(ctx), cleanupRun.flow, started))
+		}
 	}()
 
 	namingScopeExplicit = cmd.Flags().Changed("naming-scope")
@@ -195,6 +204,7 @@ func runServiceCommand(cmd *cobra.Command, args []string) (returnErr error) {
 	derivedOverrides = derived.Overrides
 
 	run := &serviceRun{workspace: workspace, module: module, service: service, serviceName: serviceName}
+	cleanupRun = run
 
 	var serverResult chan error
 	defer func() {
@@ -629,6 +639,26 @@ func (run *serviceRun) interactive(ctx context.Context) (runErr error) {
 	return runErr
 }
 
+// removeFailedRunContainers takes back the containers a failed run's own
+// agents created, once they are gone: they hold no dev data worth reusing,
+// and they hold the ports the next run needs (#1233). A run stopped on
+// purpose never calls it, so the next run still reuses what that run left.
+func removeFailedRunContainers(ctx context.Context, flow *orchestration.Flow, started time.Time) error {
+	if flow == nil || !slices.ContainsFunc(flow.SelectableRuntimeContexts(), shouldSweepStaleContainers) {
+		return nil
+	}
+	removeCtx, cancel := context.WithTimeout(ctx, 30*time.Second)
+	defer cancel()
+	removed, err := flow.RemoveContainersCreatedByThisRun(removeCtx, started)
+	if len(removed) > 0 {
+		cli.Info("removed %d containers the failed run started: %v", len(removed), removed)
+	}
+	if err != nil {
+		return fmt.Errorf("remove the failed run's containers: %w", err)
+	}
+	return nil
+}
+
 // shouldSweepStaleContainers keeps explicit Local/Nix runs independent of the
 // Docker control plane. A free run may select Docker and an explicit container
 // run requires it, so those startup paths retain orphan cleanup.

pkg/orchestration/flow.go

@@ -244,6 +244,9 @@ type World struct {
 	// Explicit disposable output policy, shared by all origin/dependency builders.
 	disposableImageScope string
 
+	// agents is every agent pid this run spawned (run_containers.go).
+	agents spawnedAgents
+
 	// beforeSourceRestart validates the accepted route inventory before stopping a service.
 	beforeSourceRestart func(context.Context) error
 

pkg/orchestration/manager.go

@@ -225,6 +225,7 @@ func (manager *Manager) Load(ctx context.Context) error {
 	}
 
 	w.Debug("load agent", wool.Field("agent-pid", instance.AgentPID))
+	manager.world.agents.add(instance.AgentPID)
 	if err = manager.validateContainerRecovery(instance); err != nil {
 		return w.Wrap(err)
 	}

pkg/orchestration/run_containers.go

@@ -0,0 +1,132 @@
+package orchestration
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"strconv"
+	"strings"
+	"sync"
+	"time"
+
+	"github.com/codefly-dev/core/runners/dockerrun"
+	"github.com/docker/docker/api/types/container"
+
+	"github.com/codefly-dev/cli/pkg/processgroup"
+)
+
+// A run that fails takes back the containers its own agents created (#1233).
+// Containers outlive a run on purpose: a running one whose owner is gone is
+// kept so the next run reuses its dev data (dockerrun.shouldReapContainer).
+// A run that failed holds no such data, and its half-started containers hold
+// the ports the next run needs.
+//
+// What a run created is told apart with no new label: each container carries
+// the pid of the agent that created it (codefly.session), the run records the
+// pid of every agent it spawned, and a container created before the run
+// started is never its own, whatever pid it carries.
+
+// ownedLabel is the value Core sets codefly.owner to on every container.
+const ownedLabel = "true"
+
+// spawnedAgents is every agent pid one run spawned, recorded as each agent
+// loads, so a manager that fails after its agent started still counts.
+type spawnedAgents struct {
+	mu   sync.Mutex
+	pids map[int]bool
+}
+
+func (s *spawnedAgents) add(pid int) {
+	if pid <= 0 {
+		return
+	}
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	if s.pids == nil {
+		s.pids = map[int]bool{}
+	}
+	s.pids[pid] = true
+}
+
+func (s *spawnedAgents) has(pid int) bool {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	return s.pids[pid]
+}
+
+// createdByRun reports whether c is a container this run's agents created:
+// codefly-owned, labelled with one of the run's agent pids, created no
+// earlier than the run started (a recycled pid on an adopted container does
+// not count), and not a ledgered data container, which no sweep removes.
+func createdByRun(c *container.Summary, agents *spawnedAgents, started time.Time) bool {
+	if c.Labels[dockerrun.LabelCodeflyOwner] != ownedLabel || c.Labels[dockerrun.LabelCodeflyInvocation] != "" {
+		return false
+	}
+	pid, err := strconv.Atoi(c.Labels[dockerrun.LabelCodeflySession])
+	if err != nil || !agents.has(pid) {
+		return false
+	}
+	return c.Created >= started.Unix()
+}
+
+// removeRunContainer force-removes one container. It is a variable so a test
+// can stand in for the Docker engine, as listOwnedContainers is.
+var removeRunContainer = func(ctx context.Context, id string) error {
+	cli, err := dockerrun.NewClient()
+	if err != nil {
+		return err
+	}
+	defer cli.Close()
+	removeCtx, cancel := context.WithTimeout(ctx, 10*time.Second)
+	defer cancel()
+	return cli.ContainerRemove(removeCtx, id, container.RemoveOptions{Force: true})
+}
+
+// otherRunsLive reports whether another run's managed process groups are
+// live in the run's workspace. Nothing records that a run adopted a container
+// another created, so while one is live this run removes nothing. A group
+// whose ownership cannot be decided is not counted: it is no live run.
+var otherRunsLive = func(ctx context.Context, world *World) bool {
+	groups, _ := processgroup.PreviewManagedProcessGroups(ctx, processgroup.InWorkspace(world.Workspace.Dir()))
+	return len(groups) > 0
+}
+
+// RemoveContainersCreatedByThisRun removes the containers this run's agents
+// created since started, in the run's recovery scope, once its agents are
+// gone; it answers their names. Call it only for a run that failed: a run
+// stopped on purpose keeps them for the next run to reuse.
+func (flow *Flow) RemoveContainersCreatedByThisRun(ctx context.Context, started time.Time) ([]string, error) {
+	scopeID, _, ok := strings.Cut(flow.containerRecoveryIdentity, ":")
+	if !ok || scopeID == "" || flow.world == nil || flow.world.Workspace == nil {
+		return nil, nil
+	}
+	if otherRunsLive(ctx, flow.world) {
+		return nil, nil
+	}
+	listed, err := listOwnedContainers(ctx, scopeID)
+	if err != nil {
+		return nil, fmt.Errorf("list the run's containers: %w", err)
+	}
+	var removed []string
+	var failures []error
+	for i := range listed {
+		c := &listed[i]
+		if !createdByRun(c, &flow.world.agents, started) {
+			continue
+		}
+		name := strings.TrimPrefix(firstName(c), "/")
+		if err := removeRunContainer(ctx, c.ID); err != nil {
+			failures = append(failures, fmt.Errorf("remove %s: %w", name, err))
+			continue
+		}
+		removed = append(removed, name)
+	}
+	return removed, errors.Join(failures...)
+}
+
+func firstName(c *container.Summary) string {
+	if len(c.Names) > 0 {
+		return c.Names[0]
+	}
+	return c.ID
+}

pkg/orchestration/run_containers_test.go

@@ -0,0 +1,114 @@
+package orchestration
+
+import (
+	"context"
+	"slices"
+	"strconv"
+	"testing"
+	"time"
+
+	"github.com/codefly-dev/core/resources"
+	"github.com/codefly-dev/core/runners/dockerrun"
+	"github.com/docker/docker/api/types/container"
+)
+
+// summary is a container as the Docker engine lists it: created at created,
+// labelled by the agent with pid, and ledgered when invocation is set.
+func summary(id string, pid int, created time.Time, invocation string) container.Summary {
+	labels := map[string]string{
+		dockerrun.LabelCodeflyOwner:   "true",
+		dockerrun.LabelCodeflySession: strconv.Itoa(pid),
+	}
+	if invocation != "" {
+		labels[dockerrun.LabelCodeflyInvocation] = invocation
+	}
+	return container.Summary{ID: id, Names: []string{"/" + id}, Created: created.Unix(), Labels: labels}
+}
+
+func TestOnlyContainersThisRunsAgentsCreatedAreItsOwn(t *testing.T) {
+	started := time.Unix(1_800_000_000, 0)
+	var agents spawnedAgents
+	agents.add(4100)
+	agents.add(4200)
+	for name, tc := range map[string]struct {
+		c    container.Summary
+		want bool
+	}{
+		"created by one of this run's agents":     {summary("ours", 4100, started.Add(time.Second), ""), true},
+		"created in the run's first second":       {summary("ours-early", 4200, started, ""), true},
+		"by another agent":                        {summary("theirs", 5100, started.Add(time.Second), ""), false},
+		"before the run started (a recycled pid)": {summary("adopted", 4100, started.Add(-time.Hour), ""), false},
+		"a ledgered (data) container of this run": {summary("ledger", 4100, started.Add(time.Second), "inv-1"), false},
+		"with no owner label":                     {container.Summary{ID: "bare", Created: started.Add(time.Second).Unix(), Labels: map[string]string{dockerrun.LabelCodeflySession: "4100"}}, false},
+		"with a session label that is not a pid":  {summary("odd", 0, started.Add(time.Second), ""), false},
+	} {
+		if tc.c.ID == "odd" {
+			tc.c.Labels[dockerrun.LabelCodeflySession] = "cli"
+		}
+		if got := createdByRun(&tc.c, &agents, started); got != tc.want {
+			t.Errorf("%s: createdByRun = %v, want %v", name, got, tc.want)
+		}
+	}
+}
+
+// failedRunFlow is a flow whose run started at started, spawned agents pids,
+// and resolved one recovery scope; the engine and the process-group view are
+// stood in for, and every removal is recorded.
+func failedRunFlow(t *testing.T, started time.Time, listed []container.Summary, otherRunLive bool, pids ...int) (*Flow, *[]string) {
+	t.Helper()
+	world := &World{Workspace: &resources.Workspace{}}
+	for _, pid := range pids {
+		world.agents.add(pid)
+	}
+	flow := &Flow{world: world, containerRecoveryIdentity: "scope-1:namespace-1"}
+	previousList, previousRemove, previousLive := listOwnedContainers, removeRunContainer, otherRunsLive
+	t.Cleanup(func() {
+		listOwnedContainers, removeRunContainer, otherRunsLive = previousList, previousRemove, previousLive
+	})
+	listOwnedContainers = func(_ context.Context, scopeID string) ([]container.Summary, error) {
+		if scopeID != "scope-1" {
+			t.Errorf("listed scope %q, want the run's", scopeID)
+		}
+		return listed, nil
+	}
+	removed := &[]string{}
+	removeRunContainer = func(_ context.Context, id string) error {
+		*removed = append(*removed, id)
+		return nil
+	}
+	otherRunsLive = func(context.Context, *World) bool { return otherRunLive }
+	return flow, removed
+}
+
+// A run that failed takes back the containers its own agents created, and
+// only those: what it adopted, what another run made and its data are kept.
+func TestAFailedRunRemovesOnlyTheContainersItCreated(t *testing.T) {
+	started := time.Unix(1_800_000_000, 0)
+	listed := []container.Summary{
+		summary("minio", 4100, started.Add(2*time.Second), ""),
+		summary("gateway", 4100, started.Add(3*time.Second), ""),
+		summary("adopted", 4100, started.Add(-time.Hour), ""),
+		summary("theirs", 5100, started.Add(time.Second), ""),
+		summary("ledger", 4200, started.Add(time.Second), "inv-1"),
+	}
+	flow, removed := failedRunFlow(t, started, listed, false, 4100, 4200)
+	got, err := flow.RemoveContainersCreatedByThisRun(context.Background(), started)
+	if err != nil {
+		t.Fatal(err)
+	}
+	slices.Sort(*removed)
+	if !slices.Equal(*removed, []string{"gateway", "minio"}) || len(got) != 2 {
+		t.Fatalf("removed %v (reported %v), want only the run's own minio and gateway", *removed, got)
+	}
+}
+
+// While another run is live in the workspace it may have adopted what this
+// one created; nothing records that, so nothing is removed.
+func TestAFailedRunKeepsItsContainersWhileAnotherRunIsLive(t *testing.T) {
+	started := time.Unix(1_800_000_000, 0)
+	flow, removed := failedRunFlow(t, started, []container.Summary{summary("minio", 4100, started.Add(time.Second), "")}, true, 4100)
+	got, err := flow.RemoveContainersCreatedByThisRun(context.Background(), started)
+	if err != nil || len(*removed) != 0 || len(got) != 0 {
+		t.Fatalf("removed %v (reported %v, err %v), want nothing while another run is live", *removed, got, err)
+	}
+}