mirror of
https://github.com/mulgadc/spinifex.git
synced 2026-10-15 17:11:30 +00:00
fix: daemon reboot
This commit is contained in:
+19
-7
@@ -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
|
||||
|
||||
@@ -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.<id> 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.<id>).
|
||||
// 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)
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user