From efb44283372518d03e30b91fc16f94f192e0604d Mon Sep 17 00:00:00 2001 From: mark wang Date: Thu, 29 Oct 2026 13:51:14 -0400 Subject: [PATCH] fix(run): a failed run removes the containers its own agents created A codefly run that fails partway leaves its half-started containers running, and they hold the ports the next run needs (#1233, part B). A failed run now force-removes the containers its own agents created since it started, in its recovery scope. A run stopped on purpose keeps them, as before, so the next run reuses their dev data. A container counts as the run's own only when it is codefly-owned, carries one of the run's agent pids as its session, was created after the run started, and is not a ledgered data container. While another run is live in the workspace nothing is removed, since nothing records that it adopted one of these containers. Co-Authored-By: Claude Opus 5.5 --- cmd/run/service.go | 30 ++++++ pkg/orchestration/flow.go | 3 + pkg/orchestration/manager.go | 1 + pkg/orchestration/run_containers.go | 132 +++++++++++++++++++++++ pkg/orchestration/run_containers_test.go | 114 ++++++++++++++++++++ 5 files changed, 280 insertions(+) create mode 100644 pkg/orchestration/run_containers.go create mode 100644 pkg/orchestration/run_containers_test.go diff --git a/cmd/run/service.go b/cmd/run/service.go index 976deda7..44c4ab20 100644 --- a/cmd/run/service.go +++ b/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. diff --git a/pkg/orchestration/flow.go b/pkg/orchestration/flow.go index e11c38a6..0535726b 100644 --- a/pkg/orchestration/flow.go +++ b/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 diff --git a/pkg/orchestration/manager.go b/pkg/orchestration/manager.go index 7d16b5c4..da91083a 100644 --- a/pkg/orchestration/manager.go +++ b/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) } diff --git a/pkg/orchestration/run_containers.go b/pkg/orchestration/run_containers.go new file mode 100644 index 00000000..271919fe --- /dev/null +++ b/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 +} diff --git a/pkg/orchestration/run_containers_test.go b/pkg/orchestration/run_containers_test.go new file mode 100644 index 00000000..121eced1 --- /dev/null +++ b/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) + } +}