refactor: move compose and deploy packages under client

pkg/client/deploy/container.go

@@ -0,0 +1,29 @@
+package deploy
+
+import (
+	"fmt"
+	"github.com/psviderski/uncloud/pkg/api"
+)
+
+type ContainerSpecStatus string
+
+const ContainerUpToDate ContainerSpecStatus = "up-to-date"
+const ContainerNeedsUpdate ContainerSpecStatus = "needs-update"
+const ContainerNeedsRecreate ContainerSpecStatus = "needs-recreate"
+
+func CompareContainerToSpec(ctr api.Container, spec api.ServiceSpec) (ContainerSpecStatus, error) {
+	specHash, err := spec.ImmutableHash()
+	if err != nil {
+		return "", fmt.Errorf("calculate immutable hash for service spec: %w", err)
+	}
+
+	// Is the hash label is unset, there is no easy way to compare its configuration with the spec,
+	// so let's recreate as well.
+	if ctr.Config.Labels[api.LabelServiceSpecHash] != specHash {
+		return ContainerNeedsRecreate, nil
+	}
+
+	// TODO: compare mutable properties such as memory or CPU limits when they are implemented.
+
+	return ContainerUpToDate, nil
+}

pkg/client/deploy/deploy.go

@@ -0,0 +1,120 @@
+package deploy
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"github.com/psviderski/uncloud/internal/machine/api/pb"
+	"github.com/psviderski/uncloud/pkg/api"
+)
+
+type Client interface {
+	api.ContainerClient
+	api.MachineClient
+	api.ServiceClient
+}
+
+// Deployment manages the process of creating or updating a service to match a desired state.
+// It coordinates the validation, planning, and execution of deployment operations.
+type Deployment struct {
+	Service  *api.Service
+	Spec     api.ServiceSpec
+	Strategy Strategy
+	cli      Client
+	plan     *Plan
+}
+
+type Plan struct {
+	ServiceID   string
+	ServiceName string
+	SequenceOperation
+}
+
+// MachineFilter determines which machines participate in a deployment operation by returning true for
+// machines that should be included.
+type MachineFilter func(m *pb.MachineInfo) bool
+
+var ErrNoMatchingMachines = errors.New("no machines match the filter")
+
+// NewDeployment creates a new deployment for the given service specification.
+// If strategy is nil, a default RollingStrategy will be used.
+func NewDeployment(cli Client, spec api.ServiceSpec, strategy Strategy) *Deployment {
+	if strategy == nil {
+		strategy = &RollingStrategy{}
+	}
+
+	return &Deployment{
+		Spec:     spec,
+		Strategy: strategy,
+		cli:      cli,
+	}
+}
+
+// Plan returns a plan of operations to reconcile the service to the desired state.
+// If a plan has already been created, the same plan will be returned.
+func (d *Deployment) Plan(ctx context.Context) (Plan, error) {
+	if d.plan != nil {
+		return *d.plan, nil
+	}
+
+	// Validate the new spec before planning.
+	if err := d.Validate(ctx); err != nil {
+		return Plan{}, fmt.Errorf("invalid deployment: %w", err)
+	}
+
+	plan, err := d.Strategy.Plan(ctx, d.cli, d.Service, d.Spec)
+	if err != nil {
+		return Plan{}, fmt.Errorf("create plan using %s strategy: %w", d.Strategy.Type(), err)
+	}
+	d.plan = &plan
+
+	return plan, nil
+}
+
+// Validate checks if the deployment specification is valid.
+func (d *Deployment) Validate(ctx context.Context) error {
+	if err := d.Spec.Validate(); err != nil {
+		return fmt.Errorf("invalid service spec: %w", err)
+	}
+	if d.Spec.Name == "" {
+		return errors.New("service name is required")
+	}
+
+	if d.Service == nil {
+		svc, err := d.cli.InspectService(ctx, d.Spec.Name)
+		if err == nil {
+			d.Service = &svc
+		} else if !errors.Is(err, api.ErrNotFound) {
+			return fmt.Errorf("inspect service: %w", err)
+		}
+	}
+	// d.Service is nil if the service doesn't exist yet (first deployment).
+	if d.Service == nil {
+		return nil
+	}
+
+	if d.Service.Name != d.Spec.Name {
+		return errors.New("service name cannot be changed")
+	}
+	if d.Service.Mode != d.Spec.Mode {
+		return errors.New("service mode cannot be changed")
+	}
+	if d.Spec.Mode == api.ServiceModeReplicated && d.Spec.Replicas < 1 {
+		return errors.New("number of replicas must be at least 1")
+	}
+
+	return nil
+}
+
+// Run executes the deployment plan and returns the ID of the created or updated service.
+// It will create a new plan if one hasn't been created yet. The deployment will either create a new service or update
+// the existing one to match the desired specification.
+// TODO: forbid to run the same deployment more than once.
+func (d *Deployment) Run(ctx context.Context) (Plan, error) {
+	plan, err := d.Plan(ctx)
+	if err != nil {
+		return plan, fmt.Errorf("create plan: %w", err)
+	}
+
+	return plan, plan.Execute(ctx, d.cli)
+}

pkg/client/deploy/operation.go

@@ -0,0 +1,142 @@
+package deploy
+
+import (
+	"context"
+	"fmt"
+	"github.com/docker/docker/api/types/container"
+	"github.com/psviderski/uncloud/pkg/api"
+	"strings"
+)
+
+// Operation represents a single atomic operation in a deployment process.
+// Operations can be composed to form complex deployment strategies.
+type Operation interface {
+	// Execute performs the operation using the provided client.
+	// TODO: Encapsulate the client in the operation as otherwise it gives an impression that different clients
+	//  can be provided. But in reality, the operation is tightly coupled with the client that was used to create it.
+	Execute(ctx context.Context, cli Client) error
+	// Format returns a human-readable representation of the operation.
+	Format(resolver NameResolver) string
+	String() string
+}
+
+// NameResolver resolves machine and container IDs to their names.
+type NameResolver interface {
+	MachineName(machineID string) string
+	ContainerName(containerID string) string
+}
+
+// RunContainerOperation creates and starts a new container on a specific machine.
+type RunContainerOperation struct {
+	ServiceID string
+	Spec      api.ServiceSpec
+	MachineID string
+}
+
+func (o *RunContainerOperation) Execute(ctx context.Context, cli Client) error {
+	resp, err := cli.CreateContainer(ctx, o.ServiceID, o.Spec, o.MachineID)
+	if err != nil {
+		return fmt.Errorf("create container: %w", err)
+	}
+	if err = cli.StartContainer(ctx, o.ServiceID, resp.ID); err != nil {
+		return fmt.Errorf("start container: %w", err)
+	}
+
+	// TODO: wait for the container to become healthy
+
+	return nil
+}
+
+func (o *RunContainerOperation) Format(resolver NameResolver) string {
+	machineName := resolver.MachineName(o.MachineID)
+	return fmt.Sprintf("%s: Run container [image=%s]", machineName, o.Spec.Container.Image)
+}
+
+func (o *RunContainerOperation) String() string {
+	return fmt.Sprintf("RunContainerOperation[service_id=%s, image=%s, machine_id=%s]",
+		o.ServiceID, o.Spec.Container.Image, o.MachineID)
+}
+
+// StopContainerOperation stops a container on a specific machine.
+type StopContainerOperation struct {
+	ServiceID   string
+	ContainerID string
+	MachineID   string
+}
+
+func (o *StopContainerOperation) Execute(ctx context.Context, cli Client) error {
+	if err := cli.StopContainer(ctx, o.ServiceID, o.ContainerID, container.StopOptions{}); err != nil {
+		return fmt.Errorf("stop container: %w", err)
+	}
+	return nil
+}
+
+func (o *StopContainerOperation) Format(resolver NameResolver) string {
+	machineName := resolver.MachineName(o.MachineID)
+	return fmt.Sprintf("%s: Stop container [name=%s]", machineName, resolver.ContainerName(o.ContainerID))
+}
+
+func (o *StopContainerOperation) String() string {
+	return fmt.Sprintf("StopContainerOperation[service_id=%s, container_id=%s, machine_id=%s]",
+		o.ServiceID, o.ContainerID, o.MachineID)
+}
+
+// RemoveContainerOperation stops and removes a container from a specific machine.
+type RemoveContainerOperation struct {
+	ServiceID   string
+	ContainerID string
+	MachineID   string
+}
+
+func (o *RemoveContainerOperation) Execute(ctx context.Context, cli Client) error {
+	if err := cli.StopContainer(ctx, o.ServiceID, o.ContainerID, container.StopOptions{}); err != nil {
+		return fmt.Errorf("stop container: %w", err)
+	}
+	if err := cli.RemoveContainer(ctx, o.ServiceID, o.ContainerID, container.RemoveOptions{}); err != nil {
+		return fmt.Errorf("remove container: %w", err)
+	}
+
+	return nil
+}
+
+func (o *RemoveContainerOperation) Format(resolver NameResolver) string {
+	machineName := resolver.MachineName(o.MachineID)
+	return fmt.Sprintf("%s: Remove container [name=%s]", machineName, resolver.ContainerName(o.ContainerID))
+}
+
+func (o *RemoveContainerOperation) String() string {
+	return fmt.Sprintf("RemoveContainerOperation[service_id=%s, container_id=%s, machine_id=%s]",
+		o.ServiceID, o.ContainerID, o.MachineID)
+}
+
+// SequenceOperation is a composite operation that executes a sequence of operations in order.
+type SequenceOperation struct {
+	Operations []Operation
+}
+
+func (o *SequenceOperation) Execute(ctx context.Context, cli Client) error {
+	for _, op := range o.Operations {
+		if err := op.Execute(ctx, cli); err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
+func (o *SequenceOperation) Format(resolver NameResolver) string {
+	ops := make([]string, len(o.Operations))
+	for i, op := range o.Operations {
+		ops[i] = "- " + op.Format(resolver)
+	}
+
+	return strings.Join(ops, "\n")
+}
+
+func (o *SequenceOperation) String() string {
+	ops := make([]string, len(o.Operations))
+	for i, op := range o.Operations {
+		ops[i] = op.String()
+	}
+
+	return fmt.Sprintf("SequenceOperation[%s]", strings.Join(ops, ", "))
+}

pkg/client/deploy/resolver.go

@@ -0,0 +1,133 @@
+package deploy
+
+import (
+	"fmt"
+	"github.com/distribution/reference"
+	"github.com/psviderski/uncloud/internal/secret"
+	"github.com/psviderski/uncloud/pkg/api"
+	"strings"
+)
+
+type ImageDigestResolver interface {
+	Resolve(image string) (string, error)
+}
+
+// ServiceSpecResolver transforms user-provided service specs into deployment-ready form.
+type ServiceSpecResolver struct {
+	ClusterDomain string
+	ImageResolver ImageDigestResolver
+}
+
+// Resolve transforms a service spec into its fully resolved form ready for deployment.
+func (r *ServiceSpecResolver) Resolve(spec *api.ServiceSpec) error {
+	if err := spec.Validate(); err != nil {
+		return fmt.Errorf("invalid service spec: %w", err)
+	}
+
+	steps := []func(*api.ServiceSpec) error{
+		r.applyDefaults,
+		r.resolveServiceName,
+		r.resolveImageDigest,
+		r.expandIngressPorts,
+	}
+
+	for _, step := range steps {
+		if err := step(spec); err != nil {
+			return err
+		}
+	}
+
+	return nil
+}
+
+func (r *ServiceSpecResolver) applyDefaults(spec *api.ServiceSpec) error {
+	if spec.Mode == "" {
+		spec.Mode = api.ServiceModeReplicated
+	}
+	// Ensure the replicated service has at least one replica.
+	if spec.Mode == api.ServiceModeReplicated && spec.Replicas == 0 {
+		spec.Replicas = 1
+	}
+
+	return nil
+}
+
+func (r *ServiceSpecResolver) resolveServiceName(spec *api.ServiceSpec) error {
+	if spec.Name != "" {
+		return nil
+	}
+
+	// Generate a random service name from the image when not provided.
+	img, err := reference.ParseDockerRef(spec.Container.Image)
+	if err != nil {
+		return fmt.Errorf("invalid image: %w", err)
+	}
+	// Get the image name without the repository and tag/digest parts.
+	imageName := reference.FamiliarName(img)
+	// Get the last part of the image name (path), e.g. "nginx" from "bitnami/nginx".
+	if i := strings.LastIndex(imageName, "/"); i != -1 {
+		imageName = imageName[i+1:]
+	}
+	// Append a random suffix to the image name to generate an optimistically unique service name.
+	suffix, err := secret.RandomAlphaNumeric(4)
+	if err != nil {
+		return fmt.Errorf("generate random suffix: %w", err)
+	}
+	spec.Name = fmt.Sprintf("%s-%s", imageName, suffix)
+
+	return nil
+}
+
+func (r *ServiceSpecResolver) resolveImageDigest(spec *api.ServiceSpec) error {
+	if r.ImageResolver == nil {
+		// Skip digest resolution when no resolver is provided.
+		return nil
+	}
+
+	imageDigest, err := r.ImageResolver.Resolve(spec.Container.Image)
+	if err != nil {
+		return fmt.Errorf("resolve image digest: %w", err)
+	}
+	spec.Container.Image = imageDigest
+
+	return nil
+}
+
+// expandIngressPorts processes HTTP(S) ingress ports in a service spec by:
+// 1. Setting a default hostname (service-name.cluster-domain) for ports without a hostname.
+// 2. Duplicating a port with a cluster domain hostname for ports with external domains.
+// This ensures every ingress port is accessible via the cluster domain, while preserving any custom domains specified
+// by the user.
+func (r *ServiceSpecResolver) expandIngressPorts(spec *api.ServiceSpec) error {
+	for i, port := range spec.Ports {
+		if port.Protocol != api.ProtocolHTTP && port.Protocol != api.ProtocolHTTPS {
+			continue
+		}
+
+		if port.Hostname == "" {
+			if r.ClusterDomain == "" {
+				return fmt.Errorf("cluster domain must be reserved to generate hostname for ingress port: %d/%s",
+					port.ContainerPort, port.Protocol)
+			}
+			// Assign the default hostname (service-name.cluster-domain).
+			spec.Ports[i].Hostname = fmt.Sprintf("%s.%s", spec.Name, r.ClusterDomain)
+		} else {
+			if r.ClusterDomain == "" {
+				// When no cluster domain is reserved, use only the provided hostname.
+				continue
+			}
+
+			if strings.HasSuffix(port.Hostname, "."+r.ClusterDomain) {
+				// If the hostname is already a cluster subdomain, use as is.
+				continue
+			}
+			// For external domains, duplicate the port with a service-name.cluster-domain hostname so the service
+			// can be accessed via both hostnames.
+			newPort := port
+			newPort.Hostname = fmt.Sprintf("%s.%s", spec.Name, r.ClusterDomain)
+			spec.Ports = append(spec.Ports, newPort)
+		}
+	}
+
+	return nil
+}

pkg/client/deploy/strategy.go

@@ -0,0 +1,377 @@
+package deploy
+
+import (
+	"context"
+	"fmt"
+	"github.com/psviderski/uncloud/internal/machine/api/pb"
+	"github.com/psviderski/uncloud/internal/secret"
+	"github.com/psviderski/uncloud/pkg/api"
+	"math/rand/v2"
+	"slices"
+)
+
+// Strategy defines how a service should be deployed or updated. Different implementations can provide various
+// deployment patterns such as rolling updates, blue/green deployments, etc.
+type Strategy interface {
+	// Type returns the type of the deployment strategy, e.g. "rolling", "blue-green".
+	Type() string
+	// Plan returns the operation to reconcile the service to the desired state.
+	// If the service does not exist (new deployment), svc will be nil.
+	Plan(ctx context.Context, cli api.MachineClient, svc *api.Service, spec api.ServiceSpec) (Plan, error)
+}
+
+// RollingStrategy implements a rolling update deployment pattern where containers are updated one at a time
+// to minimize service disruption.
+type RollingStrategy struct {
+	// MachineFilter optionally restricts which machines can be used for deployment.
+	MachineFilter MachineFilter
+}
+
+func (s *RollingStrategy) Type() string {
+	return "rolling"
+}
+
+func (s *RollingStrategy) Plan(
+	ctx context.Context, cli api.MachineClient, svc *api.Service, spec api.ServiceSpec,
+) (Plan, error) {
+	// We can assume that the spec is valid at this point because it has been validated by the deployment.
+	switch spec.Mode {
+	case api.ServiceModeReplicated:
+		return s.planReplicated(ctx, cli, svc, spec)
+	case api.ServiceModeGlobal:
+		return s.planGlobal(ctx, cli, svc, spec)
+	default:
+		return Plan{}, fmt.Errorf("unsupported service mode: '%s'", spec.Mode)
+	}
+}
+
+// planReplicated creates a plan for a replicated service deployment.
+// For replicated services, we want to maintain a specific number of containers (replicas) across the available machines
+// in the cluster.
+func (s *RollingStrategy) planReplicated(
+	ctx context.Context, cli api.MachineClient, svc *api.Service, spec api.ServiceSpec,
+) (Plan, error) {
+	plan, err := newEmptyPlan(svc, spec)
+	if err != nil {
+		return plan, err
+	}
+
+	machines, err := cli.ListMachines(ctx)
+	if err != nil {
+		return plan, fmt.Errorf("list machines: %w", err)
+	}
+	// Filter machines that are not DOWN and match the machine filter if provided.
+	var availableMachines []*pb.MachineInfo
+	var unmatchedMachines []*pb.MachineInfo
+	var downMachines []*pb.MachineInfo
+	for _, m := range machines {
+		if m.State == pb.MachineMember_DOWN {
+			downMachines = append(downMachines, m.Machine)
+		} else {
+			if s.MachineFilter == nil || s.MachineFilter(m.Machine) {
+				availableMachines = append(availableMachines, m.Machine)
+			} else {
+				unmatchedMachines = append(unmatchedMachines, m.Machine)
+			}
+		}
+	}
+
+	if len(availableMachines) == 0 {
+		if s.MachineFilter != nil {
+			return plan, ErrNoMatchingMachines
+		}
+		return plan, fmt.Errorf("no available machines to deploy service")
+	}
+	// Randomise the order of machines to avoid always deploying to the same machines first.
+	rand.Shuffle(len(availableMachines), func(i, j int) {
+		availableMachines[i], availableMachines[j] = availableMachines[j], availableMachines[i]
+	})
+
+	// Organise existing containers by machine.
+	containersOnMachine := make(map[string][]api.Container)
+	upToDateContainersOnMachine := make(map[string]int)
+	containerSpecStatuses := make(map[string]ContainerSpecStatus)
+	if svc != nil {
+		for _, c := range svc.Containers {
+			if !c.Container.State.Running || c.Container.State.Paused {
+				// Skip containers that are not running.
+				continue
+			}
+
+			status, err := CompareContainerToSpec(c.Container, spec)
+			if err != nil {
+				return plan, fmt.Errorf("compare container to spec: %w", err)
+			}
+			containerSpecStatuses[c.Container.ID] = status
+
+			if status == ContainerUpToDate {
+				upToDateContainersOnMachine[c.MachineID] += 1
+			}
+		}
+
+		// Sort containers such that running containers with the desired spec are first.
+		slices.SortFunc(svc.Containers, func(c1, c2 api.MachineContainer) int {
+			if status, ok := containerSpecStatuses[c1.Container.ID]; ok && status == ContainerUpToDate {
+				return -1
+			}
+			if status, ok := containerSpecStatuses[c2.Container.ID]; ok && status == ContainerUpToDate {
+				return 1
+			}
+			return 0
+		})
+
+		for _, c := range svc.Containers {
+			containersOnMachine[c.MachineID] = append(containersOnMachine[c.MachineID], c.Container)
+		}
+
+		// Sort machines such that machines with the most up-to-date containers are first, followed by machines with
+		// existing containers, and finally machines without containers.
+		slices.SortFunc(availableMachines, func(m1, m2 *pb.MachineInfo) int {
+			if upToDateContainersOnMachine[m1.Id] > 0 && upToDateContainersOnMachine[m2.Id] > 0 {
+				return upToDateContainersOnMachine[m2.Id] - upToDateContainersOnMachine[m1.Id]
+			}
+			if upToDateContainersOnMachine[m1.Id] > 0 {
+				return -1
+			}
+			if upToDateContainersOnMachine[m2.Id] > 0 {
+				return 1
+			}
+			return len(containersOnMachine[m2.Id]) - len(containersOnMachine[m1.Id])
+		})
+	}
+
+	// Spread the containers across the available machines evenly using a simple round-robin approach, starting with
+	// machines that already have containers and prioritising machines with containers that match the desired spec.
+	for i := 0; i < int(spec.Replicas); i++ {
+		m := availableMachines[i%len(availableMachines)]
+		containers := containersOnMachine[m.Id]
+
+		if len(containers) == 0 {
+			// No more existing containers on this machine, create a new one.
+			plan.Operations = append(plan.Operations, &RunContainerOperation{
+				ServiceID: plan.ServiceID,
+				Spec:      spec,
+				MachineID: m.Id,
+			})
+			continue
+		}
+
+		ctr := containers[0]
+		containersOnMachine[m.Id] = containers[1:]
+
+		if status, ok := containerSpecStatuses[ctr.ID]; ok { // Contains statuses for only running containers.
+			if status == ContainerUpToDate {
+				continue
+			}
+			// TODO: handle ContainerNeedsUpdate when update of mutable fields on a container is supported.
+
+			conflictingPorts, portsErr := ctr.ConflictingServicePorts(spec.Ports)
+			if portsErr != nil || len(conflictingPorts) > 0 {
+				// Stop the malformed container or the container with conflicting ports.
+				plan.Operations = append(plan.Operations, &StopContainerOperation{
+					ServiceID:   plan.ServiceID,
+					ContainerID: ctr.ID,
+					MachineID:   m.Id,
+				})
+			}
+		}
+
+		// Run a new container.
+		plan.Operations = append(plan.Operations, &RunContainerOperation{
+			ServiceID: plan.ServiceID,
+			Spec:      spec,
+			MachineID: m.Id,
+		})
+
+		// Remove the old container.
+		plan.Operations = append(plan.Operations, &RemoveContainerOperation{
+			ServiceID:   plan.ServiceID,
+			ContainerID: ctr.ID,
+			MachineID:   m.Id,
+		})
+	}
+
+	// Remove any remaining containers that are not needed.
+	for mid, containers := range containersOnMachine {
+		for _, c := range containers {
+			plan.Operations = append(plan.Operations, &RemoveContainerOperation{
+				ServiceID:   plan.ServiceID,
+				ContainerID: c.ID,
+				MachineID:   mid,
+			})
+		}
+	}
+
+	return plan, nil
+}
+
+// planGlobal creates a plan for a global service deployment, ensuring one container runs on each available machine.
+// For machines with an existing container, it attempts to start a new container before removing the old one if
+// possible. If the new container would have port conflicts with the existing one, the old container is removed first.
+// It handles multiple containers per machine (though this should not occur in normal operation) and skips machines
+// that are down.
+func (s *RollingStrategy) planGlobal(
+	ctx context.Context, cli api.MachineClient, svc *api.Service, spec api.ServiceSpec,
+) (Plan, error) {
+	plan, err := newEmptyPlan(svc, spec)
+	if err != nil {
+		return plan, err
+	}
+
+	// Map machineID to service containers on that machine. For the global mode, there should be at most one
+	// container per machine but we use a slice to handle multiple containers that may exist due to a bug
+	// or interruption in the previous deployment.
+	containersOnMachine := make(map[string][]api.MachineContainer)
+	if svc != nil {
+		for _, c := range svc.Containers {
+			containersOnMachine[c.MachineID] = append(containersOnMachine[c.MachineID], c)
+		}
+	}
+
+	machines, err := cli.ListMachines(ctx)
+	if err != nil {
+		return plan, fmt.Errorf("list machines: %w", err)
+	}
+	// Filter machines if a machine filter is provided.
+	// TODO: not sure this is the right behaviour to ignore other machines that might run service containers.
+	//  Maybe there should be another filter to specify which machines to deploy to but keep the rest running.
+	//  Could be useful to test a new version on a subset of machines before rolling out to all.
+	if s.MachineFilter != nil {
+		machines = slices.DeleteFunc(machines, func(m *pb.MachineMember) bool {
+			return !s.MachineFilter(m.Machine)
+		})
+		if len(machines) == 0 {
+			return plan, ErrNoMatchingMachines
+		}
+	}
+
+	// TODO: figure out how to return a warning if there are machines down. Embed the machinesDown in the plan?
+	var machinesDown []*pb.MachineInfo
+	for _, m := range machines {
+		// Skip machines that are down but collect them to report a warning later.
+		if m.State == pb.MachineMember_DOWN {
+			machinesDown = append(machinesDown, m.Machine)
+			fmt.Printf("WARNING: failed to run a service container on machine '%s' which is Down.\n", m.Machine.Id)
+			continue
+		}
+
+		containers := containersOnMachine[m.Machine.Id]
+		ops, err := reconcileGlobalContainer(containers, spec, plan.ServiceID, m.Machine.Id)
+		if err != nil {
+			return plan, err
+		}
+		plan.Operations = append(plan.Operations, ops...)
+	}
+
+	return plan, nil
+}
+
+// reconcileGlobalContainer returns a sequence of operations to reconcile containers on a machine for a global service.
+// It ensures exactly one container with the desired spec is running on the machine by creating a new container and
+// removing old ones. If there is a host port conflict, it stops the old container before starting a new one.
+func reconcileGlobalContainer(
+	containers []api.MachineContainer, spec api.ServiceSpec, serviceID, machineID string,
+) ([]Operation, error) {
+	var ops []Operation
+
+	if len(containers) == 0 {
+		// No containers on this machine, create a new one.
+		ops = append(ops, &RunContainerOperation{
+			ServiceID: serviceID,
+			Spec:      spec,
+			MachineID: machineID,
+		})
+		return ops, nil
+	}
+
+	// Check if there is a container with the same spec already running. If so, remove the rest.
+	upToDate := false
+	for i, c := range containers {
+		if !c.Container.State.Running || c.Container.State.Paused {
+			// Skip containers that are not running.
+			continue
+		}
+
+		status, err := CompareContainerToSpec(c.Container, spec)
+		if err != nil {
+			return nil, fmt.Errorf("compare container to spec: %w", err)
+		}
+		if status == ContainerUpToDate {
+			// The container is already running with the same spec.
+			upToDate = true
+			for j, old := range containers {
+				if i == j {
+					continue
+				}
+				ops = append(ops, &RemoveContainerOperation{
+					ServiceID:   serviceID,
+					ContainerID: old.Container.ID,
+					MachineID:   old.MachineID,
+				})
+			}
+			break
+		}
+		// TODO: handle ContainerNeedsUpdate when update of mutable fields on a container is supported.
+	}
+	if upToDate {
+		return ops, nil
+	}
+
+	// The machine has containers but none of them match the new spec.
+	// Stop the old running containers that have conflicting ports with the new spec before running a new one.
+	for _, c := range containers {
+		if c.Container.State.Running {
+			conflictingPorts, err := c.Container.ConflictingServicePorts(spec.Ports)
+			if err != nil {
+				return nil, fmt.Errorf("check conflicting ports: %w", err)
+			}
+
+			if len(conflictingPorts) > 0 {
+				// Stop the running container with conflicting ports.
+				ops = append(ops, &StopContainerOperation{
+					ServiceID:   serviceID,
+					ContainerID: c.Container.ID,
+					MachineID:   c.MachineID,
+				})
+			}
+		}
+	}
+
+	// Run a new container.
+	ops = append(ops, &RunContainerOperation{
+		ServiceID: serviceID,
+		Spec:      spec,
+		MachineID: machineID,
+	})
+
+	// Remove the old containers.
+	for _, c := range containers {
+		ops = append(ops, &RemoveContainerOperation{
+			ServiceID:   serviceID,
+			ContainerID: c.Container.ID,
+			MachineID:   c.MachineID,
+		})
+	}
+
+	return ops, nil
+}
+
+// newEmptyPlan creates a new empty plan for a service deployment with initialised service ID and name.
+func newEmptyPlan(svc *api.Service, spec api.ServiceSpec) (Plan, error) {
+	var plan Plan
+
+	// Generate a new service ID for the initial service deployment if it doesn't exist yet.
+	if svc != nil {
+		plan.ServiceID = svc.ID
+		plan.ServiceName = svc.Name
+	} else {
+		var err error
+		plan.ServiceID, err = secret.NewID()
+		if err != nil {
+			return plan, fmt.Errorf("generate service ID: %w", err)
+		}
+		plan.ServiceName = spec.Name
+	}
+
+	return plan, nil
+}