From c2be7f047c686df8b6c22d79ee1386db3d629a8a Mon Sep 17 00:00:00 2001 From: Julian Sommer <115516918+juliansommer@users.noreply.github.com> Date: Tue, 5 May 2026 14:27:59 +1000 Subject: [PATCH] refactor(vm): move AttachVolume/DetachVolume/Reboot onto Manager (#189) --- spinifex/daemon/daemon.go | 110 ---- spinifex/daemon/daemon_handlers_instance.go | 33 +- spinifex/daemon/daemon_handlers_volume.go | 476 ++------------- .../daemon/daemon_handlers_volume_test.go | 88 +++ spinifex/daemon/daemon_test.go | 177 ++---- spinifex/daemon/vm_adapters.go | 63 ++ spinifex/{daemon => vm}/device_map.go | 70 ++- spinifex/{daemon => vm}/device_map_test.go | 2 +- spinifex/vm/errors.go | 23 + spinifex/vm/lifecycle.go | 21 + spinifex/vm/manager.go | 6 + spinifex/vm/manager_deps_test.go | 24 +- spinifex/vm/volume_mounter.go | 9 + spinifex/vm/volumes.go | 416 +++++++++++++ spinifex/vm/volumes_test.go | 564 ++++++++++++++++++ 15 files changed, 1365 insertions(+), 717 deletions(-) create mode 100644 spinifex/daemon/daemon_handlers_volume_test.go rename spinifex/{daemon => vm}/device_map.go (77%) rename spinifex/{daemon => vm}/device_map_test.go (99%) create mode 100644 spinifex/vm/volumes.go create mode 100644 spinifex/vm/volumes_test.go diff --git a/spinifex/daemon/daemon.go b/spinifex/daemon/daemon.go index 91d8d0f2a..fd9f23519 100644 --- a/spinifex/daemon/daemon.go +++ b/spinifex/daemon/daemon.go @@ -1687,82 +1687,6 @@ func (d *Daemon) ebsTopic(action string) string { return fmt.Sprintf("ebs.%s.%s", d.node, action) } -// MountVolumes mounts the volumes for an instance -func (d *Daemon) MountVolumes(instance *vm.VM) error { - instance.EBSRequests.Mu.Lock() - defer instance.EBSRequests.Mu.Unlock() - - for k, v := range instance.EBSRequests.Requests { - // Send the volume payload as JSON - ebsMountRequest, err := json.Marshal(v) - - if err != nil { - slog.Error("Failed to marshal volume payload", "err", err) - return err - } - - reply, err := d.natsConn.Request(d.ebsTopic("mount"), ebsMountRequest, 30*time.Second) - - slog.Info("Mounting volume", "Vol", v.Name, "NBDURI", v.NBDURI) - - // TODO: Improve timeout handling - if err != nil { - slog.Error("Failed to request EBS mount", "err", err) - return err - } - - // Unmarshal the response - var ebsMountResponse types.EBSMountResponse - err = json.Unmarshal(reply.Data, &ebsMountResponse) - - if err != nil { - slog.Error("Failed to unmarshal volume response:", "err", err) - return err - } - - if ebsMountResponse.Error == "" { - slog.Debug("Mounted volume successfully", "response", ebsMountResponse.URI) - - // Append the NBD URI to the request - instance.EBSRequests.Requests[k].NBDURI = ebsMountResponse.URI - } else { - slog.Error("Failed to mount volume", "error", ebsMountResponse.Error) - return fmt.Errorf("failed to mount volume: %s", ebsMountResponse.Error) - } - } - - return nil -} - -// rollbackEBSMount sends an ebs.unmount request to undo a previously successful ebs.mount. -// Rollback failures are logged but not propagated; callers treat this as best-effort cleanup. -func (d *Daemon) rollbackEBSMount(req types.EBSRequest) { - data, err := json.Marshal(req) - if err != nil { - slog.Error("rollbackEBSMount: failed to marshal unmount request", "volume", req.Name, "err", err) - return - } - msg, err := d.natsConn.Request(d.ebsTopic("unmount"), data, 10*time.Second) - if err != nil { - slog.Error("rollbackEBSMount: ebs.unmount NATS request failed", "volume", req.Name, "err", err) - return - } - var resp types.EBSUnMountResponse - if err := json.Unmarshal(msg.Data, &resp); err != nil { - slog.Error("rollbackEBSMount: failed to unmarshal response", "volume", req.Name, "err", err) - return - } - if resp.Error != "" { - slog.Error("rollbackEBSMount: ebs.unmount returned error", "volume", req.Name, "err", resp.Error) - return - } - if resp.Mounted { - slog.Error("rollbackEBSMount: volume still mounted after unmount", "volume", req.Name) - return - } - slog.Info("rollbackEBSMount: volume unmounted successfully", "volume", req.Name) -} - // respondWithVolumeAttachment builds an ec2.VolumeAttachment, marshals it to JSON, and // responds on the NATS message. Used by both AttachVolume and DetachVolume handlers. func (d *Daemon) respondWithVolumeAttachment(msg *nats.Msg, volumeID, instanceID, device, state string) { @@ -1787,40 +1711,6 @@ func (d *Daemon) respondWithVolumeAttachment(msg *nats.Msg, volumeID, instanceID } } -// nextAvailableDevice finds the next available /dev/sd[f-p] device name for an instance. -// It checks both EBSRequests and BlockDeviceMappings to avoid conflicts. -func nextAvailableDevice(instance *vm.VM) string { - usedDevices := make(map[string]bool) - - // Collect devices from existing BlockDeviceMappings - if instance.Instance != nil { - for _, bdm := range instance.Instance.BlockDeviceMappings { - if bdm.DeviceName != nil { - usedDevices[*bdm.DeviceName] = true - } - } - } - - // Collect devices from EBSRequests (may not yet be in BlockDeviceMappings) - instance.EBSRequests.Mu.Lock() - for _, req := range instance.EBSRequests.Requests { - if req.DeviceName != "" { - usedDevices[req.DeviceName] = true - } - } - instance.EBSRequests.Mu.Unlock() - - // AWS convention: /dev/sd[f-p] for attached volumes - for c := 'f'; c <= 'p'; c++ { - dev := fmt.Sprintf("/dev/sd%c", c) - if !usedDevices[dev] { - return dev - } - } - - return "" -} - // canAllocate checks how many instances of the given type can be allocated. // Returns the count that can actually be allocated (0 to count). func (rm *ResourceManager) canAllocate(instanceType *ec2.InstanceTypeInfo, count int) int { diff --git a/spinifex/daemon/daemon_handlers_instance.go b/spinifex/daemon/daemon_handlers_instance.go index a64e4962a..deafbd180 100644 --- a/spinifex/daemon/daemon_handlers_instance.go +++ b/spinifex/daemon/daemon_handlers_instance.go @@ -3,6 +3,7 @@ package daemon import ( "encoding/base64" "encoding/json" + "errors" "fmt" "log/slog" "strings" @@ -12,7 +13,6 @@ import ( "github.com/aws/aws-sdk-go/service/ec2" "github.com/mulgadc/spinifex/spinifex/awserrors" "github.com/mulgadc/spinifex/spinifex/filterutil" - "github.com/mulgadc/spinifex/spinifex/qmp" "github.com/mulgadc/spinifex/spinifex/types" "github.com/mulgadc/spinifex/spinifex/utils" "github.com/mulgadc/spinifex/spinifex/vm" @@ -391,7 +391,7 @@ func (d *Daemon) handleEC2RunInstances(msg *nats.Msg) { } // Discover actual guest device names via QMP query-block - d.updateGuestDeviceNames(instance) + d.vmMgr.UpdateGuestDeviceNames(instance) successCount++ slog.Info("handleEC2RunInstances launched instance", "instanceId", instance.ID) @@ -403,19 +403,18 @@ func (d *Daemon) handleEC2RunInstances(msg *nats.Msg) { func (d *Daemon) handleRebootInstance(msg *nats.Msg, command types.EC2InstanceCommand, instance *vm.VM) { slog.Info("Rebooting instance", "id", command.ID) - var status vm.InstanceState - d.vmMgr.Inspect(instance, func(v *vm.VM) { status = v.Status }) - - if status != vm.StateRunning { - slog.Error("RebootInstance: instance not in running state", "instanceId", command.ID, "status", status) - respondWithError(msg, awserrors.ErrorIncorrectInstanceState) - return - } - - _, err := d.SendQMPCommand(instance.QMPClient, qmp.QMPCommand{Execute: "system_reset"}, command.ID) - if err != nil { - slog.Error("RebootInstance: QMP system_reset failed", "instanceId", command.ID, "err", err) - respondWithError(msg, awserrors.ErrorServerInternal) + if err := d.vmMgr.Reboot(instance.ID); err != nil { + switch { + case errors.Is(err, vm.ErrInstanceNotFound): + respondWithError(msg, awserrors.ErrorInvalidInstanceIDNotFound) + case errors.Is(err, vm.ErrInvalidTransition): + slog.Error("RebootInstance: instance not in running state", + "instanceId", command.ID, "err", err) + respondWithError(msg, awserrors.ErrorIncorrectInstanceState) + default: + slog.Error("RebootInstance: reboot failed", "instanceId", command.ID, "err", err) + respondWithError(msg, awserrors.ErrorServerInternal) + } return } @@ -465,7 +464,7 @@ func (d *Daemon) handleStartInstance(msg *nats.Msg, command types.EC2InstanceCom } // Discover actual guest device names via QMP query-block - d.updateGuestDeviceNames(instance) + d.vmMgr.UpdateGuestDeviceNames(instance) slog.Info("Instance started", "instanceId", instance.ID) @@ -878,7 +877,7 @@ func (d *Daemon) handleEC2StartStoppedInstance(msg *nats.Msg) { } // Discover actual guest device names via QMP query-block - d.updateGuestDeviceNames(instance) + d.vmMgr.UpdateGuestDeviceNames(instance) // Remove from shared KV now that it's running locally. // Retry once on failure — a stale KV entry risks duplicate starts. diff --git a/spinifex/daemon/daemon_handlers_volume.go b/spinifex/daemon/daemon_handlers_volume.go index b70a17f74..e3656c290 100644 --- a/spinifex/daemon/daemon_handlers_volume.go +++ b/spinifex/daemon/daemon_handlers_volume.go @@ -2,30 +2,26 @@ package daemon import ( "encoding/json" - "fmt" + "errors" "log/slog" - "strconv" - "strings" "time" "github.com/aws/aws-sdk-go/service/ec2" "github.com/mulgadc/spinifex/spinifex/awserrors" - "github.com/mulgadc/spinifex/spinifex/qmp" "github.com/mulgadc/spinifex/spinifex/types" "github.com/mulgadc/spinifex/spinifex/utils" "github.com/mulgadc/spinifex/spinifex/vm" "github.com/nats-io/nats.go" ) -// handleAttachVolume performs a three-phase hot-plug: -// -// Phase 1: ebs.mount via NATS (rolls back with ebs.unmount) -// Phase 2: QMP blockdev-add (rolls back Phase 1) -// Phase 3: QMP device_add (rolls back Phase 2 + Phase 1) +// handleAttachVolume validates the volume request against the volume +// service (existence, ownership, availability, AZ) and dispatches the +// QMP/state-machine pipeline to vm.Manager.AttachVolume. The manager owns +// every QMP and persistence side-effect; the daemon only emits the AWS API +// response. func (d *Daemon) handleAttachVolume(msg *nats.Msg, command types.EC2InstanceCommand, instance *vm.VM) { slog.Info("Attaching volume to instance", "instanceId", command.ID) - // Validate AttachVolumeData if command.AttachVolumeData == nil || command.AttachVolumeData.VolumeID == "" { slog.Error("AttachVolume: missing attach volume data") respondWithError(msg, awserrors.ErrorInvalidParameterValue) @@ -33,19 +29,18 @@ func (d *Daemon) handleAttachVolume(msg *nats.Msg, command types.EC2InstanceComm } volumeID := command.AttachVolumeData.VolumeID - device := command.AttachVolumeData.Device - // Validate instance is running + // Surface IncorrectInstanceState before any volume-side lookups so the + // AWS-SDK error matches the pre-2d ordering. var status vm.InstanceState d.vmMgr.Inspect(instance, func(v *vm.VM) { status = v.Status }) - if status != vm.StateRunning { - slog.Error("AttachVolume: instance not running", "instanceId", command.ID, "status", status) + slog.Error("AttachVolume: instance not running", + "instanceId", command.ID, "status", status) respondWithError(msg, awserrors.ErrorIncorrectInstanceState) return } - // Validate volume exists and is available volCfg, err := d.volumeService.GetVolumeConfig(volumeID) if err != nil { slog.Error("AttachVolume: failed to get volume config", "volumeId", volumeID, "err", err) @@ -53,400 +48,86 @@ func (d *Daemon) handleAttachVolume(msg *nats.Msg, command types.EC2InstanceComm return } - // Account scoping: verify the caller owns this volume. - // Pre-Phase4 volumes (empty TenantID) are root-only; otherwise the caller - // must match the recorded tenant exactly. callerAccountID := utils.AccountIDFromMsg(msg) if !volumeVisibleTo(volCfg.VolumeMetadata.TenantID, callerAccountID) { - slog.Warn("AttachVolume: account does not own volume", "volumeId", volumeID, "callerAccount", callerAccountID, "ownerAccount", volCfg.VolumeMetadata.TenantID) + slog.Warn("AttachVolume: account does not own volume", + "volumeId", volumeID, + "callerAccount", callerAccountID, + "ownerAccount", volCfg.VolumeMetadata.TenantID) respondWithError(msg, awserrors.ErrorInvalidVolumeNotFound) return } if volCfg.VolumeMetadata.State != "available" { - slog.Error("AttachVolume: volume not available", "volumeId", volumeID, "state", volCfg.VolumeMetadata.State) + slog.Error("AttachVolume: volume not available", + "volumeId", volumeID, "state", volCfg.VolumeMetadata.State) respondWithError(msg, awserrors.ErrorVolumeInUse) return } - // Check AZ compatibility — volume and instance must be in the same AZ if volCfg.VolumeMetadata.AvailabilityZone != "" && d.config.AZ != "" && volCfg.VolumeMetadata.AvailabilityZone != d.config.AZ { slog.Error("AttachVolume: volume and instance are in different AZs", - "volumeId", volumeID, "volumeAZ", volCfg.VolumeMetadata.AvailabilityZone, + "volumeId", volumeID, + "volumeAZ", volCfg.VolumeMetadata.AvailabilityZone, "instanceAZ", d.config.AZ) respondWithError(msg, awserrors.ErrorInvalidVolumeZoneMismatch) return } - // Determine device name - if device == "" { - d.vmMgr.Inspect(instance, func(v *vm.VM) { device = nextAvailableDevice(v) }) - if device == "" { - slog.Error("AttachVolume: no available device names") - respondWithError(msg, awserrors.ErrorAttachmentLimitExceeded) - return - } - } - - // Create EBS mount request - ebsRequest := types.EBSRequest{ - Name: volumeID, - DeviceName: device, - } - - ebsMountData, err := json.Marshal(ebsRequest) + res, err := d.vmMgr.AttachVolume(instance.ID, volumeID, command.AttachVolumeData.Device) if err != nil { - slog.Error("AttachVolume: failed to marshal ebs.mount request", "err", err) - respondWithError(msg, awserrors.ErrorServerInternal) + respondWithError(msg, attachDetachErrorCode(err)) return } - mountReply, err := d.natsConn.Request(d.ebsTopic("mount"), ebsMountData, 30*time.Second) - if err != nil { - slog.Error("AttachVolume: ebs.mount failed", "volumeId", volumeID, "err", err) - respondWithError(msg, awserrors.ErrorServerInternal) - return - } - - var mountResp types.EBSMountResponse - if err := json.Unmarshal(mountReply.Data, &mountResp); err != nil { - slog.Error("AttachVolume: failed to unmarshal mount response", "err", err) - respondWithError(msg, awserrors.ErrorServerInternal) - return - } - - if mountResp.Error != "" { - slog.Error("AttachVolume: mount returned error", "volumeId", volumeID, "err", mountResp.Error) - respondWithError(msg, awserrors.ErrorServerInternal) - return - } - - nbdURI := mountResp.URI - if nbdURI == "" { - slog.Error("AttachVolume: mount response has empty URI", "volumeId", volumeID) - d.rollbackEBSMount(ebsRequest) - respondWithError(msg, awserrors.ErrorServerInternal) - return - } - ebsRequest.NBDURI = nbdURI - - // Parse NBDURI for QMP blockdev-add - serverType, socketPath, nbdHost, nbdPort, err := utils.ParseNBDURI(nbdURI) - if err != nil { - slog.Error("AttachVolume: failed to parse NBDURI", "uri", nbdURI, "err", err) - // Rollback: unmount - d.rollbackEBSMount(ebsRequest) - respondWithError(msg, awserrors.ErrorServerInternal) - return - } - - // Build QMP server argument - var serverArg map[string]any - if serverType == "unix" { - serverArg = map[string]any{"type": "unix", "path": socketPath} - } else { - serverArg = map[string]any{"type": "inet", "host": nbdHost, "port": strconv.Itoa(nbdPort)} - } - - nodeName := fmt.Sprintf("nbd-%s", volumeID) - deviceID := fmt.Sprintf("vdisk-%s", volumeID) - iothreadID := fmt.Sprintf("ioth-%s", volumeID) - - // QMP object-add: create iothread for this volume - iothreadCmd := qmp.QMPCommand{ - Execute: "object-add", - Arguments: map[string]any{ - "qom-type": "iothread", - "id": iothreadID, - }, - } - _, err = d.SendQMPCommand(instance.QMPClient, iothreadCmd, instance.ID) - if err != nil { - slog.Error("AttachVolume: QMP object-add iothread failed", "volumeId", volumeID, "err", err) - d.rollbackEBSMount(ebsRequest) - respondWithError(msg, awserrors.ErrorServerInternal) - return - } - - // QMP blockdev-add - blockdevCmd := qmp.QMPCommand{ - Execute: "blockdev-add", - Arguments: map[string]any{ - "node-name": nodeName, - "driver": "nbd", - "server": serverArg, - "export": "", - "read-only": false, - }, - } - - _, err = d.SendQMPCommand(instance.QMPClient, blockdevCmd, instance.ID) - if err != nil { - slog.Error("AttachVolume: QMP blockdev-add failed", "volumeId", volumeID, "err", err) - d.rollbackEBSMount(ebsRequest) - respondWithError(msg, awserrors.ErrorServerInternal) - return - } - - // Determine which hotplug root port to use based on device letter. - // /dev/sdf -> hotplug1, /dev/sdg -> hotplug2, etc. - hotplugBus := "" - if len(device) > 0 { - letter := device[len(device)-1] - if letter >= 'f' && letter <= 'p' { - hotplugBus = fmt.Sprintf("hotplug%d", letter-'f'+1) - } - } - - // QMP device_add - deviceAddArgs := map[string]any{ - "driver": "virtio-blk-pci", - "id": deviceID, - "drive": nodeName, - "iothread": iothreadID, - } - if hotplugBus != "" { - deviceAddArgs["bus"] = hotplugBus - } - deviceAddCmd := qmp.QMPCommand{ - Execute: "device_add", - Arguments: deviceAddArgs, - } - - _, err = d.SendQMPCommand(instance.QMPClient, deviceAddCmd, instance.ID) - if err != nil { - slog.Error("AttachVolume: QMP device_add failed, rolling back blockdev", "volumeId", volumeID, "err", err) - // Rollback: blockdev-del then unmount - if _, delErr := d.SendQMPCommand(instance.QMPClient, qmp.QMPCommand{ - Execute: "blockdev-del", - Arguments: map[string]any{"node-name": nodeName}, - }, instance.ID); delErr != nil { - slog.Error("AttachVolume: rollback blockdev-del failed, skipping EBS unmount", "volumeId", volumeID, "err", delErr) - } else { - d.rollbackEBSMount(ebsRequest) - } - respondWithError(msg, awserrors.ErrorServerInternal) - return - } - - // Discover actual guest device name via QMP query-block. - // Use the retry variant because query-block may not include the device - // immediately after device_add. - guestDevice := device // fallback to AWS API name - deviceMap, qmpErr := queryGuestDeviceMapWait(d, instance.QMPClient, instance.ID, deviceID) - if qmpErr != nil { - slog.Warn("AttachVolume: failed to query guest device map, using API device name", "volumeId", volumeID, "err", qmpErr) - } else if gd, ok := deviceMap[deviceID]; ok { - guestDevice = gd - slog.Info("AttachVolume: discovered guest device", "volumeId", volumeID, "qemuDevice", deviceID, "guestDevice", guestDevice) - } else { - slog.Error("AttachVolume: device not found in QMP device map after retries, using API device name", - "volumeId", volumeID, "qemuDevice", deviceID, "deviceMap", deviceMap) - } - - // Update instance state: replace existing entry for this volume (handles - // stop/start cycles that keep stale entries) or append a new one. - instance.EBSRequests.Mu.Lock() - replaced := false - for i, req := range instance.EBSRequests.Requests { - if req.Name == volumeID { - instance.EBSRequests.Requests[i] = ebsRequest - replaced = true - break - } - } - if !replaced { - instance.EBSRequests.Requests = append(instance.EBSRequests.Requests, ebsRequest) - } - instance.EBSRequests.Mu.Unlock() - - // Update BlockDeviceMappings on the ec2.Instance using actual guest device name - d.vmMgr.Inspect(instance, func(v *vm.VM) { - if v.Instance == nil { - return - } - now := time.Now() - mapping := &ec2.InstanceBlockDeviceMapping{} - mapping.SetDeviceName(guestDevice) - mapping.Ebs = &ec2.EbsInstanceBlockDevice{} - mapping.Ebs.SetVolumeId(volumeID) - mapping.Ebs.SetAttachTime(now) - mapping.Ebs.SetDeleteOnTermination(false) - mapping.Ebs.SetStatus("attached") - v.Instance.BlockDeviceMappings = append(v.Instance.BlockDeviceMappings, mapping) - }) - - // Update volume metadata in S3 - if err := d.volumeService.UpdateVolumeState(volumeID, "in-use", command.ID, guestDevice); err != nil { - slog.Error("AttachVolume: failed to update volume metadata", "volumeId", volumeID, "err", err) - } - - // Persist state - if err := d.WriteState(); err != nil { - slog.Error("AttachVolume: failed to write state", "err", err) - } - - d.respondWithVolumeAttachment(msg, volumeID, command.ID, guestDevice, "attached") - slog.Info("Volume attached successfully", "volumeId", volumeID, "instanceId", command.ID, "apiDevice", device, "guestDevice", guestDevice) + d.respondWithVolumeAttachment(msg, volumeID, command.ID, res.GuestDevice, "attached") } -// handleDetachVolume performs a three-phase hot-unplug (reverse of attach): -// -// Phase 1: QMP device_del (remove guest device) -// Phase 2: QMP blockdev-del (remove block node) -// Phase 3: ebs.unmount NATS (stop NBD server) +// handleDetachVolume dispatches the QMP/state-machine pipeline to +// vm.Manager.DetachVolume and emits the AWS API response. func (d *Daemon) handleDetachVolume(msg *nats.Msg, command types.EC2InstanceCommand, instance *vm.VM) { slog.Info("Detaching volume from instance", "instanceId", command.ID) - // Validate DetachVolumeData if command.DetachVolumeData == nil || command.DetachVolumeData.VolumeID == "" { slog.Error("DetachVolume: missing detach volume data") respondWithError(msg, awserrors.ErrorInvalidParameterValue) return } - volumeID := command.DetachVolumeData.VolumeID - device := command.DetachVolumeData.Device - force := command.DetachVolumeData.Force - - // Validate instance is running - var status vm.InstanceState - d.vmMgr.Inspect(instance, func(v *vm.VM) { status = v.Status }) - - if status != vm.StateRunning { - slog.Error("DetachVolume: instance not running", "instanceId", command.ID, "status", status) - respondWithError(msg, awserrors.ErrorIncorrectInstanceState) + deviceName, err := d.vmMgr.DetachVolume( + instance.ID, + command.DetachVolumeData.VolumeID, + command.DetachVolumeData.Device, + command.DetachVolumeData.Force, + ) + if err != nil { + respondWithError(msg, attachDetachErrorCode(err)) return } - // Find the volume in EBSRequests - instance.EBSRequests.Mu.Lock() - var ebsReq types.EBSRequest - found := false - for _, req := range instance.EBSRequests.Requests { - if req.Name == volumeID { - ebsReq = req - found = true - break - } - } - instance.EBSRequests.Mu.Unlock() + d.respondWithVolumeAttachment(msg, command.DetachVolumeData.VolumeID, command.ID, deviceName, "detaching") +} - if !found { - slog.Error("DetachVolume: volume not attached to instance", "volumeId", volumeID, "instanceId", command.ID) - respondWithError(msg, awserrors.ErrorIncorrectState) - return - } - - // Reject detaching boot/EFI/CloudInit volumes - if ebsReq.Boot || ebsReq.EFI || ebsReq.CloudInit { - slog.Error("DetachVolume: cannot detach boot/EFI/CloudInit volume", "volumeId", volumeID) - respondWithError(msg, awserrors.ErrorOperationNotPermitted) - return - } - - // Optional device cross-check - if device != "" && ebsReq.DeviceName != "" && device != ebsReq.DeviceName { - slog.Error("DetachVolume: device mismatch", "requested", device, "actual", ebsReq.DeviceName) - respondWithError(msg, awserrors.ErrorInvalidParameterValue) - return - } - - deviceID := fmt.Sprintf("vdisk-%s", volumeID) - nodeName := fmt.Sprintf("nbd-%s", volumeID) - iothreadID := fmt.Sprintf("ioth-%s", volumeID) - - // Phase 1: QMP device_del (remove guest device). - // Idempotent: a prior detach may have already removed the device but - // failed at blockdev-del, leaving the volume's external state intact. - // Treat DeviceNotFound as success so the retry can drive blockdev-del - // to completion without bailing on the now-absent guest device. - _, err := d.SendQMPCommand(instance.QMPClient, qmp.QMPCommand{ - Execute: "device_del", - Arguments: map[string]any{"id": deviceID}, - }, instance.ID) +// attachDetachErrorCode maps a vm.Manager error returned by AttachVolume +// or DetachVolume to the AWS API error code that the SDK expects. +func attachDetachErrorCode(err error) string { switch { - case err == nil: - case isQMPDeviceNotFound(err): - slog.Info("DetachVolume: guest device already removed (resuming detach)", "volumeId", volumeID, "err", err) - case force: - slog.Warn("DetachVolume: QMP device_del failed (force=true, continuing)", "volumeId", volumeID, "err", err) + case errors.Is(err, vm.ErrInstanceNotFound): + return awserrors.ErrorInvalidInstanceIDNotFound + case errors.Is(err, vm.ErrInvalidTransition): + return awserrors.ErrorIncorrectInstanceState + case errors.Is(err, vm.ErrAttachmentLimitExceeded): + return awserrors.ErrorAttachmentLimitExceeded + case errors.Is(err, vm.ErrVolumeNotAttached): + return awserrors.ErrorIncorrectState + case errors.Is(err, vm.ErrVolumeNotDetachable): + return awserrors.ErrorOperationNotPermitted + case errors.Is(err, vm.ErrVolumeDeviceMismatch): + return awserrors.ErrorInvalidParameterValue default: - slog.Error("DetachVolume: QMP device_del failed", "volumeId", volumeID, "err", err) - respondWithError(msg, awserrors.ErrorServerInternal) - return + return awserrors.ErrorServerInternal } - - // Brief pause for guest to acknowledge PCI removal - time.Sleep(d.detachDelay) - - // Phase 2: QMP blockdev-del (remove block node). - // Poll-retry on "node is in use": after device_del, NBD client teardown - // and any in-flight I/O can hold the block node briefly. Bailing here - // without retry leaves the guest device gone but the block node intact, - // and the next AWS-CLI retry hits DeviceNotFound on device_del with no - // path forward (mulga-siv-41). - blockdevErr := d.tryBlockdevDel(instance, nodeName) - if blockdevErr != nil { - // Block node still referenced after retry budget exhausted; do not - // clean up state or unmount — tearing down the NBD server would - // crash the VM, and removing metadata would allow the volume to be - // double-attached. - slog.Error("DetachVolume: QMP blockdev-del failed, leaving volume state intact", "volumeId", volumeID, "err", blockdevErr) - respondWithError(msg, awserrors.ErrorServerInternal) - return - } - - // Phase 2b: QMP object-del (remove iothread, best-effort) - _, iothreadErr := d.SendQMPCommand(instance.QMPClient, qmp.QMPCommand{ - Execute: "object-del", - Arguments: map[string]any{"id": iothreadID}, - }, instance.ID) - if iothreadErr != nil { - slog.Warn("DetachVolume: QMP object-del iothread failed (non-fatal)", "volumeId", volumeID, "err", iothreadErr) - } - - // Phase 3: ebs.unmount via NATS (best-effort) - d.rollbackEBSMount(ebsReq) - - // State cleanup: remove volume from EBSRequests (search by name to avoid stale index) - instance.EBSRequests.Mu.Lock() - for i, req := range instance.EBSRequests.Requests { - if req.Name == volumeID { - instance.EBSRequests.Requests = append(instance.EBSRequests.Requests[:i], instance.EBSRequests.Requests[i+1:]...) - break - } - } - instance.EBSRequests.Mu.Unlock() - - // Remove from BlockDeviceMappings - d.vmMgr.Inspect(instance, func(v *vm.VM) { - if v.Instance == nil { - return - } - filtered := make([]*ec2.InstanceBlockDeviceMapping, 0, len(v.Instance.BlockDeviceMappings)) - for _, bdm := range v.Instance.BlockDeviceMappings { - if bdm.Ebs != nil && bdm.Ebs.VolumeId != nil && *bdm.Ebs.VolumeId == volumeID { - continue - } - filtered = append(filtered, bdm) - } - v.Instance.BlockDeviceMappings = filtered - }) - - // Update volume metadata to "available" - if err := d.volumeService.UpdateVolumeState(volumeID, "available", "", ""); err != nil { - slog.Error("DetachVolume: failed to update volume metadata", "volumeId", volumeID, "err", err) - } - - // Persist state - if err := d.WriteState(); err != nil { - slog.Error("DetachVolume: failed to write state", "err", err) - } - - d.respondWithVolumeAttachment(msg, volumeID, command.ID, ebsReq.DeviceName, "detaching") - slog.Info("Volume detached successfully", "volumeId", volumeID, "instanceId", command.ID) } func (d *Daemon) handleEC2CreateVolume(msg *nats.Msg) { @@ -515,62 +196,3 @@ func (d *Daemon) handleEC2ModifyVolume(msg *nats.Msg) { func (d *Daemon) handleEC2DeleteVolume(msg *nats.Msg) { handleNATSRequest(msg, d.volumeService.DeleteVolume) } - -// tryBlockdevDel issues blockdev-del with bounded retry on "is in use" -// errors. After device_del, NBD client teardown and any in-flight guest -// I/O can briefly hold the block node; QEMU surfaces this as a -// GenericError carrying "Node is in use". Polling at d.detachDelay -// gives the I/O drain enough time without a hardcoded long sleep. -func (d *Daemon) tryBlockdevDel(instance *vm.VM, nodeName string) error { - const maxAttempts = 20 // 20 × detachDelay (default 1s) = 20s budget - var lastErr error - for attempt := 1; attempt <= maxAttempts; attempt++ { - _, err := d.SendQMPCommand(instance.QMPClient, qmp.QMPCommand{ - Execute: "blockdev-del", - Arguments: map[string]any{"node-name": nodeName}, - }, instance.ID) - if err == nil { - if attempt > 1 { - slog.Info("DetachVolume: blockdev-del succeeded after retry", - "nodeName", nodeName, "attempts", attempt) - } - return nil - } - lastErr = err - if !isQMPNodeInUse(err) { - return err - } - if attempt == maxAttempts { - break - } - slog.Debug("DetachVolume: blockdev-del busy, retrying", - "nodeName", nodeName, "attempt", attempt, "err", err) - time.Sleep(d.detachDelay) - } - return lastErr -} - -// isQMPDeviceNotFound returns true when err is a QMP DeviceNotFound class -// error from device_del. Used to make device_del idempotent across detach -// retries — the second AWS-CLI call must not bail when QEMU reports the -// device is already gone (mulga-siv-41). -func isQMPDeviceNotFound(err error) bool { - if err == nil { - return false - } - s := err.Error() - return strings.Contains(s, "DeviceNotFound") -} - -// isQMPNodeInUse returns true when err is a QMP GenericError reporting a -// block node still in use. blockdev-del fails this way while NBD client -// teardown and queued I/O drain after device_del; the caller polls until -// the node is released or the budget expires. Match "in use" rather than -// "is in use" — QEMU phrases this as both "X is in use" and "X is still -// in use" depending on the path. -func isQMPNodeInUse(err error) bool { - if err == nil { - return false - } - return strings.Contains(err.Error(), "in use") -} diff --git a/spinifex/daemon/daemon_handlers_volume_test.go b/spinifex/daemon/daemon_handlers_volume_test.go new file mode 100644 index 000000000..a06a87439 --- /dev/null +++ b/spinifex/daemon/daemon_handlers_volume_test.go @@ -0,0 +1,88 @@ +package daemon + +import ( + "errors" + "fmt" + "testing" + + "github.com/mulgadc/spinifex/spinifex/awserrors" + "github.com/mulgadc/spinifex/spinifex/vm" + "github.com/stretchr/testify/assert" +) + +// TestAttachDetachErrorCode locks down the manager-error → AWS-API-code +// mapping that handleAttachVolume and handleDetachVolume both call. Wrong +// mapping silently breaks AWS-SDK retry semantics: clients expect 4xx +// codes for caller-fixable problems and 5xx for server faults. A future +// edit that drops a sentinel branch would otherwise pass the mechanical +// "tests still compile" bar with no signal. +func TestAttachDetachErrorCode(t *testing.T) { + tests := []struct { + name string + err error + want string + }{ + { + name: "ErrInstanceNotFound maps to InvalidInstanceID.NotFound", + err: vm.ErrInstanceNotFound, + want: awserrors.ErrorInvalidInstanceIDNotFound, + }, + { + name: "wrapped ErrInstanceNotFound still matches via errors.Is", + err: fmt.Errorf("manager: %w", vm.ErrInstanceNotFound), + want: awserrors.ErrorInvalidInstanceIDNotFound, + }, + { + name: "ErrInvalidTransition maps to IncorrectInstanceState", + err: vm.ErrInvalidTransition, + want: awserrors.ErrorIncorrectInstanceState, + }, + { + name: "wrapped ErrInvalidTransition still matches", + err: fmt.Errorf("cannot attach in state stopped: %w", vm.ErrInvalidTransition), + want: awserrors.ErrorIncorrectInstanceState, + }, + { + name: "ErrAttachmentLimitExceeded maps to AttachmentLimitExceeded", + err: vm.ErrAttachmentLimitExceeded, + want: awserrors.ErrorAttachmentLimitExceeded, + }, + { + name: "ErrVolumeNotAttached maps to IncorrectState", + err: vm.ErrVolumeNotAttached, + want: awserrors.ErrorIncorrectState, + }, + { + name: "wrapped ErrVolumeNotAttached still matches", + err: fmt.Errorf("%w: vol-1", vm.ErrVolumeNotAttached), + want: awserrors.ErrorIncorrectState, + }, + { + name: "ErrVolumeNotDetachable maps to OperationNotPermitted", + err: vm.ErrVolumeNotDetachable, + want: awserrors.ErrorOperationNotPermitted, + }, + { + name: "ErrVolumeDeviceMismatch maps to InvalidParameterValue", + err: vm.ErrVolumeDeviceMismatch, + want: awserrors.ErrorInvalidParameterValue, + }, + { + name: "unknown error falls through to ServerInternal", + err: errors.New("QMP blockdev-add: connection refused"), + want: awserrors.ErrorServerInternal, + }, + { + name: "wrapped unknown error falls through to ServerInternal", + err: fmt.Errorf("manager: %w", errors.New("nbdkit timeout")), + want: awserrors.ErrorServerInternal, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + got := attachDetachErrorCode(tt.err) + assert.Equal(t, tt.want, got) + }) + } +} diff --git a/spinifex/daemon/daemon_test.go b/spinifex/daemon/daemon_test.go index e01825b62..831f0a32d 100644 --- a/spinifex/daemon/daemon_test.go +++ b/spinifex/daemon/daemon_test.go @@ -104,6 +104,18 @@ func createTestDaemon(t *testing.T, natsURL string) *Daemon { daemon.instanceService = handlers_ec2_instance.NewInstanceServiceImpl(cfg, daemon.resourceMgr.instanceTypes, nc, objectstore.NewMemoryObjectStore()) daemon.volumeService = handlers_ec2_volume.NewVolumeServiceImplWithStore(cfg, objectstore.NewMemoryObjectStore(), nc) + // Wire the minimum vm.Deps that handler tests rely on. Lifecycle (Run/Start/ + // Stop/Terminate) tests still set up their own deps; this gives the + // post-2d AttachVolume / DetachVolume manager methods enough plumbing to + // drive ebs.mount/unmount over NATS using the test's connection. + volState := newVolumeStateUpdaterAdapter(daemon.volumeService) + daemon.vmMgr.SetDeps(vm.Deps{ + NodeID: daemon.node, + VolumeMounter: newVolumeMounterAdapter(daemon.natsConn, daemon.node, volState), + VolumeStateUpdater: volState, + DetachDelay: daemon.detachDelay, + }) + t.Cleanup(func() { if daemon.natsConn != nil { drainAndClose(t, daemon.natsConn) @@ -3136,16 +3148,18 @@ func TestMarkInstanceFailed_NilInstance(t *testing.T) { }, 5*time.Second, 10*time.Millisecond, "cleanup goroutine should reach terminated") } -// TestRollbackEBSMount_Success verifies that rollbackEBSMount sends an ebs.unmount -// request and handles a successful unmount response. -func TestRollbackEBSMount_Success(t *testing.T) { +// TestVolumeMounterAdapter_UnmountOne_Success verifies that the adapter's +// UnmountOne sends an ebs.unmount NATS request and handles a successful +// response. UnmountOne is the post-2d successor to Daemon.rollbackEBSMount; +// it is the AttachVolume rollback path and DetachVolume Phase 3. +func TestVolumeMounterAdapter_UnmountOne_Success(t *testing.T) { natsURL := sharedNATSURL daemon := createTestDaemon(t, natsURL) + adapter := newVolumeMounterAdapter(daemon.natsConn, daemon.node, nil) unmountCalled := make(chan string, 1) - // Mock ebs.unmount subscriber that returns success sub, err := daemon.natsConn.Subscribe("ebs.node-1.unmount", func(msg *nats.Msg) { var req types.EBSRequest json.Unmarshal(msg.Data, &req) @@ -3157,12 +3171,10 @@ func TestRollbackEBSMount_Success(t *testing.T) { require.NoError(t, err) defer sub.Unsubscribe() - ebsReq := types.EBSRequest{ + adapter.UnmountOne(types.EBSRequest{ Name: "vol-rollback-test", DeviceName: "/dev/sdf", - } - - daemon.rollbackEBSMount(ebsReq) + }) select { case volName := <-unmountCalled: @@ -3172,14 +3184,14 @@ func TestRollbackEBSMount_Success(t *testing.T) { } } -// TestRollbackEBSMount_UnmountError verifies that rollbackEBSMount handles -// an error response from ebs.unmount gracefully (no panic, just logs). -func TestRollbackEBSMount_UnmountError(t *testing.T) { +// TestVolumeMounterAdapter_UnmountOne_UnmountError verifies that UnmountOne +// tolerates an error response from ebs.unmount (logs only, no propagation). +func TestVolumeMounterAdapter_UnmountOne_UnmountError(t *testing.T) { natsURL := sharedNATSURL daemon := createTestDaemon(t, natsURL) + adapter := newVolumeMounterAdapter(daemon.natsConn, daemon.node, nil) - // Mock ebs.unmount subscriber that returns an error sub, err := daemon.natsConn.Subscribe("ebs.node-1.unmount", func(msg *nats.Msg) { resp := types.EBSUnMountResponse{Error: "unmount failed: device busy"} data, _ := json.Marshal(resp) @@ -3188,45 +3200,37 @@ func TestRollbackEBSMount_UnmountError(t *testing.T) { require.NoError(t, err) defer sub.Unsubscribe() - ebsReq := types.EBSRequest{Name: "vol-rollback-err"} - - // Should not panic — errors are logged but not propagated - daemon.rollbackEBSMount(ebsReq) + adapter.UnmountOne(types.EBSRequest{Name: "vol-rollback-err"}) } -// TestRollbackEBSMount_StillMounted verifies that rollbackEBSMount handles -// the case where the unmount response says the volume is still mounted. -func TestRollbackEBSMount_StillMounted(t *testing.T) { +// TestVolumeMounterAdapter_UnmountOne_StillMounted verifies that UnmountOne +// tolerates an unmount response that reports the volume still mounted. +func TestVolumeMounterAdapter_UnmountOne_StillMounted(t *testing.T) { natsURL := sharedNATSURL daemon := createTestDaemon(t, natsURL) + adapter := newVolumeMounterAdapter(daemon.natsConn, daemon.node, nil) sub, err := daemon.natsConn.Subscribe("ebs.node-1.unmount", func(msg *nats.Msg) { - resp := types.EBSUnMountResponse{Mounted: true} // still mounted + resp := types.EBSUnMountResponse{Mounted: true} data, _ := json.Marshal(resp) msg.Respond(data) }) require.NoError(t, err) defer sub.Unsubscribe() - ebsReq := types.EBSRequest{Name: "vol-still-mounted"} - - // Should not panic - daemon.rollbackEBSMount(ebsReq) + adapter.UnmountOne(types.EBSRequest{Name: "vol-still-mounted"}) } -// TestRollbackEBSMount_NATSTimeout verifies that rollbackEBSMount handles -// NATS request timeout gracefully (no subscriber on ebs.unmount). -func TestRollbackEBSMount_NATSTimeout(t *testing.T) { +// TestVolumeMounterAdapter_UnmountOne_NATSTimeout verifies that UnmountOne +// tolerates NATS request timeout (no subscriber on ebs.unmount). +func TestVolumeMounterAdapter_UnmountOne_NATSTimeout(t *testing.T) { natsURL := sharedNATSURL daemon := createTestDaemon(t, natsURL) + adapter := newVolumeMounterAdapter(daemon.natsConn, daemon.node, nil) - // No ebs.unmount subscriber — will timeout - ebsReq := types.EBSRequest{Name: "vol-timeout"} - - // Should not panic, just log the timeout - daemon.rollbackEBSMount(ebsReq) + adapter.UnmountOne(types.EBSRequest{Name: "vol-timeout"}) } // TestDescribeInstances_InvalidInstanceIDMalformed verifies that DescribeInstances @@ -3328,115 +3332,6 @@ func TestStopTerminate_IncorrectInstanceState(t *testing.T) { }) } -// TestNextAvailableDevice tests the device name auto-assignment logic -func TestNextAvailableDevice(t *testing.T) { - tests := []struct { - name string - instance *vm.VM - wantDevice string - }{ - { - name: "empty instance returns first device", - instance: &vm.VM{ - Instance: &ec2.Instance{}, - }, - wantDevice: "/dev/sdf", - }, - { - name: "nil Instance returns first device", - instance: &vm.VM{}, - wantDevice: "/dev/sdf", - }, - { - name: "existing BlockDeviceMappings skipped", - instance: &vm.VM{ - Instance: &ec2.Instance{ - BlockDeviceMappings: []*ec2.InstanceBlockDeviceMapping{ - {DeviceName: aws.String("/dev/sdf")}, - {DeviceName: aws.String("/dev/sdg")}, - }, - }, - }, - wantDevice: "/dev/sdh", - }, - { - name: "existing EBSRequests skipped", - instance: &vm.VM{ - Instance: &ec2.Instance{}, - EBSRequests: types.EBSRequests{ - Requests: []types.EBSRequest{ - {Name: "vol-1", DeviceName: "/dev/sdf"}, - }, - }, - }, - wantDevice: "/dev/sdg", - }, - { - name: "mixed sources all skipped", - instance: &vm.VM{ - Instance: &ec2.Instance{ - BlockDeviceMappings: []*ec2.InstanceBlockDeviceMapping{ - {DeviceName: aws.String("/dev/sdf")}, - }, - }, - EBSRequests: types.EBSRequests{ - Requests: []types.EBSRequest{ - {Name: "vol-1", DeviceName: "/dev/sdg"}, - }, - }, - }, - wantDevice: "/dev/sdh", - }, - { - name: "all devices f-p used returns empty", - instance: func() *vm.VM { - var bdms []*ec2.InstanceBlockDeviceMapping - for c := 'f'; c <= 'p'; c++ { - dev := fmt.Sprintf("/dev/sd%c", c) - bdms = append(bdms, &ec2.InstanceBlockDeviceMapping{ - DeviceName: aws.String(dev), - }) - } - return &vm.VM{ - Instance: &ec2.Instance{BlockDeviceMappings: bdms}, - } - }(), - wantDevice: "", - }, - { - name: "nil DeviceName in BlockDeviceMappings ignored", - instance: &vm.VM{ - Instance: &ec2.Instance{ - BlockDeviceMappings: []*ec2.InstanceBlockDeviceMapping{ - {DeviceName: nil}, - {DeviceName: aws.String("/dev/sdf")}, - }, - }, - }, - wantDevice: "/dev/sdg", - }, - { - name: "empty DeviceName in EBSRequests ignored", - instance: &vm.VM{ - EBSRequests: types.EBSRequests{ - Requests: []types.EBSRequest{ - {DeviceName: ""}, - {DeviceName: "/dev/sdf"}, - }, - }, - }, - wantDevice: "/dev/sdg", - }, - } - - for _, tt := range tests { - t.Run(tt.name, func(t *testing.T) { - got := nextAvailableDevice(tt.instance) - assert.Equal(t, tt.wantDevice, got) - }) - } -} - func TestCanAllocate(t *testing.T) { makeInstanceType := func(vCPUs int64, memoryMiB int64) *ec2.InstanceTypeInfo { return &ec2.InstanceTypeInfo{ diff --git a/spinifex/daemon/vm_adapters.go b/spinifex/daemon/vm_adapters.go index 1eca9aac5..e31a9c7cf 100644 --- a/spinifex/daemon/vm_adapters.go +++ b/spinifex/daemon/vm_adapters.go @@ -159,6 +159,68 @@ func (a *volumeMounterAdapter) Unmount(instance *vm.VM) error { return nil } +// MountOne sends ebs.mount for a single request and writes the resolved +// NBDURI back into req.NBDURI. Used by hot-attach (Manager.AttachVolume). +func (a *volumeMounterAdapter) MountOne(req *types.EBSRequest) error { + payload, err := json.Marshal(req) + if err != nil { + return fmt.Errorf("marshal ebs.mount request: %w", err) + } + + reply, err := a.nc.Request(a.topic("mount"), payload, 30*time.Second) + if err != nil { + return fmt.Errorf("ebs.mount NATS request: %w", err) + } + + var resp types.EBSMountResponse + if err := json.Unmarshal(reply.Data, &resp); err != nil { + return fmt.Errorf("unmarshal ebs.mount response: %w", err) + } + if resp.Error != "" { + return fmt.Errorf("ebs.mount returned error: %s", resp.Error) + } + if resp.URI == "" { + return vm.ErrMountAmbiguous + } + + req.NBDURI = resp.URI + return nil +} + +// UnmountOne sends ebs.unmount for a single request. Best-effort: errors +// are logged. Mirrors the pre-2d Daemon.rollbackEBSMount semantics so +// AttachVolume rollback and DetachVolume Phase 3 share one code path. +func (a *volumeMounterAdapter) UnmountOne(req types.EBSRequest) { + payload, err := json.Marshal(req) + if err != nil { + slog.Error("UnmountOne: failed to marshal unmount request", + "volume", req.Name, "err", err) + return + } + msg, err := a.nc.Request(a.topic("unmount"), payload, 10*time.Second) + if err != nil { + slog.Error("UnmountOne: ebs.unmount NATS request failed", + "volume", req.Name, "err", err) + return + } + var resp types.EBSUnMountResponse + if err := json.Unmarshal(msg.Data, &resp); err != nil { + slog.Error("UnmountOne: failed to unmarshal response", + "volume", req.Name, "err", err) + return + } + if resp.Error != "" { + slog.Error("UnmountOne: ebs.unmount returned error", + "volume", req.Name, "err", resp.Error) + return + } + if resp.Mounted { + slog.Error("UnmountOne: volume still mounted after unmount", "volume", req.Name) + return + } + slog.Info("UnmountOne: volume unmounted successfully", "volume", req.Name) +} + // qmpClientFactoryAdapter satisfies vm.QMPClientFactory. It performs only // the connect + qmp_capabilities handshake; the manager owns starting the // heartbeat goroutine on the returned client. @@ -409,6 +471,7 @@ func (d *Daemon) buildVMManagerDeps() vm.Deps { TransitionState: d.TransitionState, DevNetworking: d.config.Daemon.DevNetworking, BindHost: d.config.Host, + DetachDelay: d.detachDelay, } } diff --git a/spinifex/daemon/device_map.go b/spinifex/vm/device_map.go similarity index 77% rename from spinifex/daemon/device_map.go rename to spinifex/vm/device_map.go index 1ed97b9d6..9f19fec9d 100644 --- a/spinifex/daemon/device_map.go +++ b/spinifex/vm/device_map.go @@ -1,4 +1,4 @@ -package daemon +package vm import ( "encoding/json" @@ -11,7 +11,6 @@ import ( "time" "github.com/mulgadc/spinifex/spinifex/qmp" - "github.com/mulgadc/spinifex/spinifex/vm" ) // pciAddrRegexp extracts the device index from a boot-time QDev path. @@ -33,8 +32,8 @@ var hotplugPortRegexp = regexp.MustCompile(`hotplug(\d+)`) // The mapping is derived from PCI address order: virtio-blk-pci devices are // enumerated by the guest kernel in PCI bus order, which corresponds to the // device index in the QDev path. -func queryGuestDeviceMap(d *Daemon, qmpClient *qmp.QMPClient, instanceID string) (map[string]string, error) { - resp, err := d.SendQMPCommand(qmpClient, qmp.QMPCommand{Execute: "query-block"}, instanceID) +func queryGuestDeviceMap(qmpClient *qmp.QMPClient, instanceID string) (map[string]string, error) { + resp, err := sendQMPCommand(qmpClient, qmp.QMPCommand{Execute: "query-block"}, instanceID) if err != nil { return nil, fmt.Errorf("query-block failed: %w", err) } @@ -50,12 +49,12 @@ func queryGuestDeviceMap(d *Daemon, qmpClient *qmp.QMPClient, instanceID string) // queryGuestDeviceMapWait retries queryGuestDeviceMap until expectedDevice // appears in the result. This handles the race where query-block is called // immediately after device_add, before QEMU has registered the new device. -func queryGuestDeviceMapWait(d *Daemon, qmpClient *qmp.QMPClient, instanceID, expectedDevice string) (map[string]string, error) { +func queryGuestDeviceMapWait(qmpClient *qmp.QMPClient, instanceID, expectedDevice string) (map[string]string, error) { const maxAttempts = 5 const retryDelay = 200 * time.Millisecond for attempt := range maxAttempts { - deviceMap, err := queryGuestDeviceMap(d, qmpClient, instanceID) + deviceMap, err := queryGuestDeviceMap(qmpClient, instanceID) if err != nil { return nil, err } @@ -70,7 +69,7 @@ func queryGuestDeviceMapWait(d *Daemon, qmpClient *qmp.QMPClient, instanceID, ex } // Return the last result even though the device wasn't found — caller decides fallback. - return queryGuestDeviceMap(d, qmpClient, instanceID) + return queryGuestDeviceMap(qmpClient, instanceID) } // buildDeviceMap takes a list of BlockDevices from QMP and returns a map @@ -140,16 +139,18 @@ func buildDeviceMap(devices []qmp.BlockDevice) map[string]string { return result } -// updateGuestDeviceNames queries the running VM's QMP to discover actual guest device -// paths and updates the instance's BlockDeviceMappings accordingly. -func (d *Daemon) updateGuestDeviceNames(instance *vm.VM) { +// UpdateGuestDeviceNames queries the running VM's QMP to discover actual +// guest device paths and updates the instance's BlockDeviceMappings +// accordingly. Persists running state on success. No-op when QMPClient is +// nil. Errors are logged; callers do not need to react. +func (m *Manager) UpdateGuestDeviceNames(instance *VM) { if instance.QMPClient == nil { - slog.Warn("updateGuestDeviceNames: QMPClient is nil, cannot discover guest device names", + slog.Warn("UpdateGuestDeviceNames: QMPClient is nil, cannot discover guest device names", "instanceId", instance.ID) return } - deviceMap, err := queryGuestDeviceMap(d, instance.QMPClient, instance.ID) + deviceMap, err := queryGuestDeviceMap(instance.QMPClient, instance.ID) if err != nil { slog.Warn("Failed to query guest device map, BlockDeviceMappings will use API names", "instanceId", instance.ID, "err", err) @@ -157,8 +158,8 @@ func (d *Daemon) updateGuestDeviceNames(instance *vm.VM) { } // Build volume ID → guest device path mapping from EBSRequests. - // Collect under EBSRequests lock, then release before acquiring Instances lock - // to maintain consistent lock ordering. + // Collect under EBSRequests lock, then release before acquiring the + // manager lock to maintain consistent lock ordering. instance.EBSRequests.Mu.Lock() volToGuest := make(map[string]string, len(instance.EBSRequests.Requests)) for _, req := range instance.EBSRequests.Requests { @@ -176,8 +177,7 @@ func (d *Daemon) updateGuestDeviceNames(instance *vm.VM) { } instance.EBSRequests.Mu.Unlock() - // Update BlockDeviceMappings using the mapping - d.vmMgr.Inspect(instance, func(v *vm.VM) { + m.Inspect(instance, func(v *VM) { if v.Instance == nil { return } @@ -191,8 +191,9 @@ func (d *Daemon) updateGuestDeviceNames(instance *vm.VM) { } }) - if err := d.WriteState(); err != nil { - slog.Error("Failed to persist state after guest device name update", "instanceId", instance.ID, "err", err) + if err := m.writeRunningState(); err != nil { + slog.Error("Failed to persist state after guest device name update", + "instanceId", instance.ID, "err", err) } slog.Info("Updated guest device names", "instanceId", instance.ID, "deviceMap", deviceMap) @@ -215,7 +216,6 @@ func extractPCIIndex(qdev string) int { // extractPeripheralName extracts the device ID from a hot-plugged QDev path. // For example: "/machine/peripheral/vdisk-vol-abc123/virtio-backend" → "vdisk-vol-abc123" func extractPeripheralName(qdev string) string { - // Path format: /machine/peripheral//virtio-backend const prefix = "/machine/peripheral/" if !strings.HasPrefix(qdev, prefix) { return "" @@ -241,3 +241,35 @@ func extractHotplugPort(qdev string) int { } return idx } + +// nextAvailableDevice finds the next available /dev/sd[f-p] device name +// for instance. AWS convention reserves f-p for attached volumes. Returns +// an empty string when every slot is occupied. +func nextAvailableDevice(instance *VM) string { + usedDevices := make(map[string]bool) + + if instance.Instance != nil { + for _, bdm := range instance.Instance.BlockDeviceMappings { + if bdm.DeviceName != nil { + usedDevices[*bdm.DeviceName] = true + } + } + } + + instance.EBSRequests.Mu.Lock() + for _, req := range instance.EBSRequests.Requests { + if req.DeviceName != "" { + usedDevices[req.DeviceName] = true + } + } + instance.EBSRequests.Mu.Unlock() + + for c := 'f'; c <= 'p'; c++ { + dev := fmt.Sprintf("/dev/sd%c", c) + if !usedDevices[dev] { + return dev + } + } + + return "" +} diff --git a/spinifex/daemon/device_map_test.go b/spinifex/vm/device_map_test.go similarity index 99% rename from spinifex/daemon/device_map_test.go rename to spinifex/vm/device_map_test.go index daea929f7..0a1274709 100644 --- a/spinifex/daemon/device_map_test.go +++ b/spinifex/vm/device_map_test.go @@ -1,4 +1,4 @@ -package daemon +package vm import ( "testing" diff --git a/spinifex/vm/errors.go b/spinifex/vm/errors.go index f70dec724..0b6795c71 100644 --- a/spinifex/vm/errors.go +++ b/spinifex/vm/errors.go @@ -10,3 +10,26 @@ var ErrInstanceNotFound = errors.New("instance not found") // because the instance's current state does not permit the target // transition (e.g. Stop on an already-stopped instance). var ErrInvalidTransition = errors.New("invalid state transition") + +// ErrAttachmentLimitExceeded is returned by AttachVolume when the +// instance has no free /dev/sd[f-p] slot to assign. +var ErrAttachmentLimitExceeded = errors.New("attachment limit exceeded") + +// ErrVolumeNotAttached is returned by DetachVolume when the supplied +// volumeID is not present in the instance's EBSRequests. +var ErrVolumeNotAttached = errors.New("volume not attached to instance") + +// ErrVolumeNotDetachable is returned by DetachVolume when the target +// volume is a boot, EFI, or cloud-init volume that cannot be hot-unplugged. +var ErrVolumeNotDetachable = errors.New("volume cannot be detached") + +// ErrVolumeDeviceMismatch is returned by DetachVolume when the caller +// supplies a device name that does not match the recorded attachment. +var ErrVolumeDeviceMismatch = errors.New("volume device name mismatch") + +// ErrMountAmbiguous is returned by VolumeMounter.MountOne when the +// backend acknowledged the mount but returned an empty URI. The mount may +// or may not have started serving NBD; AttachVolume must invoke UnmountOne +// defensively so a half-started backend mount is not orphaned. Mirrors the +// pre-2d daemon-side rollback that triggered on this exact response shape. +var ErrMountAmbiguous = errors.New("ebs.mount succeeded with empty URI") diff --git a/spinifex/vm/lifecycle.go b/spinifex/vm/lifecycle.go index 1117ada77..b70a62bc1 100644 --- a/spinifex/vm/lifecycle.go +++ b/spinifex/vm/lifecycle.go @@ -38,6 +38,27 @@ func (m *Manager) Start(id string) error { return m.launch(instance) } +// Reboot issues a QMP system_reset to a running instance. The VM stays +// in StateRunning across the reset; QEMU re-runs firmware and the guest +// kernel reboots in place. Returns ErrInstanceNotFound when id is unknown +// and ErrInvalidTransition when the instance is not Running. +func (m *Manager) Reboot(id string) error { + instance, ok := m.Get(id) + if !ok { + return ErrInstanceNotFound + } + var status InstanceState + m.Inspect(instance, func(v *VM) { status = v.Status }) + if status != StateRunning { + return fmt.Errorf("%w: cannot reboot instance %s in state %s", + ErrInvalidTransition, id, status) + } + if _, err := sendQMPCommand(instance.QMPClient, qmp.QMPCommand{Execute: "system_reset"}, id); err != nil { + return fmt.Errorf("QMP system_reset: %w", err) + } + return nil +} + // launchStillValid returns true while the launch pipeline may continue // setting up resources for instance. Returns false if a concurrent terminate // has flipped status out of pending/stopped/provisioning — at that point the diff --git a/spinifex/vm/manager.go b/spinifex/vm/manager.go index 9df95b298..c415e993f 100644 --- a/spinifex/vm/manager.go +++ b/spinifex/vm/manager.go @@ -3,6 +3,7 @@ package vm import ( "maps" "sync" + "time" ) // ManagerHooks are callbacks the manager fires synchronously on every @@ -48,6 +49,11 @@ type Deps struct { // BindHost is the daemon's listen IP, used when wiring user-mode // hostfwd rules. Empty / "0.0.0.0" falls back to 127.0.0.1. BindHost string + + // DetachDelay is the pause between QMP device_del and blockdev-del + // during DetachVolume, and the polling interval for blockdev-del retry. + // Zero is acceptable in tests; production uses 1s. + DetachDelay time.Duration } // Manager owns the in-memory map of running VMs on this node and every diff --git a/spinifex/vm/manager_deps_test.go b/spinifex/vm/manager_deps_test.go index 98218b669..c593c0367 100644 --- a/spinifex/vm/manager_deps_test.go +++ b/spinifex/vm/manager_deps_test.go @@ -4,6 +4,8 @@ import ( "errors" "maps" "testing" + + "github.com/mulgadc/spinifex/spinifex/types" ) // fakeStateStore is a minimal in-memory StateStore used to verify Deps wiring. @@ -135,8 +137,11 @@ func TestManagerHooks_InitiallyNil(t *testing.T) { // fakeVolumeMounter records every call so lifecycle tests can assert ordering. type fakeVolumeMounter struct { - mounted, unmounted []string - mountErr error + mounted, unmounted []string + mountedOne, unmountedOne []string + mountErr error + mountOneErr error + mountOneURI string } func (f *fakeVolumeMounter) Mount(v *VM) error { @@ -149,6 +154,21 @@ func (f *fakeVolumeMounter) Unmount(v *VM) error { return nil } +func (f *fakeVolumeMounter) MountOne(req *types.EBSRequest) error { + f.mountedOne = append(f.mountedOne, req.Name) + if f.mountOneErr != nil { + return f.mountOneErr + } + if f.mountOneURI != "" { + req.NBDURI = f.mountOneURI + } + return nil +} + +func (f *fakeVolumeMounter) UnmountOne(req types.EBSRequest) { + f.unmountedOne = append(f.unmountedOne, req.Name) +} + var _ VolumeMounter = (*fakeVolumeMounter)(nil) func TestDeps_VolumeMounterSatisfiesInterface(t *testing.T) { diff --git a/spinifex/vm/volume_mounter.go b/spinifex/vm/volume_mounter.go index 0b200b7ef..e79809c2a 100644 --- a/spinifex/vm/volume_mounter.go +++ b/spinifex/vm/volume_mounter.go @@ -1,5 +1,7 @@ package vm +import "github.com/mulgadc/spinifex/spinifex/types" + // VolumeMounter mounts and unmounts the EBS volumes attached to a VM. The // real implementation routes ebs.mount / ebs.unmount NATS requests; the // abstraction keeps NATS out of the manager. @@ -10,4 +12,11 @@ type VolumeMounter interface { // Unmount sends ebs.unmount for each attached volume. Errors are logged // per volume and aggregated; partial failure is tolerated. Unmount(v *VM) error + + // MountOne sends ebs.mount for a single request and writes the resolved + // NBDURI back into req.NBDURI on success. Used by hot-attach. + MountOne(req *types.EBSRequest) error + // UnmountOne sends ebs.unmount for a single request. Best-effort: errors + // are logged and tolerated. + UnmountOne(req types.EBSRequest) } diff --git a/spinifex/vm/volumes.go b/spinifex/vm/volumes.go new file mode 100644 index 000000000..8d4948b07 --- /dev/null +++ b/spinifex/vm/volumes.go @@ -0,0 +1,416 @@ +package vm + +import ( + "errors" + "fmt" + "log/slog" + "strconv" + "strings" + "time" + + "github.com/aws/aws-sdk-go/service/ec2" + "github.com/mulgadc/spinifex/spinifex/qmp" + "github.com/mulgadc/spinifex/spinifex/types" + "github.com/mulgadc/spinifex/spinifex/utils" +) + +// blockdevDelMaxAttempts caps the bounded retry on "node is in use" while +// QEMU drains pending I/O after device_del. 20 × DetachDelay (default 1s) +// gives a 20s budget that matches the pre-2d daemon-side helper. +const blockdevDelMaxAttempts = 20 + +// AttachVolumeResult carries the device names produced by AttachVolume: +// the AWS-API name (/dev/sdf, possibly auto-allocated) and the discovered +// guest device path (/dev/vdb…). Daemon handlers use GuestDevice in the +// AWS API response; the discovered name eventually appears in +// DescribeInstances. +type AttachVolumeResult struct { + DeviceName string + GuestDevice string +} + +// AttachVolume hot-plugs a volume into a running instance via the +// three-phase QMP pipeline (mount → blockdev-add → device_add). On +// failure mid-pipeline, partial state is rolled back. The instance must +// be in StateRunning. If device is empty, the next free /dev/sd[f-p] +// slot is allocated. +// +// Daemon handlers retain volume-side validation (existence, ownership, +// availability, AZ) and emit the AWS API response; the manager owns +// every QMP and persistence side-effect. +func (m *Manager) AttachVolume(id, volumeID, device string) (AttachVolumeResult, error) { + instance, ok := m.Get(id) + if !ok { + return AttachVolumeResult{}, ErrInstanceNotFound + } + + var status InstanceState + m.Inspect(instance, func(v *VM) { status = v.Status }) + if status != StateRunning { + return AttachVolumeResult{}, fmt.Errorf("%w: cannot attach to instance %s in state %s", + ErrInvalidTransition, id, status) + } + + if device == "" { + m.Inspect(instance, func(v *VM) { device = nextAvailableDevice(v) }) + if device == "" { + return AttachVolumeResult{}, ErrAttachmentLimitExceeded + } + } + + ebsRequest := types.EBSRequest{ + Name: volumeID, + DeviceName: device, + } + + if m.deps.VolumeMounter == nil { + return AttachVolumeResult{}, fmt.Errorf("VolumeMounter not wired") + } + if err := m.deps.VolumeMounter.MountOne(&ebsRequest); err != nil { + slog.Error("AttachVolume: ebs.mount failed", "volumeId", volumeID, "err", err) + // Empty-URI response leaves backend NBD state ambiguous; unmount + // defensively to avoid orphaning a half-started mount. + if errors.Is(err, ErrMountAmbiguous) { + m.deps.VolumeMounter.UnmountOne(ebsRequest) + } + return AttachVolumeResult{}, fmt.Errorf("mount volume %s: %w", volumeID, err) + } + + serverType, socketPath, nbdHost, nbdPort, err := utils.ParseNBDURI(ebsRequest.NBDURI) + if err != nil { + slog.Error("AttachVolume: failed to parse NBDURI", "uri", ebsRequest.NBDURI, "err", err) + m.deps.VolumeMounter.UnmountOne(ebsRequest) + return AttachVolumeResult{}, fmt.Errorf("parse NBDURI: %w", err) + } + + var serverArg map[string]any + if serverType == "unix" { + serverArg = map[string]any{"type": "unix", "path": socketPath} + } else { + serverArg = map[string]any{"type": "inet", "host": nbdHost, "port": strconv.Itoa(nbdPort)} + } + + nodeName := fmt.Sprintf("nbd-%s", volumeID) + deviceID := fmt.Sprintf("vdisk-%s", volumeID) + iothreadID := fmt.Sprintf("ioth-%s", volumeID) + + if _, err := sendQMPCommand(instance.QMPClient, qmp.QMPCommand{ + Execute: "object-add", + Arguments: map[string]any{ + "qom-type": "iothread", + "id": iothreadID, + }, + }, instance.ID); err != nil { + slog.Error("AttachVolume: QMP object-add iothread failed", "volumeId", volumeID, "err", err) + m.deps.VolumeMounter.UnmountOne(ebsRequest) + return AttachVolumeResult{}, fmt.Errorf("QMP object-add iothread: %w", err) + } + + if _, err := sendQMPCommand(instance.QMPClient, qmp.QMPCommand{ + Execute: "blockdev-add", + Arguments: map[string]any{ + "node-name": nodeName, + "driver": "nbd", + "server": serverArg, + "export": "", + "read-only": false, + }, + }, instance.ID); err != nil { + slog.Error("AttachVolume: QMP blockdev-add failed", "volumeId", volumeID, "err", err) + m.deps.VolumeMounter.UnmountOne(ebsRequest) + return AttachVolumeResult{}, fmt.Errorf("QMP blockdev-add: %w", err) + } + + // /dev/sdf -> hotplug1, /dev/sdg -> hotplug2, etc. + hotplugBus := "" + if len(device) > 0 { + letter := device[len(device)-1] + if letter >= 'f' && letter <= 'p' { + hotplugBus = fmt.Sprintf("hotplug%d", letter-'f'+1) + } + } + + deviceAddArgs := map[string]any{ + "driver": "virtio-blk-pci", + "id": deviceID, + "drive": nodeName, + "iothread": iothreadID, + } + if hotplugBus != "" { + deviceAddArgs["bus"] = hotplugBus + } + + if _, err := sendQMPCommand(instance.QMPClient, qmp.QMPCommand{ + Execute: "device_add", + Arguments: deviceAddArgs, + }, instance.ID); err != nil { + slog.Error("AttachVolume: QMP device_add failed, rolling back blockdev", + "volumeId", volumeID, "err", err) + if _, delErr := sendQMPCommand(instance.QMPClient, qmp.QMPCommand{ + Execute: "blockdev-del", + Arguments: map[string]any{"node-name": nodeName}, + }, instance.ID); delErr != nil { + slog.Error("AttachVolume: rollback blockdev-del failed, skipping EBS unmount", + "volumeId", volumeID, "err", delErr) + } else { + m.deps.VolumeMounter.UnmountOne(ebsRequest) + } + return AttachVolumeResult{}, fmt.Errorf("QMP device_add: %w", err) + } + + // Discover guest device. query-block may not include the device + // immediately after device_add; queryGuestDeviceMapWait retries. + guestDevice := device // fallback to AWS API name + deviceMap, qmpErr := queryGuestDeviceMapWait(instance.QMPClient, instance.ID, deviceID) + if qmpErr != nil { + slog.Warn("AttachVolume: failed to query guest device map, using API device name", + "volumeId", volumeID, "err", qmpErr) + } else if gd, ok := deviceMap[deviceID]; ok { + guestDevice = gd + slog.Info("AttachVolume: discovered guest device", + "volumeId", volumeID, "qemuDevice", deviceID, "guestDevice", guestDevice) + } else { + slog.Error("AttachVolume: device not found in QMP device map after retries, using API device name", + "volumeId", volumeID, "qemuDevice", deviceID, "deviceMap", deviceMap) + } + + // Replace stale entry for volumeID (covers stop/start cycles that keep + // the original request) or append a new one. + instance.EBSRequests.Mu.Lock() + replaced := false + for i, req := range instance.EBSRequests.Requests { + if req.Name == volumeID { + instance.EBSRequests.Requests[i] = ebsRequest + replaced = true + break + } + } + if !replaced { + instance.EBSRequests.Requests = append(instance.EBSRequests.Requests, ebsRequest) + } + instance.EBSRequests.Mu.Unlock() + + m.Inspect(instance, func(v *VM) { + if v.Instance == nil { + return + } + now := time.Now() + mapping := &ec2.InstanceBlockDeviceMapping{} + mapping.SetDeviceName(guestDevice) + mapping.Ebs = &ec2.EbsInstanceBlockDevice{} + mapping.Ebs.SetVolumeId(volumeID) + mapping.Ebs.SetAttachTime(now) + mapping.Ebs.SetDeleteOnTermination(false) + mapping.Ebs.SetStatus("attached") + v.Instance.BlockDeviceMappings = append(v.Instance.BlockDeviceMappings, mapping) + }) + + if m.deps.VolumeStateUpdater != nil { + if err := m.deps.VolumeStateUpdater.UpdateVolumeState(volumeID, "in-use", instance.ID, guestDevice); err != nil { + slog.Error("AttachVolume: failed to update volume metadata", + "volumeId", volumeID, "err", err) + } + } + + if err := m.writeRunningState(); err != nil { + slog.Error("AttachVolume: failed to write state", "err", err) + } + + slog.Info("Volume attached successfully", + "volumeId", volumeID, "instanceId", instance.ID, + "apiDevice", device, "guestDevice", guestDevice) + + return AttachVolumeResult{DeviceName: device, GuestDevice: guestDevice}, nil +} + +// DetachVolume hot-unplugs a volume from a running instance via the +// three-phase QMP pipeline (device_del → blockdev-del → object-del → +// ebs.unmount). Returns the recorded AWS-API device name on success so +// the daemon handler can echo it in the response. +// +// device may be empty (no cross-check) or must equal the recorded +// DeviceName. force=true tolerates a failing device_del; blockdev-del +// is always required because tearing down the NBD server underneath a +// live block node would crash the VM. +func (m *Manager) DetachVolume(id, volumeID, device string, force bool) (string, error) { + instance, ok := m.Get(id) + if !ok { + return "", ErrInstanceNotFound + } + + var status InstanceState + m.Inspect(instance, func(v *VM) { status = v.Status }) + if status != StateRunning { + return "", fmt.Errorf("%w: cannot detach from instance %s in state %s", + ErrInvalidTransition, id, status) + } + + instance.EBSRequests.Mu.Lock() + var ebsReq types.EBSRequest + found := false + for _, req := range instance.EBSRequests.Requests { + if req.Name == volumeID { + ebsReq = req + found = true + break + } + } + instance.EBSRequests.Mu.Unlock() + + if !found { + return "", fmt.Errorf("%w: %s", ErrVolumeNotAttached, volumeID) + } + + if ebsReq.Boot || ebsReq.EFI || ebsReq.CloudInit { + return "", fmt.Errorf("%w: %s", ErrVolumeNotDetachable, volumeID) + } + + if device != "" && ebsReq.DeviceName != "" && device != ebsReq.DeviceName { + return "", fmt.Errorf("%w: requested %s, recorded %s", + ErrVolumeDeviceMismatch, device, ebsReq.DeviceName) + } + + deviceID := fmt.Sprintf("vdisk-%s", volumeID) + nodeName := fmt.Sprintf("nbd-%s", volumeID) + iothreadID := fmt.Sprintf("ioth-%s", volumeID) + + // Phase 1: device_del. Idempotent on DeviceNotFound so the second + // AWS-CLI retry can drive blockdev-del to completion when a prior + // detach left the guest device gone but the block node intact + // (mulga-siv-41). + _, err := sendQMPCommand(instance.QMPClient, qmp.QMPCommand{ + Execute: "device_del", + Arguments: map[string]any{"id": deviceID}, + }, instance.ID) + switch { + case err == nil: + case isQMPDeviceNotFound(err): + slog.Info("DetachVolume: guest device already removed (resuming detach)", + "volumeId", volumeID, "err", err) + case force: + slog.Warn("DetachVolume: QMP device_del failed (force=true, continuing)", + "volumeId", volumeID, "err", err) + default: + slog.Error("DetachVolume: QMP device_del failed", "volumeId", volumeID, "err", err) + return "", fmt.Errorf("QMP device_del: %w", err) + } + + if m.deps.DetachDelay > 0 { + time.Sleep(m.deps.DetachDelay) + } + + // Phase 2: blockdev-del with bounded retry on "node is in use". + if blockdevErr := m.tryBlockdevDel(instance, nodeName); blockdevErr != nil { + slog.Error("DetachVolume: QMP blockdev-del failed, leaving volume state intact", + "volumeId", volumeID, "err", blockdevErr) + return "", fmt.Errorf("QMP blockdev-del: %w", blockdevErr) + } + + // Phase 2b: object-del (best-effort). + if _, iothreadErr := sendQMPCommand(instance.QMPClient, qmp.QMPCommand{ + Execute: "object-del", + Arguments: map[string]any{"id": iothreadID}, + }, instance.ID); iothreadErr != nil { + slog.Warn("DetachVolume: QMP object-del iothread failed (non-fatal)", + "volumeId", volumeID, "err", iothreadErr) + } + + // Phase 3: ebs.unmount (best-effort). + if m.deps.VolumeMounter != nil { + m.deps.VolumeMounter.UnmountOne(ebsReq) + } + + // State cleanup. + instance.EBSRequests.Mu.Lock() + for i, req := range instance.EBSRequests.Requests { + if req.Name == volumeID { + instance.EBSRequests.Requests = append(instance.EBSRequests.Requests[:i], instance.EBSRequests.Requests[i+1:]...) + break + } + } + instance.EBSRequests.Mu.Unlock() + + m.Inspect(instance, func(v *VM) { + if v.Instance == nil { + return + } + filtered := make([]*ec2.InstanceBlockDeviceMapping, 0, len(v.Instance.BlockDeviceMappings)) + for _, bdm := range v.Instance.BlockDeviceMappings { + if bdm.Ebs != nil && bdm.Ebs.VolumeId != nil && *bdm.Ebs.VolumeId == volumeID { + continue + } + filtered = append(filtered, bdm) + } + v.Instance.BlockDeviceMappings = filtered + }) + + if m.deps.VolumeStateUpdater != nil { + if err := m.deps.VolumeStateUpdater.UpdateVolumeState(volumeID, "available", "", ""); err != nil { + slog.Error("DetachVolume: failed to update volume metadata", + "volumeId", volumeID, "err", err) + } + } + + if err := m.writeRunningState(); err != nil { + slog.Error("DetachVolume: failed to write state", "err", err) + } + + slog.Info("Volume detached successfully", "volumeId", volumeID, "instanceId", instance.ID) + return ebsReq.DeviceName, nil +} + +// tryBlockdevDel issues blockdev-del with bounded retry on "is in use" +// errors. After device_del, NBD client teardown and any in-flight guest +// I/O can briefly hold the block node; QEMU surfaces this as a +// GenericError carrying "Node is in use". Polling at DetachDelay +// gives the I/O drain enough time without a hardcoded long sleep. +func (m *Manager) tryBlockdevDel(instance *VM, nodeName string) error { + var lastErr error + for attempt := 1; attempt <= blockdevDelMaxAttempts; attempt++ { + _, err := sendQMPCommand(instance.QMPClient, qmp.QMPCommand{ + Execute: "blockdev-del", + Arguments: map[string]any{"node-name": nodeName}, + }, instance.ID) + if err == nil { + if attempt > 1 { + slog.Info("DetachVolume: blockdev-del succeeded after retry", + "nodeName", nodeName, "attempts", attempt) + } + return nil + } + lastErr = err + if !isQMPNodeInUse(err) { + return err + } + if attempt == blockdevDelMaxAttempts { + break + } + slog.Debug("DetachVolume: blockdev-del busy, retrying", + "nodeName", nodeName, "attempt", attempt, "err", err) + if m.deps.DetachDelay > 0 { + time.Sleep(m.deps.DetachDelay) + } + } + return lastErr +} + +// isQMPDeviceNotFound returns true when err is a QMP DeviceNotFound class +// error from device_del. Used to make device_del idempotent across detach +// retries (mulga-siv-41). +func isQMPDeviceNotFound(err error) bool { + if err == nil { + return false + } + return strings.Contains(err.Error(), "DeviceNotFound") +} + +// isQMPNodeInUse returns true when err is a QMP GenericError reporting a +// block node still in use. blockdev-del fails this way while NBD client +// teardown and queued I/O drain after device_del. QEMU phrases this as +// both "X is in use" and "X is still in use" — match "in use". +func isQMPNodeInUse(err error) bool { + if err == nil { + return false + } + return strings.Contains(err.Error(), "in use") +} diff --git a/spinifex/vm/volumes_test.go b/spinifex/vm/volumes_test.go new file mode 100644 index 000000000..5c9d96990 --- /dev/null +++ b/spinifex/vm/volumes_test.go @@ -0,0 +1,564 @@ +package vm + +import ( + "encoding/json" + "errors" + "fmt" + "net" + "sync" + "testing" + + "github.com/aws/aws-sdk-go/aws" + "github.com/aws/aws-sdk-go/service/ec2" + "github.com/mulgadc/spinifex/spinifex/qmp" + "github.com/mulgadc/spinifex/spinifex/types" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// newMockQMPClient creates a QMPClient backed by an in-memory pipe. The +// responder is invoked for every command and returns the JSON object the +// server should reply with (e.g. {"return": {}} or {"error": {...}}). A +// nil responder reply defaults to an empty success. +func newMockQMPClient(t *testing.T, responder func(qmp.QMPCommand) map[string]any) (*qmp.QMPClient, func()) { + t.Helper() + clientConn, serverConn := net.Pipe() + + client := &qmp.QMPClient{ + Conn: clientConn, + Decoder: json.NewDecoder(clientConn), + Encoder: json.NewEncoder(clientConn), + } + + done := make(chan struct{}) + go func() { + defer close(done) + dec := json.NewDecoder(serverConn) + enc := json.NewEncoder(serverConn) + for { + var cmd qmp.QMPCommand + if err := dec.Decode(&cmd); err != nil { + return + } + var resp map[string]any + if responder != nil { + resp = responder(cmd) + } + if resp == nil { + resp = map[string]any{"return": map[string]any{}} + } + if err := enc.Encode(resp); err != nil { + return + } + } + }() + + cancel := func() { + _ = clientConn.Close() + _ = serverConn.Close() + <-done + } + return client, cancel +} + +// qmpRecorder records the order of QMP commands for assertion in rollback +// tests. Safe for concurrent use because sendQMPCommand serializes via the +// QMPClient mutex, but we lock anyway for clarity. +type qmpRecorder struct { + mu sync.Mutex + commands []qmp.QMPCommand +} + +func (r *qmpRecorder) record(cmd qmp.QMPCommand) { + r.mu.Lock() + defer r.mu.Unlock() + r.commands = append(r.commands, cmd) +} + +func (r *qmpRecorder) executes() []string { + r.mu.Lock() + defer r.mu.Unlock() + out := make([]string, len(r.commands)) + for i, c := range r.commands { + out[i] = c.Execute + } + return out +} + +func TestNextAvailableDevice(t *testing.T) { + tests := []struct { + name string + instance *VM + wantDevice string + }{ + { + name: "empty instance returns first device", + instance: &VM{ + Instance: &ec2.Instance{}, + }, + wantDevice: "/dev/sdf", + }, + { + name: "nil Instance returns first device", + instance: &VM{}, + wantDevice: "/dev/sdf", + }, + { + name: "existing BlockDeviceMappings skipped", + instance: &VM{ + Instance: &ec2.Instance{ + BlockDeviceMappings: []*ec2.InstanceBlockDeviceMapping{ + {DeviceName: aws.String("/dev/sdf")}, + {DeviceName: aws.String("/dev/sdg")}, + }, + }, + }, + wantDevice: "/dev/sdh", + }, + { + name: "existing EBSRequests skipped", + instance: &VM{ + Instance: &ec2.Instance{}, + EBSRequests: types.EBSRequests{ + Requests: []types.EBSRequest{ + {Name: "vol-1", DeviceName: "/dev/sdf"}, + }, + }, + }, + wantDevice: "/dev/sdg", + }, + { + name: "mixed sources all skipped", + instance: &VM{ + Instance: &ec2.Instance{ + BlockDeviceMappings: []*ec2.InstanceBlockDeviceMapping{ + {DeviceName: aws.String("/dev/sdf")}, + }, + }, + EBSRequests: types.EBSRequests{ + Requests: []types.EBSRequest{ + {Name: "vol-1", DeviceName: "/dev/sdg"}, + }, + }, + }, + wantDevice: "/dev/sdh", + }, + { + name: "all devices f-p used returns empty", + instance: func() *VM { + var bdms []*ec2.InstanceBlockDeviceMapping + for c := 'f'; c <= 'p'; c++ { + dev := fmt.Sprintf("/dev/sd%c", c) + bdms = append(bdms, &ec2.InstanceBlockDeviceMapping{ + DeviceName: aws.String(dev), + }) + } + return &VM{ + Instance: &ec2.Instance{BlockDeviceMappings: bdms}, + } + }(), + wantDevice: "", + }, + { + name: "nil DeviceName in BlockDeviceMappings ignored", + instance: &VM{ + Instance: &ec2.Instance{ + BlockDeviceMappings: []*ec2.InstanceBlockDeviceMapping{ + {DeviceName: nil}, + {DeviceName: aws.String("/dev/sdf")}, + }, + }, + }, + wantDevice: "/dev/sdg", + }, + { + name: "empty DeviceName in EBSRequests ignored", + instance: &VM{ + EBSRequests: types.EBSRequests{ + Requests: []types.EBSRequest{ + {DeviceName: ""}, + {DeviceName: "/dev/sdf"}, + }, + }, + }, + wantDevice: "/dev/sdg", + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + got := nextAvailableDevice(tt.instance) + assert.Equal(t, tt.wantDevice, got) + }) + } +} + +func TestIsQMPDeviceNotFound(t *testing.T) { + tests := []struct { + name string + err error + want bool + }{ + {name: "nil", err: nil, want: false}, + {name: "matches DeviceNotFound", err: errors.New("QMP error: DeviceNotFound: device 'vdisk-x' not found"), want: true}, + {name: "other QMP error", err: errors.New("QMP error: GenericError: bad command"), want: false}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + assert.Equal(t, tt.want, isQMPDeviceNotFound(tt.err)) + }) + } +} + +func TestIsQMPNodeInUse(t *testing.T) { + tests := []struct { + name string + err error + want bool + }{ + {name: "nil", err: nil, want: false}, + {name: "matches 'is in use'", err: errors.New("QMP error: GenericError: Node 'nbd-x' is in use"), want: true}, + {name: "matches 'still in use'", err: errors.New("QMP error: GenericError: Node 'nbd-x' is still in use"), want: true}, + {name: "other QMP error", err: errors.New("QMP error: GenericError: not found"), want: false}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + assert.Equal(t, tt.want, isQMPNodeInUse(tt.err)) + }) + } +} + +func TestAttachVolume_InstanceNotFound(t *testing.T) { + m := NewManager() + _, err := m.AttachVolume("i-missing", "vol-1", "") + require.Error(t, err) + assert.ErrorIs(t, err, ErrInstanceNotFound) +} + +func TestAttachVolume_NotRunning(t *testing.T) { + m := NewManager() + m.Insert(&VM{ID: "i-1", Status: StateStopped, Instance: &ec2.Instance{}}) + + _, err := m.AttachVolume("i-1", "vol-1", "") + require.Error(t, err) + assert.ErrorIs(t, err, ErrInvalidTransition) +} + +func TestAttachVolume_AttachmentLimitExceeded(t *testing.T) { + bdms := make([]*ec2.InstanceBlockDeviceMapping, 0, 11) + for c := 'f'; c <= 'p'; c++ { + dev := fmt.Sprintf("/dev/sd%c", c) + bdms = append(bdms, &ec2.InstanceBlockDeviceMapping{DeviceName: aws.String(dev)}) + } + m := NewManager() + m.Insert(&VM{ID: "i-1", Status: StateRunning, Instance: &ec2.Instance{BlockDeviceMappings: bdms}}) + + _, err := m.AttachVolume("i-1", "vol-1", "") + require.Error(t, err) + assert.ErrorIs(t, err, ErrAttachmentLimitExceeded) +} + +func TestDetachVolume_InstanceNotFound(t *testing.T) { + m := NewManager() + _, err := m.DetachVolume("i-missing", "vol-1", "", false) + require.Error(t, err) + assert.ErrorIs(t, err, ErrInstanceNotFound) +} + +func TestDetachVolume_NotRunning(t *testing.T) { + m := NewManager() + m.Insert(&VM{ID: "i-1", Status: StateStopped}) + + _, err := m.DetachVolume("i-1", "vol-1", "", false) + require.Error(t, err) + assert.ErrorIs(t, err, ErrInvalidTransition) +} + +func TestDetachVolume_VolumeNotAttached(t *testing.T) { + m := NewManager() + m.Insert(&VM{ID: "i-1", Status: StateRunning, Instance: &ec2.Instance{}}) + + _, err := m.DetachVolume("i-1", "vol-missing", "", false) + require.Error(t, err) + assert.ErrorIs(t, err, ErrVolumeNotAttached) +} + +func TestDetachVolume_BootVolumeRejected(t *testing.T) { + m := NewManager() + m.Insert(&VM{ + ID: "i-1", + Status: StateRunning, + Instance: &ec2.Instance{}, + EBSRequests: types.EBSRequests{ + Requests: []types.EBSRequest{{Name: "vol-boot", Boot: true}}, + }, + }) + + _, err := m.DetachVolume("i-1", "vol-boot", "", false) + require.Error(t, err) + assert.ErrorIs(t, err, ErrVolumeNotDetachable) +} + +func TestDetachVolume_DeviceMismatch(t *testing.T) { + m := NewManager() + m.Insert(&VM{ + ID: "i-1", + Status: StateRunning, + Instance: &ec2.Instance{}, + EBSRequests: types.EBSRequests{ + Requests: []types.EBSRequest{{Name: "vol-1", DeviceName: "/dev/sdf"}}, + }, + }) + + _, err := m.DetachVolume("i-1", "vol-1", "/dev/sdg", false) + require.Error(t, err) + assert.ErrorIs(t, err, ErrVolumeDeviceMismatch) +} + +func TestReboot_InstanceNotFound(t *testing.T) { + m := NewManager() + err := m.Reboot("i-missing") + require.Error(t, err) + assert.ErrorIs(t, err, ErrInstanceNotFound) +} + +func TestReboot_NotRunning(t *testing.T) { + m := NewManager() + m.Insert(&VM{ID: "i-1", Status: StateStopped}) + + err := m.Reboot("i-1") + require.Error(t, err) + assert.ErrorIs(t, err, ErrInvalidTransition) +} + +// TestReboot_DoesNotFireHooks asserts that Manager.Reboot fires neither +// OnInstanceUp nor OnInstanceDown. The plan's "Hook firing contract" requires +// hooks to fire only on Pending→Running and terminal transitions; reboot keeps +// the VM in StateRunning across system_reset, so firing OnInstanceDown + +// OnInstanceUp would tear down per-instance NATS subs on every reboot. +func TestReboot_DoesNotFireHooks(t *testing.T) { + recorder := &qmpRecorder{} + qmpClient, cancel := newMockQMPClient(t, func(cmd qmp.QMPCommand) map[string]any { + recorder.record(cmd) + return nil // success + }) + defer cancel() + + var upCalls, downCalls int + hooks := ManagerHooks{ + OnInstanceUp: func(*VM) { upCalls++ }, + OnInstanceDown: func(string) { downCalls++ }, + } + m := NewManagerWithDeps(Deps{Hooks: hooks}) + m.Insert(&VM{ID: "i-1", Status: StateRunning, QMPClient: qmpClient}) + + require.NoError(t, m.Reboot("i-1")) + + assert.Equal(t, 0, upCalls, "Reboot must not fire OnInstanceUp") + assert.Equal(t, 0, downCalls, "Reboot must not fire OnInstanceDown") + assert.Equal(t, []string{"system_reset"}, recorder.executes(), + "Reboot must issue exactly one system_reset QMP command") +} + +// TestReboot_DoesNotChangeStatus asserts the VM stays in StateRunning across +// reboot. Pairs with TestReboot_DoesNotFireHooks to lock down the hook +// contract: status doesn't transition, so hooks shouldn't fire. +func TestReboot_DoesNotChangeStatus(t *testing.T) { + qmpClient, cancel := newMockQMPClient(t, nil) + defer cancel() + + m := NewManager() + m.Insert(&VM{ID: "i-1", Status: StateRunning, QMPClient: qmpClient}) + + require.NoError(t, m.Reboot("i-1")) + + v, ok := m.Get("i-1") + require.True(t, ok) + var status InstanceState + m.Inspect(v, func(v *VM) { status = v.Status }) + assert.Equal(t, StateRunning, status) +} + +func TestReboot_QMPFailureSurfacesError(t *testing.T) { + qmpClient, cancel := newMockQMPClient(t, func(cmd qmp.QMPCommand) map[string]any { + return map[string]any{ + "error": map[string]any{ + "class": "GenericError", + "desc": "guest is not running", + }, + } + }) + defer cancel() + + m := NewManager() + m.Insert(&VM{ID: "i-1", Status: StateRunning, QMPClient: qmpClient}) + + err := m.Reboot("i-1") + require.Error(t, err) + assert.Contains(t, err.Error(), "QMP system_reset") +} + +// attachVolumeRunningInstance returns a manager with a single running VM +// wired with the supplied QMP client. Used by AttachVolume rollback tests. +func attachVolumeRunningInstance(t *testing.T, qmpClient *qmp.QMPClient, mounter VolumeMounter) (*Manager, *VM) { + t.Helper() + m := NewManagerWithDeps(Deps{VolumeMounter: mounter}) + v := &VM{ + ID: "i-1", + Status: StateRunning, + Instance: &ec2.Instance{}, + QMPClient: qmpClient, + } + m.Insert(v) + return m, v +} + +// TestAttachVolume_MountAmbiguous_TriggersUnmountOne covers the regression +// from issue #1: pre-2d the daemon called rollbackEBSMount when the backend +// acknowledged the mount with an empty URI. 2d collapsed the empty-URI case +// into an opaque MountOne error and dropped the rollback. The fix introduces +// ErrMountAmbiguous so AttachVolume can detect this specific case and call +// UnmountOne defensively. +func TestAttachVolume_MountAmbiguous_TriggersUnmountOne(t *testing.T) { + mounter := &fakeVolumeMounter{mountOneErr: ErrMountAmbiguous} + m, _ := attachVolumeRunningInstance(t, nil, mounter) + + _, err := m.AttachVolume("i-1", "vol-1", "/dev/sdf") + require.Error(t, err) + assert.ErrorIs(t, err, ErrMountAmbiguous) + assert.Equal(t, []string{"vol-1"}, mounter.unmountedOne, + "AttachVolume must call UnmountOne to clean up the ambiguous backend mount") +} + +// TestAttachVolume_GenericMountError_DoesNotUnmount confirms that other +// MountOne failures (NATS timeout, backend explicit Error, marshal errors) +// do NOT trigger UnmountOne — matches pre-2d behaviour where the daemon +// only rolled back on the empty-URI case, not on mountResp.Error != "". +func TestAttachVolume_GenericMountError_DoesNotUnmount(t *testing.T) { + mounter := &fakeVolumeMounter{mountOneErr: errors.New("ebs.mount NATS request: timeout")} + m, _ := attachVolumeRunningInstance(t, nil, mounter) + + _, err := m.AttachVolume("i-1", "vol-1", "/dev/sdf") + require.Error(t, err) + assert.NotErrorIs(t, err, ErrMountAmbiguous) + assert.Empty(t, mounter.unmountedOne, + "non-ambiguous MountOne errors must not trigger UnmountOne (matches pre-2d)") +} + +// TestAttachVolume_NBDURIParseFailure_TriggersUnmountOne covers the rollback +// after a successful mount that returns a malformed NBDURI. AttachVolume must +// unmount the backend before bailing. +func TestAttachVolume_NBDURIParseFailure_TriggersUnmountOne(t *testing.T) { + mounter := &fakeVolumeMounter{mountOneURI: "not-a-valid-nbd-uri"} + m, _ := attachVolumeRunningInstance(t, nil, mounter) + + _, err := m.AttachVolume("i-1", "vol-1", "/dev/sdf") + require.Error(t, err) + assert.Contains(t, err.Error(), "parse NBDURI") + assert.Equal(t, []string{"vol-1"}, mounter.unmountedOne) +} + +// TestAttachVolume_ObjectAddFailure_TriggersUnmountOne covers QMP iothread +// object-add failure: AttachVolume must unmount and bail before issuing +// blockdev-add (no further QMP traffic past the failed object-add). +func TestAttachVolume_ObjectAddFailure_TriggersUnmountOne(t *testing.T) { + recorder := &qmpRecorder{} + qmpClient, cancel := newMockQMPClient(t, func(cmd qmp.QMPCommand) map[string]any { + recorder.record(cmd) + if cmd.Execute == "object-add" { + return map[string]any{"error": map[string]any{"class": "GenericError", "desc": "iothread limit"}} + } + return nil + }) + defer cancel() + + mounter := &fakeVolumeMounter{mountOneURI: "nbd:unix:/tmp/test.sock"} + m, _ := attachVolumeRunningInstance(t, qmpClient, mounter) + + _, err := m.AttachVolume("i-1", "vol-1", "/dev/sdf") + require.Error(t, err) + assert.Contains(t, err.Error(), "object-add") + 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) +} + +// TestAttachVolume_BlockdevAddFailure_TriggersUnmountOne covers QMP +// blockdev-add failure: AttachVolume must unmount and bail before issuing +// device_add. +func TestAttachVolume_BlockdevAddFailure_TriggersUnmountOne(t *testing.T) { + recorder := &qmpRecorder{} + qmpClient, cancel := newMockQMPClient(t, func(cmd qmp.QMPCommand) map[string]any { + recorder.record(cmd) + if cmd.Execute == "blockdev-add" { + return map[string]any{"error": map[string]any{"class": "GenericError", "desc": "nbd connect failed"}} + } + return nil + }) + defer cancel() + + mounter := &fakeVolumeMounter{mountOneURI: "nbd:unix:/tmp/test.sock"} + m, _ := attachVolumeRunningInstance(t, qmpClient, mounter) + + _, err := m.AttachVolume("i-1", "vol-1", "/dev/sdf") + require.Error(t, err) + assert.Contains(t, err.Error(), "blockdev-add") + assert.Equal(t, []string{"object-add", "blockdev-add"}, recorder.executes()) + assert.Equal(t, []string{"vol-1"}, mounter.unmountedOne) +} + +// TestAttachVolume_DeviceAddFailure_BlockdevDelOK_Unmounts covers the +// device_add-fail-then-rollback path where blockdev-del succeeds. The +// rollback chain must run blockdev-del then UnmountOne in order. +func TestAttachVolume_DeviceAddFailure_BlockdevDelOK_Unmounts(t *testing.T) { + recorder := &qmpRecorder{} + qmpClient, cancel := newMockQMPClient(t, func(cmd qmp.QMPCommand) map[string]any { + recorder.record(cmd) + if cmd.Execute == "device_add" { + return map[string]any{"error": map[string]any{"class": "GenericError", "desc": "no PCI slot"}} + } + return nil + }) + defer cancel() + + mounter := &fakeVolumeMounter{mountOneURI: "nbd:unix:/tmp/test.sock"} + m, _ := attachVolumeRunningInstance(t, qmpClient, mounter) + + _, err := m.AttachVolume("i-1", "vol-1", "/dev/sdf") + require.Error(t, err) + assert.Contains(t, err.Error(), "device_add") + assert.Equal(t, []string{"object-add", "blockdev-add", "device_add", "blockdev-del"}, recorder.executes()) + assert.Equal(t, []string{"vol-1"}, mounter.unmountedOne, + "successful blockdev-del rollback must be followed by UnmountOne") +} + +// TestAttachVolume_DeviceAddFailure_BlockdevDelFails_SkipsUnmountOne +// preserves the load-bearing invariant in the rollback chain: when +// blockdev-del fails, AttachVolume must NOT call UnmountOne. Tearing down +// the NBD server while QEMU still holds the block node would crash the VM. +// A future refactor that "fixes" the apparent bug by always unmounting +// would crash production VMs — this test is the regression guard. +func TestAttachVolume_DeviceAddFailure_BlockdevDelFails_SkipsUnmountOne(t *testing.T) { + recorder := &qmpRecorder{} + qmpClient, cancel := newMockQMPClient(t, func(cmd qmp.QMPCommand) map[string]any { + recorder.record(cmd) + switch cmd.Execute { + case "device_add": + return map[string]any{"error": map[string]any{"class": "GenericError", "desc": "no PCI slot"}} + case "blockdev-del": + return map[string]any{"error": map[string]any{"class": "GenericError", "desc": "node still in use"}} + } + return nil + }) + defer cancel() + + mounter := &fakeVolumeMounter{mountOneURI: "nbd:unix:/tmp/test.sock"} + m, _ := attachVolumeRunningInstance(t, qmpClient, mounter) + + _, err := m.AttachVolume("i-1", "vol-1", "/dev/sdf") + require.Error(t, err) + assert.Contains(t, err.Error(), "device_add") + 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") +}