From db22082becffae6ab0669902a2996822e7280eae Mon Sep 17 00:00:00 2001 From: Nicolas De Loof Date: Tue, 1 Sep 2026 09:32:40 +0200 Subject: [PATCH 1/3] feat: reconciler plans the start phase MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The plan learns the start vocabulary — inert until a caller opts in (ReconcileOptions.Scope, zero value keeps today's create-only plans byte-identical): - OpWaitCondition, one node per (awaited service, condition), deduplicated across dependents like networkNodes deduplicates networks; required:false marks the shared node best-effort, one required dependent upgrades it. service_started needs no node — a plain DAG edge to the dependency's chain end expresses it. Health is deliberately re-observed at execution time: the plan encodes what to wait for, never a stale observation. - OpRunPreStart, emitted at plan time only when no replica was running at observation — the imperative gating — targeting the lowest-numbered replica. - OpRunPostStart per container, after its start. - replica chains: inject+start+post_start of replica n+1 depends on the end of replica n's chain, today's sequential start order made visible in golden plans; startChainEnds points at the chain end so a service_started dependent waits for the whole service, matching InDependencyOrder semantics. - scope Start plans starting observed exited/created containers without converging them (the future compose start); scope CreateStart appends the start phase to the create plan, start nodes resolving their target from the create node that materializes the replica (CreateNodeID, the mechanism OpRenameContainer already uses). Golden tests only; no executor support yet and no caller passes the scope. Epic #14081, Lot 1 (1/3). Signed-off-by: Nicolas De Loof --- pkg/compose/plan.go | 30 ++- pkg/compose/reconcile.go | 312 ++++++++++++++++++++++- pkg/compose/reconcile_start_test.go | 381 ++++++++++++++++++++++++++++ 3 files changed, 709 insertions(+), 14 deletions(-) create mode 100644 pkg/compose/reconcile_start_test.go diff --git a/pkg/compose/plan.go b/pkg/compose/plan.go index 6a118f12e0..d1b8624af1 100644 --- a/pkg/compose/plan.go +++ b/pkg/compose/plan.go @@ -53,6 +53,23 @@ const ( // Provider operations OpRunProvider OperationType = 30 + + // Start-phase operations + OpWaitCondition OperationType = 40 + OpRunPreStart OperationType = 41 + OpRunPostStart OperationType = 42 +) + +// PlanPhase situates a node in the plan lifecycle. The Create phase converges +// resources and containers to their desired shape; the Start phase brings +// containers to running — dependency waits, pre_start hooks, starts, +// post_start hooks. The zero value is Create, so plans built before the start +// phase existed render unchanged. +type PlanPhase int + +const ( + PhaseCreate PlanPhase = iota + PhaseStart ) // String returns the human-readable name of an OperationType. @@ -82,6 +99,12 @@ func (o OperationType) String() string { return "RenameContainer" case OpRunProvider: return "RunProvider" + case OpWaitCondition: + return "WaitCondition" + case OpRunPreStart: + return "RunPreStart" + case OpRunPostStart: + return "RunPostStart" default: return fmt.Sprintf("Unknown(%d)", int(o)) } @@ -102,7 +125,8 @@ type Operation struct { Network *types.NetworkConfig // for network operations Volume *types.VolumeConfig // for volume operations Timeout *time.Duration // for stop operations - CreateNodeID int // for OpRenameContainer: ID of the CreateContainer node whose result to rename + 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 @@ -118,6 +142,7 @@ type PlanNode struct { Operation Operation DependsOn []*PlanNode // prerequisite operations Group string // event grouping key (e.g. "recreate:web:1"); empty for ungrouped nodes + Phase PlanPhase // lifecycle phase this node belongs to; zero is Create } // Plan is a directed acyclic graph of operations produced by the reconciler. @@ -171,6 +196,9 @@ func (p *Plan) String() string { if node.Group != "" { fmt.Fprintf(&sb, " [%s]", node.Group) } + if node.Phase == PhaseStart { + sb.WriteString(" {start}") + } sb.WriteByte('\n') } return sb.String() diff --git a/pkg/compose/reconcile.go b/pkg/compose/reconcile.go index 1d6b7f98e5..6c5cbb377f 100644 --- a/pkg/compose/reconcile.go +++ b/pkg/compose/reconcile.go @@ -21,6 +21,7 @@ import ( "fmt" "slices" "sort" + "strconv" "strings" "time" @@ -46,7 +47,21 @@ func toReconcileOptions(options api.CreateOptions) ReconcileOptions { } // ReconcileOptions controls how the reconciler compares desired and observed state. +// ReconcileScope selects which lifecycle phases the plan covers. The zero +// value plans the Create phase only — the historical behavior, and what +// `compose create` keeps using. ScopeCreateStart adds the Start phase to the +// same plan; ScopeStart plans starting the observed containers without +// converging them first. +type ReconcileScope int + +const ( + ScopeCreate ReconcileScope = iota + ScopeCreateStart + ScopeStart +) + type ReconcileOptions struct { + Scope ReconcileScope // lifecycle phases to plan; zero = Create only Services []string // targeted services (empty = all) Recreate string // "diverged", "force", "never" for targeted services RecreateDependencies string // same for non-targeted services @@ -77,6 +92,19 @@ type reconciler struct { // serviceNodes tracks the last plan node per service, so dependent // services can order their operations after dependencies. serviceNodes map[string]*PlanNode + // containerNodes tracks, per replica ResourceID, the create-phase node + // that materializes that container (creation, recreation, or the + // exceptional-state restart), so start-phase nodes can depend on it and + // resolve their target from its result. + containerNodes map[string]*PlanNode + // startChainEnds tracks the last start-phase node of each service's + // replica chain: what a service_started dependent (or a wait node) hooks + // onto, matching the whole-service semantics of the imperative engine. + startChainEnds map[string]*PlanNode + // waitNodes deduplicates OpWaitCondition nodes per (service, condition): + // several dependents awaiting the same condition share one node, like + // networkNodes deduplicates network creations. + waitNodes map[string]*PlanNode // stoppedByPlan records containers already stopped by an earlier stage // of the plan (typically planRecreateNetwork) so that downstream stages // can chain on the existing OpStopContainer instead of emitting a second @@ -123,6 +151,9 @@ func reconcile(_ context.Context, project *types.Project, observed *ObservedStat networkNodes: map[string]*PlanNode{}, volumeNodes: map[string]*PlanNode{}, serviceNodes: map[string]*PlanNode{}, + containerNodes: map[string]*PlanNode{}, + startChainEnds: map[string]*PlanNode{}, + waitNodes: map[string]*PlanNode{}, stoppedByPlan: map[string]*PlanNode{}, connectNodes: map[string][]*PlanNode{}, recreatedServices: map[string]bool{}, @@ -131,20 +162,28 @@ func reconcile(_ context.Context, project *types.Project, observed *ObservedStat r.resolveObserved() - if err := r.reconcileNetworks(); err != nil { - return nil, err - } + if options.Scope != ScopeStart { + if err := r.reconcileNetworks(); err != nil { + return nil, err + } - if err := r.reconcileVolumes(); err != nil { - return nil, err - } + if err := r.reconcileVolumes(); err != nil { + return nil, err + } + + if err := r.reconcileContainers(); err != nil { + return nil, err + } - if err := r.reconcileContainers(); err != nil { - return nil, err + if r.options.RemoveOrphans { + r.reconcileOrphans() + } } - if r.options.RemoveOrphans { - r.reconcileOrphans() + if options.Scope != ScopeCreate { + if err := r.planStartPhase(); err != nil { + return nil, err + } } return r.plan, nil @@ -593,12 +632,12 @@ func (r *reconciler) reconcileContainers() error { } // Visit in dependency order (leaves first = services with no deps) - return r.visitInDependencyOrder(graph) + return r.visitInDependencyOrder(graph, r.reconcileService) } // visitInDependencyOrder processes services from leaves to roots so that // dependencies are reconciled before the services that depend on them. -func (r *reconciler) visitInDependencyOrder(g *Graph) error { +func (r *reconciler) visitInDependencyOrder(g *Graph, visit func(types.ServiceConfig) error) error { visited := map[string]bool{} // Sort vertex keys for deterministic plan output in tests keys := sortedKeys(g.Vertices) @@ -631,7 +670,7 @@ func (r *reconciler) visitInDependencyOrder(g *Graph) error { if err != nil { return err } - if err := r.reconcileService(service); err != nil { + if err := visit(service); err != nil { return err } } @@ -713,6 +752,7 @@ func (r *reconciler) reconcileService(service types.ServiceConfig) error { if r.mustRecreate(service, expectedHash, parentRecreated, oc, strategy) { lastNode = r.planRecreateContainer(service, &containers[i], infraDeps) + r.containerNodes[fmt.Sprintf("service:%s:%d", service.Name, oc.Number)] = lastNode r.recreatedServices[service.Name] = true continue } @@ -733,6 +773,7 @@ func (r *reconciler) reconcileService(service types.ServiceConfig) error { Cause: "not running", Container: &containers[i].Summary, }, "", infraDeps...) + r.containerNodes[fmt.Sprintf("service:%s:%d", service.Name, oc.Number)] = lastNode } } @@ -750,6 +791,7 @@ func (r *reconciler) reconcileService(service types.ServiceConfig) error { Number: number, Name: name, }, "", infraDeps...) + r.containerNodes[fmt.Sprintf("service:%s:%d", service.Name, number)] = lastNode } if lastNode != nil { @@ -877,6 +919,250 @@ func (r *reconciler) hasVolumeMismatch(expected types.ServiceConfig, oc Observed return false } +// planStartPhase appends the Start phase to the plan: for every service, a +// replica chain of start-phase operations reproducing the imperative +// engine's semantics — dependency conditions first, one optional +// OpRunPreStart when no replica was running at observation, then per +// replica inject+start (OpStartContainer, enriched by the executor) and +// post_start hooks, each replica chained after the previous one to keep +// today's sequential start order, now visible in the plan. +// +// Conditions other than service_started materialize as OpWaitCondition +// nodes, deduplicated per (service, condition) across dependents; health is +// deliberately re-observed at execution time — the plan only encodes what to +// wait for, never a stale observation. service_started needs no node: a +// plain DAG edge to the dependency's chain end expresses it. +func (r *reconciler) planStartPhase() error { + graph, err := NewGraph(r.project, ServiceStopped) + if err != nil { + return err + } + return r.visitInDependencyOrder(graph, r.planServiceStart) +} + +// startReplica is a container the start phase must bring to running: either +// materialized by a create-phase node (create) or already observed (container). +type startReplica struct { + resID string + number int // replica number, the start order within the service + // container is the observed container to start; nil when the create + // phase materializes it. + container *container.Summary + // after is the create-phase node this replica's start must follow, when + // the create phase planned one. + after *PlanNode + // createNodeID is the ID of the node whose execution result carries the + // materialized container (the OpCreateContainer node); 0 when the + // container is observed. + createNodeID int +} + +// plannedReplica builds the startReplica for a container the create phase +// already planned a node for. The node registered in containerNodes is not +// always the one whose execution result carries the container ID: a recreate +// chain registers its final rename node (whose CreateNodeID names the actual +// create node), and an exceptional-state restart (paused, dead, ...) plans +// no create at all — the observed container sits on the node itself. +func plannedReplica(resID string, number int, node *PlanNode) startReplica { + rep := startReplica{resID: resID, number: number, after: node} + switch node.Operation.Type { + case OpCreateContainer: + rep.createNodeID = node.ID + case OpRenameContainer: + rep.createNodeID = node.Operation.CreateNodeID + case OpStartContainer: + // exceptional-state restart (paused, dead, ...): that registration + // always carries the observed container on the node itself + rep.container = node.Operation.Container + default: + // no other node type is registered in containerNodes today; leave + // the target unresolved so execution fails with a clean "no + // materialized container" error instead of panicking here + } + return rep +} + +// replicaNumber extracts the numeric replica suffix from a +// "service::" resource ID. +func replicaNumber(resID string) int { + number, _ := strconv.Atoi(resID[strings.LastIndexByte(resID, ':')+1:]) + return number +} + +// startPhaseReplicas collects the replicas to start, ascending number: +// containers materialized by the create phase plus observed up-to-date +// containers not running — exactly the isNotRunning set the imperative start +// phase acts on. anyRunning reports whether a replica was running at +// observation and left untouched by the plan, which gates pre_start. +func (r *reconciler) startPhaseReplicas(service types.ServiceConfig) (replicas []startReplica, anyRunning bool) { + seen := map[string]bool{} + for i := range r.observed.Containers[service.Name] { + oc := &r.observed.Containers[service.Name][i] + resID := fmt.Sprintf("service:%s:%d", service.Name, oc.Number) + if node, planned := r.containerNodes[resID]; planned { + seen[resID] = true + replicas = append(replicas, plannedReplica(resID, oc.Number, node)) + continue + } + if oc.State == container.StateRunning { + anyRunning = true + continue + } + seen[resID] = true + replicas = append(replicas, startReplica{resID: resID, number: oc.Number, container: &oc.Summary}) + } + for resID, node := range r.containerNodes { + if !seen[resID] && strings.HasPrefix(resID, fmt.Sprintf("service:%s:", service.Name)) { + replicas = append(replicas, plannedReplica(resID, replicaNumber(resID), node)) + } + } + sort.Slice(replicas, func(i, j int) bool { return replicas[i].number < replicas[j].number }) + return replicas, anyRunning +} + +// startPhaseDependencies turns the service's depends_on into plan +// prerequisites: a service_started condition is a plain edge to the +// dependency's chain end, any other condition a deduplicated OpWaitCondition +// node re-evaluated at execution time. +func (r *reconciler) startPhaseDependencies(service types.ServiceConfig) []*PlanNode { + var depNodes []*PlanNode + for _, dep := range sortedKeys(service.DependsOn) { + cfg := service.DependsOn[dep] + target, ok := r.startChainEnds[dep] + if !ok { + target = r.serviceNodes[dep] + } + depService, err := r.project.GetService(dep) + waitless := cfg.Condition == types.ServiceConditionStarted || + // mirror shouldWaitForDependency: nothing to wait for on + // disabled, scale-0 or provider dependencies + err != nil || depService.GetScale() == 0 || depService.Provider != nil + if waitless { + if target != nil { + depNodes = append(depNodes, target) + } + continue + } + depNodes = append(depNodes, r.waitConditionNode(dep, cfg, target)) + } + return depNodes +} + +// waitConditionNode returns the shared wait node for (dep, condition), +// creating it on first use. required:false marks it best-effort — a missing +// dependency is skipped, not fatal; one required dependent upgrades the +// shared node for everyone. +func (r *reconciler) waitConditionNode(dep string, cfg types.ServiceDependency, target *PlanNode) *PlanNode { + key := dep + ":" + cfg.Condition + wait, ok := r.waitNodes[key] + if !ok { + var waitDeps []*PlanNode + if target != nil { + waitDeps = append(waitDeps, target) + } + wait = r.plan.addNode(Operation{ + Type: OpWaitCondition, + ResourceID: fmt.Sprintf("wait:%s:%s", dep, cfg.Condition), + Cause: "depends_on condition", + Name: dep, + Condition: cfg.Condition, + BestEffort: !cfg.Required, + }, "", waitDeps...) + wait.Phase = PhaseStart + r.waitNodes[key] = wait + } else if cfg.Required && wait.Operation.BestEffort { + wait.Operation.BestEffort = false + } + return wait +} + +func (r *reconciler) planServiceStart(service types.ServiceConfig) error { + if service.Provider != nil { + // a provider has no container to start: its create-phase RunProvider + // node is what dependents hook onto + if node, ok := r.serviceNodes[service.Name]; ok { + r.startChainEnds[service.Name] = node + } + return nil + } + if service.GetScale() == 0 { + return nil + } + + replicas, anyRunning := r.startPhaseReplicas(service) + if len(replicas) == 0 { + if node, ok := r.serviceNodes[service.Name]; ok { + // nothing to start (all replicas already running): dependents + // still order after whatever the create phase did + r.startChainEnds[service.Name] = node + } + return nil + } + + prev := r.startPhaseDependencies(service) + + // pre_start runs once per service, only when no replica was running at + // observation — the imperative gating (initial up, force-recreate, or + // spec change), decided at plan time + if len(service.PreStart) > 0 && !anyRunning { + serviceCopy := service + first := replicas[0] + op := Operation{ + Type: OpRunPreStart, + ResourceID: first.resID, + Cause: "pre_start hooks", + Service: &serviceCopy, + Container: first.container, + CreateNodeID: first.createNodeID, + } + deps := prev + if first.after != nil { + deps = append(deps, first.after) + } + preStart := r.plan.addNode(op, fmt.Sprintf("start%s", strings.TrimPrefix(first.resID, "service")), deps...) + preStart.Phase = PhaseStart + prev = []*PlanNode{preStart} + } + + // the replica chain: inject+start then post_start of replica n+1 waits + // for the end of replica n's chain — today's sequential start order + var chainEnd *PlanNode + for _, rep := range replicas { + serviceCopy := service + group := fmt.Sprintf("start%s", strings.TrimPrefix(rep.resID, "service")) + op := Operation{ + Type: OpStartContainer, + ResourceID: rep.resID, + Cause: "start", + Service: &serviceCopy, + Container: rep.container, + CreateNodeID: rep.createNodeID, + } + deps := slices.Clone(prev) + if rep.after != nil { + deps = append(deps, rep.after) + } + start := r.plan.addNode(op, group, deps...) + start.Phase = PhaseStart + chainEnd = start + if len(service.PostStart) > 0 { + post := r.plan.addNode(Operation{ + Type: OpRunPostStart, + ResourceID: rep.resID, + Cause: "post_start hooks", + Service: &serviceCopy, + Container: rep.container, + CreateNodeID: op.CreateNodeID, + }, group, start) + post.Phase = PhaseStart + chainEnd = post + } + prev = []*PlanNode{chainEnd} + } + r.startChainEnds[service.Name] = chainEnd + return nil +} + // planRecreateContainer decomposes container recreation into 4 atomic operations: // CreateContainer(tmpName) → StopContainer → RemoveContainer → RenameContainer func (r *reconciler) planRecreateContainer(service types.ServiceConfig, oc *ObservedContainer, infraDeps []*PlanNode) *PlanNode { diff --git a/pkg/compose/reconcile_start_test.go b/pkg/compose/reconcile_start_test.go new file mode 100644 index 0000000000..ed5cd4e002 --- /dev/null +++ b/pkg/compose/reconcile_start_test.go @@ -0,0 +1,381 @@ +/* + Copyright 2020 Docker Compose CLI authors + + Licensed under the Apache License, Version 2.0 (the "License"); + you may not use this file except in compliance with the License. + You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + + Unless required by applicable law or agreed to in writing, software + distributed under the License is distributed on an "AS IS" BASIS, + WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + See the License for the specific language governing permissions and + limitations under the License. +*/ + +package compose + +import ( + "slices" + "strconv" + "strings" + "testing" + + "github.com/compose-spec/compose-go/v2/types" + "github.com/moby/moby/api/types/container" + "gotest.tools/v3/assert" +) + +func startScopeOptions(scope ReconcileScope) ReconcileOptions { + options := defaultReconcileOptions() + options.Scope = scope + return options +} + +func emptyObserved() *ObservedState { + return &ObservedState{ + ProjectName: "myproject", + Containers: map[string][]ObservedContainer{}, + Networks: map[string][]ObservedNetwork{}, + Volumes: map[string][]ObservedVolume{}, + } +} + +func observedServiceContainer(service string, number int, state container.ContainerState, hash string) ObservedContainer { + name := "myproject-" + service + "-" + strconv.Itoa(number) + return ObservedContainer{ + ID: name + "-id", + Name: name, + State: state, + ConfigHash: hash, + Number: number, + Summary: container.Summary{ID: name + "-id", Names: []string{"/" + name}, State: state}, + } +} + +// A fresh up plans Create and Start as one DAG: each start depends on its +// own create, and the service_started dependency is a plain edge — the +// dependent's start waits for the dependency's chain end, no wait node. +func TestPlanStart_FreshUpWithStartedDependency(t *testing.T) { + project := &types.Project{ + Name: "myproject", + Services: types.Services{ + "db": {Name: "db"}, + "web": { + Name: "web", + DependsOn: types.DependsOnConfig{ + "db": {Condition: types.ServiceConditionStarted, Required: true}, + }, + }, + }, + } + + plan, err := reconcile(t.Context(), project, emptyObserved(), startScopeOptions(ScopeCreateStart), noPrompt) + assert.NilError(t, err) + + assert.Equal(t, plan.String(), strings.TrimSpace(` +[] -> #1 service:db:1, CreateContainer, no existing container +[1] -> #2 service:web:1, CreateContainer, no existing container +[1] -> #3 service:db:1, StartContainer, start [start:db:1] {start} +[2,3] -> #4 service:web:1, StartContainer, start [start:web:1] {start} +`)+"\n") +} + +// A condition other than service_started materializes as one wait node per +// (service, condition), shared by every dependent; health is re-observed at +// execution time, the plan only encodes what to wait for. +func TestPlanStart_HealthyConditionDeduplicated(t *testing.T) { + project := &types.Project{ + Name: "myproject", + Services: types.Services{ + "db": {Name: "db"}, + "web": { + Name: "web", + DependsOn: types.DependsOnConfig{ + "db": {Condition: types.ServiceConditionHealthy, Required: true}, + }, + }, + "worker": { + Name: "worker", + DependsOn: types.DependsOnConfig{ + "db": {Condition: types.ServiceConditionHealthy, Required: true}, + }, + }, + }, + } + + plan, err := reconcile(t.Context(), project, emptyObserved(), startScopeOptions(ScopeCreateStart), noPrompt) + assert.NilError(t, err) + + assert.Equal(t, plan.String(), strings.TrimSpace(` +[] -> #1 service:db:1, CreateContainer, no existing container +[1] -> #2 service:web:1, CreateContainer, no existing container +[1] -> #3 service:worker:1, CreateContainer, no existing container +[1] -> #4 service:db:1, StartContainer, start [start:db:1] {start} +[4] -> #5 wait:db:service_healthy, WaitCondition, depends_on condition {start} +[2,5] -> #6 service:web:1, StartContainer, start [start:web:1] {start} +[3,5] -> #7 service:worker:1, StartContainer, start [start:worker:1] {start} +`)+"\n") +} + +// pre_start runs once per service before the first replica start, only when +// no replica was running at observation; replicas start sequentially, each +// chain link (start, then post_start when declared) gating the next. +func TestPlanStart_HooksAndReplicaChain(t *testing.T) { + two := 2 + project := &types.Project{ + Name: "myproject", + Services: types.Services{ + "app": { + Name: "app", + PreStart: []types.ServiceHook{{}}, + PostStart: []types.ServiceHook{{}}, + Scale: &two, + }, + }, + } + + plan, err := reconcile(t.Context(), project, emptyObserved(), startScopeOptions(ScopeCreateStart), noPrompt) + assert.NilError(t, err) + + assert.Equal(t, plan.String(), strings.TrimSpace(` +[] -> #1 service:app:1, CreateContainer, no existing container +[] -> #2 service:app:2, CreateContainer, no existing container +[1] -> #3 service:app:1, RunPreStart, pre_start hooks [start:app:1] {start} +[1,3] -> #4 service:app:1, StartContainer, start [start:app:1] {start} +[4] -> #5 service:app:1, RunPostStart, post_start hooks [start:app:1] {start} +[2,5] -> #6 service:app:2, StartContainer, start [start:app:2] {start} +[6] -> #7 service:app:2, RunPostStart, post_start hooks [start:app:2] {start} +`)+"\n") +} + +// With a replica already running and untouched by the plan, pre_start is +// gated off and the running replica gets no node — only the non-running one +// starts, the imperative isNotRunning role expressed in the plan. +func TestPlanStart_RunningReplicaGatesPreStart(t *testing.T) { + two := 2 + service := types.ServiceConfig{ + Name: "app", + PreStart: []types.ServiceHook{{}}, + Scale: &two, + } + project := &types.Project{ + Name: "myproject", + Services: types.Services{"app": service}, + } + hash, err := serviceHashWithResolvedRefs(service, nil) + assert.NilError(t, err) + observed := emptyObserved() + observed.Containers["app"] = []ObservedContainer{ + observedServiceContainer("app", 1, container.StateRunning, hash), + observedServiceContainer("app", 2, container.StateExited, hash), + } + + plan, err := reconcile(t.Context(), project, observed, startScopeOptions(ScopeCreateStart), noPrompt) + assert.NilError(t, err) + + assert.Equal(t, plan.String(), strings.TrimSpace(` +[] -> #1 service:app:2, StartContainer, start [start:app:2] {start} +`)+"\n") +} + +// A recreated replica's start-phase node resolves its container from the +// recreate chain's create node — not the rename node registered in +// containerNodes, whose execution stores no result — and orders after the +// chain's end. +func TestPlanStart_RecreatedReplicaTargetsCreateNode(t *testing.T) { + service := types.ServiceConfig{Name: "app"} + project := &types.Project{ + Name: "myproject", + Services: types.Services{"app": service}, + } + observed := emptyObserved() + observed.Containers["app"] = []ObservedContainer{ + observedServiceContainer("app", 1, container.StateRunning, "stale-hash"), + } + + plan, err := reconcile(t.Context(), project, observed, startScopeOptions(ScopeCreateStart), noPrompt) + assert.NilError(t, err) + + var create, rename, start *PlanNode + for _, n := range plan.Nodes { + switch { + case n.Operation.Type == OpCreateContainer: + create = n + case n.Operation.Type == OpRenameContainer: + rename = n + case n.Operation.Type == OpStartContainer && n.Phase == PhaseStart: + start = n + } + } + if create == nil || rename == nil || start == nil { + t.Fatalf("plan misses expected nodes (create=%v rename=%v start=%v):\n%s", create, rename, start, plan) + } + assert.Equal(t, start.Operation.CreateNodeID, create.ID) + assert.Assert(t, start.Operation.Container == nil) + assert.Assert(t, slices.Contains(start.DependsOn, rename)) +} + +// An exceptional-state container (paused, dead, ...) gets a bare create-phase +// restart, not a create: its start-phase node targets the observed container +// itself and orders after that restart node. +func TestPlanStart_ExceptionalStateReplicaKeepsObservedTarget(t *testing.T) { + service := types.ServiceConfig{Name: "app"} + project := &types.Project{ + Name: "myproject", + Services: types.Services{"app": service}, + } + hash, err := serviceHashWithResolvedRefs(service, nil) + assert.NilError(t, err) + observed := emptyObserved() + observed.Containers["app"] = []ObservedContainer{ + observedServiceContainer("app", 1, container.StatePaused, hash), + } + + plan, err := reconcile(t.Context(), project, observed, startScopeOptions(ScopeCreateStart), noPrompt) + assert.NilError(t, err) + + var createPhase, startPhase *PlanNode + for _, n := range plan.Nodes { + if n.Operation.Type != OpStartContainer { + continue + } + if n.Phase == PhaseStart { + startPhase = n + } else { + createPhase = n + } + } + if createPhase == nil || startPhase == nil { + t.Fatalf("plan misses expected start nodes (createPhase=%v startPhase=%v):\n%s", createPhase, startPhase, plan) + } + assert.Equal(t, startPhase.Operation.CreateNodeID, 0) + assert.Assert(t, startPhase.Operation.Container != nil) + assert.Equal(t, startPhase.Operation.Container.ID, "myproject-app-1-id") + assert.Assert(t, slices.Contains(startPhase.DependsOn, createPhase)) +} + +// Replica start order is numeric, not lexicographic: with 10+ replicas, +// replica 2 starts before replica 10. +func TestPlanStart_ReplicaOrderIsNumeric(t *testing.T) { + eleven := 11 + project := &types.Project{ + Name: "myproject", + Services: types.Services{ + "app": {Name: "app", Scale: &eleven}, + }, + } + + plan, err := reconcile(t.Context(), project, emptyObserved(), startScopeOptions(ScopeCreateStart), noPrompt) + assert.NilError(t, err) + + var order []string + for _, n := range plan.Nodes { + if n.Operation.Type == OpStartContainer { + order = append(order, n.Operation.ResourceID) + } + } + expected := make([]string, 0, 11) + for i := 1; i <= 11; i++ { + expected = append(expected, "service:app:"+strconv.Itoa(i)) + } + assert.DeepEqual(t, order, expected) +} + +// Scope Start plans only the start phase over observed containers — the +// future `compose start`: no convergence, exited containers start in +// dependency order, running ones are left alone. +func TestPlanStart_StartOnlyScope(t *testing.T) { + db := types.ServiceConfig{Name: "db"} + web := types.ServiceConfig{ + Name: "web", + DependsOn: types.DependsOnConfig{ + "db": {Condition: types.ServiceConditionStarted, Required: true}, + }, + } + project := &types.Project{ + Name: "myproject", + Services: types.Services{"db": db, "web": web}, + } + dbHash, err := serviceHashWithResolvedRefs(db, nil) + assert.NilError(t, err) + webHash, err := serviceHashWithResolvedRefs(web, nil) + assert.NilError(t, err) + observed := emptyObserved() + observed.Containers["db"] = []ObservedContainer{observedServiceContainer("db", 1, container.StateExited, dbHash)} + observed.Containers["web"] = []ObservedContainer{observedServiceContainer("web", 1, container.StateCreated, webHash)} + + plan, err := reconcile(t.Context(), project, observed, startScopeOptions(ScopeStart), noPrompt) + assert.NilError(t, err) + + assert.Equal(t, plan.String(), strings.TrimSpace(` +[] -> #1 service:db:1, StartContainer, start [start:db:1] {start} +[1] -> #2 service:web:1, StartContainer, start [start:web:1] {start} +`)+"\n") +} + +// An optional (required: false) condition marks the shared wait node +// best-effort — a missing dependency is skipped, not fatal; one required +// dependent upgrades the node for everyone. +func TestPlanStart_OptionalConditionIsBestEffort(t *testing.T) { + project := &types.Project{ + Name: "myproject", + Services: types.Services{ + "db": {Name: "db"}, + "web": { + Name: "web", + DependsOn: types.DependsOnConfig{ + "db": {Condition: types.ServiceConditionHealthy, Required: false}, + }, + }, + }, + } + + plan, err := reconcile(t.Context(), project, emptyObserved(), startScopeOptions(ScopeCreateStart), noPrompt) + assert.NilError(t, err) + + var wait *PlanNode + for _, n := range plan.Nodes { + if n.Operation.Type == OpWaitCondition { + wait = n + } + } + assert.Assert(t, wait != nil) + assert.Assert(t, wait.Operation.BestEffort) + + // a second dependent requiring the same condition upgrades the node + project.Services["worker"] = types.ServiceConfig{ + Name: "worker", + DependsOn: types.DependsOnConfig{ + "db": {Condition: types.ServiceConditionHealthy, Required: true}, + }, + } + plan, err = reconcile(t.Context(), project, emptyObserved(), startScopeOptions(ScopeCreateStart), noPrompt) + assert.NilError(t, err) + wait = nil + for _, n := range plan.Nodes { + if n.Operation.Type == OpWaitCondition { + wait = n + } + } + assert.Assert(t, wait != nil) + assert.Assert(t, !wait.Operation.BestEffort) +} + +// The default scope keeps yesterday's plans byte-identical: no start-phase +// node ever appears unless a caller opts in. +func TestPlanStart_DefaultScopeIsInert(t *testing.T) { + project := &types.Project{ + Name: "myproject", + Services: types.Services{"app": {Name: "app"}}, + } + + plan, err := reconcile(t.Context(), project, emptyObserved(), defaultReconcileOptions(), noPrompt) + assert.NilError(t, err) + + for _, n := range plan.Nodes { + assert.Assert(t, n.Phase == PhaseCreate) + } + assert.Equal(t, plan.String(), "[] -> #1 service:app:1, CreateContainer, no existing container\n") +} From 57a06839a6be43ab4f20fc24bea2f31be5435de0 Mon Sep 17 00:00:00 2001 From: Nicolas De Loof Date: Tue, 1 Sep 2026 11:36:20 +0200 Subject: [PATCH 2/3] feat: executor runs start-phase operations MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The executor learns the start vocabulary planned by the previous commit — still without any caller passing the scope: - execStartContainer, enriched for start-phase nodes: secret and config injection folds in (they always run as a pair right before the start — part of starting a container, not a separate decision), the target resolves from the observed summary or from the result of the create node that materialized the replica (CreateNodeID), and the ContainerStart goes through the call site holding startMx. Create-phase starts (exceptional states) keep the bare historical behavior. - execWaitCondition re-observes the depends_on condition at execution time by delegating to the imperative waitDependency primitive — no polling rewrite, and the Waiting/Healthy/Exited/Skipped events are word-for-word the vocabulary users see today. A best-effort node (every dependent optional) absorbs a missing dependency as a Skipped event. - execRunPreStart/execRunPostStart delegate to the runPreStart/runHook primitives, with listener plumbing for hook output. - start:: event groups: Starting on the chain's first node, Started once the whole chain completed — after post_start, exactly the imperative sequence (groupTracker texts are now derived from the group family instead of hardcoding recreate's). Epic #14081, Lot 1 (2/3). Signed-off-by: Nicolas De Loof --- pkg/compose/executor.go | 15 +- pkg/compose/executor_events.go | 58 +++++-- pkg/compose/executor_ops.go | 91 ++++++++++- pkg/compose/executor_start_test.go | 237 +++++++++++++++++++++++++++++ pkg/compose/reconcile.go | 11 +- 5 files changed, 393 insertions(+), 19 deletions(-) create mode 100644 pkg/compose/executor_start_test.go diff --git a/pkg/compose/executor.go b/pkg/compose/executor.go index c23b3f7b21..a54111fd48 100644 --- a/pkg/compose/executor.go +++ b/pkg/compose/executor.go @@ -23,6 +23,8 @@ import ( "github.com/compose-spec/compose-go/v2/types" "golang.org/x/sync/errgroup" + + "github.com/docker/compose/v5/pkg/api" ) // planExecutor executes a reconciliation Plan by walking the DAG and performing @@ -33,6 +35,11 @@ type planExecutor struct { project *types.Project pctx *reconciliationContext + // listener receives lifecycle-hook output (pre_start/post_start) when a + // caller attaches one; nil discards it, like the imperative engine when + // running detached. + listener api.ContainerEventListener + // containersByService is a live view used to resolve service references // (network_mode: service:x, volumes_from, ipc, pid) without a daemon // round-trip per create. @@ -153,7 +160,7 @@ func (exec *planExecutor) executeNode(ctx context.Context, node *PlanNode) error case OpCreateContainer: return exec.execCreateContainer(ctx, node) case OpStartContainer: - return exec.execStartContainer(ctx, op) + return exec.execStartContainer(ctx, node) case OpStopContainer: return exec.execStopContainer(ctx, op) case OpRemoveContainer: @@ -162,6 +169,12 @@ func (exec *planExecutor) executeNode(ctx context.Context, node *PlanNode) error return exec.execRenameContainer(ctx, node) case OpRunProvider: return exec.compose.runPlugin(ctx, exec.project, *op.Service, "up") + case OpWaitCondition: + return exec.execWaitCondition(ctx, node) + case OpRunPreStart: + return exec.execRunPreStart(ctx, node) + case OpRunPostStart: + return exec.execRunPostStart(ctx, node) default: return fmt.Errorf("unknown operation type: %s", op.Type) } diff --git a/pkg/compose/executor_events.go b/pkg/compose/executor_events.go index 5177d09313..ff43e53315 100644 --- a/pkg/compose/executor_events.go +++ b/pkg/compose/executor_events.go @@ -17,6 +17,7 @@ package compose import ( + "strings" "sync" "github.com/docker/compose/v5/pkg/api" @@ -30,29 +31,52 @@ type groupTracker struct { } type groupState struct { - eventName string // e.g. "Container myproject-web-1" - total int // total nodes in this group - started int // nodes that have started - done int // nodes that have completed + eventName string // e.g. "Container myproject-web-1" + workingText string // event text while the group runs, e.g. "Recreate" + doneText string // event text once the whole group completed + total int // total nodes in this group + started int // nodes that have started + done int // nodes that have completed } func (exec *planExecutor) buildGroupTracker(plan *Plan) *groupTracker { gt := &groupTracker{groups: map[string]*groupState{}} + nameFallback := map[string]string{} for _, node := range plan.Nodes { if node.Group == "" { continue } if _, ok := gt.groups[node.Group]; !ok { - gt.groups[node.Group] = &groupState{} + gs := &groupState{workingText: "Recreate", doneText: "Recreated"} + if strings.HasPrefix(node.Group, "start:") { + // the group closes on the chain's last node (post_start when + // declared), so Started is emitted after the hooks ran — + // word-for-word the imperative sequence + gs.workingText = api.StatusStarting + gs.doneText = api.StatusStarted + } + gt.groups[node.Group] = gs } gt.groups[node.Group].total++ - // Pick the event name from a node that has the existing container reference + // Pick the event name from a node that has the existing container + // reference — its canonical name. A name computed at plan time is + // only a fallback (below): in a recreate group the create node + // carries the temporary name, and must never win over the observed + // container's canonical name whatever the node order. if gt.groups[node.Group].eventName == "" && node.Operation.Container != nil { gt.groups[node.Group].eventName = getContainerProgressName(*node.Operation.Container) } + if nameFallback[node.Group] == "" && node.Operation.Name != "" { + nameFallback[node.Group] = "Container " + node.Operation.Name + } } - // Fallback for groups where no node had a Container (shouldn't happen for recreate) + // Fallback for groups where no node had a Container: the deterministic + // name computed at plan time (start groups over plan-materialized + // replicas), else the group key. for name, gs := range gt.groups { + if gs.eventName == "" { + gs.eventName = nameFallback[name] + } if gs.eventName == "" { gs.eventName = name } @@ -71,7 +95,7 @@ func (gt *groupTracker) onNodeStart(node *PlanNode, events api.EventProcessor) { gs := gt.groups[node.Group] gs.started++ if gs.started == 1 { - events.On(newEvent(gs.eventName, api.Working, "Recreate")) + events.On(newEvent(gs.eventName, api.Working, gs.workingText)) } } @@ -85,7 +109,7 @@ func (gt *groupTracker) onNodeDone(node *PlanNode, events api.EventProcessor) { gs := gt.groups[node.Group] gs.done++ if gs.done == gs.total { - events.On(newEvent(gs.eventName, api.Done, "Recreated")) + events.On(newEvent(gs.eventName, api.Done, gs.doneText)) } } @@ -111,8 +135,7 @@ func emitStartEvent(node *PlanNode, events api.EventProcessor) { case OpCreateContainer: events.On(creatingEvent("Container " + op.Name)) case OpStartContainer: - name := getContainerProgressName(*op.Container) - events.On(newEvent(name, api.Working, api.StatusStarting)) + events.On(newEvent(startEventName(op), api.Working, api.StatusStarting)) case OpStopContainer: events.On(stoppingEvent(getContainerProgressName(*op.Container))) case OpRemoveContainer: @@ -135,8 +158,7 @@ func emitDoneEvent(node *PlanNode, events api.EventProcessor) { case OpCreateContainer: events.On(createdEvent("Container " + op.Name)) case OpStartContainer: - name := getContainerProgressName(*op.Container) - events.On(newEvent(name, api.Done, api.StatusStarted)) + events.On(newEvent(startEventName(op), api.Done, api.StatusStarted)) case OpStopContainer: events.On(stoppedEvent(getContainerProgressName(*op.Container))) case OpRemoveContainer: @@ -152,6 +174,16 @@ func emitDoneEvent(node *PlanNode, events api.EventProcessor) { } } +// startEventName names the progress event of a container start: the +// observed container when the plan carries one, else the deterministic name +// computed at plan time for a container the same plan materializes. +func startEventName(op Operation) string { + if op.Container != nil { + return getContainerProgressName(*op.Container) + } + return "Container " + op.Name +} + // emitErrorEvent emits an error event for an ungrouped node. func emitErrorEvent(node *PlanNode, events api.EventProcessor, err error) { op := node.Operation diff --git a/pkg/compose/executor_ops.go b/pkg/compose/executor_ops.go index 13738d045b..49dfb31dad 100644 --- a/pkg/compose/executor_ops.go +++ b/pkg/compose/executor_ops.go @@ -21,6 +21,7 @@ import ( "fmt" "slices" + "github.com/compose-spec/compose-go/v2/types" "github.com/containerd/errdefs" "github.com/moby/moby/api/types/container" "github.com/moby/moby/client" @@ -127,13 +128,99 @@ func (exec *planExecutor) execCreateContainer(ctx context.Context, node *PlanNod return nil } -func (exec *planExecutor) execStartContainer(ctx context.Context, op Operation) error { +func (exec *planExecutor) execStartContainer(ctx context.Context, node *PlanNode) error { + op := node.Operation + if node.Phase != PhaseStart { + // create-phase start (exceptional states: paused, dead, ...): a bare + // engine start, exactly the historical behavior + startMx.Lock() + defer startMx.Unlock() + _, err := exec.compose.apiClient().ContainerStart(ctx, op.Container.ID, client.ContainerStartOptions{}) + return err + } + + ctr, err := exec.startTarget(node) + if err != nil { + return err + } + // secrets and configs always inject as a pair right before the start — + // they are part of starting a container, not a separate decision + if err := exec.compose.injectSecrets(ctx, exec.project, *op.Service, ctr.ID); err != nil { + return err + } + if err := exec.compose.injectConfigs(ctx, exec.project, *op.Service, ctr.ID); err != nil { + return err + } startMx.Lock() defer startMx.Unlock() - _, err := exec.compose.apiClient().ContainerStart(ctx, op.Container.ID, client.ContainerStartOptions{}) + _, err = exec.compose.apiClient().ContainerStart(ctx, ctr.ID, client.ContainerStartOptions{}) return err } +// startTarget resolves the container a start-phase node acts on: the +// observed summary carried by the plan, or the result of the create node +// that materialized the replica earlier in the same plan. +func (exec *planExecutor) startTarget(node *PlanNode) (container.Summary, error) { + op := node.Operation + if op.Container != nil { + return *op.Container, nil + } + res := exec.pctx.get(op.CreateNodeID) + if res.ContainerID == "" { + return container.Summary{}, fmt.Errorf("internal: %s has no materialized container to act on", op.ResourceID) + } + return container.Summary{ID: res.ContainerID, Names: []string{"/" + res.ContainerName}}, nil +} + +// execWaitCondition re-observes the depends_on condition at execution time, +// delegating the polling to the same waitDependency primitive the imperative +// engine uses — Waiting, Healthy, Exited and Skipped events included. A +// best-effort node (every dependent optional) absorbs a missing dependency +// as a Skipped event instead of failing. +func (exec *planExecutor) execWaitCondition(ctx context.Context, node *PlanNode) error { + op := node.Operation + exec.containersMu.Lock() + waitingFor := exec.containersByService[op.Name].filter(isNotOneOff) + exec.containersMu.Unlock() + if len(waitingFor) == 0 { + if op.BestEffort { + exec.compose.events.On(skippedEvent("Service "+op.Name, "no container to wait for")) + return nil + } + return fmt.Errorf("required dependency %q has no container to wait for", op.Name) + } + exec.compose.events.On(containerEvents(waitingFor, waiting)...) + return exec.compose.waitDependency(ctx, op.Name, types.ServiceDependency{ + Condition: op.Condition, + Required: !op.BestEffort, + }, waitingFor) +} + +// execRunPreStart runs the service's pre_start hooks against the designated +// replica, through the same primitive as the imperative engine. +func (exec *planExecutor) execRunPreStart(ctx context.Context, node *PlanNode) error { + ctr, err := exec.startTarget(node) + if err != nil { + return err + } + return exec.compose.runPreStart(ctx, exec.project, *node.Operation.Service, ctr, exec.listener) +} + +// execRunPostStart runs the service's post_start hooks inside the started +// container, through the same primitive as the imperative engine. +func (exec *planExecutor) execRunPostStart(ctx context.Context, node *PlanNode) error { + ctr, err := exec.startTarget(node) + if err != nil { + return err + } + for _, hook := range node.Operation.Service.PostStart { + if err := exec.compose.runHook(ctx, ctr, *node.Operation.Service, hook, exec.listener); err != nil { + return err + } + } + return nil +} + func (exec *planExecutor) execStopContainer(ctx context.Context, op Operation) error { _, err := exec.compose.apiClient().ContainerStop(ctx, op.Container.ID, client.ContainerStopOptions{ Timeout: utils.DurationSecondToInt(op.Timeout), diff --git a/pkg/compose/executor_start_test.go b/pkg/compose/executor_start_test.go new file mode 100644 index 0000000000..3fd2ddc1c8 --- /dev/null +++ b/pkg/compose/executor_start_test.go @@ -0,0 +1,237 @@ +/* + Copyright 2020 Docker Compose CLI authors + + Licensed under the Apache License, Version 2.0 (the "License"); + you may not use this file except in compliance with the License. + You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + + Unless required by applicable law or agreed to in writing, software + distributed under the License is distributed on an "AS IS" BASIS, + WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + See the License for the specific language governing permissions and + limitations under the License. +*/ + +package compose + +import ( + "testing" + + "github.com/compose-spec/compose-go/v2/types" + "github.com/moby/moby/api/types/container" + "github.com/moby/moby/client" + "go.uber.org/mock/gomock" + "gotest.tools/v3/assert" + + "github.com/docker/compose/v5/pkg/api" + "github.com/docker/compose/v5/pkg/mocks" +) + +func newRecordingTestService(t *testing.T) (*composeService, *mocks.MockAPIClient, *recordingEventProcessor) { + t.Helper() + mockCtrl := gomock.NewController(t) + cli := mocks.NewMockCli(mockCtrl) + apiClient := mocks.NewMockAPIClient(mockCtrl) + cli.EXPECT().Client().Return(apiClient).AnyTimes() + + recorder := &recordingEventProcessor{} + svc, err := NewComposeService(cli, WithEventProcessor(recorder)) + assert.NilError(t, err) + return svc.(*composeService), apiClient, recorder +} + +func startPhaseNode(plan *Plan, op Operation, group string, deps ...*PlanNode) *PlanNode { + node := plan.addNode(op, group, deps...) + node.Phase = PhaseStart + return node +} + +// A start-phase node on an observed container starts it through the engine +// and emits the imperative event pair, Starting then Started. +func TestExecuteStartPhase_ObservedContainer(t *testing.T) { + svc, apiClient, recorder := newRecordingTestService(t) + + ctr := container.Summary{ + ID: "c1", + Names: []string{"/test-web-1"}, + Labels: map[string]string{ + api.ServiceLabel: "web", + api.ContainerNumberLabel: "1", + }, + } + apiClient.EXPECT().ContainerStart(gomock.Any(), "c1", gomock.Any()). + Return(client.ContainerStartResult{}, nil) + + web := types.ServiceConfig{Name: "web"} + plan := &Plan{} + startPhaseNode(plan, Operation{ + Type: OpStartContainer, + ResourceID: "service:web:1", + Cause: "start", + Service: &web, + Container: &ctr, + Name: "test-web-1", + }, "start:web:1") + + err := svc.executePlan(t.Context(), &types.Project{Name: "test"}, emptyObservedState("test"), plan) + assert.NilError(t, err) + assert.DeepEqual(t, recorder.summary(), []string{ + "Container test-web-1: Starting", + "Container test-web-1: Started", + }) +} + +// A start-phase node for a replica the same plan materialized resolves its +// target from the create node's result — the CreateNodeID mechanism. +func TestExecuteStartPhase_TargetFromCreateNode(t *testing.T) { + svc, apiClient, _ := newRecordingTestService(t) + + apiClient.EXPECT().ContainerStart(gomock.Any(), "created-id", gomock.Any()). + Return(client.ContainerStartResult{}, nil) + + web := types.ServiceConfig{Name: "web"} + plan := &Plan{} + startPhaseNode(plan, Operation{ + Type: OpStartContainer, + ResourceID: "service:web:1", + Cause: "start", + Service: &web, + Name: "test-web-1", + CreateNodeID: 7, + }, "start:web:1") + + exec := svc.newPlanExecutor(&types.Project{Name: "test"}, emptyObservedState("test")) + exec.pctx.set(7, operationResult{ContainerID: "created-id", ContainerName: "test-web-1"}) + assert.NilError(t, exec.run(t.Context(), plan)) +} + +// A best-effort wait (every dependent optional) absorbs a missing dependency +// as a Skipped event; a required one fails without touching the engine. +func TestExecuteStartPhase_WaitConditionMissingDependency(t *testing.T) { + svc, _, recorder := newRecordingTestService(t) + + plan := &Plan{} + startPhaseNode(plan, Operation{ + Type: OpWaitCondition, + ResourceID: "wait:db:service_healthy", + Cause: "depends_on condition", + Name: "db", + Condition: types.ServiceConditionHealthy, + BestEffort: true, + }, "") + + err := svc.executePlan(t.Context(), &types.Project{Name: "test"}, emptyObservedState("test"), plan) + assert.NilError(t, err) + assert.DeepEqual(t, recorder.summary(), []string{"Service db: Skipped: no container to wait for"}) + + required := &Plan{} + startPhaseNode(required, Operation{ + Type: OpWaitCondition, + ResourceID: "wait:db:service_healthy", + Cause: "depends_on condition", + Name: "db", + Condition: types.ServiceConditionHealthy, + }, "") + err = svc.executePlan(t.Context(), &types.Project{Name: "test"}, emptyObservedState("test"), required) + assert.ErrorContains(t, err, `required dependency "db" has no container to wait for`) +} + +// execWaitCondition delegates the polling to the imperative waitDependency +// primitive: the Waiting and Healthy events are the exact vocabulary users +// see today. +func TestExecuteStartPhase_WaitHealthyDelegates(t *testing.T) { + svc, apiClient, recorder := newRecordingTestService(t) + + db := container.Summary{ + ID: "db1", + Names: []string{"/test-db-1"}, + Labels: map[string]string{ + api.ServiceLabel: "db", + api.ContainerNumberLabel: "1", + }, + } + apiClient.EXPECT().ContainerInspect(gomock.Any(), "db1", gomock.Any()). + Return(client.ContainerInspectResult{Container: container.InspectResponse{ + ID: "db1", + Name: "/test-db-1", + Config: &container.Config{Healthcheck: &container.HealthConfig{Test: []string{"CMD", "true"}}}, + State: &container.State{ + Status: container.StateRunning, + Health: &container.Health{Status: container.Healthy}, + }, + }}, nil).MinTimes(1) + + observed := emptyObservedState("test") + observed.Containers["db"] = []ObservedContainer{{ID: "db1", Name: "test-db-1", Number: 1, State: container.StateRunning, Summary: db}} + + plan := &Plan{} + startPhaseNode(plan, Operation{ + Type: OpWaitCondition, + ResourceID: "wait:db:service_healthy", + Cause: "depends_on condition", + Name: "db", + Condition: types.ServiceConditionHealthy, + }, "") + + err := svc.executePlan(t.Context(), &types.Project{Name: "test"}, observed, plan) + assert.NilError(t, err) + assert.DeepEqual(t, recorder.summary(), []string{ + "Container test-db-1: Waiting", + "Container test-db-1: Healthy", + }) +} + +// The recreate group's event name is the observed container's canonical +// name, even when the create node — carrying the temporary recreate name — +// appears first in the group (regression: TestRestartWithDependencies). +func TestGroupEventNameIgnoresTemporaryName(t *testing.T) { + old := container.Summary{ + ID: "8cdaa8cc3322", + Names: []string{"/test-web-1"}, + Labels: map[string]string{ + api.ServiceLabel: "web", + api.ContainerNumberLabel: "1", + }, + } + plan := &Plan{} + create := plan.addNode(Operation{ + Type: OpCreateContainer, ResourceID: "service:web:1", Name: "8cdaa8cc3322_test-web-1", + }, "recreate:web:1") + plan.addNode(Operation{ + Type: OpStopContainer, ResourceID: "service:web:1", Container: &old, + }, "recreate:web:1", create) + + exec := &planExecutor{} + groups := exec.buildGroupTracker(plan) + assert.Equal(t, groups.groups["recreate:web:1"].eventName, "Container test-web-1") +} + +// The start group closes on the chain's last node, so Started is emitted +// after post_start hooks ran — word-for-word the imperative sequence. +func TestStartGroupEmitsStartedAfterPostStart(t *testing.T) { + web := types.ServiceConfig{Name: "web"} + plan := &Plan{} + start := startPhaseNode(plan, Operation{ + Type: OpStartContainer, ResourceID: "service:web:1", Service: &web, Name: "test-web-1", + }, "start:web:1") + post := startPhaseNode(plan, Operation{ + Type: OpRunPostStart, ResourceID: "service:web:1", Service: &web, Name: "test-web-1", + }, "start:web:1", start) + + exec := &planExecutor{} + groups := exec.buildGroupTracker(plan) + recorder := &recordingEventProcessor{} + + groups.onNodeStart(start, recorder) + groups.onNodeDone(start, recorder) + assert.DeepEqual(t, recorder.summary(), []string{"Container test-web-1: Starting"}) + + groups.onNodeStart(post, recorder) + groups.onNodeDone(post, recorder) + assert.DeepEqual(t, recorder.summary(), []string{ + "Container test-web-1: Starting", + "Container test-web-1: Started", + }) +} diff --git a/pkg/compose/reconcile.go b/pkg/compose/reconcile.go index 6c5cbb377f..202dfc2464 100644 --- a/pkg/compose/reconcile.go +++ b/pkg/compose/reconcile.go @@ -944,7 +944,8 @@ func (r *reconciler) planStartPhase() error { // materialized by a create-phase node (create) or already observed (container). type startReplica struct { resID string - number int // replica number, the start order within the service + number int // replica number, the start order within the service + name string // container name, for progress events // container is the observed container to start; nil when the create // phase materializes it. container *container.Summary @@ -964,7 +965,7 @@ type startReplica struct { // create node), and an exceptional-state restart (paused, dead, ...) plans // no create at all — the observed container sits on the node itself. func plannedReplica(resID string, number int, node *PlanNode) startReplica { - rep := startReplica{resID: resID, number: number, after: node} + rep := startReplica{resID: resID, number: number, name: node.Operation.Name, after: node} switch node.Operation.Type { case OpCreateContainer: rep.createNodeID = node.ID @@ -974,6 +975,7 @@ func plannedReplica(resID string, number int, node *PlanNode) startReplica { // exceptional-state restart (paused, dead, ...): that registration // always carries the observed container on the node itself rep.container = node.Operation.Container + rep.name = getCanonicalContainerName(*node.Operation.Container) default: // no other node type is registered in containerNodes today; leave // the target unresolved so execution fails with a clean "no @@ -1009,7 +1011,7 @@ func (r *reconciler) startPhaseReplicas(service types.ServiceConfig) (replicas [ continue } seen[resID] = true - replicas = append(replicas, startReplica{resID: resID, number: oc.Number, container: &oc.Summary}) + replicas = append(replicas, startReplica{resID: resID, number: oc.Number, name: getCanonicalContainerName(oc.Summary), container: &oc.Summary}) } for resID, node := range r.containerNodes { if !seen[resID] && strings.HasPrefix(resID, fmt.Sprintf("service:%s:", service.Name)) { @@ -1113,6 +1115,7 @@ func (r *reconciler) planServiceStart(service types.ServiceConfig) error { Cause: "pre_start hooks", Service: &serviceCopy, Container: first.container, + Name: first.name, CreateNodeID: first.createNodeID, } deps := prev @@ -1136,6 +1139,7 @@ func (r *reconciler) planServiceStart(service types.ServiceConfig) error { Cause: "start", Service: &serviceCopy, Container: rep.container, + Name: rep.name, CreateNodeID: rep.createNodeID, } deps := slices.Clone(prev) @@ -1152,6 +1156,7 @@ func (r *reconciler) planServiceStart(service types.ServiceConfig) error { Cause: "post_start hooks", Service: &serviceCopy, Container: rep.container, + Name: rep.name, CreateNodeID: op.CreateNodeID, }, group, start) post.Phase = PhaseStart From 391c2ee8fd0bdf0539bf960f99c44dd29aae98dc Mon Sep 17 00:00:00 2001 From: Nicolas De Loof Date: Tue, 1 Sep 2026 10:48:47 +0200 Subject: [PATCH 3/3] refactor: split create() into preparePlan + execute MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Pure extraction: preparePlan holds everything that precedes execution — model preparation (images, models, networks, volumes, use_api_socket), state observation, and reconciliation — and returns the canonical project, the observed snapshot, and the plan. create() becomes preparePlan → emitRunningEvents → executePlan. No behavior change; this hands a future caller (up) the three values it needs to plan create+start as one DAG and execute it once. Epic #14081, Lot 1 (3/3). Signed-off-by: Nicolas De Loof --- pkg/compose/create.go | 40 +++++++++++++++++++++++++++------------- 1 file changed, 27 insertions(+), 13 deletions(-) diff --git a/pkg/compose/create.go b/pkg/compose/create.go index b30ce1a65c..b7ea6a7ecb 100644 --- a/pkg/compose/create.go +++ b/pkg/compose/create.go @@ -66,6 +66,24 @@ func (s *composeService) Create(ctx context.Context, project *types.Project, opt } func (s *composeService) create(ctx context.Context, project *types.Project, options api.CreateOptions) error { + project, observed, plan, err := s.preparePlan(ctx, project, options) + if err != nil { + return err + } + + // Emit "Running" events for containers that are already up-to-date, so + // the progress display accounts for containers the plan will not touch. + emitRunningEvents(project, observed, plan, s.events) + + return s.executePlan(ctx, project, observed, plan) +} + +// preparePlan runs everything that precedes the execution of a create: model +// preparation (images, models, networks, volumes, use_api_socket), state +// observation, and reconciliation. It returns the canonical project (the +// use_api_socket rewrite happens here), the observed snapshot, and the plan, +// so a caller can hold all three before executing. +func (s *composeService) preparePlan(ctx context.Context, project *types.Project, options api.CreateOptions) (*types.Project, *ObservedState, *Plan, error) { if len(options.Services) == 0 { options.Services = project.ServiceNames() } @@ -77,40 +95,40 @@ func (s *composeService) create(ctx context.Context, project *types.Project, opt err := project.CheckContainerNameUnicity() if err != nil { - return err + return nil, nil, nil, err } err = s.ensureImagesExists(ctx, project, options.Build, options.QuietPull) if err != nil { - return err + return nil, nil, nil, err } err = s.ensureModels(ctx, project, options.QuietPull) if err != nil { - return err + return nil, nil, nil, err } prepareNetworks(project) externalNetworks, err := s.checkExternalNetworks(ctx, project) if err != nil { - return err + return nil, nil, nil, err } prepareVolumes(project) externalVolumes, err := s.checkExternalVolumes(ctx, project) if err != nil { - return err + return nil, nil, nil, err } // Temporary implementation of use_api_socket until we get actual support inside docker engine project, err = s.useAPISocket(project) if err != nil { - return err + return nil, nil, nil, err } observed, err := s.collectObservedState(ctx, project) if err != nil { - return err + return nil, nil, nil, err } observed.setResolvedNetworks(externalNetworks, project) observed.setResolvedVolumes(externalVolumes) @@ -126,14 +144,10 @@ func (s *composeService) create(ctx context.Context, project *types.Project, opt plan, err := reconcile(ctx, project, observed, toReconcileOptions(options), s.prompt) if err != nil { - return err + return nil, nil, nil, err } - // Emit "Running" events for containers that are already up-to-date, so - // the progress display accounts for containers the plan will not touch. - emitRunningEvents(project, observed, plan, s.events) - - return s.executePlan(ctx, project, observed, plan) + return project, observed, plan, nil } func prepareNetworks(project *types.Project) {