pool: bring a claim's volumes up concurrently

sandboxd/pool/pool_test.go

@@ -809,6 +809,7 @@ type fakeEngine struct {
 	staleReconciles                   []string // VM names ReconcileStaleCreate was called on
 	installCAErr                      error
 	diskAttachErr                     error
+	diskAttachErrFor                  string // volume name whose DiskAttach fails; "" = never
 	diskAttachCancel                  context.CancelFunc
 	mountVolumeErr                    error
 	unmountVolumeErr                  error
@@ -826,6 +827,8 @@ type fakeEngine struct {
 	tap                                                                                string // non-empty: lifecycle records carry this NIC tap
 	listCount                                                                          int
 
+	attachRendezvous *sync.WaitGroup // non-nil: DiskAttach waits there until every attach has arrived
+
 	cloneStall      chan struct{} // non-nil: Clone blocks until closed
 	probeStall      chan struct{} // non-nil: Probe blocks until closed
 	hibernateStall  chan struct{} // non-nil: Hibernate blocks until closed
@@ -1043,14 +1046,22 @@ func (f *fakeEngine) InstallCACert(_ context.Context, vsockSocket string, _ []by
 
 func (f *fakeEngine) DiskAttach(_ context.Context, _ string, spec engine.VolumeSpec) error {
 	f.mu.Lock()
-	defer f.mu.Unlock()
 	f.volumeSpecs = append(f.volumeSpecs, spec)
 	f.volumeOps = append(f.volumeOps, "attach:"+spec.Name)
 	f.attachDirty[spec.Name] = volumeDirty(spec.Path)
-	if f.diskAttachCancel != nil {
-		f.diskAttachCancel()
+	cancel, rendezvous, err := f.diskAttachCancel, f.attachRendezvous, f.diskAttachErr
+	if spec.Name == f.diskAttachErrFor {
+		err = errors.New("attach failed")
 	}
-	return f.diskAttachErr
+	f.mu.Unlock()
+	if cancel != nil {
+		cancel()
+	}
+	if rendezvous != nil {
+		rendezvous.Done()
+		rendezvous.Wait()
+	}
+	return err
 }
 
 func (f *fakeEngine) MountVolume(_ context.Context, _, name, mount string, rw bool) error {

sandboxd/pool/volume.go

@@ -11,6 +11,7 @@ import (
 	"time"
 
 	"github.com/projecteru2/core/log"
+	"golang.org/x/sync/errgroup"
 
 	"github.com/cocoonstack/sandbox/sandboxd/engine"
 	"github.com/cocoonstack/sandbox/sandboxd/types"
@@ -177,30 +178,47 @@ func (m *Manager) confirmVolumesClean(volumes []resolvedVolume) error {
 	return nil
 }
 
-// applyVolumes attaches and mounts the resolved set; applied is that same set
-// in request shape, recorded on the sandbox only once every mount is up.
+// applyVolumes brings the resolved set up concurrently: cocoon serializes the
+// attach per VM, so what overlaps is the CLI spawns, settle waits and guest
+// mounts. applied is the request-shaped set, recorded once every mount is up.
 func (m *Manager) applyVolumes(ctx context.Context, sb *types.Sandbox, volumes []resolvedVolume, applied []types.Volume) error {
-	if len(volumes) == 0 {
+	switch len(volumes) {
+	case 0:
 		return nil
-	}
-	for _, volume := range volumes {
-		// Write-ahead: the marker must be durable before any guest write can be.
-		if volume.disk.RW {
-			if err := markVolumeDirty(volume.disk.Path); err != nil {
-				return fmt.Errorf("mark volume %q dirty: %w", volume.applied.Name, err)
-			}
+	case 1:
+		if err := m.applyVolume(ctx, sb, volumes[0]); err != nil {
+			return err
 		}
-		if err := m.eng.DiskAttach(ctx, sb.VMName, volume.disk); err != nil {
-			return fmt.Errorf("attach volume %q: %w", volume.applied.Name, err)
+	default:
+		group, groupCtx := errgroup.WithContext(ctx)
+		for _, volume := range volumes {
+			group.Go(func() error { return m.applyVolume(groupCtx, sb, volume) })
 		}
-		if err := m.eng.MountVolume(ctx, sb.VsockSocket, volume.applied.Name, volume.applied.Mount, volume.disk.RW); err != nil {
-			return fmt.Errorf("setup volume %q: %w", volume.applied.Name, err)
+		if err := group.Wait(); err != nil {
+			return err
 		}
 	}
 	sb.Volumes = applied
 	return nil
 }
 
+// applyVolume keeps one volume's steps strictly ordered; siblings overlap freely.
+func (m *Manager) applyVolume(ctx context.Context, sb *types.Sandbox, volume resolvedVolume) error {
+	// Write-ahead: the marker must be durable before any guest write can be.
+	if volume.disk.RW {
+		if err := markVolumeDirty(volume.disk.Path); err != nil {
+			return fmt.Errorf("mark volume %q dirty: %w", volume.applied.Name, err)
+		}
+	}
+	if err := m.eng.DiskAttach(ctx, sb.VMName, volume.disk); err != nil {
+		return fmt.Errorf("attach volume %q: %w", volume.applied.Name, err)
+	}
+	if err := m.eng.MountVolume(ctx, sb.VsockSocket, volume.applied.Name, volume.applied.Mount, volume.disk.RW); err != nil {
+		return fmt.Errorf("setup volume %q: %w", volume.applied.Name, err)
+	}
+	return nil
+}
+
 // quiesceVolumes unmounts a claim's writable mounts in reverse order and
 // returns the teardown its VM removal must finish. A failed unmount never
 // blocks teardown: the image keeps its marker and waits for a recovering

sandboxd/pool/volume_test.go

@@ -8,13 +8,15 @@ import (
 	"path/filepath"
 	"slices"
 	"strings"
+	"sync"
 	"testing"
 
 	"github.com/cocoonstack/sandbox/sandboxd/config"
+	"github.com/cocoonstack/sandbox/sandboxd/engine"
 	"github.com/cocoonstack/sandbox/sandboxd/types"
 )
 
-func TestClaimProvisionAppliesVolumesInOrder(t *testing.T) {
+func TestClaimProvisionAppliesVolumesInRequestOrder(t *testing.T) {
 	first := writeVolumeImage(t, "imagenet.img", "first")
 	second := writeVolumeImage(t, "weights.img", "second")
 	eng := newFakeEngine()
@@ -35,19 +37,20 @@ func TestClaimProvisionAppliesVolumesInOrder(t *testing.T) {
 	if !slices.Equal(sb.Volumes, wantApplied) {
 		t.Errorf("volumes=%v, want %v", sb.Volumes, wantApplied)
 	}
-	wantOps := []string{
-		"provision", "probe",
-		"attach:weights", "mount:weights:/models",
-		"attach:imagenet", "mount:imagenet:/volumes/imagenet",
+	wantLead := []string{"provision", "probe"}
+	if ops := eng.volumeOpsLog(); len(ops) < len(wantLead) || !slices.Equal(ops[:len(wantLead)], wantLead) {
+		t.Errorf("operations=%v, want %v ahead of every volume op", ops, wantLead)
 	}
-	if !slices.Equal(eng.volumeOps, wantOps) {
-		t.Errorf("operations=%v, want %v", eng.volumeOps, wantOps)
-	}
-	if !slices.Equal(eng.volumeMounts, wantApplied) {
-		t.Errorf("mounts=%v, want %v", eng.volumeMounts, wantApplied)
+	assertVolumeBringUp(t, eng, wantApplied)
+	wantSpecs := []engine.VolumeSpec{
+		{Name: "imagenet", Path: first, DirectIO: "off"},
+		{Name: "weights", Path: second, DirectIO: "on"},
 	}
-	if got := eng.volumeSpecs; len(got) != 2 || got[0].Path != second || got[0].DirectIO != "on" || got[1].Path != first {
-		t.Errorf("attached specs=%+v", got)
+	specs := slices.SortedFunc(slices.Values(eng.volumeSpecs), func(a, b engine.VolumeSpec) int {
+		return strings.Compare(a.Name, b.Name)
+	})
+	if !slices.Equal(specs, wantSpecs) {
+		t.Errorf("attached specs=%+v, want %+v", specs, wantSpecs)
 	}
 	persisted, err := newClaimStore(m.dataDir).load()
 	if err != nil {
@@ -70,6 +73,90 @@ func TestClaimProvisionAppliesVolumesInOrder(t *testing.T) {
 	}
 }
 
+func TestClaimProvisionBringsMixedVolumesUpConcurrently(t *testing.T) {
+	dataset := writeVolumeImage(t, "dataset.img", "dataset")
+	weights := writeVolumeImage(t, "weights.img", "weights")
+	scratch := writeVolumeImage(t, "scratch.img", "scratch")
+	cache := writeVolumeImage(t, "cache.img", "cache")
+	eng := newFakeEngine()
+	m := newVolumeManager(t, eng, []config.VolumeSpec{
+		{Name: "dataset", Path: dataset},
+		{Name: "weights", Path: weights},
+		{Name: "scratch", Path: scratch, Writable: true},
+		{Name: "cache", Path: cache, Writable: true},
+	})
+	requested := []types.Volume{
+		{Name: "weights", Mount: "/models"},
+		{Name: "scratch", Mode: types.VolumeModeRW},
+		{Name: "dataset"},
+		{Name: "cache", Mount: "/cache", Mode: types.VolumeModeRW},
+	}
+	wantApplied := []types.Volume{
+		{Name: "weights", Mount: "/models"},
+		{Name: "scratch", Mount: "/volumes/scratch", Mode: types.VolumeModeRW},
+		{Name: "dataset", Mount: "/volumes/dataset"},
+		{Name: "cache", Mount: "/cache", Mode: types.VolumeModeRW},
+	}
+	// Every attach must be in flight at once: a sequential apply blocks here.
+	var attaches sync.WaitGroup
+	attaches.Add(len(requested))
+	eng.attachRendezvous = &attaches
+
+	sb, err := m.ClaimProvision(t.Context(), testKey, 0, "", "", requested)
+	if err != nil {
+		t.Fatalf("ClaimProvision: %v", err)
+	}
+	if !slices.Equal(sb.Volumes, wantApplied) {
+		t.Errorf("volumes=%v, want the request order %v", sb.Volumes, wantApplied)
+	}
+	assertVolumeBringUp(t, eng, wantApplied)
+	if !volumeDirty(scratch) || !volumeDirty(cache) {
+		t.Error("live writable claim left an image unmarked")
+	}
+	if volumeDirty(dataset) || volumeDirty(weights) {
+		t.Error("read-only volume of a mixed claim was marked dirty")
+	}
+}
+
+func TestClaimProvisionOneVolumeAttachFailureFailsWholeClaim(t *testing.T) {
+	first := writeVolumeImage(t, "first.img", "first")
+	broken := writeVolumeImage(t, "broken.img", "broken")
+	third := writeVolumeImage(t, "third.img", "third")
+	eng := newFakeEngine()
+	eng.diskAttachErrFor = "broken"
+	m := newVolumeManager(t, eng, []config.VolumeSpec{
+		{Name: "first", Path: first, Writable: true},
+		{Name: "broken", Path: broken, Writable: true},
+		{Name: "third", Path: third},
+	})
+
+	_, err := m.ClaimProvision(t.Context(), testKey, 0, "", "", []types.Volume{
+		{Name: "first", Mode: types.VolumeModeRW},
+		{Name: "broken", Mode: types.VolumeModeRW},
+		{Name: "third"},
+	})
+	if err == nil || !strings.Contains(err.Error(), `attach volume "broken"`) {
+		t.Fatalf("ClaimProvision: %v, want the failing volume's attach error", err)
+	}
+	if len(eng.removes) != 1 {
+		t.Errorf("removes=%v, want the failed VM destroyed", eng.removes)
+	}
+	if !volumeDirty(first) || !volumeDirty(broken) {
+		t.Error("failed claim cleared a writable marker, opening an image no writer flushed")
+	}
+	if volumeDirty(third) {
+		t.Error("read-only volume was marked dirty")
+	}
+	for _, name := range []string{"first", "broken", "third"} {
+		if holders := volumeHoldersOf(m, name); holders != (volumeHolders{}) {
+			t.Errorf("registry for %s after the failure=%+v, want empty", name, holders)
+		}
+	}
+	if _, gauges := m.Info(); gauges.Claimed != 0 {
+		t.Errorf("claimed=%d, want 0", gauges.Claimed)
+	}
+}
+
 func TestClaimWarmAppliesVolumesAndRefillsAfterFailure(t *testing.T) {
 	path := writeVolumeImage(t, "data.img", "data")
 	for _, tt := range []struct {
@@ -400,6 +487,37 @@ func TestClaimProvisionRejectsMissingVolumePathBeforeProvision(t *testing.T) {
 	}
 }
 
+// assertVolumeBringUp pins each volume's own marker→attach→mount order and the
+// completeness of the set; cross-volume order is free.
+func assertVolumeBringUp(t *testing.T, eng *fakeEngine, applied []types.Volume) {
+	t.Helper()
+	ops := eng.volumeOpsLog()
+	brought := 0
+	for _, op := range ops {
+		if strings.HasPrefix(op, "attach:") || strings.HasPrefix(op, "mount:") {
+			brought++
+		}
+	}
+	if brought != 2*len(applied) {
+		t.Errorf("bring-up ops=%v, want one attach and one mount per volume of %v", ops, applied)
+	}
+	for _, volume := range applied {
+		attach := slices.Index(ops, "attach:"+volume.Name)
+		mount := slices.Index(ops, "mount:"+volume.Name+":"+volume.Mount)
+		if attach < 0 || mount < attach {
+			t.Errorf("volume %s: attach=%d mount=%d in %v, want the attach first", volume.Name, attach, mount, ops)
+		}
+		if volume.RW() && !eng.dirtyAtAttach(volume.Name) {
+			t.Errorf("volume %s attached before its dirty marker was durable", volume.Name)
+		}
+	}
+	byName := func(a, b types.Volume) int { return strings.Compare(a.Name, b.Name) }
+	mounts := slices.SortedFunc(slices.Values(eng.volumeMounts), byName)
+	if want := slices.SortedFunc(slices.Values(applied), byName); !slices.Equal(mounts, want) {
+		t.Errorf("mounts=%v, want %v", mounts, want)
+	}
+}
+
 func newVolumeManager(t *testing.T, eng *fakeEngine, volumes []config.VolumeSpec) *Manager {
 	t.Helper()
 	return newVolumeManagerAt(t, eng, t.TempDir(), volumes)