refactor(executor): let a caller drive the create and start phases itself

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

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"})
+}