diff --git a/spinifex/daemon/daemon.go b/spinifex/daemon/daemon.go index bbc039ba6..4a9957325 100644 --- a/spinifex/daemon/daemon.go +++ b/spinifex/daemon/daemon.go @@ -49,7 +49,6 @@ import ( handlers_elbv2 "github.com/mulgadc/spinifex/spinifex/handlers/elbv2" "github.com/mulgadc/spinifex/spinifex/instancetypes" "github.com/mulgadc/spinifex/spinifex/objectstore" - "github.com/mulgadc/spinifex/spinifex/qmp" "github.com/mulgadc/spinifex/spinifex/tags" "github.com/mulgadc/spinifex/spinifex/types" "github.com/mulgadc/spinifex/spinifex/utils" @@ -149,14 +148,19 @@ type Daemon struct { // JetStream manager for KV state storage (nil if JetStream disabled) jsManager *JetStreamManager + // stateStore is the vm.StateStore-shaped view over jsManager. Both the + // vm.Manager and daemon-side handlers route VM-instance state I/O + // through it. Initialized after initJetStream succeeds. + stateStore vm.StateStore + // Delay after QMP device_del before blockdev-del (default 1s, 0 in tests) detachDelay time.Duration // NATS connect retry options (nil uses defaults: 5min max, 500ms initial delay) natsRetryOpts []utils.RetryOption - // NetworkPlumber handles tap device lifecycle for VPC networking - networkPlumber NetworkPlumber + // networkPlumber handles tap device lifecycle for VPC networking. + networkPlumber vm.NetworkPlumber // Management NIC infrastructure: bridge IP + IP allocator for system instances. // Populated at startup when br-mgmt is detected; nil/empty otherwise. @@ -897,7 +901,7 @@ func (d *Daemon) Start() error { d.waitForClusterReady() d.upgradeJetStreamReplicas() - d.restoreInstances() + d.vmMgr.Restore() // Rebuild mgmt IP allocator from restored VMs so we don't re-allocate IPs // that are already in use by running system instances. @@ -982,6 +986,8 @@ func (d *Daemon) initJetStream() error { retryDelay = min(retryDelay*2, 10*time.Second) } + d.stateStore = newStateStoreAdapter(d.jsManager) + // Replica upgrade is deferred to after all services have created their // KV buckets and the cluster is ready (see upgradeJetStreamReplicas). @@ -1094,19 +1100,6 @@ func (d *Daemon) checkPredastoreReady() bool { return true } -// restoreInstances delegates to vm.Manager.Restore. Daemon-side shim -// preserved so existing tests keep their bare-method call shape; phase 2f -// migrates them into vm/ and removes the wrapper. -func (d *Daemon) restoreInstances() { - d.vmMgr.Restore() -} - -// areVolumeSocketsValid is a daemon-side shim around vm.AreVolumeSocketsValid -// kept so the existing TestAreVolumeSocketsValid_* cases call the same name. -func (d *Daemon) areVolumeSocketsValid(instance *vm.VM) bool { - return vm.AreVolumeSocketsValid(instance) -} - // awaitShutdown blocks until the daemon's shutdown wait group completes. func (d *Daemon) awaitShutdown() { done := make(chan struct{}) @@ -1201,7 +1194,7 @@ func (d *Daemon) ClusterManager() error { } }) - // Load TLS certificate (C-5: serve over HTTPS instead of plaintext HTTP) + // Load TLS certificate. // Resolve relative cert paths against config directory (cert lives alongside spinifex.toml). // For binary installs, systemd sets absolute paths via env vars; for dev, the config // stores relative paths like "config/server.pem" which need resolution. @@ -1270,77 +1263,6 @@ func (d *Daemon) WriteState() error { return nil } -// LoadState loads the instance state from JetStream KV store (required). -// Replaces the manager's running set with the loaded map. -func (d *Daemon) LoadState() error { - if d.jsManager == nil { - return fmt.Errorf("JetStream manager not initialized - cannot load state") - } - - loaded, err := d.jsManager.LoadState(d.node) - if err != nil { - slog.Error("JetStream load failed", "error", err) - return fmt.Errorf("failed to load state from JetStream: %w", err) - } - - d.vmMgr.Replace(loaded) - return nil -} - -func (d *Daemon) SendQMPCommand(q *qmp.QMPClient, cmd qmp.QMPCommand, instanceId string) (*qmp.QMPResponse, error) { - // Confirm QMP client is initialized - if q == nil || q.Encoder == nil || q.Decoder == nil { - return nil, fmt.Errorf("QMP client is not initialized") - } - - // Lock the QMP client - q.Mu.Lock() - defer q.Mu.Unlock() - - // Set a read deadline so we don't block forever on a hung QEMU process - if err := q.Conn.SetReadDeadline(time.Now().Add(30 * time.Second)); err != nil { - return nil, fmt.Errorf("set read deadline: %w", err) - } - defer func() { _ = q.Conn.SetReadDeadline(time.Time{}) }() // clear deadline after command - - if err := q.Encoder.Encode(cmd); err != nil { - return nil, fmt.Errorf("encode error: %w", err) - } - - for { - var msg map[string]any - if err := q.Decoder.Decode(&msg); err != nil { - return nil, fmt.Errorf("decode error: %w", err) - } - - if _, ok := msg["event"]; ok { - // QMP events are informational only — state transitions are driven - // by the command handlers that initiate the action, avoiding races - // between event-driven and command-driven transitions. - slog.Info("QMP event", "event", msg["event"], "instanceId", instanceId) - // Extend deadline after receiving an event (QEMU is alive, just chatty) - if err := q.Conn.SetReadDeadline(time.Now().Add(30 * time.Second)); err != nil { - return nil, fmt.Errorf("set read deadline: %w", err) - } - continue - } - if errObj, ok := msg["error"].(map[string]any); ok { - return nil, fmt.Errorf("QMP error: %s: %s", errObj["class"], errObj["desc"]) - } - if _, ok := msg["return"]; ok { - respBytes, err := json.Marshal(msg) - if err != nil { - return nil, fmt.Errorf("marshal QMP response: %w", err) - } - var resp qmp.QMPResponse - if err := json.Unmarshal(respBytes, &resp); err != nil { - return nil, fmt.Errorf("unmarshal error: %w", err) - } - return &resp, nil - } - } -} - // setupReload registers a SIGHUP handler that reloads GPU config without restarting. func (d *Daemon) setupReload() { d.shutdownWg.Go(func() { diff --git a/spinifex/daemon/daemon_handlers_instance.go b/spinifex/daemon/daemon_handlers_instance.go index b553eee32..37168a487 100644 --- a/spinifex/daemon/daemon_handlers_instance.go +++ b/spinifex/daemon/daemon_handlers_instance.go @@ -72,8 +72,9 @@ func (d *Daemon) handleEC2RunInstances(msg *nats.Msg) { respondWithError(msg, awserrors.ErrorInvalidAMIIDNotFound) return } - // Verify the caller can use this AMI: must own it or it must be a system/pre-phase4 AMI. - // System AMIs have non-account-ID owner aliases (e.g. "self", "spinifex", empty). + // Verify the caller can use this AMI: must own it or the AMI must + // have a non-account-ID owner alias (e.g. "self", "spinifex", empty) + // indicating a system/legacy AMI. amiOwner := amiMeta.ImageOwnerAlias if amiOwner != "" && amiOwner != accountID { if utils.IsAccountID(amiOwner) { @@ -690,7 +691,8 @@ func (d *Daemon) handleEC2DescribeInstances(msg *nats.Msg) { d.vmMgr.View(func(vms map[string]*vm.VM) { for _, instance := range vms { // Skip instances not owned by the caller's account. - // Pre-Phase4 instances (empty AccountID) are only visible to root. + // Instances with an empty AccountID (legacy/migration data) + // are only visible to root. if !isInstanceVisible(accountID, instance.AccountID) { continue } @@ -837,14 +839,14 @@ func (d *Daemon) handleEC2StartStoppedInstance(msg *nats.Msg) { return } - if d.jsManager == nil { - slog.Error("handleEC2StartStoppedInstance: JetStream not available") + if d.stateStore == nil { + slog.Error("handleEC2StartStoppedInstance: state store not available") respondWithError(msg, awserrors.ErrorServerInternal) return } // Load instance from shared KV - instance, err := d.jsManager.LoadStoppedInstance(req.InstanceID) + instance, err := d.stateStore.LoadStoppedInstance(req.InstanceID) if err != nil { slog.Error("handleEC2StartStoppedInstance: failed to load stopped instance", "instanceId", req.InstanceID, "err", err) respondWithError(msg, awserrors.ErrorServerInternal) @@ -927,10 +929,10 @@ func (d *Daemon) handleEC2StartStoppedInstance(msg *nats.Msg) { // Remove from shared KV now that it's running locally. // Retry once on failure — a stale KV entry risks duplicate starts. - if err := d.jsManager.DeleteStoppedInstance(req.InstanceID); err != nil { + if err := d.stateStore.DeleteStoppedInstance(req.InstanceID); err != nil { slog.Warn("handleEC2StartStoppedInstance: first KV delete failed, retrying", "instanceId", req.InstanceID, "err", err) - if retryErr := d.jsManager.DeleteStoppedInstance(req.InstanceID); retryErr != nil { + if retryErr := d.stateStore.DeleteStoppedInstance(req.InstanceID); retryErr != nil { slog.Error("handleEC2StartStoppedInstance: KV delete failed after retry, instance is running locally but stale entry remains in shared KV", "instanceId", req.InstanceID, "err", retryErr) } @@ -964,14 +966,14 @@ func (d *Daemon) handleEC2TerminateStoppedInstance(msg *nats.Msg) { return } - if d.jsManager == nil { - slog.Error("handleEC2TerminateStoppedInstance: JetStream not available") + if d.stateStore == nil { + slog.Error("handleEC2TerminateStoppedInstance: state store not available") respondWithError(msg, awserrors.ErrorServerInternal) return } // Load instance from shared KV - instance, err := d.jsManager.LoadStoppedInstance(req.InstanceID) + instance, err := d.stateStore.LoadStoppedInstance(req.InstanceID) if err != nil { slog.Error("handleEC2TerminateStoppedInstance: failed to load stopped instance", "instanceId", req.InstanceID, "err", err) respondWithError(msg, awserrors.ErrorServerInternal) @@ -1066,7 +1068,7 @@ func (d *Daemon) handleEC2TerminateStoppedInstance(msg *nats.Msg) { // Write to terminated KV bucket FIRST so the instance is visible in DescribeInstances. // If this fails, the instance remains in the stopped bucket (safe to retry). instance.Status = vm.StateTerminated - if err := d.jsManager.WriteTerminatedInstance(req.InstanceID, instance); err != nil { + if err := d.stateStore.WriteTerminatedInstance(req.InstanceID, instance); err != nil { slog.Error("handleEC2TerminateStoppedInstance: failed to write to terminated KV, aborting", "instanceId", req.InstanceID, "err", err) respondWithError(msg, awserrors.ErrorServerInternal) return @@ -1074,10 +1076,10 @@ func (d *Daemon) handleEC2TerminateStoppedInstance(msg *nats.Msg) { // Now safe to remove from shared stopped KV — instance already exists in terminated bucket. // Retry once on failure to avoid duplicate entries in DescribeInstances. - if err := d.jsManager.DeleteStoppedInstance(req.InstanceID); err != nil { + if err := d.stateStore.DeleteStoppedInstance(req.InstanceID); err != nil { slog.Warn("handleEC2TerminateStoppedInstance: first stopped KV delete failed, retrying", "instanceId", req.InstanceID, "err", err) - if retryErr := d.jsManager.DeleteStoppedInstance(req.InstanceID); retryErr != nil { + if retryErr := d.stateStore.DeleteStoppedInstance(req.InstanceID); retryErr != nil { slog.Error("handleEC2TerminateStoppedInstance: stopped KV delete failed after retry, instance may appear in both buckets", "instanceId", req.InstanceID, "err", retryErr) } @@ -1092,22 +1094,25 @@ func (d *Daemon) handleEC2TerminateStoppedInstance(msg *nats.Msg) { // handleEC2DescribeStoppedInstances returns stopped instances from shared KV. func (d *Daemon) handleEC2DescribeStoppedInstances(msg *nats.Msg) { - d.describeInstancesFromKV(msg, d.jsManager.ListStoppedInstances, 80, "stopped", "handleEC2DescribeStoppedInstances") + if d.stateStore == nil { + respondWithError(msg, awserrors.ErrorServerInternal) + return + } + d.describeInstancesFromKV(msg, d.stateStore.ListStoppedInstances, 80, "stopped", "handleEC2DescribeStoppedInstances") } // handleEC2DescribeTerminatedInstances returns terminated instances from the terminated KV bucket. func (d *Daemon) handleEC2DescribeTerminatedInstances(msg *nats.Msg) { - d.describeInstancesFromKV(msg, d.jsManager.ListTerminatedInstances, 48, "terminated", "handleEC2DescribeTerminatedInstances") + if d.stateStore == nil { + respondWithError(msg, awserrors.ErrorServerInternal) + return + } + d.describeInstancesFromKV(msg, d.stateStore.ListTerminatedInstances, 48, "terminated", "handleEC2DescribeTerminatedInstances") } // describeInstancesFromKV is a shared helper for DescribeStopped/TerminatedInstances handlers. // It lists instances from a KV bucket, filters by account/instance ID, and responds with reservations. func (d *Daemon) describeInstancesFromKV(msg *nats.Msg, listFn func() ([]*vm.VM, error), fallbackCode int64, fallbackName, handlerName string) { - if d.jsManager == nil { - respondWithError(msg, awserrors.ErrorServerInternal) - return - } - accountID := utils.AccountIDFromMsg(msg) describeInput := &ec2.DescribeInstancesInput{} @@ -1227,13 +1232,13 @@ func (d *Daemon) handleEC2ModifyInstanceAttribute(msg *nats.Msg) { return } - if d.jsManager == nil { - slog.Error("handleEC2ModifyInstanceAttribute: JetStream not available") + if d.stateStore == nil { + slog.Error("handleEC2ModifyInstanceAttribute: state store not available") respondWithError(msg, awserrors.ErrorServerInternal) return } - instance, err := d.jsManager.LoadStoppedInstance(instanceID) + instance, err := d.stateStore.LoadStoppedInstance(instanceID) if err != nil { slog.Error("handleEC2ModifyInstanceAttribute: failed to load stopped instance", "instanceId", instanceID, "err", err) respondWithError(msg, awserrors.ErrorServerInternal) @@ -1289,7 +1294,7 @@ func (d *Daemon) handleEC2ModifyInstanceAttribute(msg *nats.Msg) { } } - if err := d.jsManager.WriteStoppedInstance(instanceID, instance); err != nil { + if err := d.stateStore.WriteStoppedInstance(instanceID, instance); err != nil { slog.Error("handleEC2ModifyInstanceAttribute: failed to write modified instance to KV", "instanceId", instanceID, "err", err) respondWithError(msg, awserrors.ErrorServerInternal) @@ -1335,12 +1340,12 @@ func (d *Daemon) handleEC2DescribeInstanceAttribute(msg *nats.Msg) { } if instance == nil { - if d.jsManager == nil { - slog.Error("handleEC2DescribeInstanceAttribute: JetStream not available") + if d.stateStore == nil { + slog.Error("handleEC2DescribeInstanceAttribute: state store not available") respondWithError(msg, awserrors.ErrorServerInternal) return } - stopped, err := d.jsManager.LoadStoppedInstance(instanceID) + stopped, err := d.stateStore.LoadStoppedInstance(instanceID) if err != nil { slog.Error("handleEC2DescribeInstanceAttribute: failed to load stopped instance", "instanceId", instanceID, "err", err) diff --git a/spinifex/daemon/daemon_handlers_test.go b/spinifex/daemon/daemon_handlers_test.go index 898055cbd..952d8e5e3 100644 --- a/spinifex/daemon/daemon_handlers_test.go +++ b/spinifex/daemon/daemon_handlers_test.go @@ -86,6 +86,7 @@ func createFullTestDaemonWithJetStream(t *testing.T, natsURL string) *Daemon { require.NoError(t, err) err = daemon.jsManager.InitTerminatedInstanceBucket() require.NoError(t, err) + daemon.stateStore = newStateStoreAdapter(daemon.jsManager) return daemon } @@ -660,7 +661,7 @@ func TestHandleEC2Events_RebootRunningInstance(t *testing.T) { require.NoError(t, err) require.NotNil(t, reply) - // With nil QMPClient encoder/decoder, SendQMPCommand returns error, + // With nil QMPClient encoder/decoder, the QMP send returns error, // so we expect an error response (ServerInternal). var errResp map[string]any err = json.Unmarshal(reply.Data, &errResp) diff --git a/spinifex/daemon/daemon_handlers_volume.go b/spinifex/daemon/daemon_handlers_volume.go index 2ca7911b1..dbbecda41 100644 --- a/spinifex/daemon/daemon_handlers_volume.go +++ b/spinifex/daemon/daemon_handlers_volume.go @@ -31,7 +31,7 @@ func (d *Daemon) handleAttachVolume(msg *nats.Msg, command types.EC2InstanceComm volumeID := command.AttachVolumeData.VolumeID // Surface IncorrectInstanceState before any volume-side lookups so the - // AWS-SDK error matches the pre-2d ordering. + // caller sees the instance-state error first when both apply. status := d.vmMgr.Status(instance) if status != vm.StateRunning { slog.Error("AttachVolume: instance not running", diff --git a/spinifex/daemon/daemon_system_instance.go b/spinifex/daemon/daemon_system_instance.go index d796c91f0..6a8a6b779 100644 --- a/spinifex/daemon/daemon_system_instance.go +++ b/spinifex/daemon/daemon_system_instance.go @@ -243,7 +243,7 @@ func (d *Daemon) LaunchSystemInstance(input *handlers_elbv2.SystemInstanceInput) slog.Warn("LaunchSystemInstance: failed to write state", "instanceId", instance.ID, "err", err) } - // Subscribe to per-instance NATS topic for terminate commands + // Subscribe to per-instance NATS topic for terminate commands. d.mu.Lock() sub, subErr := d.natsConn.Subscribe(fmt.Sprintf("ec2.cmd.%s", instance.ID), d.handleEC2Events) if subErr != nil { diff --git a/spinifex/daemon/daemon_test.go b/spinifex/daemon/daemon_test.go index b2795e487..303bfd0fb 100644 --- a/spinifex/daemon/daemon_test.go +++ b/spinifex/daemon/daemon_test.go @@ -107,13 +107,12 @@ func createTestDaemon(t *testing.T, natsURL string) *Daemon { // 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) + // AttachVolume / DetachVolume manager methods enough plumbing to drive + // ebs.mount/unmount over NATS using the test's connection. daemon.vmMgr.SetDeps(vm.Deps{ NodeID: daemon.node, - VolumeMounter: newVolumeMounterAdapter(daemon.natsConn, daemon.node, volState), - VolumeStateUpdater: volState, + VolumeMounter: newVolumeMounterAdapter(daemon.natsConn, daemon.node, daemon.volumeService), + VolumeStateUpdater: daemon.volumeService, DetachDelay: daemon.detachDelay, }) @@ -960,9 +959,10 @@ func TestDaemon_BootAllocation(t *testing.T) { err = daemon.jsManager.WriteState("node-1", vms) require.NoError(t, err) - // Manually trigger the LoadState and allocation logic normally found in Start() - err = daemon.LoadState() + // Manually trigger the load and allocation logic normally found in Start() + loaded, err := daemon.jsManager.LoadState(daemon.node) require.NoError(t, err) + daemon.vmMgr.Replace(loaded) // Simulate the allocation loop in Start() for _, instance := range daemon.vmMgr.Snapshot() { @@ -2269,8 +2269,8 @@ func TestHandleEC2Events_DetachVolume(t *testing.T) { }) t.Run("QMPDeviceDelFails_NoForce", func(t *testing.T) { - // With nil QMPClient encoder/decoder, SendQMPCommand returns error. - // Without force=true, this should return ServerInternal. + // With nil QMPClient encoder/decoder, the QMP device_del returns + // error. Without force=true, this should return ServerInternal. command := types.EC2InstanceCommand{ ID: instanceID, Attributes: types.EC2CommandAttributes{ @@ -2998,7 +2998,7 @@ func TestInstanceTypeMemoryMiB_NilSafety(t *testing.T) { })) } -// --- Daemon.WriteState / Daemon.LoadState nil jsManager --- +// --- Daemon.WriteState nil jsManager --- func TestDaemon_WriteState_NilJSManager(t *testing.T) { d := &Daemon{jsManager: nil} @@ -3007,13 +3007,6 @@ func TestDaemon_WriteState_NilJSManager(t *testing.T) { assert.Contains(t, err.Error(), "JetStream manager not initialized") } -func TestDaemon_LoadState_NilJSManager(t *testing.T) { - d := &Daemon{jsManager: nil} - err := d.LoadState() - assert.Error(t, err) - assert.Contains(t, err.Error(), "JetStream manager not initialized") -} - // --- GetAvailableInstanceTypeInfos edge cases --- func TestGetAvailableInstanceTypeInfos_Overcommitted(t *testing.T) { @@ -3098,8 +3091,8 @@ func TestNewDaemon_WalDirPreservedIfSet(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. +// response. UnmountOne is shared by the AttachVolume rollback path and the +// DetachVolume ebs.unmount step. func TestVolumeMounterAdapter_UnmountOne_Success(t *testing.T) { natsURL := sharedNATSURL diff --git a/spinifex/daemon/network.go b/spinifex/daemon/network.go index 579cd13e0..2f1e3716d 100644 --- a/spinifex/daemon/network.go +++ b/spinifex/daemon/network.go @@ -26,24 +26,11 @@ var sudoCommand = func(name string, args ...string) *exec.Cmd { return exec.Command("sudo", append([]string{name}, args...)...) } -// NetworkPlumber handles tap device and OVS bridge operations for VM networking. -// The live implementation runs system commands (ip, ovs-vsctl); tests use a mock. -type NetworkPlumber interface { - // SetupTapDevice creates a tap device and adds it to the OVS br-int bridge - // with the correct iface-id for OVN port binding. - SetupTapDevice(eniId, mac string) error - - // CleanupTapDevice removes the tap device from br-int and deletes it. - CleanupTapDevice(eniId string) error -} - -// OVSNetworkPlumber implements NetworkPlumber using system commands. +// OVSNetworkPlumber implements vm.NetworkPlumber using system commands +// (ip, ovs-vsctl); tests use a mock satisfying the same interface. type OVSNetworkPlumber struct{} -var ( - _ NetworkPlumber = (*OVSNetworkPlumber)(nil) - _ vm.NetworkPlumber = (*OVSNetworkPlumber)(nil) -) +var _ vm.NetworkPlumber = (*OVSNetworkPlumber)(nil) func (p *OVSNetworkPlumber) SetupTapDevice(eniId, mac string) error { tapName := vm.TapDeviceName(eniId) diff --git a/spinifex/daemon/network_test.go b/spinifex/daemon/network_test.go index 7ee3bfc59..1e630ae4f 100644 --- a/spinifex/daemon/network_test.go +++ b/spinifex/daemon/network_test.go @@ -44,7 +44,7 @@ type MockNetworkPlumber struct { CleanupErr error } -var _ NetworkPlumber = (*MockNetworkPlumber)(nil) +var _ vm.NetworkPlumber = (*MockNetworkPlumber)(nil) type mockSetupCall struct { ENIId string diff --git a/spinifex/daemon/ownership.go b/spinifex/daemon/ownership.go index e9d6ae645..ea6c38820 100644 --- a/spinifex/daemon/ownership.go +++ b/spinifex/daemon/ownership.go @@ -10,14 +10,15 @@ import ( // checkInstanceOwnership verifies the caller owns the instance. // Returns true if access is allowed, false if denied (error already sent). -// Pre-Phase4 instances (empty AccountID) are only visible to root (GlobalAccountID). +// Instances with an empty AccountID (legacy/migration data) are only +// visible to root (GlobalAccountID). func checkInstanceOwnership(msg *nats.Msg, instanceID, ownerAccountID string) bool { callerAccountID := utils.AccountIDFromMsg(msg) - // Pre-Phase4 instance: only root can access + // Untenanted instance: only root can access. if ownerAccountID == "" { if callerAccountID != utils.GlobalAccountID { - slog.Warn("Pre-Phase4 instance access denied (not root)", + slog.Warn("Untenanted instance access denied (not root)", "instanceId", instanceID, "callerAccount", callerAccountID) respondWithError(msg, awserrors.ErrorInvalidInstanceIDNotFound) return false @@ -36,7 +37,8 @@ func checkInstanceOwnership(msg *nats.Msg, instanceID, ownerAccountID string) bo } // isInstanceVisible checks if the caller can see this instance (for Describe operations). -// Pre-Phase4 instances (empty AccountID) are only visible to root (GlobalAccountID). +// Instances with an empty AccountID (legacy/migration data) are only +// visible to root (GlobalAccountID). func isInstanceVisible(callerAccountID, ownerAccountID string) bool { if ownerAccountID == "" { return callerAccountID == utils.GlobalAccountID @@ -45,10 +47,10 @@ func isInstanceVisible(callerAccountID, ownerAccountID string) bool { } // volumeVisibleTo reports whether callerAccountID may operate on a volume with -// the given tenantID. Pre-Phase4 volumes (empty tenantID) are root-only — -// without this, the legacy short-circuit `tenantID != "" && tenantID != caller` -// matched every caller and let any tenant attach a legacy/migration volume by -// ID alone. +// the given tenantID. Volumes with an empty tenantID (legacy/migration +// data) are root-only — without this, the short-circuit +// `tenantID != "" && tenantID != caller` matches every caller and lets any +// tenant attach a legacy/migration volume by ID alone. func volumeVisibleTo(tenantID, callerAccountID string) bool { if tenantID == "" { return callerAccountID == utils.GlobalAccountID diff --git a/spinifex/daemon/ownership_test.go b/spinifex/daemon/ownership_test.go index cdd4f80c8..c8188568d 100644 --- a/spinifex/daemon/ownership_test.go +++ b/spinifex/daemon/ownership_test.go @@ -26,13 +26,13 @@ func TestIsInstanceVisible(t *testing.T) { expected: false, }, { - name: "pre-Phase4 instance visible to root", + name: "untenanted instance visible to root", callerAccountID: utils.GlobalAccountID, ownerAccountID: "", expected: true, }, { - name: "pre-Phase4 instance hidden from non-root", + name: "untenanted instance hidden from non-root", callerAccountID: "123456789012", ownerAccountID: "", expected: false, @@ -82,19 +82,19 @@ func TestVolumeVisibleTo(t *testing.T) { expected: false, }, { - name: "pre-Phase4 volume visible to root", + name: "untenanted volume visible to root", tenantID: "", callerAccountID: utils.GlobalAccountID, expected: true, }, { - name: "pre-Phase4 volume hidden from non-root tenant", + name: "untenanted volume hidden from non-root tenant", tenantID: "", callerAccountID: "123456789012", expected: false, }, { - name: "pre-Phase4 volume denied to empty caller", + name: "untenanted volume denied to empty caller", tenantID: "", callerAccountID: "", expected: false, diff --git a/spinifex/daemon/state_test.go b/spinifex/daemon/state_test.go index 1bf93f818..0d6d47bf3 100644 --- a/spinifex/daemon/state_test.go +++ b/spinifex/daemon/state_test.go @@ -40,6 +40,7 @@ func createDaemonWithJetStream(t *testing.T) *Daemon { require.NoError(t, daemon.jsManager.InitKVBucket()) require.NoError(t, daemon.jsManager.InitClusterStateBucket()) require.NoError(t, daemon.jsManager.InitTerminatedInstanceBucket()) + daemon.stateStore = newStateStoreAdapter(daemon.jsManager) // Wire just enough vm.Deps for manager-driven state operations to work // (migrate, MarkFailed, Restore classification). Full wiring (network @@ -48,7 +49,7 @@ func createDaemonWithJetStream(t *testing.T) *Daemon { // a real VM lifecycle. daemon.vmMgr.SetDeps(vm.Deps{ NodeID: daemon.node, - StateStore: newStateStoreAdapter(daemon.jsManager), + StateStore: daemon.stateStore, TransitionState: daemon.TransitionState, InstanceTypes: newInstanceTypeResolverAdapter(daemon.resourceMgr), Resources: newResourceControllerAdapter(daemon.resourceMgr), @@ -569,17 +570,17 @@ func TestTransitionState_MultipleInstancesIndependent(t *testing.T) { // --- Recovery / restoreInstances tests --- // // These tests simulate a daemon restart by writing instance state to JetStream, -// then calling restoreInstances on a fresh daemon. Since no QEMU process is +// then calling vmMgr.Restore() on a fresh daemon. Since no QEMU process is // running, isInstanceProcessRunning returns false for all instances, which // exercises the recovery state resolution logic. -// simulateCleanRestore writes a clean shutdown marker then calls restoreInstances. -// Without the marker, restoreInstances sleeps 3s waiting for stale QEMU PIDs — -// unnecessary in tests since no QEMU process ever runs. +// simulateCleanRestore writes a clean shutdown marker then drives the +// manager's restore path. Without the marker, Restore sleeps 3s waiting for +// stale QEMU PIDs — unnecessary in tests since no QEMU process ever runs. func simulateCleanRestore(t *testing.T, daemon *Daemon) { t.Helper() require.NoError(t, daemon.jsManager.WriteShutdownMarker(daemon.node)) - daemon.restoreInstances() + daemon.vmMgr.Restore() } // TestRestoreInstances_StoppingFinalizedToStopped verifies that an instance @@ -905,7 +906,9 @@ func TestStatePersistence_RoundTrip(t *testing.T) { // Simulate restart: clear and reload. daemon.vmMgr.Replace(map[string]*vm.VM{}) - require.NoError(t, daemon.LoadState()) + snapshot, err := daemon.jsManager.LoadState(daemon.node) + require.NoError(t, err) + daemon.vmMgr.Replace(snapshot) loaded, _ := daemon.vmMgr.Get("i-roundtrip") require.NotNil(t, loaded) @@ -1037,9 +1040,6 @@ func TestPendingWatchdog_MarksStuckInstanceFailed(t *testing.T) { } func TestAreVolumeSocketsValid_ValidSocket(t *testing.T) { - daemon := createDaemonWithJetStream(t) - - // Create a real Unix socket listener sockPath := filepath.Join(t.TempDir(), "test.sock") listener, err := net.Listen("unix", sockPath) require.NoError(t, err) @@ -1053,12 +1053,10 @@ func TestAreVolumeSocketsValid_ValidSocket(t *testing.T) { }, }, } - assert.True(t, daemon.areVolumeSocketsValid(instance)) + assert.True(t, vm.AreVolumeSocketsValid(instance)) } func TestAreVolumeSocketsValid_MissingSocket(t *testing.T) { - daemon := createDaemonWithJetStream(t) - instance := &vm.VM{ ID: "i-missing-sock", EBSRequests: types.EBSRequests{ @@ -1067,18 +1065,16 @@ func TestAreVolumeSocketsValid_MissingSocket(t *testing.T) { }, }, } - assert.False(t, daemon.areVolumeSocketsValid(instance)) + assert.False(t, vm.AreVolumeSocketsValid(instance)) } func TestAreVolumeSocketsValid_StaleSocketFile(t *testing.T) { - daemon := createDaemonWithJetStream(t) - - // Create a socket file with no listener — simulates viperblock crash - // leaving a stale socket on disk + // Socket file with no listener — simulates viperblock crash leaving a + // stale socket on disk. os.Stat would return true here, but dial fails. sockPath := filepath.Join(t.TempDir(), "stale.sock") listener, err := net.Listen("unix", sockPath) require.NoError(t, err) - listener.Close() // close listener but leave file on disk + listener.Close() instance := &vm.VM{ ID: "i-stale-sock", @@ -1088,23 +1084,19 @@ func TestAreVolumeSocketsValid_StaleSocketFile(t *testing.T) { }, }, } - // os.Stat would return true here, but dial should fail - assert.False(t, daemon.areVolumeSocketsValid(instance)) + assert.False(t, vm.AreVolumeSocketsValid(instance)) } func TestAreVolumeSocketsValid_EmptyRequests(t *testing.T) { - daemon := createDaemonWithJetStream(t) - instance := &vm.VM{ ID: "i-no-vols", EBSRequests: types.EBSRequests{}, } - assert.True(t, daemon.areVolumeSocketsValid(instance)) + assert.True(t, vm.AreVolumeSocketsValid(instance)) } func TestAreVolumeSocketsValid_TCPTransport(t *testing.T) { - daemon := createDaemonWithJetStream(t) - + // TCP transport can't be validated locally — should return true. instance := &vm.VM{ ID: "i-tcp", EBSRequests: types.EBSRequests{ @@ -1113,8 +1105,7 @@ func TestAreVolumeSocketsValid_TCPTransport(t *testing.T) { }, }, } - // TCP transport can't be validated locally — should return true - assert.True(t, daemon.areVolumeSocketsValid(instance)) + assert.True(t, vm.AreVolumeSocketsValid(instance)) } func TestMarkInstanceFailed_AlreadyShuttingDown(t *testing.T) { @@ -1168,10 +1159,12 @@ func TestRestoreInstances_ProvisioningInstanceLaunchTimeReset(t *testing.T) { before := time.Now() simulateCleanRestore(t, daemon) - // LoadState inside restoreInstances creates fresh objects, so look up - // from the map. The instance may have been removed by markInstanceFailed - // (launch fails in test env), so load directly from JetStream state. - require.NoError(t, daemon.LoadState()) + // Restore creates fresh objects, so look up from the map. The instance + // may have been removed by markInstanceFailed (launch fails in test + // env), so reload directly from JetStream state. + loaded, err := daemon.jsManager.LoadState(daemon.node) + require.NoError(t, err) + daemon.vmMgr.Replace(loaded) instance, _ := daemon.vmMgr.Get("i-provisioning") if instance == nil { // Instance was cleaned up by markInstanceFailed — that's fine. diff --git a/spinifex/daemon/vm_adapters.go b/spinifex/daemon/vm_adapters.go index ab7239f07..953b94d70 100644 --- a/spinifex/daemon/vm_adapters.go +++ b/spinifex/daemon/vm_adapters.go @@ -2,17 +2,14 @@ package daemon import ( "encoding/json" - "errors" "fmt" "log/slog" - "os/exec" "strings" "time" "github.com/aws/aws-sdk-go/service/ec2" "github.com/mulgadc/spinifex/spinifex/awserrors" handlers_ec2_placementgroup "github.com/mulgadc/spinifex/spinifex/handlers/ec2/placementgroup" - "github.com/mulgadc/spinifex/spinifex/qmp" "github.com/mulgadc/spinifex/spinifex/types" "github.com/mulgadc/spinifex/spinifex/vm" "github.com/nats-io/nats.go" @@ -188,8 +185,8 @@ func (a *volumeMounterAdapter) MountOne(req *types.EBSRequest) error { } // 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. +// are logged. Shared by AttachVolume rollback and the DetachVolume +// ebs.unmount step. func (a *volumeMounterAdapter) UnmountOne(req types.EBSRequest) { payload, err := json.Marshal(req) if err != nil { @@ -221,86 +218,6 @@ func (a *volumeMounterAdapter) UnmountOne(req types.EBSRequest) { 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. -type qmpClientFactoryAdapter struct{} - -var _ vm.QMPClientFactory = (*qmpClientFactoryAdapter)(nil) - -func newQMPClientFactoryAdapter() *qmpClientFactoryAdapter { return &qmpClientFactoryAdapter{} } - -func (a *qmpClientFactoryAdapter) Create(v *vm.VM) (*qmp.QMPClient, error) { - client, err := qmp.NewQMPClient(v.Config.QMPSocket) - if err != nil { - return nil, fmt.Errorf("connect QMP socket %s: %w", v.Config.QMPSocket, err) - } - - if _, err := sendQMPHandshake(client, v.ID); err != nil { - _ = client.Conn.Close() - return nil, err - } - - return client, nil -} - -// sendQMPHandshake issues qmp_capabilities under the QMPClient's mutex. Mirrors -// the slim version of Daemon.SendQMPCommand for the handshake; the daemon's -// full helper still owns long-running command dispatch. -func sendQMPHandshake(client *qmp.QMPClient, instanceID string) (*qmp.QMPResponse, error) { - if client == nil || client.Encoder == nil || client.Decoder == nil { - return nil, errors.New("QMP client is not initialized") - } - - client.Mu.Lock() - defer client.Mu.Unlock() - - if err := client.Conn.SetReadDeadline(time.Now().Add(30 * time.Second)); err != nil { - return nil, fmt.Errorf("set read deadline: %w", err) - } - defer func() { _ = client.Conn.SetReadDeadline(time.Time{}) }() - - if err := client.Encoder.Encode(qmp.QMPCommand{Execute: "qmp_capabilities"}); err != nil { - return nil, fmt.Errorf("encode qmp_capabilities: %w", err) - } - - for { - var msg map[string]any - if err := client.Decoder.Decode(&msg); err != nil { - return nil, fmt.Errorf("decode qmp_capabilities response: %w", err) - } - if _, ok := msg["event"]; ok { - continue - } - raw, err := json.Marshal(msg) - if err != nil { - return nil, fmt.Errorf("marshal qmp response: %w", err) - } - var resp qmp.QMPResponse - if err := json.Unmarshal(raw, &resp); err != nil { - return nil, fmt.Errorf("unmarshal qmp response: %w", err) - } - if resp.Error != nil { - return nil, fmt.Errorf("qmp_capabilities failed: %s", resp.Error.Desc) - } - slog.Debug("QMP handshake complete", "instance", instanceID) - return &resp, nil - } -} - -// processLauncherAdapter satisfies vm.ProcessLauncher. The real -// implementation is a 1:1 wrapper over Config.Execute; the seam exists so -// tests can substitute scripted commands. -type processLauncherAdapter struct{} - -var _ vm.ProcessLauncher = (*processLauncherAdapter)(nil) - -func newProcessLauncherAdapter() *processLauncherAdapter { return &processLauncherAdapter{} } - -func (a *processLauncherAdapter) Launch(cfg *vm.Config) (*exec.Cmd, error) { - return cfg.Execute() -} - // instanceTypeResolverAdapter satisfies vm.InstanceTypeResolver. It looks up // the EC2 InstanceTypeInfo on the daemon's ResourceManager and projects the // QEMU-relevant numbers (vCPU, memory MiB, architecture) into the @@ -368,37 +285,13 @@ func (a *resourceControllerAdapter) CanAllocate(instanceType string, count int) return a.rm.canAllocate(it, count) } -// volumeStateUpdaterAdapter satisfies vm.VolumeStateUpdater by delegating to -// the daemon's volume service. -type volumeStateUpdaterAdapter struct { - svc volumeStateUpdater -} - -// volumeStateUpdater is the narrow slice of handlers_ec2_volume.VolumeServiceImpl -// that the manager touches. Defining it locally avoids dragging the full -// volume-service surface into the adapter. -type volumeStateUpdater interface { - UpdateVolumeState(volumeID, state, instanceID, attachmentDevice string) error -} - -var _ vm.VolumeStateUpdater = (*volumeStateUpdaterAdapter)(nil) - -func newVolumeStateUpdaterAdapter(svc volumeStateUpdater) *volumeStateUpdaterAdapter { - return &volumeStateUpdaterAdapter{svc: svc} -} - -func (a *volumeStateUpdaterAdapter) UpdateVolumeState(volumeID, state, instanceID, attachmentDevice string) error { - return a.svc.UpdateVolumeState(volumeID, state, instanceID, attachmentDevice) -} - // onInstanceRecoveringHook returns the daemon's OnInstanceRecovering // callback. Restore fires it once per instance about to be relaunched so // concurrent terminate commands can land on this node before launch // completes — without it, ec2.cmd. would only be subscribed by // onInstanceUpHook after launch success, leaving a window where the // instance is reachable by DescribeInstances (in StatePending) but not -// by EC2 commands. Mirrors the pre-2e early-subscribe block in -// daemon.restoreInstances. +// by EC2 commands. func (d *Daemon) onInstanceRecoveringHook() func(*vm.VM) { return func(instance *vm.VM) { d.mu.Lock() @@ -520,17 +413,14 @@ func (d *Daemon) onInstanceDownHook() func(string) { // All collaborators must already be initialized; callers are expected to // invoke this from Daemon.Start after services and JetStream are ready. func (d *Daemon) buildVMManagerDeps() vm.Deps { - volState := newVolumeStateUpdaterAdapter(d.volumeService) return vm.Deps{ NodeID: d.node, - StateStore: newStateStoreAdapter(d.jsManager), - VolumeMounter: newVolumeMounterAdapter(d.natsConn, d.node, volState), - QMPClientFactory: newQMPClientFactoryAdapter(), - ProcessLauncher: newProcessLauncherAdapter(), + StateStore: d.stateStore, + VolumeMounter: newVolumeMounterAdapter(d.natsConn, d.node, d.volumeService), NetworkPlumber: d.networkPlumber, InstanceTypes: newInstanceTypeResolverAdapter(d.resourceMgr), Resources: newResourceControllerAdapter(d.resourceMgr), - VolumeStateUpdater: volState, + VolumeStateUpdater: d.volumeService, InstanceCleaner: newInstanceCleanerAdapter(d), Hooks: vm.ManagerHooks{ OnInstanceUp: d.onInstanceUpHook(), @@ -563,8 +453,7 @@ func newInstanceCleanerAdapter(d *Daemon) *instanceCleanerAdapter { // DeleteVolumes deletes EFI / cloud-init internal volumes via ebs.delete // and user volumes flagged DeleteOnTermination via the volume service. -// Errors are logged per volume; partial failure is tolerated to match -// pre-2c stopInstance behaviour. +// Errors are logged per volume; partial failure is tolerated. func (a *instanceCleanerAdapter) DeleteVolumes(instance *vm.VM) { instance.EBSRequests.Mu.Lock() defer instance.EBSRequests.Mu.Unlock() diff --git a/spinifex/handlers/ec2/volume/service_impl.go b/spinifex/handlers/ec2/volume/service_impl.go index 090e1b44b..c47a1cf25 100644 --- a/spinifex/handlers/ec2/volume/service_impl.go +++ b/spinifex/handlers/ec2/volume/service_impl.go @@ -22,6 +22,7 @@ import ( "github.com/mulgadc/spinifex/spinifex/objectstore" "github.com/mulgadc/spinifex/spinifex/types" "github.com/mulgadc/spinifex/spinifex/utils" + "github.com/mulgadc/spinifex/spinifex/vm" "github.com/mulgadc/viperblock/viperblock" s3backend "github.com/mulgadc/viperblock/viperblock/backends/s3" "github.com/nats-io/nats.go" @@ -32,6 +33,10 @@ const defaultGP3IOPS = 3000 // Ensure VolumeServiceImpl implements VolumeService var _ VolumeService = (*VolumeServiceImpl)(nil) +// Ensure VolumeServiceImpl satisfies vm.VolumeStateUpdater so the manager +// can call UpdateVolumeState directly without a daemon-side adapter. +var _ vm.VolumeStateUpdater = (*VolumeServiceImpl)(nil) + // VolumeServiceImpl handles EBS volume operations with S3 storage type VolumeServiceImpl struct { config *config.Config diff --git a/spinifex/vm/crash_recovery.go b/spinifex/vm/crash_recovery.go index 5db3059e9..2942ab8ba 100644 --- a/spinifex/vm/crash_recovery.go +++ b/spinifex/vm/crash_recovery.go @@ -110,8 +110,6 @@ func (m *Manager) HandleCrash(instance *VM, waitErr error) { if v.Health.FirstCrashTime.IsZero() { v.Health.FirstCrashTime = now } - v.Running = false - v.PID = 0 }) if m.deps.Resources != nil && instance.InstanceType != "" { diff --git a/spinifex/vm/crash_recovery_test.go b/spinifex/vm/crash_recovery_test.go index 41dfc278e..c9cb078bb 100644 --- a/spinifex/vm/crash_recovery_test.go +++ b/spinifex/vm/crash_recovery_test.go @@ -128,8 +128,6 @@ func TestHandleCrash_CoreFlow(t *testing.T) { instance := &VM{ ID: "i-core", Status: StateRunning, - Running: true, - PID: 12345, InstanceType: "t3.micro", Config: Config{QMPSocket: qmpPath}, } @@ -147,9 +145,6 @@ func TestHandleCrash_CoreFlow(t *testing.T) { assert.Equal(t, "unknown", instance.Health.LastCrashReason) assert.False(t, instance.Health.FirstCrashTime.IsZero()) - assert.False(t, instance.Running, "Running must be cleared") - assert.Zero(t, instance.PID, "PID must be cleared") - assert.Equal(t, 1, rc.deallocateCount("t3.micro"), "crashed instance resources must be deallocated") _, err := os.Stat(qmpPath) @@ -162,8 +157,6 @@ func TestHandleCrash_FirstCrashSetsTime(t *testing.T) { instance := &VM{ ID: "i-firstcrash", Status: StateRunning, - Running: true, - PID: 111, InstanceType: "t3.micro", } m.Insert(instance) @@ -175,8 +168,6 @@ func TestHandleCrash_FirstCrashSetsTime(t *testing.T) { m.UpdateState(instance.ID, func(v *VM) { v.Status = StateRunning - v.Running = true - v.PID = 222 }) time.Sleep(time.Millisecond) @@ -188,17 +179,15 @@ func TestHandleCrash_FirstCrashSetsTime(t *testing.T) { } // TestHandleCrash_UnknownInstanceType verifies a crash on a VM with an -// instance type not in the resolver still completes the state transition, -// health bookkeeping, and pid/running clear — even though MaybeRestart will -// later refuse to restart it. +// instance type not in the resolver still completes the state transition +// and health bookkeeping — even though MaybeRestart will later refuse to +// restart it. func TestHandleCrash_UnknownInstanceType(t *testing.T) { m, _, _, _ := crashTestManager(t) instance := &VM{ ID: "i-unknown", Status: StateRunning, - Running: true, - PID: 333, InstanceType: "z99.nonexistent", } m.Insert(instance) @@ -207,8 +196,6 @@ func TestHandleCrash_UnknownInstanceType(t *testing.T) { assert.Equal(t, StateError, m.Status(instance)) assert.Equal(t, 1, instance.Health.CrashCount) - assert.False(t, instance.Running) - assert.Zero(t, instance.PID) } func TestMaybeRestart_ExceedsMaxInWindow(t *testing.T) { diff --git a/spinifex/vm/errors.go b/spinifex/vm/errors.go index 0b6795c71..55e886894 100644 --- a/spinifex/vm/errors.go +++ b/spinifex/vm/errors.go @@ -30,6 +30,5 @@ 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. +// defensively so a half-started backend mount is not orphaned. var ErrMountAmbiguous = errors.New("ebs.mount succeeded with empty URI") diff --git a/spinifex/vm/lifecycle.go b/spinifex/vm/lifecycle.go index 9fedd8cd1..4921ee96d 100644 --- a/spinifex/vm/lifecycle.go +++ b/spinifex/vm/lifecycle.go @@ -72,8 +72,7 @@ func (m *Manager) launchStillValid(instance *VM) bool { } // launch is the orchestrator: pid check, mount volumes, exec QEMU, attach -// QMP, fire OnInstanceUp, transition to Running. Mirrors the pre-2b -// Daemon.LaunchInstance flow. +// QMP, fire OnInstanceUp, transition to Running. func (m *Manager) launch(instance *VM) error { if !m.launchStillValid(instance) { return nil @@ -109,7 +108,7 @@ func (m *Manager) launch(instance *VM) error { return err } - qmpClient, err := m.deps.QMPClientFactory.Create(instance) + qmpClient, err := newQMPClientWithHandshake(instance) if err != nil { slog.Error("Failed to create QMP client", "err", err) return err @@ -149,8 +148,7 @@ func (m *Manager) launch(instance *VM) error { if m.deps.Hooks.OnInstanceUp != nil { // Launch path: per-instance subscribe failures are logged and the - // launch still succeeds, mirroring the pre-2e early-subscribe block - // in handleEC2RunInstances. The instance is reachable via cluster + // launch still succeeds. The instance is reachable via cluster // fan-out (DescribeInstances) and the next OnInstanceUp on a // state-touching event will reinstall the subs idempotently. if err := m.deps.Hooks.OnInstanceUp(instance); err != nil { @@ -163,7 +161,7 @@ func (m *Manager) launch(instance *VM) error { } // startQEMU launches the QEMU process for instance and waits for startup to -// confirm. Mirrors the pre-2b Daemon.StartInstance flow. +// confirm. func (m *Manager) startQEMU(instance *VM) error { pidFile, err := utils.GeneratePidFile(instance.ID) if err != nil { @@ -281,7 +279,7 @@ func (m *Manager) startQEMU(instance *VM) error { startupConfirmed := make(chan bool, 1) go func() { - cmd, err := m.deps.ProcessLauncher.Launch(&instance.Config) + cmd, err := instance.Config.Execute() if err != nil { slog.Error("Failed to execute VM", "err", err) processChan <- 0 @@ -384,7 +382,6 @@ func (m *Manager) startQEMU(instance *VM) error { } // appendDevHostfwdNIC adds a user-mode NIC with SSH hostfwd for dev access. -// Mirrors the pre-2b inline DEV_NETWORKING block in StartInstance. func (m *Manager) appendDevHostfwdNIC(instance *VM) { sshDebugAddr, err := viperblock.FindFreePort() if err != nil { @@ -436,10 +433,10 @@ func (m *Manager) appendDevHostfwdNIC(instance *VM) { // AttachQMP connects a QMP client to a QEMU process that already exists // (e.g. a daemon restart finding a still-running QEMU) and starts the -// heartbeat goroutine. The launch path uses QMPClientFactory + qmpHeartbeat -// directly; AttachQMP is the matching seam for reconnect callers. +// heartbeat goroutine. AttachQMP is the matching seam for reconnect +// callers; the launch path calls the same helper inline. func (m *Manager) AttachQMP(instance *VM) error { - client, err := m.deps.QMPClientFactory.Create(instance) + client, err := newQMPClientWithHandshake(instance) if err != nil { return err } @@ -448,6 +445,22 @@ func (m *Manager) AttachQMP(instance *VM) error { return nil } +// newQMPClientWithHandshake dials the instance's QMP socket and runs the +// qmp_capabilities handshake. The caller owns starting the heartbeat +// goroutine on the returned client. +func newQMPClientWithHandshake(v *VM) (*qmp.QMPClient, error) { + client, err := qmp.NewQMPClient(v.Config.QMPSocket) + if err != nil { + return nil, fmt.Errorf("connect QMP socket %s: %w", v.Config.QMPSocket, err) + } + if _, err := sendQMPCommand(client, qmp.QMPCommand{Execute: "qmp_capabilities"}, v.ID); err != nil { + _ = client.Conn.Close() + return nil, err + } + slog.Debug("QMP handshake complete", "instance", v.ID) + return client, nil +} + // qmpHeartbeat sends a query-status QMP command every 30s to confirm the // QEMU process is still healthy. Exits when the instance reaches a terminal // or transitional state. Closes the QMP connection on exit. @@ -479,9 +492,7 @@ func (m *Manager) qmpHeartbeat(instance *VM) { } // sendQMPCommand encodes cmd onto client and decodes the matching response, -// skipping informational events. Mirrors the pre-2b Daemon.SendQMPCommand -// helper but without the daemon receiver — used by manager-internal callers -// (heartbeat). +// skipping informational events. func sendQMPCommand(q *qmp.QMPClient, cmd qmp.QMPCommand, instanceID string) (*qmp.QMPResponse, error) { if q == nil || q.Encoder == nil || q.Decoder == nil { return nil, fmt.Errorf("QMP client is not initialized") diff --git a/spinifex/vm/manager.go b/spinifex/vm/manager.go index 73f2e9f22..e71d94273 100644 --- a/spinifex/vm/manager.go +++ b/spinifex/vm/manager.go @@ -34,8 +34,6 @@ type Deps struct { StateStore StateStore VolumeMounter VolumeMounter - QMPClientFactory QMPClientFactory - ProcessLauncher ProcessLauncher NetworkPlumber NetworkPlumber InstanceTypes InstanceTypeResolver Resources ResourceController diff --git a/spinifex/vm/manager_deps_test.go b/spinifex/vm/manager_deps_test.go index 7e251346b..6f32aa225 100644 --- a/spinifex/vm/manager_deps_test.go +++ b/spinifex/vm/manager_deps_test.go @@ -213,12 +213,8 @@ func TestDeps_StateStoreSatisfiesInterface(t *testing.T) { if err := m.deps.StateStore.WriteStoppedInstance("i-1", &VM{ID: "i-1"}); err != nil { t.Fatalf("WriteStoppedInstance: %v", err) } - got, err := m.deps.StateStore.LoadStoppedInstance("i-1") - if err != nil { - t.Fatalf("LoadStoppedInstance: %v", err) - } - if got == nil || got.ID != "i-1" { - t.Fatalf("LoadStoppedInstance: got %v, want VM{ID:i-1}", got) + if got := store.stopped["i-1"]; got == nil || got.ID != "i-1" { + t.Fatalf("WriteStoppedInstance: got %v, want VM{ID:i-1}", got) } if err := m.deps.StateStore.SaveRunningState("n", map[string]*VM{"i-1": {ID: "i-1"}}); err != nil { diff --git a/spinifex/vm/process_launcher.go b/spinifex/vm/process_launcher.go deleted file mode 100644 index 77ddc02b2..000000000 --- a/spinifex/vm/process_launcher.go +++ /dev/null @@ -1,10 +0,0 @@ -package vm - -import "os/exec" - -// ProcessLauncher resolves a Config into an *exec.Cmd ready to Start(). The -// real implementation is a 1-line wrapper over Config.Execute(); the seam -// exists so tests can substitute scripted commands without spawning QEMU. -type ProcessLauncher interface { - Launch(cfg *Config) (*exec.Cmd, error) -} diff --git a/spinifex/vm/qmp_client.go b/spinifex/vm/qmp_client.go deleted file mode 100644 index b50587627..000000000 --- a/spinifex/vm/qmp_client.go +++ /dev/null @@ -1,10 +0,0 @@ -package vm - -import "github.com/mulgadc/spinifex/spinifex/qmp" - -// QMPClientFactory connects a QMP client to a VM's QMP socket and runs the -// qmp_capabilities handshake. The returned client is ready for command -// dispatch. Heartbeats are owned by the manager, not the factory. -type QMPClientFactory interface { - Create(v *VM) (*qmp.QMPClient, error) -} diff --git a/spinifex/vm/restore.go b/spinifex/vm/restore.go index 6c866af42..4d0e8aa97 100644 --- a/spinifex/vm/restore.go +++ b/spinifex/vm/restore.go @@ -245,11 +245,11 @@ func (m *Manager) finalizeTransitionalRestore(instance *VM) bool { return true } -// relaunchAll kicks the recovery launch loop. Per pre-2e behaviour, we -// announce each instance via OnInstanceRecovering before launching so the -// daemon can early-subscribe ec2.cmd.; that lets a concurrent terminate -// reach this node while the relaunch is still pending. The OnInstanceUp -// hook fired after a successful launch reinstalls both per-instance subs +// relaunchAll kicks the recovery launch loop. Each instance is announced +// via OnInstanceRecovering before launching so the daemon can +// early-subscribe ec2.cmd.; that lets a concurrent terminate reach +// this node while the relaunch is still pending. The OnInstanceUp hook +// fired after a successful launch reinstalls both per-instance subs // idempotently. func (m *Manager) relaunchAll(toLaunch []*VM) { if m.deps.Hooks.OnInstanceRecovering != nil { diff --git a/spinifex/vm/shutdown.go b/spinifex/vm/shutdown.go index 352a04144..6ed1bb883 100644 --- a/spinifex/vm/shutdown.go +++ b/spinifex/vm/shutdown.go @@ -129,7 +129,7 @@ func (m *Manager) Terminate(id string) error { // launch errors mid-way: callers (NATS RunInstances handler, recovery // worker, pending watchdog, system-instance launcher) get back control // immediately and do not block on volume unmount, ENI delete, or KV -// writes. Mirrors the pre-2c `go d.finalizeTermination(instance)` pattern. +// writes. // // Tolerates instances already in a cleanup state (no-op) and instances // that may or may not be present in the running-VM map. @@ -350,9 +350,8 @@ func (m *Manager) transitionWithPrecheck(instance *VM, target InstanceState) err } // writeRunningState persists the current running-VM map via the StateStore. -// The View callback holds the manager lock across the marshal+put — same -// constraint as the daemon's pre-2c WriteState. Splitting marshal from put -// is a 2f cleanup item. +// The View callback holds the manager lock across the marshal+put so VM +// fields can't change mid-encode; splitting marshal from put is deferred. func (m *Manager) writeRunningState() error { if m.deps.StateStore == nil { return nil diff --git a/spinifex/vm/shutdown_test.go b/spinifex/vm/shutdown_test.go index 49f21c445..7d1114db6 100644 --- a/spinifex/vm/shutdown_test.go +++ b/spinifex/vm/shutdown_test.go @@ -132,9 +132,7 @@ func TestMarkFailed_AlreadyTerminated_NoOp(t *testing.T) { // TestStop_DoesNotCallDeleteVolumes locks down the architectural // invariant that Stop must never delete volumes — a regression here -// would silently destroy user data on every stop. Phase 2c removed -// TestStopInstance_NoDelete_OnStop from daemon_test.go without a -// vm-package replacement; this is that replacement. +// would silently destroy user data on every stop. func TestStop_DoesNotCallDeleteVolumes(t *testing.T) { m, _, _, cleaner, _ := shutdownTestManager(t) diff --git a/spinifex/vm/state_store.go b/spinifex/vm/state_store.go index 0df33e5d6..be9c88084 100644 --- a/spinifex/vm/state_store.go +++ b/spinifex/vm/state_store.go @@ -1,11 +1,12 @@ package vm -// StateStore persists VM state across the three locations spinifex models: -// per-node "running" state, the cluster-shared "stopped" bucket, and the -// cluster-shared "terminated" bucket. +// StateStore persists VM state to the per-node "running" bucket and to the +// cluster-shared "stopped" / "terminated" buckets. // // The interface lives in the vm package; the live implementation is a thin -// adapter in the daemon package wrapping the JetStream-backed bucket. +// adapter in the daemon package wrapping the JetStream-backed bucket. Both +// vm.Manager and daemon-side handlers route VM-instance state I/O through +// this interface so the daemon can be exercised against an in-memory fake. type StateStore interface { // SaveRunningState writes the snapshot of VMs currently running on the // given node. The supplied map is owned by the caller and must not be diff --git a/spinifex/vm/vm.go b/spinifex/vm/vm.go index f3ba3e923..3c8844ce6 100644 --- a/spinifex/vm/vm.go +++ b/spinifex/vm/vm.go @@ -35,8 +35,6 @@ type ExtraENI struct { type VM struct { ID string `json:"id"` - PID int `json:"pid"` - Running bool `json:"running"` Status InstanceState `json:"status"` InstanceType string `json:"instance_type"` Config Config `json:"config"` @@ -70,7 +68,8 @@ type VM struct { Health InstanceHealthState `json:"health"` // AccountID is the AWS account that owns this instance. - // Empty for pre-Phase4 resources (treated as visible to all accounts). + // Empty AccountID identifies legacy/migration resources that are only + // visible to the root account. AccountID string `json:"account_id,omitempty"` // VPC networking: ENI attached to this instance (set by RunInstances when VPC mode is active) @@ -130,8 +129,6 @@ type VM struct { // that last ran this instance. Must be called after deserializing a VM from // shared KV before launching it on a new node. func (v *VM) ResetNodeLocalState() { - v.PID = 0 - v.Running = false v.MetadataServerAddress = "" v.QMPClient = &qmp.QMPClient{} v.EBSRequests.Mu = sync.Mutex{} diff --git a/spinifex/vm/vm_test.go b/spinifex/vm/vm_test.go index 0a63b0556..efac707f3 100644 --- a/spinifex/vm/vm_test.go +++ b/spinifex/vm/vm_test.go @@ -233,16 +233,12 @@ func argExists(args []string, flag string) bool { func TestResetNodeLocalState(t *testing.T) { v := &VM{ ID: "i-abc123", - PID: 12345, - Running: true, MetadataServerAddress: "127.0.0.1:9999", Status: StateRunning, } v.ResetNodeLocalState() - assert.Equal(t, 0, v.PID) - assert.False(t, v.Running) assert.Empty(t, v.MetadataServerAddress) assert.NotNil(t, v.QMPClient) // ID and Status should be unchanged diff --git a/spinifex/vm/volumes.go b/spinifex/vm/volumes.go index b86c917bf..7a567c230 100644 --- a/spinifex/vm/volumes.go +++ b/spinifex/vm/volumes.go @@ -16,7 +16,7 @@ import ( // 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. +// gives a 20s budget. const blockdevDelMaxAttempts = 20 // AttachVolumeResult carries the device names produced by AttachVolume: @@ -270,10 +270,9 @@ func (m *Manager) DetachVolume(id, volumeID, device string, force bool) (string, 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). + // device_del is idempotent on DeviceNotFound so a second AWS-CLI + // retry can drive blockdev-del to completion when a prior detach left + // the guest device gone but the block node intact. _, err := sendQMPCommand(instance.QMPClient, qmp.QMPCommand{ Execute: "device_del", Arguments: map[string]any{"id": deviceID}, @@ -295,14 +294,14 @@ func (m *Manager) DetachVolume(id, volumeID, device string, force bool) (string, time.Sleep(m.deps.DetachDelay) } - // Phase 2: blockdev-del with bounded retry on "node is in use". + // 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). + // object-del (best-effort). if _, iothreadErr := sendQMPCommand(instance.QMPClient, qmp.QMPCommand{ Execute: "object-del", Arguments: map[string]any{"id": iothreadID}, @@ -311,7 +310,7 @@ func (m *Manager) DetachVolume(id, volumeID, device string, force bool) (string, "volumeId", volumeID, "err", iothreadErr) } - // Phase 3: ebs.unmount (best-effort). + // ebs.unmount (best-effort). if m.deps.VolumeMounter != nil { m.deps.VolumeMounter.UnmountOne(ebsReq) } @@ -392,7 +391,7 @@ func (m *Manager) tryBlockdevDel(instance *VM, nodeName string) error { // 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). +// retries. func isQMPDeviceNotFound(err error) bool { if err == nil { return false diff --git a/spinifex/vm/volumes_test.go b/spinifex/vm/volumes_test.go index d9524ad54..dd162a890 100644 --- a/spinifex/vm/volumes_test.go +++ b/spinifex/vm/volumes_test.go @@ -411,12 +411,10 @@ func attachVolumeRunningInstance(t *testing.T, qmpClient *qmp.QMPClient, mounter 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. +// TestAttachVolume_MountAmbiguous_TriggersUnmountOne asserts that an +// ebs.mount response with an empty URI is treated as ambiguous: AttachVolume +// returns ErrMountAmbiguous and invokes UnmountOne so a half-started +// backend mount cannot be orphaned. func TestAttachVolume_MountAmbiguous_TriggersUnmountOne(t *testing.T) { mounter := &fakeVolumeMounter{mountOneErr: ErrMountAmbiguous} m, _ := attachVolumeRunningInstance(t, nil, mounter) @@ -430,8 +428,7 @@ func TestAttachVolume_MountAmbiguous_TriggersUnmountOne(t *testing.T) { // 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 != "". +// do NOT trigger UnmountOne — only the empty-URI ambiguous case rolls back. func TestAttachVolume_GenericMountError_DoesNotUnmount(t *testing.T) { mounter := &fakeVolumeMounter{mountOneErr: errors.New("ebs.mount NATS request: timeout")} m, _ := attachVolumeRunningInstance(t, nil, mounter) @@ -440,7 +437,7 @@ func TestAttachVolume_GenericMountError_DoesNotUnmount(t *testing.T) { 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)") + "non-ambiguous MountOne errors must not trigger UnmountOne") } // TestAttachVolume_NBDURIParseFailure_TriggersUnmountOne covers the rollback