feat: reconciler plans the purge of stale pre_start hook runners

pre_start hooks run in ephemeral containers. When a run fails, the
runner is deliberately retained for post-mortem inspection, and the next
run purges it — but that purge lived entirely inside the imperative
primitive, invisible to the reconciliation plan. It is now a plan
operation: when pre_start is going to run again (hooks declared, no
replica running at observation — the imperative gating), the plan emits
one best-effort RemoveContainer per stale runner, dropping its anonymous
volumes, exactly the warn-only semantics of the imperative purge that
remains in place as backstop.

Observed state learns to tell hook containers apart: they carry no
container-number label and previously classified as a service replica
numbered 0. They now land in a dedicated HookContainers bucket the
reconciler plans purges from.

The imperative primitive is also split into its lifecycle-free execution
piece (execPreStartHook: start, wait, log streaming, retain-on-failure)
and the create/remove pieces around it, recomposed identically in
runPreStart — locked by the existing characterization tests. This
prepares moving hook-container creation and post-success removal into
the plan once the start phase lands (#14200).

Signed-off-by: Nicolas De Loof <[email protected]>

pkg/compose/executor_ops.go

@@ -142,8 +142,15 @@ func (exec *planExecutor) execStopContainer(ctx context.Context, op Operation) e
 }
 
 func (exec *planExecutor) execRemoveContainer(ctx context.Context, op Operation) error {
-	_, err := exec.compose.apiClient().ContainerRemove(ctx, op.Container.ID, client.ContainerRemoveOptions{Force: true})
+	_, err := exec.compose.apiClient().ContainerRemove(ctx, op.Container.ID, client.ContainerRemoveOptions{Force: true, RemoveVolumes: op.RemoveVolumes})
 	if err != nil {
+		if op.BestEffort {
+			// warn-only removal (stale pre_start hook runner): the container
+			// stays visible to the operator, the plan carries on — and the
+			// live view below keeps it, since it was not removed
+			logrus.Warnf("failed to remove %s: %v", op.ResourceID, err)
+			return nil
+		}
 		return err
 	}
 	// Why: a dependent service's create may resolve `network_mode: service:X`

pkg/compose/executor_test.go

@@ -174,6 +174,38 @@ func emptyObservedState(project string) *ObservedState {
 // Goes through newPlanExecutor + run (i.e. the same code path executePlan
 // uses in production) so the test exercises the errgroup, done-channel
 // wiring and group tracker — not a hand-rolled loop over executeNode.
+// A best-effort removal failure (stale pre_start hook runner purge) is
+// warn-only: the plan carries on, and the removal passes RemoveVolumes so the
+// runner's anonymous volumes go with it — the imperative purge semantics.
+func TestExecutePlanBestEffortRemoveContainerFailureTolerated(t *testing.T) {
+	svc, apiClient := newTestService(t)
+
+	ctr := container.Summary{
+		ID:     "hook1",
+		Names:  []string{"/some-hook-runner"},
+		Labels: map[string]string{api.ServiceLabel: "web"},
+	}
+
+	apiClient.EXPECT().ContainerRemove(gomock.Any(), "hook1", gomock.Any()).
+		DoAndReturn(func(_ context.Context, _ string, opts client.ContainerRemoveOptions) (client.ContainerRemoveResult, error) {
+			assert.Assert(t, opts.RemoveVolumes, "hook-runner purge must drop anonymous volumes")
+			return client.ContainerRemoveResult{}, errors.New("device or resource busy")
+		})
+
+	plan := &Plan{}
+	plan.addNode(Operation{
+		Type:          OpRemoveContainer,
+		ResourceID:    "hook:web:stale:hook1",
+		Cause:         "stale pre_start hook container",
+		Container:     &ctr,
+		RemoveVolumes: true,
+		BestEffort:    true,
+	}, "")
+
+	err := svc.executePlan(t.Context(), &types.Project{Name: "test"}, emptyObservedState("test"), plan)
+	assert.NilError(t, err)
+}
+
 func TestExecutePlanRemoveContainerDropsFromCache(t *testing.T) {
 	svc, apiClient := newTestService(t)
 

pkg/compose/observed_state.go

@@ -46,6 +46,11 @@ type ObservedState struct {
 	// others as orphans (see selectNetwork/selectVolume).
 	Networks map[string][]ObservedNetwork // compose network key → observed
 	Volumes  map[string][]ObservedVolume  // compose volume key → observed
+	// HookContainers are ephemeral lifecycle-hook runners (HookLabel set),
+	// per service. Any observed at collection time is stale by definition —
+	// a previous run failed before removing it — and the reconciler plans
+	// its purge before re-running the hooks.
+	HookContainers map[string][]ObservedContainer // service name → hook containers
 }
 
 // selectNetwork picks, among the live networks recorded for a compose key, the
@@ -148,6 +153,8 @@ func (s *composeService) collectObservedState(ctx context.Context, project *type
 		Containers:  map[string][]ObservedContainer{},
 		Networks:    map[string][]ObservedNetwork{},
 		Volumes:     map[string][]ObservedVolume{},
+
+		HookContainers: map[string][]ObservedContainer{},
 	}
 
 	// --- Containers ---
@@ -170,6 +177,18 @@ func (s *composeService) collectObservedState(ctx context.Context, project *type
 
 	for _, ctr := range raw {
 		svcName := ctr.Labels[api.ServiceLabel]
+		if ctr.Labels[api.HookLabel] != "" && knownServices[svcName] {
+			// lifecycle-hook containers (ephemeral pre_start runners) are
+			// neither service replicas nor one-offs: classified apart, so
+			// they never masquerade as a replica (they carry no
+			// container-number label and would otherwise read as number 0)
+			// and the reconciler can plan purging stale ones. A hook
+			// container whose service left the model falls through to the
+			// orphan check below instead — nothing plans purges for an
+			// unknown service, and --remove-orphans must keep cleaning it.
+			state.HookContainers[svcName] = append(state.HookContainers[svcName], toObservedContainer(ctr))
+			continue
+		}
 		if isNotOneOff(ctr) && knownServices[svcName] {
 			state.Containers[svcName] = append(state.Containers[svcName], toObservedContainer(ctr))
 		} else if isOrphaned(project)(ctr) {

pkg/compose/observed_state_test.go

@@ -160,6 +160,33 @@ func TestCollectObservedState(t *testing.T) {
 					api.OneoffLabel:  "True",
 				},
 			},
+			{
+				// Stale lifecycle-hook runner (a previous run failed before
+				// removing it): neither a replica (it has no container-number
+				// label and must not read as number 0) nor a one-off —
+				// classified apart so the reconciler can plan its purge.
+				ID:    "c5",
+				Names: []string{"/hook-runner"},
+				State: container.StateExited,
+				Labels: map[string]string{
+					api.ServiceLabel: "web",
+					api.ProjectLabel: "myproject",
+					api.HookLabel:    "pre_start",
+				},
+			},
+			{
+				// Hook runner whose service left the model: nothing plans
+				// purges for an unknown service, so it must keep flowing to
+				// the orphan path --remove-orphans cleans.
+				ID:    "c6",
+				Names: []string{"/old-hook-runner"},
+				State: container.StateExited,
+				Labels: map[string]string{
+					api.ServiceLabel: "old",
+					api.ProjectLabel: "myproject",
+					api.HookLabel:    "pre_start",
+				},
+			},
 		},
 	}, nil)
 
@@ -202,11 +229,20 @@ func TestCollectObservedState(t *testing.T) {
 	assert.Equal(t, len(state.Containers["db"]), 1)
 	assert.Equal(t, state.Containers["db"][0].ID, "c2")
 
-	// Orphans: only the model-absent service "old". The running one-off c4 is
-	// absent everywhere — not in the "web" bucket (asserted above: 1 replica),
-	// not an orphan: up leaves live `compose run` sessions alone.
-	assert.Equal(t, len(state.Orphans), 1)
+	// The hook runner is classified apart — not a "web" replica (asserted
+	// above: 1 replica), not an orphan
+	assert.Equal(t, len(state.HookContainers["web"]), 1)
+	assert.Equal(t, state.HookContainers["web"][0].ID, "c5")
+
+	// Orphans: the model-absent service "old" — its replica AND its hook
+	// runner (c6), which must not hide in HookContainers where nothing would
+	// ever purge it. The running one-off c4 is absent everywhere — not in the
+	// "web" bucket (asserted above: 1 replica), not an orphan: up leaves live
+	// `compose run` sessions alone.
+	assert.Equal(t, len(state.Orphans), 2)
 	assert.Equal(t, state.Orphans[0].ID, "c3")
+	assert.Equal(t, state.Orphans[1].ID, "c6")
+	assert.Equal(t, len(state.HookContainers["old"]), 0)
 
 	// Networks
 	assert.Equal(t, len(state.Networks), 1)

pkg/compose/plan.go

@@ -127,11 +127,15 @@ type Operation struct {
 	Timeout      *time.Duration       // for stop operations
 	CreateNodeID int                  // for OpRenameContainer/start-phase ops: ID of the CreateContainer node whose result to target
 	Condition    string               // for OpWaitCondition: depends_on condition to wait for (service_healthy, ...)
-	// BestEffort marks an operation whose failure must not abort the plan. It is
-	// used for the optional removal of the old network on a rename: if the
-	// network is still in use (by non-Compose containers) the removal is skipped
-	// with a warning instead of failing — the new network already carries a
-	// different name, so the migration does not depend on the old one going away.
+	// RemoveVolumes asks OpRemoveContainer to also remove the container's
+	// anonymous volumes — the imperative semantics for hook-runner containers.
+	RemoveVolumes bool
+	// BestEffort marks an operation whose failure must not abort the plan.
+	// Used for the optional removal of the old network on a rename (if the
+	// network is still in use by non-Compose containers the removal is skipped
+	// with a warning — the new network already carries a different name), and
+	// for purging stale pre_start hook runners (the imperative purge is
+	// warn-only: a failed removal leaves the container visible, never blocks).
 	BestEffort bool
 }
 

pkg/compose/pre_start.go

@@ -99,26 +99,38 @@ func (s *composeService) runPreStart(ctx context.Context, project *types.Project
 		logrus.Warnf("service %q: failed to remove stale pre_start hook containers: %v", service.Name, err)
 	}
 	for i, hook := range service.PreStart {
-		if err := s.runPreStartHook(ctx, project, service, ctr, i, hook, listener); err != nil {
+		created, err := s.createPreStartContainer(ctx, project, service, ctr, hook)
+		if err != nil {
 			return err
 		}
+		if err := s.execPreStartHook(ctx, service, i, created.ID, listener); err != nil {
+			return err
+		}
+		// Success: remove the hook container, mirroring the old AutoRemove behaviour
+		// (including its anonymous volumes). A removal failure is logged but does not
+		// gate service start — the hook already succeeded.
+		if _, removeErr := s.apiClient().ContainerRemove(ctx, created.ID, client.ContainerRemoveOptions{RemoveVolumes: true}); removeErr != nil {
+			logrus.Warnf("service %q pre_start[%d]: failed to remove hook container %s: %v", service.Name, i, created.ID, removeErr)
+		}
 	}
 	return nil
 }
 
-func (s *composeService) runPreStartHook(
-	ctx context.Context, project *types.Project, service types.ServiceConfig,
-	ctr container.Summary, index int, hook types.PreStartHook, listener api.ContainerEventListener,
+// execPreStartHook starts an already-created hook container, streams its logs
+// and waits for its exit. It owns only execution-failure handling: a container
+// that never started or a run cancelled by the user is removed, a genuinely
+// failed hook is retained for post-mortem inspection. Removing the container
+// after a successful run is the caller's job — the container's lifecycle
+// belongs to whoever created it (the imperative runPreStart loop today, the
+// reconciliation plan once the executor runs hook nodes).
+func (s *composeService) execPreStartHook(
+	ctx context.Context, service types.ServiceConfig,
+	index int, containerID string, listener api.ContainerEventListener,
 ) error {
-	created, err := s.createPreStartContainer(ctx, project, service, ctr, hook)
-	if err != nil {
-		return err
-	}
-
 	// Subscribe to wait before start to avoid missing the exit event for short-lived hooks.
 	// WaitConditionNotRunning would match immediately because the container is still in
 	// "created" state, so use WaitConditionNextExit to block until the run actually finishes.
-	waitRes := s.apiClient().ContainerWait(ctx, created.ID, client.ContainerWaitOptions{
+	waitRes := s.apiClient().ContainerWait(ctx, containerID, client.ContainerWaitOptions{
 		Condition: container.WaitConditionNextExit,
 	})
 
@@ -128,13 +140,13 @@ func (s *composeService) runPreStartHook(
 	// open cannot deadlock `<-logsDone`.
 	logCtx, cancelLogs := context.WithCancel(ctx)
 	defer cancelLogs()
-	logsDone, getTail := s.streamPreStartLogs(logCtx, created.ID, service, index, listener)
+	logsDone, getTail := s.streamPreStartLogs(logCtx, containerID, service, index, listener)
 
-	if _, err := s.apiClient().ContainerStart(ctx, created.ID, client.ContainerStartOptions{}); err != nil {
+	if _, err := s.apiClient().ContainerStart(ctx, containerID, client.ContainerStartOptions{}); err != nil {
 		// AutoRemove is false, so we must remove the never-started container
 		// explicitly. A failed removal is logged so the orphan is visible.
-		if _, removeErr := s.apiClient().ContainerRemove(ctx, created.ID, client.ContainerRemoveOptions{Force: true, RemoveVolumes: true}); removeErr != nil {
-			logrus.Warnf("service %q pre_start[%d]: failed to remove orphan hook container %s: %v", service.Name, index, created.ID, removeErr)
+		if _, removeErr := s.apiClient().ContainerRemove(ctx, containerID, client.ContainerRemoveOptions{Force: true, RemoveVolumes: true}); removeErr != nil {
+			logrus.Warnf("service %q pre_start[%d]: failed to remove orphan hook container %s: %v", service.Name, index, containerID, removeErr)
 		}
 		// Drain waitRes so the client's wait goroutine exits without having to
 		// wait for the parent context to be canceled.
@@ -156,15 +168,15 @@ func (s *composeService) runPreStartHook(
 		// and return the raw context error without decorating it with the tail or
 		// retaining the container for post-mortem inspection.
 		if ctx.Err() != nil {
-			if _, removeErr := s.apiClient().ContainerRemove(context.Background(), created.ID, client.ContainerRemoveOptions{Force: true, RemoveVolumes: true}); removeErr != nil {
-				logrus.Warnf("service %q pre_start[%d]: failed to remove hook container %s after cancellation: %v", service.Name, index, created.ID, removeErr)
+			if _, removeErr := s.apiClient().ContainerRemove(context.Background(), containerID, client.ContainerRemoveOptions{Force: true, RemoveVolumes: true}); removeErr != nil {
+				logrus.Warnf("service %q pre_start[%d]: failed to remove hook container %s after cancellation: %v", service.Name, index, containerID, removeErr)
 			}
 			return waitErr
 		}
 		// Genuine hook failure: retain the container so the operator can run
 		// `docker logs <id>` and `docker inspect <id>` to diagnose the failure.
 		// Include the short container ID in the error to make it actionable.
-		shortID := created.ID
+		shortID := containerID
 		if len(shortID) > 12 {
 			shortID = shortID[:12]
 		}
@@ -173,12 +185,6 @@ func (s *composeService) runPreStartHook(
 		}
 		return fmt.Errorf("%w (hook container %s retained for inspection)", waitErr, shortID)
 	}
-	// Success: remove the hook container, mirroring the old AutoRemove behaviour
-	// (including its anonymous volumes). A removal failure is logged but does not
-	// gate service start — the hook already succeeded.
-	if _, removeErr := s.apiClient().ContainerRemove(ctx, created.ID, client.ContainerRemoveOptions{RemoveVolumes: true}); removeErr != nil {
-		logrus.Warnf("service %q pre_start[%d]: failed to remove hook container %s: %v", service.Name, index, created.ID, removeErr)
-	}
 	return nil
 }
 

pkg/compose/reconcile.go

@@ -769,6 +769,8 @@ func (r *reconciler) reconcileService(service types.ServiceConfig) error {
 		return err
 	}
 
+	r.planPurgeStaleHookRunners(service, expected)
+
 	containers := r.observed.Containers[service.Name]
 	actual := len(containers)
 
@@ -872,6 +874,41 @@ func (r *reconciler) reconcileService(service types.ServiceConfig) error {
 
 // mustRecreate decides whether oc must be recreated to match expected. The
 // expectedHash and parentRecreated inputs are precomputed once per service by
+// planPurgeStaleHookRunners plans the removal of hook-runner containers left
+// behind by a previous run that failed before removing them. It mirrors the
+// imperative purge living inside the gated runPreStart call: emitted only when
+// pre_start is going to run again — hooks declared, a replica to start
+// (scale > 0: the imperative start path returns before the hooks for a
+// scale-0 service) and no replica running at observation — so a genuinely
+// failed hook container stays retained for inspection as long as its service
+// is otherwise up. Removals are best-effort (the imperative purge is
+// warn-only) and independent of every other node.
+func (r *reconciler) planPurgeStaleHookRunners(service types.ServiceConfig, expectedScale int) {
+	stale := r.observed.HookContainers[service.Name]
+	if len(stale) == 0 || len(service.PreStart) == 0 || expectedScale == 0 {
+		return
+	}
+	for _, oc := range r.observed.Containers[service.Name] {
+		if oc.State == container.StateRunning {
+			return
+		}
+	}
+	serviceCopy := service
+	stale = slices.Clone(stale)
+	slices.SortFunc(stale, func(a, b ObservedContainer) int { return strings.Compare(a.ID, b.ID) })
+	for i := range stale {
+		r.plan.addNode(Operation{
+			Type:          OpRemoveContainer,
+			ResourceID:    fmt.Sprintf("hook:%s:stale:%s", service.Name, stale[i].ID[:min(12, len(stale[i].ID))]),
+			Cause:         "stale pre_start hook container",
+			Service:       &serviceCopy,
+			Container:     &stale[i].Summary,
+			RemoveVolumes: true,
+			BestEffort:    true,
+		}, "")
+	}
+}
+
 // reconcileService — see expectedConfigHash and parentNamespaceRecreated for
 // the rationale (issue #13878).
 func (r *reconciler) mustRecreate(expected types.ServiceConfig, expectedHash string, parentRecreated bool, oc ObservedContainer, policy string) bool {

pkg/compose/reconcile_test.go

@@ -1334,6 +1334,134 @@ func TestReconcileContainers_ExitedIsNoop(t *testing.T) {
 // container creation depends on the last plan node of the service it depends
 // on (via reconciler.serviceNodes). Without this, services declared in
 // depends_on could start before their dependencies' operations complete.
+// Stale pre_start hook runners (left by a previous run that failed before
+// removing them) are purged by the plan when pre_start is going to run again:
+// hooks declared and no replica running at observation — the imperative
+// gating. Removals are best-effort and drop the runner's anonymous volumes,
+// like the warn-only imperative purge they mirror.
+func TestReconcileContainers_StaleHookRunnersPurged(t *testing.T) {
+	project := &types.Project{
+		Name: "myproject",
+		Services: types.Services{
+			"app": {Name: "app", Scale: intPtr(1), PreStart: []types.PreStartHook{{}}},
+		},
+	}
+	observed := &ObservedState{
+		ProjectName: "myproject",
+		Containers:  map[string][]ObservedContainer{},
+		HookContainers: map[string][]ObservedContainer{
+			"app": {
+				// deliberately out of ID order: the plan sorts for determinism
+				{ID: "stale-b-id", Summary: container.Summary{ID: "stale-b-id"}},
+				{ID: "stale-a-id", Summary: container.Summary{ID: "stale-a-id"}},
+			},
+		},
+		Networks: map[string][]ObservedNetwork{},
+		Volumes:  map[string][]ObservedVolume{},
+	}
+
+	plan, err := reconcile(t.Context(), project, observed, defaultReconcileOptions(), noPrompt)
+	assert.NilError(t, err)
+
+	assert.Equal(t, plan.String(), strings.TrimSpace(`
+[] -> #1 hook:app:stale:stale-a-id, RemoveContainer, stale pre_start hook container
+[] -> #2 hook:app:stale:stale-b-id, RemoveContainer, stale pre_start hook container
+[] -> #3 service:app:1, CreateContainer, no existing container
+`)+"\n")
+
+	for _, n := range plan.Nodes {
+		if n.Operation.Type != OpRemoveContainer {
+			continue
+		}
+		assert.Assert(t, n.Operation.BestEffort, "purge #%d must not abort the plan on failure", n.ID)
+		assert.Assert(t, n.Operation.RemoveVolumes, "purge #%d must drop the runner's anonymous volumes", n.ID)
+	}
+}
+
+// A scale-0 service never reaches its pre_start hooks (the imperative start
+// path returns before them), so its stale runners are not purged either.
+func TestReconcileContainers_StaleHookRunnersKeptWhenScaleZero(t *testing.T) {
+	project := &types.Project{
+		Name: "myproject",
+		Services: types.Services{
+			"app": {Name: "app", Scale: intPtr(0), PreStart: []types.PreStartHook{{}}},
+		},
+	}
+	observed := &ObservedState{
+		ProjectName: "myproject",
+		Containers:  map[string][]ObservedContainer{},
+		HookContainers: map[string][]ObservedContainer{
+			"app": {{ID: "stale-a-id", Summary: container.Summary{ID: "stale-a-id"}}},
+		},
+		Networks: map[string][]ObservedNetwork{},
+		Volumes:  map[string][]ObservedVolume{},
+	}
+
+	plan, err := reconcile(t.Context(), project, observed, defaultReconcileOptions(), noPrompt)
+	assert.NilError(t, err)
+	assert.Assert(t, plan.IsEmpty(), "unexpected plan:\n%s", plan)
+}
+
+// A running replica gates pre_start off, so the stale runner stays: the
+// imperative purge lives inside the gated runPreStart call and would not run
+// either — a genuinely failed hook container stays retained for inspection as
+// long as its service is otherwise up.
+func TestReconcileContainers_StaleHookRunnersKeptWhenReplicaRunning(t *testing.T) {
+	svc := types.ServiceConfig{Name: "app", Scale: intPtr(1), PreStart: []types.PreStartHook{{}}}
+	hash := mustServiceHash(t, svc)
+	project := &types.Project{
+		Name:     "myproject",
+		Services: types.Services{"app": svc},
+	}
+	observed := &ObservedState{
+		ProjectName: "myproject",
+		Containers: map[string][]ObservedContainer{
+			"app": {{
+				ID: "c1", Number: 1, State: container.StateRunning, ConfigHash: hash,
+				Summary: container.Summary{
+					ID: "c1", State: container.StateRunning,
+					Labels: map[string]string{api.ServiceLabel: "app", api.ContainerNumberLabel: "1", api.ConfigHashLabel: hash},
+				},
+			}},
+		},
+		HookContainers: map[string][]ObservedContainer{
+			"app": {{ID: "stale-a-id", Summary: container.Summary{ID: "stale-a-id"}}},
+		},
+		Networks: map[string][]ObservedNetwork{},
+		Volumes:  map[string][]ObservedVolume{},
+	}
+
+	plan, err := reconcile(t.Context(), project, observed, defaultReconcileOptions(), noPrompt)
+	assert.NilError(t, err)
+	assert.Assert(t, plan.IsEmpty())
+}
+
+// Without pre_start hooks in the model, stale runners are not the plan's to
+// purge: runPreStart never runs, so the imperative engine leaves them alone
+// too.
+func TestReconcileContainers_StaleHookRunnersKeptWithoutHooks(t *testing.T) {
+	project := &types.Project{
+		Name:     "myproject",
+		Services: types.Services{"app": {Name: "app", Scale: intPtr(1)}},
+	}
+	observed := &ObservedState{
+		ProjectName: "myproject",
+		Containers:  map[string][]ObservedContainer{},
+		HookContainers: map[string][]ObservedContainer{
+			"app": {{ID: "stale-a-id", Summary: container.Summary{ID: "stale-a-id"}}},
+		},
+		Networks: map[string][]ObservedNetwork{},
+		Volumes:  map[string][]ObservedVolume{},
+	}
+
+	plan, err := reconcile(t.Context(), project, observed, defaultReconcileOptions(), noPrompt)
+	assert.NilError(t, err)
+
+	assert.Equal(t, plan.String(), strings.TrimSpace(`
+[] -> #1 service:app:1, CreateContainer, no existing container
+`)+"\n")
+}
+
 func TestReconcileContainers_DependsOnChain(t *testing.T) {
 	project := &types.Project{
 		Name: "myproject",