diff --git a/hive/daemon/daemon.go b/hive/daemon/daemon.go index 27c3d3475..ebf19ecd0 100644 --- a/hive/daemon/daemon.go +++ b/hive/daemon/daemon.go @@ -621,8 +621,17 @@ func (d *Daemon) restoreInstances() { continue } - if instance.Attributes.StopInstance { - slog.Info("Instance flagged as user initiated stop, skipping", "instance", instance.ID) + if instance.Attributes.StopInstance || instance.Status == vm.StateStopped { + slog.Info("Instance is stopped, subscribing to NATS for start commands", "instance", instance.ID) + d.mu.Lock() + sub, err := d.natsConn.QueueSubscribe(fmt.Sprintf("ec2.cmd.%s", instance.ID), "hive-events", d.handleEC2Events) + if err != nil { + d.mu.Unlock() + slog.Error("Failed to subscribe stopped instance to NATS", "instance", instance.ID, "err", err) + continue + } + d.natsSubscriptions[instance.ID] = sub + d.mu.Unlock() continue } @@ -1162,12 +1171,15 @@ func (d *Daemon) stopInstance(instances map[string]*vm.VM, deleteVolume bool) er // For stop operations, keep the subscription so we can receive start commands if deleteVolume { for _, instance := range instances { - slog.Info("Unsubscribing from NATS subject", "instance", instance.ID) - if err := d.natsSubscriptions[fmt.Sprintf("ec2.cmd.%s", instance.ID)].Unsubscribe(); err != nil { - slog.Error("Failed to unsubscribe from NATS subject", "instance", instance.ID, "err", err) + d.mu.Lock() + if sub, ok := d.natsSubscriptions[instance.ID]; ok { + slog.Info("Unsubscribing from NATS subject", "instance", instance.ID) + if err := sub.Unsubscribe(); err != nil { + slog.Error("Failed to unsubscribe from NATS subject", "instance", instance.ID, "err", err) + } + delete(d.natsSubscriptions, instance.ID) } - // TODO: Remove redundant subscription if not used - //d.natsSubscriptions[fmt.Sprintf("ec2.describe.%s", instance.ID)].Unsubscribe() + d.mu.Unlock() } } return nil diff --git a/hive/daemon/state_test.go b/hive/daemon/state_test.go index 62fd72d81..de7521bf0 100644 --- a/hive/daemon/state_test.go +++ b/hive/daemon/state_test.go @@ -1,11 +1,15 @@ package daemon import ( + "encoding/json" + "fmt" "os" + "sync" "testing" "time" "github.com/mulgadc/hive/hive/config" + "github.com/mulgadc/hive/hive/qmp" "github.com/mulgadc/hive/hive/vm" "github.com/nats-io/nats-server/v2/server" "github.com/nats-io/nats.go" @@ -168,3 +172,670 @@ func TestTransitionState_NilEC2Instance(t *testing.T) { require.NoError(t, err) assert.Equal(t, vm.StateRunning, instance.Status) } + +// allStates is the canonical list of every InstanceState in the system. +var allStates = []vm.InstanceState{ + vm.StateProvisioning, + vm.StatePending, + vm.StateRunning, + vm.StateStopping, + vm.StateStopped, + vm.StateShuttingDown, + vm.StateTerminated, + vm.StateError, +} + +// stableStates are states where an instance can rest indefinitely. +var stableStates = []vm.InstanceState{ + vm.StateRunning, + vm.StateStopped, + vm.StateTerminated, + vm.StateError, +} + +// transitionalStates are states that must resolve to a stable state. +var transitionalStates = []vm.InstanceState{ + vm.StateProvisioning, + vm.StatePending, + vm.StateStopping, + vm.StateShuttingDown, +} + +// TestValidTransitions_MapCoversAllNonTerminalStates verifies that every state +// except StateTerminated has at least one valid outgoing transition. A missing +// entry would mean an instance could get permanently stuck. +func TestValidTransitions_MapCoversAllNonTerminalStates(t *testing.T) { + for _, state := range allStates { + if state == vm.StateTerminated { + continue + } + targets, ok := vm.ValidTransitions[state] + assert.True(t, ok, "state %q missing from ValidTransitions map", state) + assert.NotEmpty(t, targets, "state %q has no valid transitions — instances would get stuck", state) + } +} + +// TestTerminatedState_HasNoOutgoingTransitions ensures the terminal state is +// truly terminal — no valid transitions out. +func TestTerminatedState_HasNoOutgoingTransitions(t *testing.T) { + targets, ok := vm.ValidTransitions[vm.StateTerminated] + assert.False(t, ok || len(targets) > 0, + "StateTerminated should have no valid transitions, got %v", targets) +} + +// TestTransitionalStates_CanReachStableState verifies that every transitional +// state has at least one direct transition to a stable state. This is the +// structural guarantee that prevents the "stuck in stopping" bug. +func TestTransitionalStates_CanReachStableState(t *testing.T) { + stableSet := map[vm.InstanceState]bool{} + for _, s := range stableStates { + stableSet[s] = true + } + + for _, state := range transitionalStates { + targets := vm.ValidTransitions[state] + hasStableTarget := false + for _, target := range targets { + if stableSet[target] { + hasStableTarget = true + break + } + } + assert.True(t, hasStableTarget, + "transitional state %q has no direct path to a stable state (targets: %v)", state, targets) + } +} + +// TestEveryState_CanReachTerminated does a BFS from each non-terminated state +// to confirm there is always a path to StateTerminated. An unreachable terminal +// state means instances can never be fully cleaned up. +func TestEveryState_CanReachTerminated(t *testing.T) { + for _, start := range allStates { + if start == vm.StateTerminated { + continue + } + + visited := map[vm.InstanceState]bool{start: true} + queue := []vm.InstanceState{start} + found := false + + for len(queue) > 0 { + current := queue[0] + queue = queue[1:] + if current == vm.StateTerminated { + found = true + break + } + for _, next := range vm.ValidTransitions[current] { + if !visited[next] { + visited[next] = true + queue = append(queue, next) + } + } + } + + assert.True(t, found, + "state %q has no path to StateTerminated — instances starting here can never be cleaned up", start) + } +} + +// TestStopLifecycle_CompletesFullChain walks the full stop lifecycle: +// provisioning → running → stopping → stopped, then restart and terminate: +// stopped → running → shutting-down → terminated. +func TestStopLifecycle_CompletesFullChain(t *testing.T) { + daemon := createDaemonWithJetStream(t) + + instance := &vm.VM{ID: "i-lifecycle-stop", Status: vm.StateProvisioning} + daemon.Instances.Mu.Lock() + daemon.Instances.VMS[instance.ID] = instance + daemon.Instances.Mu.Unlock() + + chain := []vm.InstanceState{ + vm.StateRunning, // launch + vm.StateStopping, // stop requested + vm.StateStopped, // stop completed + vm.StateRunning, // restart + vm.StateShuttingDown, // terminate requested + vm.StateTerminated, // terminate completed + } + + for _, target := range chain { + err := daemon.TransitionState(instance, target) + require.NoError(t, err, "transition to %s failed (current: %s)", target, instance.Status) + } + + assert.Equal(t, vm.StateTerminated, instance.Status) +} + +// TestStopLifecycle_ErrorRecovery verifies that an instance that hits an error +// during stop can recover: running → stopping → error → shutting-down → terminated. +func TestStopLifecycle_ErrorRecovery(t *testing.T) { + daemon := createDaemonWithJetStream(t) + + instance := &vm.VM{ID: "i-lifecycle-err", Status: vm.StateRunning} + daemon.Instances.Mu.Lock() + daemon.Instances.VMS[instance.ID] = instance + daemon.Instances.Mu.Unlock() + + chain := []vm.InstanceState{ + vm.StateStopping, + vm.StateError, // stop failed + vm.StateShuttingDown, // force terminate from error + vm.StateTerminated, + } + + for _, target := range chain { + err := daemon.TransitionState(instance, target) + require.NoError(t, err, "transition to %s failed (current: %s)", target, instance.Status) + } + + assert.Equal(t, vm.StateTerminated, instance.Status) +} + +// TestShuttingDown_ErrorRecovery verifies: shutting-down → error → shutting-down → terminated. +func TestShuttingDown_ErrorRecovery(t *testing.T) { + daemon := createDaemonWithJetStream(t) + + instance := &vm.VM{ID: "i-shutdown-err", Status: vm.StateShuttingDown} + daemon.Instances.Mu.Lock() + daemon.Instances.VMS[instance.ID] = instance + daemon.Instances.Mu.Unlock() + + chain := []vm.InstanceState{ + vm.StateError, // terminate failed + vm.StateShuttingDown, // retry terminate + vm.StateTerminated, // success + } + + for _, target := range chain { + err := daemon.TransitionState(instance, target) + require.NoError(t, err, "transition to %s failed (current: %s)", target, instance.Status) + } + + assert.Equal(t, vm.StateTerminated, instance.Status) +} + +// TestTransitionState_ConcurrentTransitions hammers the same instance with +// concurrent transitions to verify the mutex protects against races and +// the instance always ends in a valid state. +func TestTransitionState_ConcurrentTransitions(t *testing.T) { + daemon := createDaemonWithJetStream(t) + + instance := &vm.VM{ID: "i-concurrent", Status: vm.StateRunning} + daemon.Instances.Mu.Lock() + daemon.Instances.VMS[instance.ID] = instance + daemon.Instances.Mu.Unlock() + + // Multiple goroutines race to transition from running. + // After one succeeds, the others may chain from the new state + // (e.g., running→error, then error→shutting-down), which is correct. + targets := []vm.InstanceState{vm.StateStopping, vm.StateShuttingDown, vm.StateError} + + var wg sync.WaitGroup + results := make([]error, len(targets)) + + for i, target := range targets { + wg.Add(1) + go func(idx int, tgt vm.InstanceState) { + defer wg.Done() + results[idx] = daemon.TransitionState(instance, tgt) + }(i, target) + } + wg.Wait() + + // At least one transition must succeed. + successes := 0 + for _, err := range results { + if err == nil { + successes++ + } + } + assert.GreaterOrEqual(t, successes, 1, "at least one concurrent transition should succeed") + + // The final state must be reachable from running via valid transitions. + finalState := instance.Status + reachable := reachableStates(vm.StateRunning) + assert.True(t, reachable[finalState], + "instance ended in state %q which is not reachable from running", finalState) +} + +// reachableStates returns all states reachable from start via BFS. +func reachableStates(start vm.InstanceState) map[vm.InstanceState]bool { + visited := map[vm.InstanceState]bool{start: true} + queue := []vm.InstanceState{start} + for len(queue) > 0 { + current := queue[0] + queue = queue[1:] + for _, next := range vm.ValidTransitions[current] { + if !visited[next] { + visited[next] = true + queue = append(queue, next) + } + } + } + return visited +} + +// TestTransitionState_DoubleTransitionSameTarget verifies that transitioning +// to the same state twice is rejected (no self-loops). +func TestTransitionState_DoubleTransitionSameTarget(t *testing.T) { + daemon := createDaemonWithJetStream(t) + + for _, state := range allStates { + t.Run(string(state), func(t *testing.T) { + instance := &vm.VM{ID: fmt.Sprintf("i-double-%s", state), Status: state} + daemon.Instances.Mu.Lock() + daemon.Instances.VMS[instance.ID] = instance + daemon.Instances.Mu.Unlock() + + err := daemon.TransitionState(instance, state) + assert.Error(t, err, "self-transition %s -> %s should be rejected", state, state) + assert.Equal(t, state, instance.Status, "state should be unchanged after rejected self-transition") + }) + } +} + +// TestTransitionState_InvalidFromTerminated exhaustively checks that +// terminated cannot transition to any state. +func TestTransitionState_InvalidFromTerminated(t *testing.T) { + daemon := createDaemonWithJetStream(t) + + for _, target := range allStates { + t.Run(fmt.Sprintf("terminated->%s", target), func(t *testing.T) { + instance := &vm.VM{ID: fmt.Sprintf("i-term-%s", target), Status: vm.StateTerminated} + daemon.Instances.Mu.Lock() + daemon.Instances.VMS[instance.ID] = instance + daemon.Instances.Mu.Unlock() + + err := daemon.TransitionState(instance, target) + assert.Error(t, err, "transition from terminated to %s should be rejected", target) + assert.Equal(t, vm.StateTerminated, instance.Status) + }) + } +} + +// TestTransitionState_StoppingMustReachStoppedOrError is the specific +// regression test for the bug where stopping never reached stopped. +// It verifies both the happy path and error path from stopping. +func TestTransitionState_StoppingMustReachStoppedOrError(t *testing.T) { + daemon := createDaemonWithJetStream(t) + + t.Run("happy_path", func(t *testing.T) { + instance := &vm.VM{ID: "i-stop-happy", Status: vm.StateStopping} + daemon.Instances.Mu.Lock() + daemon.Instances.VMS[instance.ID] = instance + daemon.Instances.Mu.Unlock() + + err := daemon.TransitionState(instance, vm.StateStopped) + require.NoError(t, err) + assert.Equal(t, vm.StateStopped, instance.Status) + }) + + t.Run("error_path", func(t *testing.T) { + instance := &vm.VM{ID: "i-stop-error", Status: vm.StateStopping} + daemon.Instances.Mu.Lock() + daemon.Instances.VMS[instance.ID] = instance + daemon.Instances.Mu.Unlock() + + err := daemon.TransitionState(instance, vm.StateError) + require.NoError(t, err) + assert.Equal(t, vm.StateError, instance.Status) + }) + + t.Run("force_terminate_path", func(t *testing.T) { + instance := &vm.VM{ID: "i-stop-term", Status: vm.StateStopping} + daemon.Instances.Mu.Lock() + daemon.Instances.VMS[instance.ID] = instance + daemon.Instances.Mu.Unlock() + + err := daemon.TransitionState(instance, vm.StateShuttingDown) + require.NoError(t, err) + assert.Equal(t, vm.StateShuttingDown, instance.Status) + + err = daemon.TransitionState(instance, vm.StateTerminated) + require.NoError(t, err) + assert.Equal(t, vm.StateTerminated, instance.Status) + }) +} + +// TestTransitionState_ShuttingDownMustReachTerminatedOrError is the parallel +// regression test for the terminate flow. +func TestTransitionState_ShuttingDownMustReachTerminatedOrError(t *testing.T) { + daemon := createDaemonWithJetStream(t) + + t.Run("happy_path", func(t *testing.T) { + instance := &vm.VM{ID: "i-shut-happy", Status: vm.StateShuttingDown} + daemon.Instances.Mu.Lock() + daemon.Instances.VMS[instance.ID] = instance + daemon.Instances.Mu.Unlock() + + err := daemon.TransitionState(instance, vm.StateTerminated) + require.NoError(t, err) + assert.Equal(t, vm.StateTerminated, instance.Status) + }) + + t.Run("error_path", func(t *testing.T) { + instance := &vm.VM{ID: "i-shut-error", Status: vm.StateShuttingDown} + daemon.Instances.Mu.Lock() + daemon.Instances.VMS[instance.ID] = instance + daemon.Instances.Mu.Unlock() + + err := daemon.TransitionState(instance, vm.StateError) + require.NoError(t, err) + assert.Equal(t, vm.StateError, instance.Status) + }) +} + +// TestValidTransitions_NoCyclesWithoutStableState ensures there is no cycle +// of only transitional states. Every cycle in the state graph must pass +// through at least one stable state. +func TestValidTransitions_NoCyclesWithoutStableState(t *testing.T) { + stableSet := map[vm.InstanceState]bool{} + for _, s := range stableStates { + stableSet[s] = true + } + + // DFS from each transitional state — if we revisit a transitional state + // without going through a stable state, that's a problem. + for _, start := range transitionalStates { + visited := map[vm.InstanceState]bool{} + var hasCycle bool + + var dfs func(vm.InstanceState) + dfs = func(current vm.InstanceState) { + if stableSet[current] { + return // reached a stable state, stop this path + } + if visited[current] { + hasCycle = true + return + } + visited[current] = true + for _, next := range vm.ValidTransitions[current] { + dfs(next) + } + } + dfs(start) + + assert.False(t, hasCycle, + "found a cycle of only transitional states starting from %q", start) + } +} + +// TestTransitionState_MultipleInstancesIndependent verifies that transitioning +// one instance does not affect another instance's state. +func TestTransitionState_MultipleInstancesIndependent(t *testing.T) { + daemon := createDaemonWithJetStream(t) + + inst1 := &vm.VM{ID: "i-multi-1", Status: vm.StateRunning} + inst2 := &vm.VM{ID: "i-multi-2", Status: vm.StateRunning} + daemon.Instances.Mu.Lock() + daemon.Instances.VMS[inst1.ID] = inst1 + daemon.Instances.VMS[inst2.ID] = inst2 + daemon.Instances.Mu.Unlock() + + err := daemon.TransitionState(inst1, vm.StateStopping) + require.NoError(t, err) + + assert.Equal(t, vm.StateStopping, inst1.Status) + assert.Equal(t, vm.StateRunning, inst2.Status, "inst2 should be unaffected by inst1's transition") +} + +// --- 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 +// running, isInstanceProcessRunning returns false for all instances, which +// exercises the recovery state resolution logic. + +// TestRestoreInstances_StoppingFinalizedToStopped verifies that an instance +// stuck in StateStopping when the daemon died gets finalized to StateStopped. +func TestRestoreInstances_StoppingFinalizedToStopped(t *testing.T) { + daemon := createDaemonWithJetStream(t) + + // Simulate: daemon was running, instance was in StateStopping, daemon crashed. + daemon.Instances.VMS["i-restore-stop"] = &vm.VM{ + ID: "i-restore-stop", + Status: vm.StateStopping, + } + require.NoError(t, daemon.WriteState()) + + // Clear in-memory state to simulate fresh daemon startup. + daemon.Instances.VMS = make(map[string]*vm.VM) + + // restoreInstances loads from JetStream and resolves transitional states. + daemon.restoreInstances() + + instance, ok := daemon.Instances.VMS["i-restore-stop"] + require.True(t, ok, "instance should be loaded from state") + assert.Equal(t, vm.StateStopped, instance.Status, + "stopping instance should be finalized to stopped on recovery") +} + +// TestRestoreInstances_ShuttingDownFinalizedToTerminated verifies that an +// instance stuck in StateShuttingDown gets finalized to StateTerminated. +func TestRestoreInstances_ShuttingDownFinalizedToTerminated(t *testing.T) { + daemon := createDaemonWithJetStream(t) + + daemon.Instances.VMS["i-restore-shut"] = &vm.VM{ + ID: "i-restore-shut", + Status: vm.StateShuttingDown, + } + require.NoError(t, daemon.WriteState()) + + daemon.Instances.VMS = make(map[string]*vm.VM) + daemon.restoreInstances() + + instance, ok := daemon.Instances.VMS["i-restore-shut"] + require.True(t, ok, "instance should be loaded from state") + assert.Equal(t, vm.StateTerminated, instance.Status, + "shutting-down instance should be finalized to terminated on recovery") +} + +// TestRestoreInstances_TerminatedSkipped verifies that terminated instances +// are loaded but not relaunched or modified. +func TestRestoreInstances_TerminatedSkipped(t *testing.T) { + daemon := createDaemonWithJetStream(t) + + daemon.Instances.VMS["i-restore-term"] = &vm.VM{ + ID: "i-restore-term", + Status: vm.StateTerminated, + } + require.NoError(t, daemon.WriteState()) + + daemon.Instances.VMS = make(map[string]*vm.VM) + daemon.restoreInstances() + + instance, ok := daemon.Instances.VMS["i-restore-term"] + require.True(t, ok, "terminated instance should still be loaded") + assert.Equal(t, vm.StateTerminated, instance.Status, + "terminated instance should remain terminated") +} + +// TestRestoreInstances_UserStoppedSkipped verifies that instances flagged as +// user-stopped are not relaunched. +func TestRestoreInstances_UserStoppedSkipped(t *testing.T) { + daemon := createDaemonWithJetStream(t) + + daemon.Instances.VMS["i-restore-userstop"] = &vm.VM{ + ID: "i-restore-userstop", + Status: vm.StateStopped, + Attributes: qmp.Attributes{ + StopInstance: true, + }, + } + require.NoError(t, daemon.WriteState()) + + daemon.Instances.VMS = make(map[string]*vm.VM) + daemon.restoreInstances() + + instance, ok := daemon.Instances.VMS["i-restore-userstop"] + require.True(t, ok, "user-stopped instance should still be loaded") + assert.Equal(t, vm.StateStopped, instance.Status, + "user-stopped instance should remain stopped") +} + +// TestRestoreInstances_RunningResetToPending verifies that a running instance +// whose QEMU process is dead gets reset to pending for relaunch. +// LaunchInstance will fail (no QEMU available in tests), but the state +// should be set to pending before the attempt. +func TestRestoreInstances_RunningResetToPending(t *testing.T) { + daemon := createDaemonWithJetStream(t) + + daemon.Instances.VMS["i-restore-run"] = &vm.VM{ + ID: "i-restore-run", + Status: vm.StateRunning, + } + require.NoError(t, daemon.WriteState()) + + daemon.Instances.VMS = make(map[string]*vm.VM) + daemon.restoreInstances() + + instance, ok := daemon.Instances.VMS["i-restore-run"] + require.True(t, ok, "running instance should still be loaded") + // LaunchInstance fails (no QEMU), so it stays pending or gets marked failed. + // Either way it should NOT remain "running" — that would be a lie. + assert.NotEqual(t, vm.StateRunning, instance.Status, + "instance should not remain running when QEMU process is dead") +} + +// TestRestoreInstances_MixedStates verifies recovery with multiple instances +// in different states, ensuring each is handled correctly and independently. +func TestRestoreInstances_MixedStates(t *testing.T) { + daemon := createDaemonWithJetStream(t) + + daemon.Instances.VMS["i-mix-stopping"] = &vm.VM{ + ID: "i-mix-stopping", Status: vm.StateStopping, + } + daemon.Instances.VMS["i-mix-shutting"] = &vm.VM{ + ID: "i-mix-shutting", Status: vm.StateShuttingDown, + } + daemon.Instances.VMS["i-mix-terminated"] = &vm.VM{ + ID: "i-mix-terminated", Status: vm.StateTerminated, + } + daemon.Instances.VMS["i-mix-stopped"] = &vm.VM{ + ID: "i-mix-stopped", Status: vm.StateStopped, + Attributes: qmp.Attributes{StopInstance: true}, + } + require.NoError(t, daemon.WriteState()) + + daemon.Instances.VMS = make(map[string]*vm.VM) + daemon.restoreInstances() + + assert.Equal(t, vm.StateStopped, daemon.Instances.VMS["i-mix-stopping"].Status, + "stopping should finalize to stopped") + assert.Equal(t, vm.StateTerminated, daemon.Instances.VMS["i-mix-shutting"].Status, + "shutting-down should finalize to terminated") + assert.Equal(t, vm.StateTerminated, daemon.Instances.VMS["i-mix-terminated"].Status, + "terminated should remain terminated") + assert.Equal(t, vm.StateStopped, daemon.Instances.VMS["i-mix-stopped"].Status, + "user-stopped should remain stopped") +} + +// TestRestoreInstances_StatePersistsAfterRecovery verifies that the finalized +// state is actually persisted to JetStream, so a second restart doesn't +// re-process the same transitional state. +func TestRestoreInstances_StatePersistsAfterRecovery(t *testing.T) { + daemon := createDaemonWithJetStream(t) + + daemon.Instances.VMS["i-persist-stop"] = &vm.VM{ + ID: "i-persist-stop", Status: vm.StateStopping, + } + daemon.Instances.VMS["i-persist-shut"] = &vm.VM{ + ID: "i-persist-shut", Status: vm.StateShuttingDown, + } + require.NoError(t, daemon.WriteState()) + + // First restart: finalize transitional states. + daemon.Instances.VMS = make(map[string]*vm.VM) + daemon.restoreInstances() + + assert.Equal(t, vm.StateStopped, daemon.Instances.VMS["i-persist-stop"].Status) + assert.Equal(t, vm.StateTerminated, daemon.Instances.VMS["i-persist-shut"].Status) + + // Second restart: load again, verify states are already finalized. + daemon.Instances.VMS = make(map[string]*vm.VM) + daemon.restoreInstances() + + assert.Equal(t, vm.StateStopped, daemon.Instances.VMS["i-persist-stop"].Status, + "stopped state should persist through second restart") + assert.Equal(t, vm.StateTerminated, daemon.Instances.VMS["i-persist-shut"].Status, + "terminated state should persist through second restart") +} + +// TestStatePersistence_RoundTrip verifies that all instance fields survive +// a write-load cycle through JetStream, simulating a daemon restart. +func TestStatePersistence_RoundTrip(t *testing.T) { + daemon := createDaemonWithJetStream(t) + + original := &vm.VM{ + ID: "i-roundtrip", + Status: vm.StateRunning, + InstanceType: "t3.micro", + Attributes: qmp.Attributes{ + StopInstance: false, + TerminateInstance: false, + }, + } + daemon.Instances.VMS[original.ID] = original + require.NoError(t, daemon.WriteState()) + + // Simulate restart: clear and reload. + daemon.Instances.VMS = make(map[string]*vm.VM) + require.NoError(t, daemon.LoadState()) + + loaded := daemon.Instances.VMS["i-roundtrip"] + require.NotNil(t, loaded) + assert.Equal(t, original.ID, loaded.ID) + assert.Equal(t, original.Status, loaded.Status) + assert.Equal(t, original.InstanceType, loaded.InstanceType) + assert.Equal(t, original.Attributes.StopInstance, loaded.Attributes.StopInstance) +} + +// TestRestoreInstances_StoppedInstanceCanReceiveStartCommand verifies that after +// a daemon restart, a stopped instance has an active ec2.cmd. subscription +// so the gateway's StartInstances can reach it (no "no responders" error). +func TestRestoreInstances_StoppedInstanceCanReceiveStartCommand(t *testing.T) { + daemon := createDaemonWithJetStream(t) + + instanceID := "i-restore-start" + daemon.Instances.VMS[instanceID] = &vm.VM{ + ID: instanceID, + InstanceType: "t3.micro", + Status: vm.StateStopped, + Attributes: qmp.Attributes{StopInstance: true}, + } + require.NoError(t, daemon.WriteState()) + + // Simulate daemon restart: clear in-memory state and restore from JetStream. + daemon.Instances.VMS = make(map[string]*vm.VM) + daemon.restoreInstances() + + // Instance should still be stopped. + instance, ok := daemon.Instances.VMS[instanceID] + require.True(t, ok, "stopped instance should be loaded after restore") + assert.Equal(t, vm.StateStopped, instance.Status) + + // Send a start command to the per-instance topic (ec2.cmd.). + // restoreInstances should have created this subscription for stopped instances. + command := qmp.Command{ + ID: instanceID, + QMPCommand: qmp.QMPCommand{ + Execute: "cont", + Arguments: map[string]any{}, + }, + Attributes: qmp.Attributes{ + StartInstance: true, + }, + } + cmdData, err := json.Marshal(command) + require.NoError(t, err) + + // The request should get a response (subscription exists), not "no responders". + // LaunchInstance will fail in tests (no QEMU), so we expect an error response, + // but the key assertion is that the subscription is active. + subject := fmt.Sprintf("ec2.cmd.%s", instanceID) + reply, err := daemon.natsConn.Request(subject, cmdData, 5*time.Second) + require.NoError(t, err, "ec2.cmd.%s should have a subscriber after restoreInstances", instanceID) + require.NotNil(t, reply) +}