diff --git a/pkg/compose/executor.go b/pkg/compose/executor.go index e3e37f1c8f..b53087b5ea 100644 --- a/pkg/compose/executor.go +++ b/pkg/compose/executor.go @@ -39,10 +39,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 @@ -116,12 +116,48 @@ 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. +// +// A planRun is single-use and driven from one goroutine: each phase runs at +// most once, Create first. +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. +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 { @@ -189,33 +225,66 @@ 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. 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 runCreate only returned once runPhase +// waited for every Create-phase node, so no goroutine can observe the +// previous value. +// +// 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. The guards below are what +// enforce it for a caller driving the phases itself; run() gets it by +// calling runCreate first. +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 diff --git a/pkg/compose/executor_start_test.go b/pkg/compose/executor_start_test.go index 392e056fb8..66b953fc55 100644 --- a/pkg/compose/executor_start_test.go +++ b/pkg/compose/executor_start_test.go @@ -1385,3 +1385,200 @@ func TestTranslateWaitTimeout(t *testing.T) { assert.Equal(t, translateWaitTimeout(context.DeadlineExceeded), context.DeadlineExceeded) assert.NilError(t, translateWaitTimeout(nil)) } + +// 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. +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{} + 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"}}, + }, "").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()) +} + +// 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, apiClient, _ := newStartPhaseTestService(t) + + var mu sync.Mutex + var lines []string + listener := func(event api.ContainerEvent) { + mu.Lock() + defer mu.Unlock() + lines = append(lines, event.Line) + } + + 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 + + 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(), listener)) + assert.DeepEqual(t, lines, []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. +func TestPlanRunRefusesAnIllegalPhaseSequence(t *testing.T) { + newRun := func(t *testing.T, createErr error) *planRun { + svc, apiClient := newTestService(t) + nw := types.NetworkConfig{Name: "test_default"} + if createErr != nil { + apiClient.EXPECT().NetworkCreate(gomock.Any(), "test_default", gomock.Any()). + Return(client.NetworkCreateResult{}, createErr) + } else { + apiClient.EXPECT().NetworkCreate(gomock.Any(), "test_default", gomock.Any()). + Return(client.NetworkCreateResult{ID: "net-id"}, nil).AnyTimes() + } + // no ContainerStart expectation: any call fails the test + plan := &Plan{} + 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"}}, + }, "").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) + 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")) + 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) + 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")) + 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) { + svc, apiClient := newTestService(t) + apiClient.EXPECT().ContainerStart(gomock.Any(), "db-id", gomock.Any()). + Return(client.ContainerStartResult{}, nil).Times(1) + plan := &Plan{} + plan.addNode(Operation{ + Type: OpStartContainer, + ResourceID: "service:db:1", + Container: &container.Summary{ID: "db-id", Names: []string{"/test-db-1"}}, + }, "").Phase = PhaseStart + 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)) + assert.ErrorContains(t, pr.runStart(t.Context(), nil), "already ran") + }) +} + +// 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, apiClient, _ := newStartPhaseTestService(t) + + var mu sync.Mutex + var lines []string + listener := func(event api.ContainerEvent) { + mu.Lock() + defer mu.Unlock() + lines = append(lines, event.Line) + } + + 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 + + exec := svc.newPlanExecutor(&types.Project{Name: "test"}, emptyObservedState("test"), listener) + assert.NilError(t, exec.run(t.Context(), plan)) + assert.DeepEqual(t, lines, []string{"post-start ok"}) +}