From bb1c232a74cb85d31e80fce9f465ff4c23144aed Mon Sep 17 00:00:00 2001 From: Nicolas De Loof Date: Mon, 17 Aug 2026 18:34:51 +0200 Subject: [PATCH] =?UTF-8?q?reconcile:=20plan=20the=20start=20phase=20?= =?UTF-8?q?=E2=80=94=20condition=20waits,=20pre=5Fstart,=20container=20sta?= =?UTF-8?q?rts?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Behind the new ReconcileOptions.PlanStart, the reconciler extends the plan past container creation: per-dependency WaitCondition nodes (service_started stays a plain DAG edge; required:false maps to Optional nodes), a once-per-service RunPreStart node — its 'only when no replica is running' rule decided at plan time from the observed state, on the lowest replica number — and one enriched StartContainer node per non-running replica, chained after its create (or reconnect) nodes, its dependencies' start nodes and its wait nodes. Start candidates are collected while walking existing containers, which surfaces a case the plan used to leave dangling: containers the plan itself stops to recreate a diverged network were never restarted by the plan — the second engine picked them up. With PlanStart they are start candidates chained after their reconnects. With PlanStart unset nothing changes: candidates are discarded and the plan stops at creation, as before (all existing plan snapshots unchanged). Covered by plan-shape unit tests: dependency wiring, Optional propagation, and both pre_start planning rules. Phase 1 groundwork for the start-in-plan convergence (#14081). Signed-off-by: Nicolas De Loof --- pkg/compose/plan.go | 17 ++- pkg/compose/reconcile.go | 234 ++++++++++++++++++++++++++++++---- pkg/compose/reconcile_test.go | 125 ++++++++++++++++++ 3 files changed, 350 insertions(+), 26 deletions(-) diff --git a/pkg/compose/plan.go b/pkg/compose/plan.go index c62c87e7f..74e5d8446 100644 --- a/pkg/compose/plan.go +++ b/pkg/compose/plan.go @@ -51,6 +51,14 @@ const ( OpRemoveContainer OperationType = 23 OpRenameContainer OperationType = 24 + // OpWaitCondition blocks until a dependency service reaches the condition + // its dependent declared (healthy, completed_successfully, ...). The + // service_started condition needs no such node: plain DAG edges express it. + OpWaitCondition OperationType = 25 + // OpRunPreStart runs a service's pre_start hooks, once per service, + // before its first container start. + OpRunPreStart OperationType = 26 + // Provider operations OpRunProvider OperationType = 30 ) @@ -80,6 +88,10 @@ func (o OperationType) String() string { return "RemoveContainer" case OpRenameContainer: return "RenameContainer" + case OpWaitCondition: + return "WaitCondition" + case OpRunPreStart: + return "RunPreStart" case OpRunProvider: return "RunProvider" default: @@ -101,8 +113,9 @@ type Operation struct { Name string // target container/resource name 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 + Timeout *time.Duration // for stop operations, and deadline of OpWaitCondition + Condition string // for OpWaitCondition: the depends_on condition to wait for + CreateNodeID int // for OpRenameContainer and OpStartContainer: ID of the CreateContainer node whose result to target (when Container is nil) // BestEffort marks an operation that tolerates one specific, expected // error inside its own execution (today: removing the old network on a // rename skips a conflict when the network is still in use — the new diff --git a/pkg/compose/reconcile.go b/pkg/compose/reconcile.go index a065969bc..e6264df14 100644 --- a/pkg/compose/reconcile.go +++ b/pkg/compose/reconcile.go @@ -54,6 +54,13 @@ type ReconcileOptions struct { Timeout *time.Duration // for stop operations RemoveOrphans bool SkipProviders bool + // PlanStart extends the plan with the start phase: dependency-condition + // waits, pre_start hooks and container starts. When false the plan stops + // at container creation and starting is left to the caller (the + // historical two-engine split). + PlanStart bool + // WaitTimeout bounds each dependency-condition wait (0 = no deadline). + WaitTimeout time.Duration } // reconciler compares a types.Project (desired state) with an ObservedState @@ -102,6 +109,11 @@ type reconciler struct { // called once per service. observedContainersByService map[string]Containers + // startNodes tracks, per service, the start-phase nodes emitted when + // options.PlanStart is set. Dependents chain their own start phase after + // these (a service is "started" once all its replicas are). + startNodes map[string][]*PlanNode + // resolvedNetworks/resolvedVolumes hold the single live resource selected per // compose key from the (possibly multi-valued) observed state — see // resolveObserved. All reconcile logic reads these, never observed.Networks/ @@ -127,6 +139,7 @@ func reconcile(_ context.Context, project *types.Project, observed *ObservedStat connectNodes: map[string][]*PlanNode{}, recreatedServices: map[string]bool{}, observedContainersByService: observed.containersByService(), + startNodes: map[string][]*PlanNode{}, } r.resolveObserved() @@ -683,9 +696,55 @@ func (r *reconciler) reconcileService(service types.ServiceConfig) error { // Collect dependency nodes that container creation should depend on infraDeps := r.infrastructureDeps(service) - var lastNode *PlanNode + lastNode, candidates := r.planExistingContainers(service, containers, expected, strategy, expectedHash, parentRecreated, infraDeps) - // Process existing containers + // Scale up: create new containers + nextNum := nextContainerNumber(r.observedSummaries(service.Name)) + for i := 0; i < expected-actual; i++ { + number := nextNum + i + name := getContainerName(r.project.Name, service, number) + svc := service // copy for pointer stability + lastNode = r.plan.addNode(Operation{ + Type: OpCreateContainer, + ResourceID: fmt.Sprintf("service:%s:%d", service.Name, number), + Cause: "no existing container", + Service: &svc, + Number: number, + Name: name, + }, "", infraDeps...) + candidates = append(candidates, startCandidate{ + number: number, + after: []*PlanNode{lastNode}, + createNodeID: lastNode.ID, + }) + } + + if r.options.PlanStart { + node, err := r.planServiceStart(service, candidates) + if err != nil { + return err + } + if node != nil { + lastNode = node + } + } + + if lastNode != nil { + r.serviceNodes[service.Name] = lastNode + } + return nil +} + +// planExistingContainers walks the service's observed containers (pre-sorted +// by sortContainers) and plans scale-downs, recreations and paused/dead +// restarts. It returns the last node emitted and the start candidates the +// start phase will act on when options.PlanStart is set. +func (r *reconciler) planExistingContainers( + service types.ServiceConfig, containers []ObservedContainer, expected int, + strategy, expectedHash string, parentRecreated bool, infraDeps []*PlanNode, +) (*PlanNode, []startCandidate) { + var lastNode *PlanNode + var candidates []startCandidate for i, oc := range containers { if i >= expected { // Scale down: stop + remove excess containers. Track the remove @@ -708,15 +767,42 @@ func (r *reconciler) reconcileService(service types.ServiceConfig) error { } if r.mustRecreate(service, expectedHash, parentRecreated, oc, strategy) { - lastNode = r.planRecreateContainer(service, &containers[i], infraDeps) + var createNode *PlanNode + lastNode, createNode = r.planRecreateContainer(service, &containers[i], infraDeps) r.recreatedServices[service.Name] = true + candidates = append(candidates, startCandidate{ + number: oc.Number, + after: []*PlanNode{lastNode}, + createNodeID: createNode.ID, + }) continue } // Container is up-to-date switch oc.State { - case container.StateRunning, container.StateCreated, container.StateRestarting, container.StateExited: - // Nothing to do (exited containers are left as-is, matching convergence.go behavior) + case container.StateRunning, container.StateRestarting: + if stopNode, stopped := r.stoppedByPlan[oc.ID]; stopped { + // The plan stops this container (e.g. to recreate a diverged + // network): it must be started again once reconnected. + after := append([]*PlanNode{stopNode}, r.connectNodes[oc.ID]...) + candidates = append(candidates, startCandidate{ + number: oc.Number, + after: after, + container: &containers[i].Summary, + }) + } else { + candidates = append(candidates, startCandidate{number: oc.Number, running: true}) + } + case container.StateCreated, container.StateExited: + // Nothing to plan on the create side: starting created/exited + // containers is the start phase's job — planned by + // planServiceStart when PlanStart is set, left to the caller's + // start engine otherwise. + candidates = append(candidates, startCandidate{ + number: oc.Number, + after: r.connectNodes[oc.ID], + container: &containers[i].Summary, + }) default: // Any other state (paused, dead, ...): attempt to (re)start lastNode = r.plan.addNode(Operation{ @@ -727,27 +813,124 @@ func (r *reconciler) reconcileService(service types.ServiceConfig) error { }, "", infraDeps...) } } + return lastNode, candidates +} - // Scale up: create new containers - nextNum := nextContainerNumber(r.observedSummaries(service.Name)) - for i := 0; i < expected-actual; i++ { - number := nextNum + i - name := getContainerName(r.project.Name, service, number) - svc := service // copy for pointer stability - lastNode = r.plan.addNode(Operation{ - Type: OpCreateContainer, - ResourceID: fmt.Sprintf("service:%s:%d", service.Name, number), - Cause: "no existing container", - Service: &svc, - Number: number, - Name: name, - }, "", infraDeps...) +// startCandidate describes one replica the start phase may have to start: +// either an existing container (container set) or one the plan creates +// (createNodeID set, resolved by the executor once the create ran). +type startCandidate struct { + number int + after []*PlanNode // nodes the start must wait for (create, reconnects, ...) + container *container.Summary // existing container, nil for plan-created ones + createNodeID int // create node whose result carries the container ID + running bool // already running and left untouched: nothing to start +} + +// planServiceStart extends the plan with the service's start phase: +// dependency-condition waits, pre_start hooks and one start per non-running +// replica. Returns the last node emitted, or nil if nothing needed starting. +func (r *reconciler) planServiceStart(service types.ServiceConfig, candidates []startCandidate) (*PlanNode, error) { + svc := service // copy for pointer stability + + var toStart []startCandidate + anyRunning := false + for _, c := range candidates { + if c.running { + anyRunning = true + continue + } + toStart = append(toStart, c) + } + if len(toStart) == 0 { + return nil, nil } - if lastNode != nil { - r.serviceNodes[service.Name] = lastNode + // Dependency conditions. service_started needs no wait node: the edges on + // the dependency's own start nodes express it. + var waitNodes []*PlanNode + var startDeps []*PlanNode + for _, dep := range sortedKeys(service.DependsOn) { + config := service.DependsOn[dep] + depNodes := r.startNodes[dep] + if len(depNodes) == 0 { + if node := r.serviceNodes[dep]; node != nil { + depNodes = []*PlanNode{node} + } + } + startDeps = append(startDeps, depNodes...) + + shouldWait, err := shouldWaitForDependency(dep, config, r.project) + if err != nil { + return nil, err + } + if !shouldWait { + continue + } + depSvc := r.project.Services[dep] + waitNodes = append(waitNodes, r.plan.addNode(Operation{ + Type: OpWaitCondition, + ResourceID: fmt.Sprintf("wait:%s:%s", service.Name, dep), + Cause: fmt.Sprintf("%s depends on %s (%s)", service.Name, dep, config.Condition), + Service: &depSvc, + Name: dep, + Condition: config.Condition, + Timeout: waitTimeout(r.options.WaitTimeout), + Optional: !config.Required, + }, "", depNodes...)) } - return nil + startDeps = append(startDeps, waitNodes...) + + // pre_start hooks run once per service, only when no replica is already + // running, on the lowest-numbered replica — decided here at plan time + // from the observed state, accepting the same observe-to-execute drift + // window as the rest of the plan. + var preStart *PlanNode + if len(service.PreStart) > 0 && !anyRunning { + lowest := toStart[0] + for _, c := range toStart[1:] { + if c.number < lowest.number { + lowest = c + } + } + deps := append(slices.Clone(startDeps), lowest.after...) + preStart = r.plan.addNode(Operation{ + Type: OpRunPreStart, + ResourceID: fmt.Sprintf("prestart:%s", service.Name), + Cause: "pre_start hooks", + Service: &svc, + Number: lowest.number, + Container: lowest.container, + CreateNodeID: lowest.createNodeID, + }, "", deps...) + } + + var last *PlanNode + for _, c := range toStart { + deps := append(slices.Clone(startDeps), c.after...) + if preStart != nil { + deps = append(deps, preStart) + } + last = r.plan.addNode(Operation{ + Type: OpStartContainer, + ResourceID: fmt.Sprintf("service:%s:%d", service.Name, c.number), + Cause: "service start", + Service: &svc, + Name: getContainerName(r.project.Name, service, c.number), + Number: c.number, + Container: c.container, + CreateNodeID: c.createNodeID, + }, "", deps...) + r.startNodes[service.Name] = append(r.startNodes[service.Name], last) + } + return last, nil +} + +func waitTimeout(d time.Duration) *time.Duration { + if d <= 0 { + return nil + } + return &d } // mustRecreate decides whether oc must be recreated to match expected. The @@ -871,7 +1054,10 @@ func (r *reconciler) hasVolumeMismatch(expected types.ServiceConfig, oc Observed // 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 { +// planRecreateContainer returns the final node of the recreate sequence and +// the create node (whose result carries the new container's ID, needed by the +// start phase). +func (r *reconciler) planRecreateContainer(service types.ServiceConfig, oc *ObservedContainer, infraDeps []*PlanNode) (*PlanNode, *PlanNode) { resID := fmt.Sprintf("service:%s:%d", service.Name, oc.Number) group := fmt.Sprintf("recreate:%s:%d", service.Name, oc.Number) tmpName := fmt.Sprintf("%s_%s", oc.ID[:min(12, len(oc.ID))], getContainerName(r.project.Name, service, oc.Number)) @@ -952,7 +1138,7 @@ func (r *reconciler) planRecreateContainer(service types.ServiceConfig, oc *Obse CreateNodeID: createNode.ID, }, group, removeNode) - return renameNode + return renameNode, createNode } // planStopDependents plans stop operations for containers of services that diff --git a/pkg/compose/reconcile_test.go b/pkg/compose/reconcile_test.go index 8f5812266..a4c2f2c8f 100644 --- a/pkg/compose/reconcile_test.go +++ b/pkg/compose/reconcile_test.go @@ -18,6 +18,7 @@ package compose import ( "fmt" + "slices" "strconv" "strings" "testing" @@ -1601,3 +1602,127 @@ func mustResolvedServiceHash(t *testing.T, svc types.ServiceConfig, containers m assert.NilError(t, err) return h } + +// --- Start-phase planning tests --- + +func TestReconcilePlanStart(t *testing.T) { + project := &types.Project{ + Name: "myproject", + Services: types.Services{ + "db": {Name: "db", Scale: intPtr(1)}, + "app": {Name: "app", Scale: intPtr(1), DependsOn: types.DependsOnConfig{ + "db": {Condition: types.ServiceConditionHealthy, Required: true}, + }}, + }, + } + observed := &ObservedState{ + ProjectName: "myproject", + Containers: map[string][]ObservedContainer{}, + Networks: map[string][]ObservedNetwork{}, + Volumes: map[string][]ObservedVolume{}, + } + + options := defaultReconcileOptions() + options.PlanStart = true + plan, err := reconcile(t.Context(), project, observed, options, declinePrompt) + assert.NilError(t, err) + assert.Equal(t, plan.String(), strings.TrimSpace(` +[] -> #1 service:db:1, CreateContainer, no existing container +[1] -> #2 service:db:1, StartContainer, service start +[2] -> #3 service:app:1, CreateContainer, no existing container +[2] -> #4 wait:app:db, WaitCondition, app depends on db (service_healthy) +[2,3,4] -> #5 service:app:1, StartContainer, service start +`)+"\n") +} + +func TestReconcilePlanStartOptionalDependency(t *testing.T) { + project := &types.Project{ + Name: "myproject", + Services: types.Services{ + "db": {Name: "db", Scale: intPtr(1)}, + "app": {Name: "app", Scale: intPtr(1), DependsOn: types.DependsOnConfig{ + "db": {Condition: types.ServiceConditionHealthy, Required: false}, + }}, + }, + } + observed := &ObservedState{ + ProjectName: "myproject", + Containers: map[string][]ObservedContainer{}, + Networks: map[string][]ObservedNetwork{}, + Volumes: map[string][]ObservedVolume{}, + } + + options := defaultReconcileOptions() + options.PlanStart = true + plan, err := reconcile(t.Context(), project, observed, options, declinePrompt) + assert.NilError(t, err) + for _, node := range plan.Nodes { + if node.Operation.Type == OpWaitCondition { + assert.Assert(t, node.Operation.Optional, "a required:false wait must be Optional") + return + } + } + t.Fatal("expected a WaitCondition node in the plan") +} + +func TestReconcilePlanStartPreStart(t *testing.T) { + project := &types.Project{ + Name: "myproject", + Services: types.Services{ + "app": {Name: "app", Scale: intPtr(2), PreStart: []types.ServiceHook{ + {Command: []string{"echo", "init"}}, + }}, + }, + } + + t.Run("hook planned once when no replica is running", func(t *testing.T) { + observed := &ObservedState{ + ProjectName: "myproject", + Containers: map[string][]ObservedContainer{}, + Networks: map[string][]ObservedNetwork{}, + Volumes: map[string][]ObservedVolume{}, + } + options := defaultReconcileOptions() + options.PlanStart = true + plan, err := reconcile(t.Context(), project, observed, options, declinePrompt) + assert.NilError(t, err) + count := 0 + for _, node := range plan.Nodes { + if node.Operation.Type == OpRunPreStart { + count++ + // every start of the service must wait for the hook + for _, other := range plan.Nodes { + if other.Operation.Type == OpStartContainer { + assert.Assert(t, slices.Contains(other.DependsOn, node), + "start %s must depend on the pre_start node", other.Operation.ResourceID) + } + } + } + } + assert.Equal(t, count, 1, "pre_start must be planned exactly once per service") + }) + + t.Run("hook not planned while a replica is already running", func(t *testing.T) { + hash, err := serviceHashWithResolvedRefs(project.Services["app"], map[string]Containers{}) + assert.NilError(t, err) + observed := &ObservedState{ + ProjectName: "myproject", + Containers: map[string][]ObservedContainer{ + "app": {{ + ID: "app1", Name: "myproject-app-1", Number: 1, + State: container.StateRunning, ConfigHash: hash, + }}, + }, + Networks: map[string][]ObservedNetwork{}, + Volumes: map[string][]ObservedVolume{}, + } + options := defaultReconcileOptions() + options.PlanStart = true + plan, err := reconcile(t.Context(), project, observed, options, declinePrompt) + assert.NilError(t, err) + for _, node := range plan.Nodes { + assert.Assert(t, node.Operation.Type != OpRunPreStart, + "scale-up beside a running replica must not re-run pre_start") + } + }) +}