plan engine: the start phase (epic #14081, Lot 1)

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) {

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

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

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),