Cleanup: remove convergence struct, extract resolveServiceReferences

- Remove the `convergence` struct and `newConvergence` constructor
- Extract `resolveServiceReferences` as a standalone function taking
  `map[string]Containers` instead of a method on convergence
- Add `getContainersByService` helper on composeService
- Update run.go and executor.go to use the new standalone function
- Remove dead code: `getObservedState`, `setObservedState`

Co-Authored-By: Claude Opus 4.6 (1M context) <[email protected]>
Signed-off-by: Nicolas De Loof <[email protected]>

Author: ndeloof

pkg/compose/containers.go

@@ -56,6 +56,20 @@ func (s *composeService) getContainers(ctx context.Context, project string, oneO
 	return containers, nil
 }
 
+// getContainersByService returns all non-oneoff containers for the project, grouped by service name.
+func (s *composeService) getContainersByService(ctx context.Context, projectName string) (map[string]Containers, error) {
+	all, err := s.getContainers(ctx, projectName, oneOffExclude, true)
+	if err != nil {
+		return nil, err
+	}
+	result := map[string]Containers{}
+	for _, c := range all.filter(isNotOneOff) {
+		svc := c.Labels[api.ServiceLabel]
+		result[svc] = append(result[svc], c)
+	}
+	return result, nil
+}
+
 func getDefaultFilters(projectName string, oneOff oneOff, selectedServices ...string) client.Filters {
 	f := projectFilter(projectName)
 	if len(selectedServices) == 1 {

pkg/compose/convergence.go

@@ -49,37 +49,6 @@ const (
 // re-creating container, adding or removing replicas, or starting stopped containers.
 // Cross services dependencies are managed by creating services in expected order and updating `service:xx` reference
 // when a service has converged, so dependent ones can be managed with resolved containers references.
-type convergence struct {
-	compose    *composeService
-	services   map[string]Containers
-	networks   map[string]string
-	volumes    map[string]string
-	stateMutex sync.Mutex
-}
-
-func (c *convergence) getObservedState(serviceName string) Containers {
-	c.stateMutex.Lock()
-	defer c.stateMutex.Unlock()
-	return c.services[serviceName]
-}
-
-func newConvergence(services []string, state Containers, networks map[string]string, volumes map[string]string, s *composeService) *convergence {
-	observedState := map[string]Containers{}
-	for _, s := range services {
-		observedState[s] = Containers{}
-	}
-	for _, c := range state.filter(isNotOneOff) {
-		service := c.Labels[api.ServiceLabel]
-		observedState[service] = append(observedState[service], c)
-	}
-	return &convergence{
-		compose:  s,
-		services: observedState,
-		networks: networks,
-		volumes:  volumes,
-	}
-}
-
 func getScale(config types.ServiceConfig) (int, error) {
 	scale := config.GetScale()
 	if scale > 1 && config.ContainerName != "" {
@@ -90,21 +59,18 @@ func getScale(config types.ServiceConfig) (int, error) {
 	return scale, nil
 }
 
-// resolveServiceReferences replaces reference to another service with reference to an actual container
-func (c *convergence) resolveServiceReferences(service *types.ServiceConfig) error {
-	err := c.resolveVolumeFrom(service)
-	if err != nil {
+// resolveServiceReferences replaces references to other services with references
+// to actual container IDs. It resolves VolumesFrom, NetworkMode, IPC and PID
+// shared namespaces. The containersByService map provides the observed containers
+// grouped by service name.
+func resolveServiceReferences(service *types.ServiceConfig, containersByService map[string]Containers) error {
+	if err := resolveVolumeFrom(service, containersByService); err != nil {
 		return err
 	}
-
-	err = c.resolveSharedNamespaces(service)
-	if err != nil {
-		return err
-	}
-	return nil
+	return resolveSharedNamespaces(service, containersByService)
 }
 
-func (c *convergence) resolveVolumeFrom(service *types.ServiceConfig) error {
+func resolveVolumeFrom(service *types.ServiceConfig, containersByService map[string]Containers) error {
 	for i, vol := range service.VolumesFrom {
 		spec := strings.Split(vol, ":")
 		if len(spec) == 0 {
@@ -115,7 +81,7 @@ func (c *convergence) resolveVolumeFrom(service *types.ServiceConfig) error {
 			continue
 		}
 		name := spec[0]
-		dependencies := c.getObservedState(name)
+		dependencies := containersByService[name]
 		if len(dependencies) == 0 {
 			return fmt.Errorf("cannot share volume with service %s: container missing", name)
 		}
@@ -124,28 +90,25 @@ func (c *convergence) resolveVolumeFrom(service *types.ServiceConfig) error {
 	return nil
 }
 
-func (c *convergence) resolveSharedNamespaces(service *types.ServiceConfig) error {
-	str := service.NetworkMode
-	if name := getDependentServiceFromMode(str); name != "" {
-		dependencies := c.getObservedState(name)
+func resolveSharedNamespaces(service *types.ServiceConfig, containersByService map[string]Containers) error {
+	if name := getDependentServiceFromMode(service.NetworkMode); name != "" {
+		dependencies := containersByService[name]
 		if len(dependencies) == 0 {
 			return fmt.Errorf("cannot share network namespace with service %s: container missing", name)
 		}
 		service.NetworkMode = types.ContainerPrefix + dependencies.sorted()[0].ID
 	}
 
-	str = service.Ipc
-	if name := getDependentServiceFromMode(str); name != "" {
-		dependencies := c.getObservedState(name)
+	if name := getDependentServiceFromMode(service.Ipc); name != "" {
+		dependencies := containersByService[name]
 		if len(dependencies) == 0 {
 			return fmt.Errorf("cannot share IPC namespace with service %s: container missing", name)
 		}
 		service.Ipc = types.ContainerPrefix + dependencies.sorted()[0].ID
 	}
 
-	str = service.Pid
-	if name := getDependentServiceFromMode(str); name != "" {
-		dependencies := c.getObservedState(name)
+	if name := getDependentServiceFromMode(service.Pid); name != "" {
+		dependencies := containersByService[name]
 		if len(dependencies) == 0 {
 			return fmt.Errorf("cannot share PID namespace with service %s: container missing", name)
 		}

pkg/compose/executor.go

@@ -198,12 +198,11 @@ func (exec *planExecutor) execCreateContainer(ctx context.Context, node *PlanNod
 	// Resolve service references (network_mode, ipc, pid, volumes_from) to actual
 	// container IDs. This must happen at execution time because the referenced
 	// containers may have just been created by earlier plan nodes.
-	observedState, err := exec.compose.getContainers(ctx, exec.project.Name, oneOffExclude, true)
+	containersByService, err := exec.compose.getContainersByService(ctx, exec.project.Name)
 	if err != nil {
 		return err
 	}
-	conv := newConvergence(exec.project.ServiceNames(), observedState, nil, nil, exec.compose)
-	if err := conv.resolveServiceReferences(&service); err != nil {
+	if err := resolveServiceReferences(&service, containersByService); err != nil {
 		return err
 	}
 

pkg/compose/run.go

@@ -181,8 +181,7 @@ func (s *composeService) prepareRun(ctx context.Context, project *types.Project,
 		Labels:            mergeLabels(service.Labels, service.CustomLabels),
 	}
 
-	err = newConvergence(project.ServiceNames(), observedState, nil, nil, s).resolveServiceReferences(&service)
-	if err != nil {
+	if err := s.resolveRunServiceReferences(ctx, project.Name, &service); err != nil {
 		return prepareRunResult{}, err
 	}
 
@@ -269,6 +268,14 @@ func applyRunOptions(project *types.Project, service *types.ServiceConfig, opts
 	}
 }
 
+func (s *composeService) resolveRunServiceReferences(ctx context.Context, projectName string, service *types.ServiceConfig) error {
+	containersByService, err := s.getContainersByService(ctx, projectName)
+	if err != nil {
+		return err
+	}
+	return resolveServiceReferences(service, containersByService)
+}
+
 func (s *composeService) startDependencies(ctx context.Context, project *types.Project, options api.RunOptions) error {
 	project = project.WithServicesDisabled(options.Service)