fix: EBS hot-plug PCIe port allocator (every attach past the first collided) (#421)

* fix: serialize EBS hot-plug per instance to prevent PCIe port race

Concurrent AttachVolume calls to one instance both allocated the lowest free
hot-plug PCIe root port via nextHotplugEBSPort (which reads live QEMU state),
then issued QMP device_add on the same hotplug-ebs bus. The second device_add
failed with 'PCI: slot 0 function 0 already occupied by virtio-blk-pci',
surfacing as nats:timeout -> ServerInternal and a CSI retry-storm
('retry quota exceeded'); volumes never attached.

Add a per-instance attachMu and hold it across the whole AttachVolume and
DetachVolume critical sections, mirroring the ENI hot-plug serialization.
The mutex is non-persisted and shared via the Manager's stable *VM.

* fix: allocate EBS hot-plug PCIe port from in-memory accounting

AttachVolume derived the free PCIe hot-plug root port from a live
query-block scan (nextHotplugEBSPort), extracting the bus with a
hotplug-ebs(\d+) regex on each device's QDev. Real QEMU reports a
hot-plugged blockdev by id (/machine/peripheral/vdisk-vol-xxx/
virtio-backend), with no hotplug-ebs component, so the regex never
matched, the used set stayed empty, and the allocator always returned
port 1. Only the first EBS volume per instance attached; every later
attach collided on "PCI: slot 0 function 0 already occupied by
virtio-blk-pci", rolling back the blockdev, killing NBDKit, and
surfacing as nats: timeout / ServerInternal to the CSI driver, which
then exhausted its retry quota.

Allocate the port from authoritative in-memory accounting instead:
record the chosen port as EBSRequest.HotplugPort (persisted, so it
survives a daemon restart) and pick the lowest unclaimed port with
freeHotplugEBSPort over the instance's recorded requests. Hold
attachMu across allocate->device_add so concurrent CSI attaches cannot
pick the same port. DetachVolume already removes the request, freeing
the port. Drops the query-block round-trip from the attach path.

spinifex/types/ebs.go

@@ -18,6 +18,10 @@ type EBSRequest struct {
 	DeleteOnTermination bool   `json:"DeleteOnTermination"`
 	NBDURI              string `json:"NBDURI"`     // NBD URI - socket path (nbd:unix:/path.sock) or TCP (nbd://host:port)
 	DeviceName          string `json:"DeviceName"` // AWS API device name (e.g. /dev/sdf) for hot-plugged volumes
+	// HotplugPort is the PCIe hot-plug root port (hotplug-ebs{N}) a hot-plugged
+	// volume occupies. 0 for boot/non-hot-plugged volumes. Persisted so port
+	// accounting survives a daemon restart.
+	HotplugPort int `json:"HotplugPort,omitempty"`
 }
 
 // NBDTransport defines the transport type for NBD connections

spinifex/vm/device_map.go

@@ -11,6 +11,7 @@ import (
 	"time"
 
 	"github.com/mulgadc/spinifex/spinifex/qmp"
+	"github.com/mulgadc/spinifex/spinifex/types"
 )
 
 // pciAddrRegexp extracts the device index from a boot-time QDev path.
@@ -260,30 +261,26 @@ func nextAvailableDevice(instance *VM) string {
 	return ""
 }
 
-// nextHotplugEBSPort picks the lowest free EBS hot-plug PCIe port
-// (1..EBSHotPlugSlotCount) from live QEMU state. The port is decoupled from the
-// AWS device name: callers (e.g. the EBS CSI driver) supply arbitrary names like
-// /dev/xvdaa that do not map onto the /dev/sd[f-p] slot range, so the physical
-// port is allocated independently. Returns 0 when the pool is exhausted.
-func nextHotplugEBSPort(qmpClient *qmp.QMPClient, instanceID string) (int, error) {
-	resp, err := sendQMPCommand(qmpClient, qmp.QMPCommand{Execute: "query-block"}, instanceID)
-	if err != nil {
-		return 0, fmt.Errorf("query-block failed: %w", err)
-	}
-	var devices []qmp.BlockDevice
-	if err := json.Unmarshal(resp.Return, &devices); err != nil {
-		return 0, fmt.Errorf("failed to parse query-block response: %w", err)
-	}
-	used := make(map[int]bool, len(devices))
-	for _, dev := range devices {
-		if port := extractHotplugPort(dev.QDev); port > 0 {
-			used[port] = true
+// freeHotplugEBSPort returns the lowest EBS hot-plug PCIe port
+// (1..EBSHotPlugSlotCount) not already claimed by an attached volume, or 0 when
+// the pool is exhausted. The port is decoupled from the AWS device name:
+// callers (e.g. the EBS CSI driver) supply arbitrary names like /dev/xvdaa that
+// do not map onto the /dev/sd[f-p] range, so the physical port is allocated
+// independently. Accounting is in-memory and authoritative: QEMU reports block
+// devices by id, not by bus, so a live query-block scan cannot tell which port
+// is occupied. Callers must hold the EBSRequests lock; serialize the matching
+// device_add under attachMu so allocation and use are atomic.
+func freeHotplugEBSPort(reqs []types.EBSRequest) int {
+	used := make(map[int]bool, len(reqs))
+	for _, r := range reqs {
+		if r.HotplugPort > 0 {
+			used[r.HotplugPort] = true
 		}
 	}
 	for p := 1; p <= EBSHotPlugSlotCount; p++ {
 		if !used[p] {
-			return p, nil
+			return p
 		}
 	}
-	return 0, nil
+	return 0
 }

spinifex/vm/device_map_test.go

@@ -6,6 +6,7 @@ import (
 	"time"
 
 	"github.com/mulgadc/spinifex/spinifex/qmp"
+	"github.com/mulgadc/spinifex/spinifex/types"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 )
@@ -326,3 +327,39 @@ func TestExtractHotplugPort(t *testing.T) {
 		})
 	}
 }
+
+func TestFreeHotplugEBSPort(t *testing.T) {
+	mk := func(ports ...int) []types.EBSRequest {
+		reqs := make([]types.EBSRequest, 0, len(ports))
+		for _, p := range ports {
+			reqs = append(reqs, types.EBSRequest{HotplugPort: p})
+		}
+		return reqs
+	}
+
+	tests := []struct {
+		name string
+		reqs []types.EBSRequest
+		want int
+	}{
+		{name: "empty pool returns 1", reqs: nil, want: 1},
+		{name: "first port taken returns 2", reqs: mk(1), want: 2},
+		{name: "lowest gap is reused", reqs: mk(1, 3), want: 2},
+		{name: "boot/zero ports ignored", reqs: mk(0, 0), want: 1},
+		{name: "contiguous run returns next", reqs: mk(1, 2, 3), want: 4},
+		{name: "out-of-order still finds gap", reqs: mk(3, 1), want: 2},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			assert.Equal(t, tt.want, freeHotplugEBSPort(tt.reqs))
+		})
+	}
+
+	t.Run("exhausted pool returns 0", func(t *testing.T) {
+		full := make([]types.EBSRequest, 0, EBSHotPlugSlotCount)
+		for p := 1; p <= EBSHotPlugSlotCount; p++ {
+			full = append(full, types.EBSRequest{HotplugPort: p})
+		}
+		assert.Equal(t, 0, freeHotplugEBSPort(full))
+	})
+}

spinifex/vm/vm.go

@@ -46,6 +46,13 @@ type VM struct {
 
 	QMPClient *qmp.QMPClient `json:"-"`
 
+	// attachMu serializes EBS volume hot-plug (attach/detach) for this instance
+	// so PCIe hot-plug port allocation and the matching device_add are atomic.
+	// Two concurrent attaches would otherwise pick the same port and the second
+	// device_add fails with "slot 0 ... already occupied". Non-persisted; the
+	// Manager hands out a stable *VM, so the lock is shared across calls.
+	attachMu sync.Mutex `json:"-"`
+
 	// User attributes (user initiated stop/delete)
 	Attributes types.EC2CommandAttributes `json:"attributes"`
 

spinifex/vm/volumes.go

@@ -65,6 +65,13 @@ func (m *Manager) AttachVolume(id, volumeID, device string) (AttachVolumeResult,
 			ErrInvalidTransition, id, status)
 	}
 
+	// Serialize hot-plug per instance so the port allocation below and the
+	// matching device_add are atomic; concurrent attaches would otherwise pick
+	// the same PCIe root port and the second device_add fails ("slot 0 already
+	// occupied"). Detach takes the same lock.
+	instance.attachMu.Lock()
+	defer instance.attachMu.Unlock()
+
 	if device == "" {
 		m.UpdateState(id, func(v *VM) { device = nextAvailableDevice(v) })
 		if device == "" {
@@ -108,21 +115,22 @@ func (m *Manager) AttachVolume(id, volumeID, device string) (AttachVolumeResult,
 	deviceID := fmt.Sprintf("vdisk-%s", volumeID)
 	iothreadID := fmt.Sprintf("ioth-%s", volumeID)
 
-	// Allocate the PCIe hot-plug port from live QEMU state before any QMP
-	// mutation, so an exhausted pool rolls back only the mount. virtio-blk-pci
-	// cannot land on pcie.0 (no hot-plug), so a free hotplug-ebs root port is
-	// required regardless of the AWS device name.
-	hotplugPort, err := nextHotplugEBSPort(instance.QMPClient, instance.ID)
-	if err != nil {
-		slog.Error("AttachVolume: failed to query hot-plug ports", "volumeId", volumeID, "err", err)
-		m.rollbackUnmount(ebsRequest)
-		return AttachVolumeResult{}, fmt.Errorf("query hot-plug ports: %w", err)
-	}
+	// Allocate the PCIe hot-plug port from in-memory accounting. QEMU reports
+	// block devices by id (/machine/peripheral/<id>/virtio-backend), not by bus,
+	// so a live query-block scan cannot tell which hotplug-ebs port is occupied
+	// and always returned the first one — every attach past the first collided
+	// on slot 0. The recorded HotplugPort per attached volume is authoritative;
+	// attachMu (held above) serializes allocation. virtio-blk-pci cannot land on
+	// pcie.0 (no hot-plug), so a free hotplug-ebs root port is always required.
+	instance.EBSRequests.Mu.Lock()
+	hotplugPort := freeHotplugEBSPort(instance.EBSRequests.Requests)
+	instance.EBSRequests.Mu.Unlock()
 	if hotplugPort == 0 {
 		slog.Error("AttachVolume: EBS hot-plug port pool exhausted", "volumeId", volumeID)
 		m.rollbackUnmount(ebsRequest)
 		return AttachVolumeResult{}, ErrAttachmentLimitExceeded
 	}
+	ebsRequest.HotplugPort = hotplugPort
 
 	if _, err := sendQMPCommand(instance.QMPClient, qmp.QMPCommand{
 		Execute: "object-add",
@@ -282,6 +290,10 @@ func (m *Manager) DetachVolume(id, volumeID, device string, force bool) (string,
 			ErrInvalidTransition, id, status)
 	}
 
+	// Serialize against concurrent attach/detach on this instance's PCIe ports.
+	instance.attachMu.Lock()
+	defer instance.attachMu.Unlock()
+
 	instance.EBSRequests.Mu.Lock()
 	var ebsReq types.EBSRequest
 	found := false

spinifex/vm/volumes_test.go

@@ -616,7 +616,7 @@ func TestAttachVolume_ObjectAddFailure_TriggersUnmountOne(t *testing.T) {
 	_, err := m.AttachVolume("i-1", "vol-1", "/dev/sdf")
 	require.Error(t, err)
 	assert.Contains(t, err.Error(), "object-add")
-	assert.Equal(t, []string{"query-block", "object-add"}, recorder.executes(),
+	assert.Equal(t, []string{"object-add"}, recorder.executes(),
 		"object-add failure must short-circuit before blockdev-add")
 	assert.Equal(t, []string{"vol-1"}, mounter.unmountedOne)
 }
@@ -644,7 +644,7 @@ func TestAttachVolume_BlockdevAddFailure_TriggersUnmountOne(t *testing.T) {
 	_, err := m.AttachVolume("i-1", "vol-1", "/dev/sdf")
 	require.Error(t, err)
 	assert.Contains(t, err.Error(), "blockdev-add")
-	assert.Equal(t, []string{"query-block", "object-add", "blockdev-add", "object-del"}, recorder.executes())
+	assert.Equal(t, []string{"object-add", "blockdev-add", "object-del"}, recorder.executes())
 	assert.Equal(t, []string{"vol-1"}, mounter.unmountedOne)
 }
 
@@ -671,7 +671,7 @@ func TestAttachVolume_DeviceAddFailure_BlockdevDelOK_Unmounts(t *testing.T) {
 	_, err := m.AttachVolume("i-1", "vol-1", "/dev/sdf")
 	require.Error(t, err)
 	assert.Contains(t, err.Error(), "device_add")
-	assert.Equal(t, []string{"query-block", "object-add", "blockdev-add", "device_add", "blockdev-del", "object-del"}, recorder.executes())
+	assert.Equal(t, []string{"object-add", "blockdev-add", "device_add", "blockdev-del", "object-del"}, recorder.executes())
 	assert.Equal(t, []string{"vol-1"}, mounter.unmountedOne,
 		"successful blockdev-del rollback must be followed by UnmountOne")
 }
@@ -704,7 +704,7 @@ func TestAttachVolume_DeviceAddFailure_BlockdevDelFails_SkipsUnmountOne(t *testi
 	_, err := m.AttachVolume("i-1", "vol-1", "/dev/sdf")
 	require.Error(t, err)
 	assert.Contains(t, err.Error(), "device_add")
-	assert.Equal(t, []string{"query-block", "object-add", "blockdev-add", "device_add", "blockdev-del"}, recorder.executes())
+	assert.Equal(t, []string{"object-add", "blockdev-add", "device_add", "blockdev-del"}, recorder.executes())
 	assert.Empty(t, mounter.unmountedOne,
 		"failed blockdev-del must skip UnmountOne — unmounting under a live block node crashes QEMU")
 }