From f1c9c86e73eb8178cd81467bc1fb901d3fda3628 Mon Sep 17 00:00:00 2001 From: Nicolas De Loof Date: Thu, 8 Oct 2026 11:03:45 +0200 Subject: [PATCH] refactor(executor): let a caller drive the create and start phases itself run() walked both phases in one call, keeping the state they share local and unreachable. Split it into begin, which prepares the run without dispatching anything, and a planRun whose runCreate and runStart each dispatch their own phase, so a caller can act between the two -- the interactive up will set up its attach/printer/monitor session there. run() is now begin followed by both phases, so every existing caller behaves as before. The start phase takes its hook-log listener when it is entered, since an attached caller only has one once its session exists, and refuses to run before a successful create phase or twice, keeping the create-phase barrier intact for callers that drive the phases themselves. Signed-off-by: Nicolas De Loof --- pkg/compose/executor.go | 147 ++++++++++++++----- pkg/compose/executor_start_test.go | 227 +++++++++++++++++++++++++++++ 2 files changed, 336 insertions(+), 38 deletions(-) diff --git a/pkg/compose/executor.go b/pkg/compose/executor.go index 3ffee1933e..f884b22b46 100644 --- a/pkg/compose/executor.go +++ b/pkg/compose/executor.go @@ -40,10 +40,10 @@ type planExecutor struct { pctx *reconciliationContext // listener streams pre_start/post_start hook logs, exactly like the - // imperative start path. nil until an attached caller wires one in (the - // interactive-up convergence, a later lot of #14081) — every caller today - // passes nil, so hook execution stays silent, matching today's detached - // behavior. + // imperative start path. nil until an attached caller wires one in, either + // at construction or, for a caller driving the phases itself, when it + // enters the Start phase (planRun.runStart) — nil leaves hook execution + // silent, matching detached behavior. listener api.ContainerEventListener // containersByService is a live view used to resolve service references @@ -93,11 +93,10 @@ func (pc *reconciliationContext) get(nodeID int) operationResult { // while respecting dependency edges. It emits progress events and handles // group-based event aggregation for composite operations like recreate. // -// No caller threads a hook-log listener through here yet: create() (its only -// caller) plans no start-phase operations today. newPlanExecutor takes one -// directly for that reason — it is what the interactive-up convergence (a -// later lot of #14081) will call once it needs to stream pre_start/post_start -// hook logs into an attached session. +// It threads no hook-log listener: its callers stream no pre_start/post_start +// logs. A caller that does, into an attached session (interactive up), builds +// the executor itself, drives a planRun, and hands its listener to runStart +// once its session exists. func (s *composeService) executePlan(ctx context.Context, project *types.Project, observed *ObservedState, plan *Plan) error { return s.newPlanExecutor(project, observed, nil).run(ctx, plan) } @@ -117,12 +116,71 @@ func (s *composeService) newPlanExecutor(project *types.Project, observed *Obser // run walks the plan DAG, executing nodes in parallel where possible while // respecting dependency edges. Emits progress events and handles group-based -// event aggregation for composite operations like recreate. +// event aggregation for composite operations like recreate. It is begin +// followed by both phases: a caller that must act between the Create and the +// Start phase drives a planRun itself. func (exec *planExecutor) run(ctx context.Context, plan *Plan) error { if plan.IsEmpty() { return nil } + pr, err := exec.begin(plan) + if err != nil { + return err + } + if err := pr.runCreate(ctx); err != nil { + return err + } + return pr.runStart(ctx, nil) +} + +// planRun is one execution of a Plan, split at the boundary between its two +// phases. begin builds the state both phases share (the done channels and +// failure bookkeeping a Start-phase node needs to see a Create-phase one +// through, the group tracker, the limiter); runCreate and runStart then each +// dispatch their own phase, so a caller can act in between -- interactive up +// sets up its attach/printer/monitor session there -- without a second +// snapshot or a second plan. +// +// The Create phase is a hard barrier: no Start-phase node is even +// dispatched until every Create-phase node has succeeded. This isn't +// just about not canceling unrelated in-flight Create work on a Start +// failure (phaseCancel already keeps that scoped per phase, see its own +// doc comment) -- it's that a Start-phase node whose own DependsOn +// edges are all satisfied has no reason to wait for a COMPLETELY +// UNRELATED Create-phase node elsewhere in the plan, so without this +// barrier a service could already be starting while a sibling it has no +// relationship with is still being created, or fails. The old +// create()-then-start() sequence never allowed that: the whole create +// phase, project-wide, always finished (or failed, with start() never +// invoked at all) before any dependency wait, or anything else in the +// start phase, even began. Interactive up's attach/printer session +// depends on this too: it takes exclusive hold of the terminal for the +// create phase's own progress display, handing it over to continuous +// container log streaming only once the create phase is entirely done +// -- a Start-phase node's log output interleaving with Create's +// progress bars has nowhere consistent to go. runStart's guards enforce it +// for a caller driving the phases itself; run() gets it by calling +// runCreate first. +// +// A planRun is single-use and driven from one goroutine: each phase runs at +// most once, Create first. There is one planRun per executor: two of them +// would share its reconciliation context and service view. +type planRun struct { + exec *planExecutor + rs *runState + createNodes []*PlanNode + startNodes []*PlanNode + + createRan bool + createErr error + startRan bool +} +// begin prepares plan for execution without dispatching anything: it builds +// the state shared by both phases, splits the nodes by phase and rejects a +// plan that could only deadlock. An empty plan, or one with a single phase, +// gives a planRun whose other phase dispatches nothing. +func (exec *planExecutor) begin(plan *Plan) (*planRun, error) { // Build a done-channel per node so dependents can wait done := make(map[int]chan struct{}, len(plan.Nodes)) for _, node := range plan.Nodes { @@ -203,33 +261,46 @@ func (exec *planExecutor) run(ctx context.Context, plan *Plan) error { for _, node := range createNodes { for _, dep := range node.DependsOn { if dep.Phase == PhaseStart { - return fmt.Errorf("invalid plan: create-phase node %d depends on start-phase node %d", node.ID, dep.ID) + return nil, fmt.Errorf("invalid plan: create-phase node %d depends on start-phase node %d", node.ID, dep.ID) } } } - // The Create phase is a hard barrier: no Start-phase node is even - // dispatched until every Create-phase node has succeeded. This isn't - // just about not canceling unrelated in-flight Create work on a Start - // failure (phaseCancel already keeps that scoped per phase, see its own - // doc comment) -- it's that a Start-phase node whose own DependsOn - // edges are all satisfied has no reason to wait for a COMPLETELY - // UNRELATED Create-phase node elsewhere in the plan, so without this - // barrier a service could already be starting while a sibling it has no - // relationship with is still being created, or fails. The old - // create()-then-start() sequence never allowed that: the whole create - // phase, project-wide, always finished (or failed, with start() never - // invoked at all) before any dependency wait, or anything else in the - // start phase, even began. Interactive up's attach/printer session - // depends on this too: it takes exclusive hold of the terminal for the - // create phase's own progress display, handing it over to continuous - // container log streaming only once the create phase is entirely done - // -- a Start-phase node's log output interleaving with Create's - // progress bars has nowhere consistent to go. - if err := exec.runPhase(createNodes, rs, newPhaseCancel(ctx)); err != nil { - return err + return &planRun{exec: exec, rs: rs, createNodes: createNodes, startNodes: startNodes}, nil +} + +// runCreate dispatches the Create phase and waits for it. The Start phase +// cannot be dispatched until it has returned without error. +func (r *planRun) runCreate(ctx context.Context) error { + if r.createRan { + return errors.New("plan run: the create phase already ran") + } + r.createRan = true + r.createErr = r.exec.runPhase(r.createNodes, r.rs, newPhaseCancel(ctx)) + return r.createErr +} + +// runStart dispatches the Start phase and waits for it. It refuses to run +// before a successful Create phase, and to run twice. A non-nil listener +// replaces the one the executor was built with and receives the +// pre_start/post_start hook logs; nil keeps the construction-time one. It +// can be given here because an attached caller only has one once it has set +// up its session, which happens after the Create phase. It is set before any +// Start-phase node is dispatched, and the planner only emits the operations +// that read it (pre_start, post_start) in the Start phase, so no goroutine can +// observe the previous value. +func (r *planRun) runStart(ctx context.Context, listener api.ContainerEventListener) error { + if !r.createRan || r.createErr != nil { + return errors.New("plan run: the start phase cannot run before the create phase succeeded") + } + if r.startRan { + return errors.New("plan run: the start phase already ran") + } + r.startRan = true + if listener != nil { + r.exec.listener = listener } - return exec.runPhase(startNodes, rs, newPhaseCancel(ctx)) + return r.exec.runPhase(r.startNodes, r.rs, newPhaseCancel(ctx)) } // runPhase dispatches every node in nodes concurrently (respecting their own @@ -237,8 +308,8 @@ func (exec *planExecutor) run(ctx context.Context, plan *Plan) error { // waits for them all to finish. phase is this call's own cancellation scope: // a node failing here cancels phase.ctx for its siblings in THIS call only, // never a concurrently-running call for a different phase (not that there -// is one today -- run() calls this sequentially -- but runNode has no way -// to tell, so this still matters if that ever changes). See phaseCancel's +// is one today -- a planRun runs its phases sequentially -- but runNode has +// no way to tell, so this still matters if that ever changes). See phaseCancel's // doc comment for why a node's failure goes through phase.fail instead of // errgroup.Group's own first-error-wins race. func (exec *planExecutor) runPhase(nodes []*PlanNode, rs *runState, phase *phaseCancel) error { @@ -324,12 +395,12 @@ func acquireNodeSlot(ctx context.Context, limiter *semaphore.Weighted, node *Pla return func() { releaseSlot(limiter) }, nil } -// runState carries the per-run() state every node's goroutine shares: the +// runState carries the per-planRun state every node's goroutine shares: the // done-channel per node, dependency-failure bookkeeping, group event // tracking, and the concurrency limiter. Shared across both phases (built -// once in run(), before either runs) since done/failures must stay visible +// once in begin, before either runs) since done/failures must stay visible // to a Start-phase node depending on a Create-phase one across the barrier -// between them. Split out of run() purely to keep runNode a plain method +// between them. Split out of begin purely to keep runNode a plain method // instead of a closure captured in a loop. type runState struct { done map[int]chan struct{} @@ -377,7 +448,7 @@ func (exec *planExecutor) runNode(node *PlanNode, rs *runState, phase *phaseCanc } // A dependency may have just failed without nodeCtx being canceled yet - // (see run()'s failures comment) -- skip this node exactly like a + // (see begin's failures comment) -- skip this node exactly like a // canceled one: no slot acquired, no event emitted, its own dependents // unblocked with the same failure recorded. if err := rs.failedDependency(node.DependsOn); err != nil { diff --git a/pkg/compose/executor_start_test.go b/pkg/compose/executor_start_test.go index 130004b6eb..64da00face 100644 --- a/pkg/compose/executor_start_test.go +++ b/pkg/compose/executor_start_test.go @@ -1895,3 +1895,230 @@ func TestGroupTrackerNothingStartsAfterTheFinalEvent(t *testing.T) { assert.Assert(t, !gt.onNodeStart(start, rec), "a node of a given up group must not go on") assert.DeepEqual(t, rec.resources, []api.Resource{{ID: "Container test-api-1", Status: api.Error, Text: "not started: interrupted"}}) } + +// TestPlanRunDrivesPhasesSeparately verifies that begin/runCreate/runStart +// leave a real gap between the two phases: after runCreate returned, the +// Create-phase node has fully run and no Start-phase node was dispatched -- +// the window interactive up uses to set up its attach/printer session. The +// Start-phase node depends on the Create-phase one, so the state the two +// phases share is exercised across the gap. +func TestPlanRunDrivesPhasesSeparately(t *testing.T) { + svc, apiClient := newTestService(t) + + var networkCreated, started atomic.Bool + nw := types.NetworkConfig{Name: "test_default"} + apiClient.EXPECT().NetworkCreate(gomock.Any(), "test_default", gomock.Any()). + DoAndReturn(func(context.Context, string, client.NetworkCreateOptions) (client.NetworkCreateResult, error) { + networkCreated.Store(true) + return client.NetworkCreateResult{ID: "net-id"}, nil + }) + apiClient.EXPECT().ContainerStart(gomock.Any(), "db-id", gomock.Any()). + DoAndReturn(func(context.Context, string, client.ContainerStartOptions) (client.ContainerStartResult, error) { + started.Store(true) + return client.ContainerStartResult{}, nil + }) + + plan := &Plan{} + network := plan.addNode(Operation{Type: OpCreateNetwork, ResourceID: "network:default", Name: nw.Name, Network: &nw}, "") + plan.addNode(Operation{ + Type: OpStartContainer, + ResourceID: "service:db:1", + Container: &container.Summary{ID: "db-id", Names: []string{"/test-db-1"}}, + }, "", network).Phase = PhaseStart + + exec := svc.newPlanExecutor(&types.Project{Name: "test"}, emptyObservedState("test"), nil) + pr, err := exec.begin(plan) + assert.NilError(t, err) + assert.Assert(t, !networkCreated.Load() && !started.Load(), "begin must not dispatch anything") + + assert.NilError(t, pr.runCreate(t.Context())) + assert.Assert(t, networkCreated.Load(), "the Create phase must be complete once runCreate returned") + assert.Assert(t, !started.Load(), "no Start-phase node may be dispatched before runStart") + + assert.NilError(t, pr.runStart(t.Context(), nil)) + assert.Assert(t, started.Load()) +} + +// hookLogs collects the lines of the hook logs a listener receives. +type hookLogs struct { + mu sync.Mutex + lines []string +} + +func (h *hookLogs) listener(event api.ContainerEvent) { + h.mu.Lock() + defer h.mu.Unlock() + h.lines = append(h.lines, event.Line) +} + +func (h *hookLogs) got() []string { + h.mu.Lock() + defer h.mu.Unlock() + return slices.Clone(h.lines) +} + +// postStartHookPlan mocks the exec round trip of a post_start hook printing +// "post-start ok", and returns the service running it and the plan of its +// Start phase. +func postStartHookPlan(t *testing.T) (*composeService, *Plan) { + svc, apiClient, _ := newStartPhaseTestService(t) + + serverConn, clientConn := net.Pipe() + go func() { + assert.NilError(t, writeStdcopyFrame(serverConn, 1, "post-start ok\n")) + serverConn.Close() //nolint:errcheck + }() + apiClient.EXPECT().ExecCreate(gomock.Any(), "c1", gomock.Any()). + Return(client.ExecCreateResult{ID: "exec1"}, nil) + apiClient.EXPECT().ExecAttach(gomock.Any(), "exec1", gomock.Any()). + Return(client.ExecAttachResult{HijackedResponse: client.NewHijackedResponse(clientConn, "")}, nil) + apiClient.EXPECT().ExecInspect(gomock.Any(), "exec1", gomock.Any()). + Return(client.ExecInspectResult{ExitCode: 0}, nil) + + service := types.ServiceConfig{ + Name: "web", + PostStart: []types.ServiceHook{{Command: types.ShellCommand{"/notify.sh"}}}, + } + plan := &Plan{} + plan.addNode(Operation{ + Type: OpRunPostStart, + ResourceID: "service:web:1", + Service: &service, + Container: &container.Summary{ID: "c1", Names: []string{"/test-web-1"}}, + }, "").Phase = PhaseStart + return svc, plan +} + +// TestPlanRunStartTakesTheListenerAtPhaseEntry verifies that a listener +// handed to runStart -- not the one the executor was built with, nil here -- +// receives the hook logs of the Start phase: an attached caller only has its +// printer once its session is set up, after the Create phase. +func TestPlanRunStartTakesTheListenerAtPhaseEntry(t *testing.T) { + svc, plan := postStartHookPlan(t) + var logs hookLogs + + exec := svc.newPlanExecutor(&types.Project{Name: "test"}, emptyObservedState("test"), nil) + pr, err := exec.begin(plan) + assert.NilError(t, err) + assert.NilError(t, pr.runCreate(t.Context())) + assert.NilError(t, pr.runStart(t.Context(), logs.listener)) + assert.DeepEqual(t, logs.got(), []string{"post-start ok"}) +} + +// TestPlanRunStartListenerReplacesTheConstructionTimeOne verifies that a +// listener handed to runStart takes over from the one the executor was built +// with: the hook logs go to the former only. +func TestPlanRunStartListenerReplacesTheConstructionTimeOne(t *testing.T) { + svc, plan := postStartHookPlan(t) + var constructed, given hookLogs + + exec := svc.newPlanExecutor(&types.Project{Name: "test"}, emptyObservedState("test"), constructed.listener) + pr, err := exec.begin(plan) + assert.NilError(t, err) + assert.NilError(t, pr.runCreate(t.Context())) + assert.NilError(t, pr.runStart(t.Context(), given.listener)) + assert.DeepEqual(t, given.got(), []string{"post-start ok"}) + assert.Equal(t, len(constructed.got()), 0) +} + +// TestExecutePlanKeepsTheConstructionTimeListener verifies that run() -- which +// hands no listener to runStart -- still streams the Start phase's hook logs +// to the one the executor was built with, and that a nil runStart listener +// never drops it. +func TestExecutePlanKeepsTheConstructionTimeListener(t *testing.T) { + svc, plan := postStartHookPlan(t) + var logs hookLogs + + exec := svc.newPlanExecutor(&types.Project{Name: "test"}, emptyObservedState("test"), logs.listener) + assert.NilError(t, exec.run(t.Context(), plan)) + assert.DeepEqual(t, logs.got(), []string{"post-start ok"}) +} + +// TestPlanRunRefusesAnIllegalPhaseSequence verifies the guards that keep the +// Create-phase barrier intact for a caller driving the phases itself: the +// Start phase never runs before the Create phase, after a failed one, or +// twice; and a phase is never run twice either. Each mocked call is expected +// exactly as often as the sequence makes it, and none when it makes none: a +// call beyond that, or a missing one, fails the test. +func TestPlanRunRefusesAnIllegalPhaseSequence(t *testing.T) { + // newRun plans a network create followed by a container start; createErr + // is what the network create returns, creates and starts how many + // NetworkCreate and ContainerStart calls the sequence under test makes. + newRun := func(t *testing.T, createErr error, creates, starts int) *planRun { + svc, apiClient := newTestService(t) + nw := types.NetworkConfig{Name: "test_default"} + if creates > 0 { + result := client.NetworkCreateResult{ID: "net-id"} + if createErr != nil { + result = client.NetworkCreateResult{} + } + apiClient.EXPECT().NetworkCreate(gomock.Any(), "test_default", gomock.Any()). + Return(result, createErr).Times(creates) + } + if starts > 0 { + apiClient.EXPECT().ContainerStart(gomock.Any(), "db-id", gomock.Any()). + Return(client.ContainerStartResult{}, nil).Times(starts) + } + plan := &Plan{} + network := plan.addNode(Operation{Type: OpCreateNetwork, ResourceID: "network:default", Name: nw.Name, Network: &nw}, "") + plan.addNode(Operation{ + Type: OpStartContainer, + ResourceID: "service:db:1", + Container: &container.Summary{ID: "db-id", Names: []string{"/test-db-1"}}, + }, "", network).Phase = PhaseStart + pr, err := svc.newPlanExecutor(&types.Project{Name: "test"}, emptyObservedState("test"), nil).begin(plan) + assert.NilError(t, err) + return pr + } + + t.Run("start before create", func(t *testing.T) { + pr := newRun(t, nil, 0, 0) + assert.ErrorContains(t, pr.runStart(t.Context(), nil), "before the create phase succeeded") + }) + t.Run("start after a failed create", func(t *testing.T) { + pr := newRun(t, errors.New("boom"), 1, 0) + assert.ErrorContains(t, pr.runCreate(t.Context()), "boom") + assert.ErrorContains(t, pr.runStart(t.Context(), nil), "before the create phase succeeded") + }) + t.Run("create twice", func(t *testing.T) { + pr := newRun(t, nil, 1, 0) + assert.NilError(t, pr.runCreate(t.Context())) + assert.ErrorContains(t, pr.runCreate(t.Context()), "already ran") + }) + t.Run("create again after a failed create", func(t *testing.T) { + pr := newRun(t, errors.New("boom"), 1, 0) + assert.ErrorContains(t, pr.runCreate(t.Context()), "boom") + assert.ErrorContains(t, pr.runCreate(t.Context()), "already ran") + }) + t.Run("start twice", func(t *testing.T) { + pr := newRun(t, nil, 1, 1) + assert.NilError(t, pr.runCreate(t.Context())) + assert.NilError(t, pr.runStart(t.Context(), nil)) + assert.ErrorContains(t, pr.runStart(t.Context(), nil), "already ran") + }) +} + +// TestPlanRunOfAnEmptyOrSinglePhasePlan verifies what +// begin documents: an empty plan, or one with a single phase, gives a planRun +// whose phases run without error and dispatch only the nodes the plan has. +func TestPlanRunOfAnEmptyOrSinglePhasePlan(t *testing.T) { + t.Run("empty plan", func(t *testing.T) { + svc, _ := newTestService(t) // any API call fails the test + pr, err := svc.newPlanExecutor(&types.Project{Name: "test"}, emptyObservedState("test"), nil).begin(&Plan{}) + assert.NilError(t, err) + assert.NilError(t, pr.runCreate(t.Context())) + assert.NilError(t, pr.runStart(t.Context(), nil)) + }) + t.Run("create phase only", func(t *testing.T) { + svc, apiClient := newTestService(t) // no ContainerStart expectation: any start fails the test + apiClient.EXPECT().NetworkCreate(gomock.Any(), "test_default", gomock.Any()). + Return(client.NetworkCreateResult{ID: "net-id"}, nil).Times(1) + nw := types.NetworkConfig{Name: "test_default"} + plan := &Plan{} + plan.addNode(Operation{Type: OpCreateNetwork, ResourceID: "network:default", Name: nw.Name, Network: &nw}, "") + pr, err := svc.newPlanExecutor(&types.Project{Name: "test"}, emptyObservedState("test"), nil).begin(plan) + assert.NilError(t, err) + assert.NilError(t, pr.runCreate(t.Context())) + assert.NilError(t, pr.runStart(t.Context(), nil)) + }) +}