chore: implement VolumeScheduler to determine what missing volumes should be created and where

pkg/api/service.go

@@ -61,6 +61,18 @@ func (s *ServiceSpec) Volume(name string) (VolumeSpec, bool) {
 	return VolumeSpec{}, false
 }
 
+// MountedDockerVolumes returns the list of volumes of VolumeTypeVolume type that are mounted into the container.
+func (s *ServiceSpec) MountedDockerVolumes() []VolumeSpec {
+	volumes := make(map[string]VolumeSpec)
+	for _, m := range s.Container.VolumeMounts {
+		if v, ok := s.Volume(m.VolumeName); ok && v.Type == VolumeTypeVolume {
+			volumes[v.Name] = v
+		}
+	}
+
+	return slices.Collect(maps.Values(volumes))
+}
+
 func (s *ServiceSpec) SetDefaults() ServiceSpec {
 	spec := s.Clone()
 

pkg/api/volume.go

@@ -18,6 +18,9 @@ const (
 	VolumeTypeVolume = "volume"
 	// VolumeTypeTmpfs is the type for mounting a temporary file system stored in the host memory.
 	VolumeTypeTmpfs = "tmpfs"
+
+	// VolumeDriverLocal is the default volume driver for local named Docker volumes.
+	VolumeDriverLocal = "local"
 )
 
 // VolumeSpec defines a volume mount specification. As of April 2025, the volume must be created before deploying
@@ -76,7 +79,7 @@ func (v *VolumeSpec) SetDefaults() VolumeSpec {
 			spec.VolumeOptions = &VolumeOptions{}
 		}
 		if spec.VolumeOptions.Driver == nil {
-			spec.VolumeOptions.Driver = &mount.Driver{Name: "local"}
+			spec.VolumeOptions.Driver = &mount.Driver{Name: VolumeDriverLocal}
 		}
 		if spec.VolumeOptions.Name == "" {
 			spec.VolumeOptions.Name = spec.Name
@@ -113,6 +116,33 @@ func (v *VolumeSpec) Equals(other VolumeSpec) bool {
 	return reflect.DeepEqual(vol, other)
 }
 
+// MatchesDockerVolume checks if this VolumeSpec is compatible with the given named Docker volume.
+// In other words, it checks if the spec could be used to create the volume.
+func (v *VolumeSpec) MatchesDockerVolume(vol volume.Volume) bool {
+	if v.Type != VolumeTypeVolume {
+		return false
+	}
+	spec := v.SetDefaults()
+
+	if spec.DockerVolumeName() != vol.Name {
+		return false
+	}
+
+	volDriver := vol.Driver
+	if volDriver == "" {
+		volDriver = VolumeDriverLocal
+	}
+	if spec.VolumeOptions.Driver.Name != volDriver {
+		return false
+	}
+
+	if !reflect.DeepEqual(spec.VolumeOptions.Driver.Options, vol.Options) {
+		return false
+	}
+
+	return true
+}
+
 func (v *VolumeSpec) Clone() VolumeSpec {
 	spec := *v
 

pkg/client/deploy/scheduler/service.go

@@ -10,29 +10,35 @@ import (
 )
 
 type ServiceScheduler struct {
-	Machines    []*Machine
-	Spec        api.ServiceSpec
-	Constraints []Constraint
+	machines    []*Machine
+	spec        api.ServiceSpec
+	constraints []Constraint
 }
 
-func NewServiceScheduler(ctx context.Context, cli Client, spec api.ServiceSpec) (*ServiceScheduler, error) {
+func NewServiceSchedulerWithClient(ctx context.Context, cli Client, spec api.ServiceSpec) (*ServiceScheduler, error) {
 	machines, err := InspectMachines(ctx, cli)
 	if err != nil {
 		return nil, fmt.Errorf("inspect machines: %w", err)
 	}
 
+	return NewServiceSchedulerWithMachines(machines, spec), nil
+}
+
+// NewServiceSchedulerWithMachines creates a new ServiceScheduler with the given machines and service specification.
+func NewServiceSchedulerWithMachines(machines []*Machine, spec api.ServiceSpec) *ServiceScheduler {
 	constraints := constraintsFromSpec(spec)
 
 	return &ServiceScheduler{
-		Machines:    machines,
-		Spec:        spec,
-		Constraints: constraints,
-	}, nil
+		machines:    machines,
+		spec:        spec,
+		constraints: constraints,
+	}
 }
 
-func (s *ServiceScheduler) AvailableMachines() ([]*Machine, error) {
+// EligibleMachines returns a list of machines that satisfy all constraints for the next scheduled container.
+func (s *ServiceScheduler) EligibleMachines() ([]*Machine, error) {
 	var available []*Machine
-	for _, machine := range s.Machines {
+	for _, machine := range s.machines {
 		if s.evaluateConstraints(machine) {
 			available = append(available, machine)
 		}
@@ -44,7 +50,7 @@ func (s *ServiceScheduler) AvailableMachines() ([]*Machine, error) {
 }
 
 func (s *ServiceScheduler) evaluateConstraints(machine *Machine) bool {
-	for _, c := range s.Constraints {
+	for _, c := range s.constraints {
 		if !c.Evaluate(machine) {
 			return false
 		}

pkg/client/deploy/scheduler/volume.go

@@ -0,0 +1,296 @@
+package scheduler
+
+import (
+	"context"
+	"fmt"
+	"slices"
+	"strings"
+
+	mapset "github.com/deckarep/golang-set/v2"
+	"github.com/psviderski/uncloud/pkg/api"
+)
+
+// VolumeScheduler determines what missing volumes should be created and where for a multi-service deployment.
+// It must satisfy the following constraints:
+//   - Services that share a volume must be placed on the same machine where the volume is located.
+//     If the volume is located on multiple machines, services can be placed on any of them.
+//   - Services must respect their individual placement constraints.
+//   - If a volume already exists on a machine, it must be used instead of creating a new one.
+//   - A missing volume must only be created on one machine.
+type VolumeScheduler struct {
+	// machines is a list of available machines in the cluster.
+	machines []*Machine
+	// serviceSpecs is a list of service specifications included in the deployment.
+	serviceSpecs []api.ServiceSpec
+	// volumeSpecs is a map of volume names to their specifications from the service specs in a canonical form.
+	volumeSpecs map[string]api.VolumeSpec
+	// volumeServices is a map of volume names to the list of service names that use the volume.
+	volumeServices map[string][]string
+	// existingVolumeMachines is a map of volume names to the set of machine IDs where those volumes are located.
+	// Contains only volumes that are used by at least one service in serviceSpecs.
+	existingVolumeMachines map[string]mapset.Set[string]
+}
+
+// NewVolumeSchedulerWithClient creates a new VolumeScheduler with the given cluster client and service specifications.
+func NewVolumeSchedulerWithClient(ctx context.Context, cli Client, specs []api.ServiceSpec) (*VolumeScheduler, error) {
+	machines, err := InspectMachines(ctx, cli)
+	if err != nil {
+		return nil, fmt.Errorf("inspect machines: %w", err)
+	}
+
+	return NewVolumeSchedulerWithMachines(machines, specs)
+}
+
+// NewVolumeSchedulerWithMachines creates a new VolumeScheduler with the given cluster machines
+// and service specifications.
+func NewVolumeSchedulerWithMachines(machines []*Machine, specs []api.ServiceSpec) (*VolumeScheduler, error) {
+	var specsWithVolumes []api.ServiceSpec
+	// Docker volume name -> VolumeSpec.
+	volumeSpecs := make(map[string]api.VolumeSpec)
+	// Volume name -> list of service names that use the volume.
+	volumeServices := make(map[string][]string)
+	// Volume name -> set of machine IDs where the volume is located.
+	existingVolumeMachines := make(map[string]mapset.Set[string])
+
+	// Validate all service names are unique to avoid scheduling conflicts.
+	serviceNames := make(map[string]struct{}, len(specs))
+	for _, spec := range specs {
+		if err := spec.Validate(); err != nil {
+			return nil, fmt.Errorf("invalid service spec: %w", err)
+		}
+
+		if _, exists := serviceNames[spec.Name]; exists {
+			return nil, fmt.Errorf("duplicate service name: '%s'", spec.Name)
+		}
+		serviceNames[spec.Name] = struct{}{}
+
+		mountedVolumes := spec.MountedDockerVolumes()
+		if len(mountedVolumes) == 0 {
+			continue
+		}
+		specsWithVolumes = append(specsWithVolumes, spec)
+
+		for _, v := range mountedVolumes {
+			v = v.SetDefaults()
+			// Reset any aliases in a service spec to the actual Docker volume name.
+			v.Name = v.DockerVolumeName()
+
+			if seenVolume, ok := volumeSpecs[v.Name]; ok {
+				if !seenVolume.Equals(v) {
+					return nil, fmt.Errorf("volume '%s' is defined multiple times with different options", v.Name)
+				}
+			} else {
+				volumeSpecs[v.Name] = v
+			}
+
+			volumeServices[v.Name] = append(volumeServices[v.Name], spec.Name)
+		}
+	}
+
+	// Validate the configurations of existing volumes on machines don't conflict with the volume specs, for example,
+	// a volume and a spec with the same name don't have different drivers.
+	for _, machine := range machines {
+		for _, vol := range machine.Volumes {
+			if spec, ok := volumeSpecs[vol.Name]; ok {
+				if !spec.MatchesDockerVolume(vol) {
+					return nil, fmt.Errorf("volume '%s' specification does not match the existing volume "+
+						"on machine '%s'", vol.Name, machine.Info.Name)
+				}
+
+				if _, setInitialised := existingVolumeMachines[vol.Name]; !setInitialised {
+					existingVolumeMachines[vol.Name] = mapset.NewSet[string]()
+				}
+				existingVolumeMachines[vol.Name].Add(machine.Info.Id)
+			}
+		}
+	}
+
+	return &VolumeScheduler{
+		machines:               machines,
+		serviceSpecs:           specsWithVolumes,
+		volumeSpecs:            volumeSpecs,
+		volumeServices:         volumeServices,
+		existingVolumeMachines: existingVolumeMachines,
+	}, nil
+}
+
+// Schedule determines what missing volumes should be created and where for services in the multi-service deployment.
+// It returns a map of machine IDs to a list of api.VolumeSpec that should be created on that machine,
+// or an error if services can't be scheduled due to scheduling constraints.
+func (s *VolumeScheduler) Schedule() (map[string][]api.VolumeSpec, error) {
+	if len(s.serviceSpecs) == 0 {
+		// No services with volume mounts, nothing to schedule.
+		return nil, nil
+	}
+
+	// Service name -> set of machine IDs where the service can be scheduled.
+	serviceEligibleMachines := make(map[string]mapset.Set[string])
+	// Volume name -> list of service names that use the volume.
+	// Get eligible machines for each service without considering its volume mounts.
+	for _, spec := range s.serviceSpecs {
+		machineIDs, err := s.serviceEligibleMachinesWithoutVolumes(spec)
+		if err != nil {
+			return nil, err
+		}
+		serviceEligibleMachines[spec.Name] = machineIDs
+	}
+
+	// For each volume that exists on any machine(s) (which shouldn't be created), intersect each service's
+	// eligible machines that use the volume with the machines the volume is located on.
+	// Service name -> list of processed volume names (quoted) to format the error message.
+	quotedServiceVolumes := make(map[string][]string)
+	for volumeName, volumeMachines := range s.existingVolumeMachines {
+		for _, serviceName := range s.volumeServices[volumeName] {
+			quotedServiceVolumes[serviceName] = append(quotedServiceVolumes[serviceName],
+				fmt.Sprintf("'%s'", volumeName))
+			newEligibleMachines := serviceEligibleMachines[serviceName].Intersect(volumeMachines)
+			if newEligibleMachines.Cardinality() == 0 {
+				volumes := strings.Join(quotedServiceVolumes[serviceName], ", ")
+				return nil, fmt.Errorf("unable to find a machine that satisfies service '%s' "+
+					"placement constraints and has all required volumes: %s", serviceName, volumes)
+			}
+			serviceEligibleMachines[serviceName] = newEligibleMachines
+		}
+	}
+
+	// Skip constraints propagation for volumes that already exist on machines as the propagation only works
+	// for missing volumes.
+	placedVolumes := make(map[string]struct{})
+	for volumeName := range s.existingVolumeMachines {
+		placedVolumes[volumeName] = struct{}{}
+	}
+
+	if err := s.propagateConstraintsUntilConvergence(serviceEligibleMachines, placedVolumes); err != nil {
+		return nil, err
+	}
+
+	// Schedule each missing volume on one of its eligible machines.
+	scheduledVolumes := make(map[string][]api.VolumeSpec)
+	for missingVolumeName, missingVolumeSpec := range s.volumeSpecs {
+		// Skip volumes that already exist on machines.
+		if _, ok := s.existingVolumeMachines[missingVolumeName]; ok {
+			continue
+		}
+
+		serviceNames := s.volumeServices[missingVolumeName]
+		if len(serviceNames) == 0 {
+			return nil, fmt.Errorf("bug detected: no services using volume '%s'", missingVolumeName)
+		}
+
+		// Get the current eligible machines (any service using the volume will have the same set after convergence).
+		eligibleMachines := serviceEligibleMachines[serviceNames[0]]
+		if eligibleMachines.Cardinality() == 0 {
+			return nil, fmt.Errorf("bug detected: no eligible machines for volume '%s'", missingVolumeName)
+		}
+
+		// Choose the first machine in the sorted eligible machines to schedule the volume on.
+		// Sort the eligible machines to ensure deterministic behavior.
+		sortedEligibleMachines := eligibleMachines.ToSlice()
+		slices.Sort(sortedEligibleMachines)
+		machineID := sortedEligibleMachines[0]
+		// Update constraints for all services that use this volume to be placed on the selected machine.
+		for _, serviceName := range serviceNames {
+			serviceEligibleMachines[serviceName] = mapset.NewSet(machineID)
+		}
+		placedVolumes[missingVolumeName] = struct{}{}
+		scheduledVolumes[machineID] = append(scheduledVolumes[machineID], missingVolumeSpec)
+
+		// Propagate the updated constraints.
+		if err := s.propagateConstraintsUntilConvergence(serviceEligibleMachines, placedVolumes); err != nil {
+			return nil, fmt.Errorf("unexpected error while propagating constraints after "+
+				"scheduling volume '%s' on machine '%s': %w", missingVolumeName, machineID, err)
+		}
+	}
+
+	return scheduledVolumes, nil
+}
+
+// serviceEligibleMachinesWithoutVolumes returns a set of machine IDs where the service can be scheduled
+// without considering its volume mounts.
+func (s *VolumeScheduler) serviceEligibleMachinesWithoutVolumes(spec api.ServiceSpec) (mapset.Set[string], error) {
+	specWithoutVolumes := spec.Clone()
+	specWithoutVolumes.Container.VolumeMounts = nil
+
+	scheduler := NewServiceSchedulerWithMachines(s.machines, specWithoutVolumes)
+	machines, err := scheduler.EligibleMachines()
+	if err != nil {
+		return nil, fmt.Errorf("schedule service '%s': %w", spec.Name, err)
+	}
+
+	machineIDs := mapset.NewSetWithSize[string](len(machines))
+	for _, m := range machines {
+		machineIDs.Add(m.Info.Id)
+	}
+
+	return machineIDs, nil
+}
+
+// propagateConstraintsUntilConvergence iteratively narrows down eligible machines for services by propagating
+// constraints through shared volumes until convergence. It only processes volumes that need to be created (not
+// existing volumes) and ensures services sharing a volume converge to the same set of eligible machines.
+// If skipVolumes is provided, those volumes are excluded from constraint propagation.
+// Returns an error if any services have no eligible machines after constraint propagation.
+func (s *VolumeScheduler) propagateConstraintsUntilConvergence(
+	serviceEligibleMachines map[string]mapset.Set[string],
+	skipVolumes map[string]struct{},
+) error {
+	changed := true
+	// Loop until no more changes occur.
+	for changed {
+		changed = false
+
+		// For each volume, find the intersection of eligible machines for all services using that volume and update
+		// their eligible machines with the intersection. This will narrow down the machines where the volume can
+		// be created.
+		for volumeName, serviceNames := range s.volumeServices {
+			if skipVolumes != nil {
+				if _, ok := skipVolumes[volumeName]; ok {
+					continue
+				}
+			}
+			// Skip if there are no services using this volume (shouldn't happen).
+			if len(serviceNames) == 0 {
+				continue
+			}
+
+			// Find the intersection of eligible machines for all services using this volume.
+			var eligibleMachinesForVolume mapset.Set[string]
+			first := true
+			for _, serviceName := range serviceNames {
+				if first {
+					// First service: initialise the intersection.
+					eligibleMachinesForVolume = serviceEligibleMachines[serviceName].Clone()
+					first = false
+				} else {
+					eligibleMachinesForVolume = serviceEligibleMachines[serviceName].Intersect(
+						eligibleMachinesForVolume)
+				}
+			}
+
+			// If no machines are eligible for this volume, we have a constraint violation.
+			//goland:noinspection GoDfaNilDereference
+			if eligibleMachinesForVolume.Cardinality() == 0 {
+				var quotedServiceNames []string // Used to format the error message.
+				for _, svcName := range serviceNames {
+					quotedServiceNames = append(quotedServiceNames, fmt.Sprintf("'%s'", svcName))
+				}
+				return fmt.Errorf("unable to find a machine that satisfies placement constraints "+
+					"for services %s that must be placed together to share volume '%s'",
+					strings.Join(quotedServiceNames, ", "), volumeName)
+			}
+
+			// Update eligible machines for all services using this volume.
+			newCount := eligibleMachinesForVolume.Cardinality()
+			for _, serviceName := range serviceNames {
+				oldCount := serviceEligibleMachines[serviceName].Cardinality()
+				serviceEligibleMachines[serviceName] = eligibleMachinesForVolume
+
+				if oldCount != newCount {
+					changed = true
+				}
+			}
+		}
+	}
+
+	return nil
+}

pkg/client/deploy/scheduler/volume_test.go

@@ -0,0 +1,816 @@
+package scheduler
+
+import (
+	"testing"
+
+	"github.com/docker/docker/api/types/mount"
+	"github.com/docker/docker/api/types/volume"
+	"github.com/psviderski/uncloud/internal/machine/api/pb"
+	"github.com/psviderski/uncloud/pkg/api"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func TestVolumeScheduler_Schedule(t *testing.T) {
+	tests := []struct {
+		name         string
+		machines     []*Machine
+		serviceSpecs []api.ServiceSpec
+		want         map[string][]api.VolumeSpec
+		wantErr      string
+	}{
+		{
+			name: "single service with missing volume",
+			machines: []*Machine{
+				{
+					Info: &pb.MachineInfo{
+						Id: "machine1",
+					},
+				},
+			},
+			serviceSpecs: []api.ServiceSpec{
+				{
+					Name: "service1",
+					Container: api.ContainerSpec{
+						Image: "portainer/pause:latest",
+						VolumeMounts: []api.VolumeMount{
+							{
+								VolumeName:    "vol1",
+								ContainerPath: "/data",
+							},
+						},
+					},
+					Volumes: []api.VolumeSpec{
+						{
+							Name: "vol1",
+							Type: api.VolumeTypeVolume,
+						},
+					},
+				},
+			},
+			want: map[string][]api.VolumeSpec{
+				"machine1": {
+					{
+						Name: "vol1",
+						Type: api.VolumeTypeVolume,
+					},
+				},
+			},
+		},
+		{
+			name: "multiple services sharing a missing volume",
+			machines: []*Machine{
+				{
+					Info: &pb.MachineInfo{
+						Id: "machine1",
+					},
+					Volumes: []volume.Volume{},
+				},
+				{
+					Info: &pb.MachineInfo{
+						Id: "machine2",
+					},
+					Volumes: []volume.Volume{},
+				},
+			},
+			serviceSpecs: []api.ServiceSpec{
+				{
+					Name: "service1",
+					Container: api.ContainerSpec{
+						Image: "portainer/pause:latest",
+						VolumeMounts: []api.VolumeMount{
+							{
+								VolumeName:    "vol1",
+								ContainerPath: "/data",
+							},
+						},
+					},
+					Volumes: []api.VolumeSpec{
+						{
+							Name: "vol1",
+							Type: api.VolumeTypeVolume,
+						},
+					},
+				},
+				{
+					Name: "service2",
+					Container: api.ContainerSpec{
+						Image: "portainer/pause:latest",
+						VolumeMounts: []api.VolumeMount{
+							{
+								VolumeName:    "vol1",
+								ContainerPath: "/data",
+							},
+						},
+					},
+					Volumes: []api.VolumeSpec{
+						{
+							Name: "vol1",
+							Type: api.VolumeTypeVolume,
+						},
+					},
+				},
+			},
+			want: map[string][]api.VolumeSpec{
+				"machine1": {
+					{
+						Name: "vol1",
+						Type: api.VolumeTypeVolume,
+					},
+				},
+			},
+		},
+		{
+			name: "service with existing volume",
+			machines: []*Machine{
+				{
+					Info: &pb.MachineInfo{
+						Id: "machine1",
+					},
+				},
+				{
+					Info: &pb.MachineInfo{
+						Id: "machine2",
+					},
+					Volumes: []volume.Volume{
+						{
+							Name: "vol1",
+						},
+					},
+				},
+			},
+			serviceSpecs: []api.ServiceSpec{
+				{
+					Name: "service1",
+					Container: api.ContainerSpec{
+						Image: "portainer/pause:latest",
+						VolumeMounts: []api.VolumeMount{
+							{
+								VolumeName:    "vol1",
+								ContainerPath: "/data",
+							},
+						},
+					},
+					Volumes: []api.VolumeSpec{
+						{
+							Name: "vol1",
+							Type: api.VolumeTypeVolume,
+						},
+					},
+				},
+			},
+			want: map[string][]api.VolumeSpec{},
+		},
+		{
+			name: "service with placement constraint and missing volume",
+			machines: []*Machine{
+				{
+					Info: &pb.MachineInfo{
+						Id: "machine1",
+					},
+					Volumes: []volume.Volume{},
+				},
+				{
+					Info: &pb.MachineInfo{
+						Id: "machine2",
+					},
+					Volumes: []volume.Volume{},
+				},
+			},
+			serviceSpecs: []api.ServiceSpec{
+				{
+					Name: "service1",
+					Placement: api.Placement{
+						Machines: []string{"machine2"},
+					},
+					Container: api.ContainerSpec{
+						Image: "portainer/pause:latest",
+						VolumeMounts: []api.VolumeMount{
+							{
+								VolumeName:    "vol1",
+								ContainerPath: "/data",
+							},
+						},
+					},
+					Volumes: []api.VolumeSpec{
+						{
+							Name: "vol1",
+							Type: api.VolumeTypeVolume,
+						},
+					},
+				},
+			},
+			want: map[string][]api.VolumeSpec{
+				"machine2": {
+					{
+						Name: "vol1",
+						Type: api.VolumeTypeVolume,
+					},
+				},
+			},
+		},
+		{
+			name: "services with conflicting placement constraints",
+			machines: []*Machine{
+				{
+					Info: &pb.MachineInfo{
+						Id: "machine1",
+					},
+					Volumes: []volume.Volume{},
+				},
+				{
+					Info: &pb.MachineInfo{
+						Id: "machine2",
+					},
+					Volumes: []volume.Volume{},
+				},
+			},
+			serviceSpecs: []api.ServiceSpec{
+				{
+					Name: "service1",
+					Placement: api.Placement{
+						Machines: []string{"machine1"},
+					},
+					Container: api.ContainerSpec{
+						Image: "portainer/pause:latest",
+						VolumeMounts: []api.VolumeMount{
+							{
+								VolumeName:    "vol1",
+								ContainerPath: "/data",
+							},
+						},
+					},
+					Volumes: []api.VolumeSpec{
+						{
+							Name: "vol1",
+							Type: api.VolumeTypeVolume,
+						},
+					},
+				},
+				{
+					Name: "service2",
+					Placement: api.Placement{
+						Machines: []string{"machine2"},
+					},
+					Container: api.ContainerSpec{
+						Image: "portainer/pause:latest",
+						VolumeMounts: []api.VolumeMount{
+							{
+								VolumeName:    "vol1",
+								ContainerPath: "/data",
+							},
+						},
+					},
+					Volumes: []api.VolumeSpec{
+						{
+							Name: "vol1",
+							Type: api.VolumeTypeVolume,
+						},
+					},
+				},
+			},
+			wantErr: "unable to find a machine that satisfies placement constraints for services " +
+				"'service1', 'service2' that must be placed together to share volume 'vol1'",
+		},
+		{
+			name: "service with existing volume on wrong machine",
+			machines: []*Machine{
+				{
+					Info: &pb.MachineInfo{
+						Id: "machine1",
+					},
+					Volumes: []volume.Volume{
+						{
+							Name: "vol1",
+						},
+					},
+				},
+				{
+					Info: &pb.MachineInfo{
+						Id: "machine2",
+					},
+					Volumes: []volume.Volume{},
+				},
+			},
+			serviceSpecs: []api.ServiceSpec{
+				{
+					Name: "service1",
+					Placement: api.Placement{
+						Machines: []string{"machine2"},
+					},
+					Container: api.ContainerSpec{
+						Image: "portainer/pause:latest",
+						VolumeMounts: []api.VolumeMount{
+							{
+								VolumeName:    "vol1",
+								ContainerPath: "/data",
+							},
+						},
+					},
+					Volumes: []api.VolumeSpec{
+						{
+							Name: "vol1",
+							Type: api.VolumeTypeVolume,
+						},
+					},
+				},
+			},
+			wantErr: "unable to find a machine that satisfies service 'service1' placement constraints " +
+				"and has all required volumes: 'vol1'",
+		},
+		{
+			name: "multiple services with multiple volumes, some shared",
+			machines: []*Machine{
+				{
+					Info: &pb.MachineInfo{
+						Id: "machine1",
+					},
+					Volumes: []volume.Volume{},
+				},
+				{
+					Info: &pb.MachineInfo{
+						Id: "machine2",
+					},
+					Volumes: []volume.Volume{},
+				},
+			},
+			serviceSpecs: []api.ServiceSpec{
+				{
+					Name: "service1",
+					Container: api.ContainerSpec{
+						Image: "portainer/pause:latest",
+						VolumeMounts: []api.VolumeMount{
+							{
+								VolumeName:    "vol1",
+								ContainerPath: "/data1",
+							},
+							{
+								VolumeName:    "vol2",
+								ContainerPath: "/data2",
+							},
+						},
+					},
+					Volumes: []api.VolumeSpec{
+						{
+							Name: "vol1",
+							Type: api.VolumeTypeVolume,
+						},
+						{
+							Name: "vol2",
+							Type: api.VolumeTypeVolume,
+						},
+					},
+				},
+				{
+					Name: "service2",
+					Container: api.ContainerSpec{
+						Image: "portainer/pause:latest",
+						VolumeMounts: []api.VolumeMount{
+							{
+								VolumeName:    "vol2",
+								ContainerPath: "/data2",
+							},
+							{
+								VolumeName:    "vol3",
+								ContainerPath: "/data3",
+							},
+						},
+					},
+					Volumes: []api.VolumeSpec{
+						{
+							Name: "vol2",
+							Type: api.VolumeTypeVolume,
+						},
+						{
+							Name: "vol3",
+							Type: api.VolumeTypeVolume,
+						},
+					},
+				},
+				{
+					Name: "service3",
+					Container: api.ContainerSpec{
+						Image: "portainer/pause:latest",
+						VolumeMounts: []api.VolumeMount{
+							{
+								VolumeName:    "vol3",
+								ContainerPath: "/data3",
+							},
+							{
+								VolumeName:    "vol4",
+								ContainerPath: "/data4",
+							},
+						},
+					},
+					Volumes: []api.VolumeSpec{
+						{
+							Name: "vol3",
+							Type: api.VolumeTypeVolume,
+						},
+						{
+							Name: "vol4",
+							Type: api.VolumeTypeVolume,
+						},
+					},
+				},
+			},
+			want: map[string][]api.VolumeSpec{
+				"machine1": {
+					{
+						Name: "vol1",
+						Type: api.VolumeTypeVolume,
+					},
+					{
+						Name: "vol2",
+						Type: api.VolumeTypeVolume,
+					},
+					{
+						Name: "vol3",
+						Type: api.VolumeTypeVolume,
+					},
+					{
+						Name: "vol4",
+						Type: api.VolumeTypeVolume,
+					},
+				},
+			},
+		},
+		{
+			name: "multiple services with multiple volumes, some shared and existing",
+			machines: []*Machine{
+				{
+					Info: &pb.MachineInfo{
+						Id: "machine1",
+					},
+				},
+				{
+					Info: &pb.MachineInfo{
+						Id: "machine2",
+					},
+					Volumes: []volume.Volume{
+						{
+							Name: "vol2",
+						},
+					},
+				},
+				{
+					Info: &pb.MachineInfo{
+						Id: "machine3",
+					},
+					Volumes: []volume.Volume{
+						{
+							Name: "vol1",
+						},
+						{
+							Name: "vol2",
+						},
+					},
+				},
+			},
+			serviceSpecs: []api.ServiceSpec{
+				{
+					Name: "service1",
+					Container: api.ContainerSpec{
+						Image: "portainer/pause:latest",
+						VolumeMounts: []api.VolumeMount{
+							{
+								VolumeName:    "vol3",
+								ContainerPath: "/data3",
+							},
+							{
+								VolumeName:    "vol4",
+								ContainerPath: "/data4",
+							},
+						},
+					},
+					Volumes: []api.VolumeSpec{
+						{
+							Name: "vol3",
+							Type: api.VolumeTypeVolume,
+						},
+						{
+							Name: "vol4",
+							Type: api.VolumeTypeVolume,
+						},
+					},
+				},
+				{
+					Name: "service2",
+					Container: api.ContainerSpec{
+						Image: "portainer/pause:latest",
+						VolumeMounts: []api.VolumeMount{
+							{
+								VolumeName:    "vol1",
+								ContainerPath: "/data1",
+							},
+							{
+								VolumeName:    "vol2",
+								ContainerPath: "/data2",
+							},
+							{
+								VolumeName:    "vol3",
+								ContainerPath: "/data3",
+							},
+						},
+					},
+					Volumes: []api.VolumeSpec{
+						{
+							Name: "vol1",
+							Type: api.VolumeTypeVolume,
+						},
+						{
+							Name: "vol2",
+							Type: api.VolumeTypeVolume,
+						},
+						{
+							Name: "vol3",
+							Type: api.VolumeTypeVolume,
+						},
+					},
+				},
+				{
+					Name: "service3",
+					Container: api.ContainerSpec{
+						Image: "portainer/pause:latest",
+						VolumeMounts: []api.VolumeMount{
+							{
+								VolumeName:    "vol2",
+								ContainerPath: "/data2",
+							},
+							{
+								VolumeName:    "vol4-alias",
+								ContainerPath: "/data4",
+							},
+						},
+					},
+					Volumes: []api.VolumeSpec{
+						{
+							Name: "vol2",
+							Type: api.VolumeTypeVolume,
+						},
+						{
+							Name: "vol4-alias",
+							Type: api.VolumeTypeVolume,
+							VolumeOptions: &api.VolumeOptions{
+								Name: "vol4",
+								Driver: &mount.Driver{
+									Name: api.VolumeDriverLocal,
+								},
+							},
+						},
+					},
+				},
+				{
+					Name: "service4",
+					Container: api.ContainerSpec{
+						Image: "portainer/pause:latest",
+						VolumeMounts: []api.VolumeMount{
+							{
+								VolumeName:    "vol2-alias",
+								ContainerPath: "/data2",
+							},
+							{
+								VolumeName:    "vol5",
+								ContainerPath: "/data5",
+							},
+						},
+					},
+					Volumes: []api.VolumeSpec{
+						{
+							Name: "vol2-alias",
+							Type: api.VolumeTypeVolume,
+							VolumeOptions: &api.VolumeOptions{
+								Name: "vol2",
+								Driver: &mount.Driver{
+									Name: api.VolumeDriverLocal,
+								},
+							},
+						},
+						{
+							Name: "vol5",
+							Type: api.VolumeTypeVolume,
+						},
+					},
+				},
+			},
+			want: map[string][]api.VolumeSpec{
+				"machine2": {
+					{
+						Name: "vol5",
+						Type: api.VolumeTypeVolume,
+					},
+				},
+				"machine3": {
+					{
+						Name: "vol3",
+						Type: api.VolumeTypeVolume,
+					},
+					{
+						Name: "vol4",
+						Type: api.VolumeTypeVolume,
+					},
+				},
+			},
+		},
+		{
+			name: "multiple services with multiple volumes, some shared, with conflicting placement constraints",
+			machines: []*Machine{
+				{
+					Info: &pb.MachineInfo{
+						Id: "machine1",
+					},
+					Volumes: []volume.Volume{},
+				},
+				{
+					Info: &pb.MachineInfo{
+						Id: "machine2",
+					},
+					Volumes: []volume.Volume{},
+				},
+			},
+			serviceSpecs: []api.ServiceSpec{
+				{
+					Name: "service1",
+					Placement: api.Placement{
+						Machines: []string{"machine1"},
+					},
+					Container: api.ContainerSpec{
+						Image: "portainer/pause:latest",
+						VolumeMounts: []api.VolumeMount{
+							{
+								VolumeName:    "vol1",
+								ContainerPath: "/data1",
+							},
+							{
+								VolumeName:    "vol2",
+								ContainerPath: "/data2",
+							},
+						},
+					},
+					Volumes: []api.VolumeSpec{
+						{
+							Name: "vol1",
+							Type: api.VolumeTypeVolume,
+						},
+						{
+							Name: "vol2",
+							Type: api.VolumeTypeVolume,
+						},
+					},
+				},
+				{
+					Name: "service2",
+					Placement: api.Placement{
+						Machines: []string{"machine2"},
+					},
+					Container: api.ContainerSpec{
+						Image: "portainer/pause:latest",
+						VolumeMounts: []api.VolumeMount{
+							{
+								VolumeName:    "vol2",
+								ContainerPath: "/data2",
+							},
+							{
+								VolumeName:    "vol3",
+								ContainerPath: "/data3",
+							},
+						},
+					},
+					Volumes: []api.VolumeSpec{
+						{
+							Name: "vol2",
+							Type: api.VolumeTypeVolume,
+						},
+						{
+							Name: "vol3",
+							Type: api.VolumeTypeVolume,
+						},
+					},
+				},
+			},
+			wantErr: "unable to find a machine that satisfies placement constraints for services " +
+				"'service1', 'service2' that must be placed together to share volume 'vol2'",
+		},
+		{
+			name: "multiple services with multiple volumes, no shared",
+			machines: []*Machine{
+				{
+					Info: &pb.MachineInfo{
+						Id: "machine1",
+					},
+					Volumes: []volume.Volume{
+						{
+							Name: "vol1",
+						},
+					},
+				},
+				{
+					Info: &pb.MachineInfo{
+						Id: "machine2",
+					},
+					Volumes: []volume.Volume{
+						{
+							Name: "vol3",
+						},
+					},
+				},
+			},
+			serviceSpecs: []api.ServiceSpec{
+				{
+					Name: "service1",
+					Container: api.ContainerSpec{
+						Image: "portainer/pause:latest",
+						VolumeMounts: []api.VolumeMount{
+							{
+								VolumeName:    "vol1",
+								ContainerPath: "/data1",
+							},
+							{
+								VolumeName:    "vol2",
+								ContainerPath: "/data2",
+							},
+						},
+					},
+					Volumes: []api.VolumeSpec{
+						{
+							Name: "vol1",
+							Type: api.VolumeTypeVolume,
+						},
+						{
+							Name: "vol2",
+							Type: api.VolumeTypeVolume,
+						},
+					},
+				},
+				{
+					Name: "service2",
+					Container: api.ContainerSpec{
+						Image: "portainer/pause:latest",
+						VolumeMounts: []api.VolumeMount{
+							{
+								VolumeName:    "vol3",
+								ContainerPath: "/data3",
+							},
+							{
+								VolumeName:    "vol4",
+								ContainerPath: "/data4",
+							},
+						},
+					},
+					Volumes: []api.VolumeSpec{
+						{
+							Name: "vol3",
+							Type: api.VolumeTypeVolume,
+						},
+						{
+							Name: "vol4",
+							Type: api.VolumeTypeVolume,
+						},
+					},
+				},
+			},
+			want: map[string][]api.VolumeSpec{
+				"machine1": {
+					{
+						Name: "vol2",
+						Type: api.VolumeTypeVolume,
+					},
+				},
+				"machine2": {
+					{
+						Name: "vol4",
+						Type: api.VolumeTypeVolume,
+					},
+				},
+			},
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			scheduler, err := NewVolumeSchedulerWithMachines(tt.machines, tt.serviceSpecs)
+			require.NoError(t, err)
+			result, err := scheduler.Schedule()
+
+			if tt.wantErr != "" {
+				assert.Error(t, err)
+				assert.Contains(t, err.Error(), tt.wantErr)
+			} else {
+				assert.NoError(t, err)
+
+				assert.Len(t, result, len(tt.want), "Number of machines with volumes to create should match")
+				for machineID, expectedVolumes := range tt.want {
+					// Transform the expected volumes to the canonical form with defaults set.
+					for i := range expectedVolumes {
+						expectedVolumes[i] = expectedVolumes[i].SetDefaults()
+					}
+
+					actualVolumes, ok := result[machineID]
+					assert.True(t, ok, "Machine %s should be in the result", machineID)
+					assert.ElementsMatch(t, expectedVolumes, actualVolumes,
+						"Volumes for machine %s should match", machineID)
+				}
+			}
+		})
+	}
+}

pkg/client/deploy/strategy.go

@@ -56,12 +56,12 @@ func (s *RollingStrategy) planReplicated(
 		return plan, err
 	}
 
-	sched, err := scheduler.NewServiceScheduler(ctx, cli, spec)
+	sched, err := scheduler.NewServiceSchedulerWithClient(ctx, cli, spec)
 	if err != nil {
 		return plan, err
 	}
 	// TODO: return a detailed report on required constraints and which ones are satisfied?
-	availableMachines, err := sched.AvailableMachines()
+	availableMachines, err := sched.EligibleMachines()
 	if err != nil {
 		return plan, err
 	}
@@ -216,12 +216,12 @@ func (s *RollingStrategy) planGlobal(
 		}
 	}
 
-	sched, err := scheduler.NewServiceScheduler(ctx, cli, spec)
+	sched, err := scheduler.NewServiceSchedulerWithClient(ctx, cli, spec)
 	if err != nil {
 		return plan, err
 	}
 
-	availableMachines, err := sched.AvailableMachines()
+	availableMachines, err := sched.EligibleMachines()
 	if err != nil {
 		return plan, err
 	}

pkg/client/service.go

@@ -11,7 +11,6 @@ import (
 	"github.com/docker/docker/api/types/container"
 	"github.com/psviderski/uncloud/internal/machine/api/pb"
 	"github.com/psviderski/uncloud/pkg/api"
-	"github.com/psviderski/uncloud/pkg/client/deploy"
 	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/metadata"
 	"google.golang.org/grpc/status"
@@ -40,7 +39,7 @@ func (cli *Client) RunService(ctx context.Context, spec api.ServiceSpec) (RunSer
 		}
 	}
 
-	deployment := cli.NewDeployment(spec, &deploy.RollingStrategy{})
+	deployment := cli.NewDeployment(spec, nil)
 	plan, err := deployment.Run(ctx)
 	if err != nil {
 		return resp, err