mirror of
https://github.com/mulgadc/spinifex.git
synced 2026-10-23 04:36:40 +00:00
Julian Sommer
61e8eb9d91
During cluster restart, restoreInstances resets running instances (with dead QEMU) to pending for relaunch, but didn't reset LaunchTime. The pending watchdog (5min timeout) saw the stale LaunchTime from the original launch and immediately marked instances as failed/terminated, preventing any relaunch from succeeding.
988 lines
34 KiB
Go
988 lines
34 KiB
Go
package daemon
|
|
|
|
import (
|
|
"fmt"
|
|
"os"
|
|
"sync"
|
|
"testing"
|
|
"time"
|
|
|
|
"github.com/aws/aws-sdk-go/service/ec2"
|
|
"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"
|
|
"github.com/stretchr/testify/assert"
|
|
"github.com/stretchr/testify/require"
|
|
)
|
|
|
|
// createDaemonWithJetStream creates a daemon backed by an in-process NATS+JetStream server
|
|
// so that TransitionState (which calls WriteState) works end-to-end.
|
|
func createDaemonWithJetStream(t *testing.T) *Daemon {
|
|
t.Helper()
|
|
|
|
jsTmpDir, err := os.MkdirTemp("", "nats-js-state-test-*")
|
|
require.NoError(t, err)
|
|
t.Cleanup(func() { os.RemoveAll(jsTmpDir) })
|
|
|
|
opts := &server.Options{
|
|
Host: "127.0.0.1",
|
|
Port: -1,
|
|
JetStream: true,
|
|
StoreDir: jsTmpDir,
|
|
NoLog: true,
|
|
NoSigs: true,
|
|
}
|
|
ns, err := server.NewServer(opts)
|
|
require.NoError(t, err)
|
|
go ns.Start()
|
|
require.True(t, ns.ReadyForConnections(5*time.Second), "NATS server failed to start")
|
|
t.Cleanup(func() { ns.Shutdown() })
|
|
|
|
natsURL := ns.ClientURL()
|
|
|
|
tmpDir, err := os.MkdirTemp("", "hive-state-test-*")
|
|
require.NoError(t, err)
|
|
t.Cleanup(func() { os.RemoveAll(tmpDir) })
|
|
|
|
clusterCfg := &config.ClusterConfig{
|
|
Node: "node-1",
|
|
Nodes: map[string]config.Config{"node-1": {BaseDir: tmpDir}},
|
|
}
|
|
daemon := NewDaemon(clusterCfg)
|
|
daemon.config = &config.Config{BaseDir: tmpDir}
|
|
|
|
nc, err := nats.Connect(natsURL)
|
|
require.NoError(t, err)
|
|
t.Cleanup(func() { nc.Close() })
|
|
|
|
daemon.natsConn = nc
|
|
daemon.jsManager, err = NewJetStreamManager(nc, 1)
|
|
require.NoError(t, err)
|
|
require.NoError(t, daemon.jsManager.InitKVBucket())
|
|
require.NoError(t, daemon.jsManager.InitClusterStateBucket())
|
|
|
|
return daemon
|
|
}
|
|
|
|
func TestTransitionState_ValidTransitions(t *testing.T) {
|
|
daemon := createDaemonWithJetStream(t)
|
|
|
|
tests := []struct {
|
|
name string
|
|
from vm.InstanceState
|
|
to vm.InstanceState
|
|
}{
|
|
{"provisioning->running", vm.StateProvisioning, vm.StateRunning},
|
|
{"provisioning->error", vm.StateProvisioning, vm.StateError},
|
|
{"provisioning->shutting-down", vm.StateProvisioning, vm.StateShuttingDown},
|
|
{"pending->running", vm.StatePending, vm.StateRunning},
|
|
{"pending->error", vm.StatePending, vm.StateError},
|
|
{"pending->shutting-down", vm.StatePending, vm.StateShuttingDown},
|
|
{"running->stopping", vm.StateRunning, vm.StateStopping},
|
|
{"running->shutting-down", vm.StateRunning, vm.StateShuttingDown},
|
|
{"running->error", vm.StateRunning, vm.StateError},
|
|
{"stopping->stopped", vm.StateStopping, vm.StateStopped},
|
|
{"stopping->shutting-down", vm.StateStopping, vm.StateShuttingDown},
|
|
{"stopping->error", vm.StateStopping, vm.StateError},
|
|
{"stopped->running", vm.StateStopped, vm.StateRunning},
|
|
{"stopped->shutting-down", vm.StateStopped, vm.StateShuttingDown},
|
|
{"stopped->error", vm.StateStopped, vm.StateError},
|
|
{"shutting-down->terminated", vm.StateShuttingDown, vm.StateTerminated},
|
|
{"shutting-down->error", vm.StateShuttingDown, vm.StateError},
|
|
{"error->running", vm.StateError, vm.StateRunning},
|
|
{"error->shutting-down", vm.StateError, vm.StateShuttingDown},
|
|
}
|
|
|
|
for _, tt := range tests {
|
|
t.Run(tt.name, func(t *testing.T) {
|
|
instance := &vm.VM{
|
|
ID: "i-test-valid",
|
|
Status: tt.from,
|
|
}
|
|
|
|
daemon.Instances.Mu.Lock()
|
|
daemon.Instances.VMS[instance.ID] = instance
|
|
daemon.Instances.Mu.Unlock()
|
|
|
|
err := daemon.TransitionState(instance, tt.to)
|
|
require.NoError(t, err)
|
|
|
|
assert.Equal(t, tt.to, instance.Status)
|
|
})
|
|
}
|
|
}
|
|
|
|
func TestTransitionState_InvalidTransitions(t *testing.T) {
|
|
daemon := createDaemonWithJetStream(t)
|
|
|
|
tests := []struct {
|
|
name string
|
|
from vm.InstanceState
|
|
to vm.InstanceState
|
|
}{
|
|
{"running->running", vm.StateRunning, vm.StateRunning},
|
|
{"running->pending", vm.StateRunning, vm.StatePending},
|
|
{"running->stopped", vm.StateRunning, vm.StateStopped},
|
|
{"running->terminated", vm.StateRunning, vm.StateTerminated},
|
|
{"stopped->stopping", vm.StateStopped, vm.StateStopping},
|
|
{"stopped->terminated", vm.StateStopped, vm.StateTerminated},
|
|
{"terminated->running", vm.StateTerminated, vm.StateRunning},
|
|
{"terminated->stopped", vm.StateTerminated, vm.StateStopped},
|
|
{"stopping->running", vm.StateStopping, vm.StateRunning},
|
|
{"shutting-down->running", vm.StateShuttingDown, vm.StateRunning},
|
|
{"shutting-down->stopped", vm.StateShuttingDown, vm.StateStopped},
|
|
}
|
|
|
|
for _, tt := range tests {
|
|
t.Run(tt.name, func(t *testing.T) {
|
|
instance := &vm.VM{
|
|
ID: "i-test-invalid",
|
|
Status: tt.from,
|
|
}
|
|
|
|
daemon.Instances.Mu.Lock()
|
|
daemon.Instances.VMS[instance.ID] = instance
|
|
daemon.Instances.Mu.Unlock()
|
|
|
|
err := daemon.TransitionState(instance, tt.to)
|
|
assert.Error(t, err)
|
|
assert.Contains(t, err.Error(), "invalid state transition")
|
|
|
|
// Status should remain unchanged
|
|
assert.Equal(t, tt.from, instance.Status)
|
|
})
|
|
}
|
|
}
|
|
|
|
func TestTransitionState_NilEC2Instance(t *testing.T) {
|
|
daemon := createDaemonWithJetStream(t)
|
|
|
|
instance := &vm.VM{
|
|
ID: "i-test-nil-ec2",
|
|
Status: vm.StateProvisioning,
|
|
Instance: nil, // no EC2 instance metadata
|
|
}
|
|
|
|
daemon.Instances.Mu.Lock()
|
|
daemon.Instances.VMS[instance.ID] = instance
|
|
daemon.Instances.Mu.Unlock()
|
|
|
|
err := daemon.TransitionState(instance, vm.StateRunning)
|
|
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.
|
|
|
|
// 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.
|
|
func simulateCleanRestore(t *testing.T, daemon *Daemon) {
|
|
t.Helper()
|
|
require.NoError(t, daemon.jsManager.WriteShutdownMarker(daemon.node))
|
|
daemon.restoreInstances()
|
|
}
|
|
|
|
// TestRestoreInstances_StoppingFinalizedToStopped verifies that an instance
|
|
// stuck in StateStopping when the daemon died gets finalized to StateStopped
|
|
// and migrated to shared KV.
|
|
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.
|
|
simulateCleanRestore(t, daemon)
|
|
|
|
// Stopped instances are migrated to shared KV and removed from local map
|
|
_, ok := daemon.Instances.VMS["i-restore-stop"]
|
|
assert.False(t, ok, "stopping→stopped instance should be migrated to shared KV, not in local map")
|
|
|
|
stoppedInst, err := daemon.jsManager.LoadStoppedInstance("i-restore-stop")
|
|
require.NoError(t, err)
|
|
require.NotNil(t, stoppedInst, "stopping→stopped instance should exist in shared KV")
|
|
assert.Equal(t, vm.StateStopped, stoppedInst.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)
|
|
simulateCleanRestore(t, daemon)
|
|
|
|
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)
|
|
simulateCleanRestore(t, daemon)
|
|
|
|
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_UserStoppedMigratedToSharedKV verifies that instances
|
|
// flagged as user-stopped are migrated to shared KV and not relaunched.
|
|
func TestRestoreInstances_UserStoppedMigratedToSharedKV(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)
|
|
simulateCleanRestore(t, daemon)
|
|
|
|
// Stopped instances should be migrated to shared KV
|
|
_, ok := daemon.Instances.VMS["i-restore-userstop"]
|
|
assert.False(t, ok, "user-stopped instance should be migrated to shared KV, not in local map")
|
|
|
|
stoppedInst, err := daemon.jsManager.LoadStoppedInstance("i-restore-userstop")
|
|
require.NoError(t, err)
|
|
require.NotNil(t, stoppedInst, "user-stopped instance should exist in shared KV")
|
|
assert.Equal(t, vm.StateStopped, stoppedInst.Status,
|
|
"user-stopped instance should remain stopped in shared KV")
|
|
}
|
|
|
|
// 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)
|
|
simulateCleanRestore(t, daemon)
|
|
|
|
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_RunningResetsLaunchTime verifies that when a running
|
|
// instance is reset to pending for relaunch, its LaunchTime is updated to now.
|
|
// Without this, the pending watchdog sees the stale LaunchTime and immediately
|
|
// marks the instance as failed.
|
|
func TestRestoreInstances_RunningResetsLaunchTime(t *testing.T) {
|
|
daemon := createDaemonWithJetStream(t)
|
|
|
|
staleTime := time.Now().Add(-30 * time.Minute)
|
|
daemon.Instances.VMS["i-stale-launch"] = &vm.VM{
|
|
ID: "i-stale-launch",
|
|
Status: vm.StateRunning,
|
|
Instance: &ec2.Instance{
|
|
LaunchTime: &staleTime,
|
|
},
|
|
}
|
|
require.NoError(t, daemon.WriteState())
|
|
|
|
daemon.Instances.VMS = make(map[string]*vm.VM)
|
|
before := time.Now()
|
|
simulateCleanRestore(t, daemon)
|
|
|
|
instance, ok := daemon.Instances.VMS["i-stale-launch"]
|
|
require.True(t, ok, "instance should be loaded from state")
|
|
require.NotNil(t, instance.Instance)
|
|
require.NotNil(t, instance.Instance.LaunchTime)
|
|
|
|
// LaunchTime should be reset to approximately now, not the stale 30min-ago time
|
|
assert.True(t, instance.Instance.LaunchTime.After(before) || instance.Instance.LaunchTime.Equal(before),
|
|
"LaunchTime should be reset to now, got %v (stale was %v)", *instance.Instance.LaunchTime, staleTime)
|
|
assert.True(t, time.Since(*instance.Instance.LaunchTime) < pendingWatchdogTimeout,
|
|
"LaunchTime should be within watchdog timeout window")
|
|
}
|
|
|
|
// TestRestoreInstances_MixedStates verifies recovery with multiple instances
|
|
// in different states, ensuring each is handled correctly and independently.
|
|
// Stopped instances (including stopping→stopped transitions) are migrated to shared KV.
|
|
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)
|
|
simulateCleanRestore(t, daemon)
|
|
|
|
// Stopped instances are migrated to shared KV and removed from local map
|
|
assert.Nil(t, daemon.Instances.VMS["i-mix-stopping"],
|
|
"stopping→stopped should be migrated to shared KV")
|
|
stoppedFromStopping, err := daemon.jsManager.LoadStoppedInstance("i-mix-stopping")
|
|
require.NoError(t, err)
|
|
require.NotNil(t, stoppedFromStopping, "stopping→stopped should exist in shared KV")
|
|
assert.Equal(t, vm.StateStopped, stoppedFromStopping.Status)
|
|
|
|
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.Nil(t, daemon.Instances.VMS["i-mix-stopped"],
|
|
"user-stopped should be migrated to shared KV")
|
|
stoppedFromUser, err := daemon.jsManager.LoadStoppedInstance("i-mix-stopped")
|
|
require.NoError(t, err)
|
|
require.NotNil(t, stoppedFromUser, "user-stopped should exist in shared KV")
|
|
assert.Equal(t, vm.StateStopped, stoppedFromUser.Status)
|
|
}
|
|
|
|
// 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)
|
|
simulateCleanRestore(t, daemon)
|
|
|
|
// stopping→stopped migrates to shared KV
|
|
assert.Nil(t, daemon.Instances.VMS["i-persist-stop"],
|
|
"stopping→stopped should be migrated to shared KV")
|
|
stoppedInst, err := daemon.jsManager.LoadStoppedInstance("i-persist-stop")
|
|
require.NoError(t, err)
|
|
require.NotNil(t, stoppedInst)
|
|
assert.Equal(t, vm.StateStopped, stoppedInst.Status)
|
|
|
|
assert.Equal(t, vm.StateTerminated, daemon.Instances.VMS["i-persist-shut"].Status)
|
|
|
|
// Second restart: stopped instance is in shared KV (not per-node state),
|
|
// so local map won't have it. Terminated instance persists in per-node state.
|
|
daemon.Instances.VMS = make(map[string]*vm.VM)
|
|
simulateCleanRestore(t, daemon)
|
|
|
|
// Stopped instance should still be in shared KV
|
|
stoppedInst2, err := daemon.jsManager.LoadStoppedInstance("i-persist-stop")
|
|
require.NoError(t, err)
|
|
require.NotNil(t, stoppedInst2, "stopped instance should persist in shared KV through second restart")
|
|
assert.Equal(t, vm.StateStopped, stoppedInst2.Status)
|
|
|
|
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_StoppedInstanceMigratedToSharedKV verifies that after
|
|
// a daemon restart, a stopped instance is migrated to shared KV and can be
|
|
// retrieved by the ec2.start handler.
|
|
func TestRestoreInstances_StoppedInstanceMigratedToSharedKV(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)
|
|
simulateCleanRestore(t, daemon)
|
|
|
|
// Instance should not be in local map (migrated to shared KV).
|
|
_, ok := daemon.Instances.VMS[instanceID]
|
|
assert.False(t, ok, "stopped instance should be migrated to shared KV, not in local map")
|
|
|
|
// Instance should be in shared KV
|
|
stoppedInst, err := daemon.jsManager.LoadStoppedInstance(instanceID)
|
|
require.NoError(t, err)
|
|
require.NotNil(t, stoppedInst, "stopped instance should exist in shared KV")
|
|
assert.Equal(t, vm.StateStopped, stoppedInst.Status)
|
|
assert.Equal(t, "node-1", stoppedInst.LastNode,
|
|
"stopped instance should record last node")
|
|
}
|
|
|
|
// TestRestoreInstances_UnknownTypeMigratedToStoppedWithReason verifies that an
|
|
// instance whose type is not in the local type map gets moved to stopped with
|
|
// an InsufficientInstanceCapacity state reason and migrated to shared KV.
|
|
func TestRestoreInstances_UnknownTypeMigratedToStoppedWithReason(t *testing.T) {
|
|
daemon := createDaemonWithJetStream(t)
|
|
|
|
instanceID := "i-restore-notype"
|
|
daemon.Instances.VMS[instanceID] = &vm.VM{
|
|
ID: instanceID,
|
|
InstanceType: "m7i.small", // type that won't exist on this host
|
|
Status: vm.StateRunning,
|
|
Instance: &ec2.Instance{},
|
|
}
|
|
require.NoError(t, daemon.WriteState())
|
|
|
|
// Simulate daemon restart
|
|
daemon.Instances.VMS = make(map[string]*vm.VM)
|
|
simulateCleanRestore(t, daemon)
|
|
|
|
// Instance should not be in local map (migrated to shared KV)
|
|
_, ok := daemon.Instances.VMS[instanceID]
|
|
assert.False(t, ok, "instance with unknown type should be migrated to shared KV")
|
|
|
|
// Instance should be stopped in shared KV with capacity reason
|
|
stoppedInst, err := daemon.jsManager.LoadStoppedInstance(instanceID)
|
|
require.NoError(t, err)
|
|
require.NotNil(t, stoppedInst, "instance should exist in shared KV")
|
|
assert.Equal(t, vm.StateStopped, stoppedInst.Status)
|
|
require.NotNil(t, stoppedInst.Instance.StateReason)
|
|
assert.Equal(t, "Server.InsufficientInstanceCapacity", *stoppedInst.Instance.StateReason.Code)
|
|
assert.Contains(t, *stoppedInst.Instance.StateReason.Message, "m7i.small")
|
|
}
|
|
|
|
// TestPendingWatchdog_MarksStuckInstanceFailed verifies that the pending watchdog
|
|
// detects instances stuck in pending beyond the timeout and marks them failed.
|
|
func TestPendingWatchdog_MarksStuckInstanceFailed(t *testing.T) {
|
|
daemon := createDaemonWithJetStream(t)
|
|
|
|
staleTime := time.Now().Add(-6 * time.Minute)
|
|
recentTime := time.Now()
|
|
|
|
daemon.Instances.VMS["i-stuck"] = &vm.VM{
|
|
ID: "i-stuck",
|
|
Status: vm.StatePending,
|
|
Instance: &ec2.Instance{
|
|
LaunchTime: &staleTime,
|
|
},
|
|
}
|
|
daemon.Instances.VMS["i-fresh"] = &vm.VM{
|
|
ID: "i-fresh",
|
|
Status: vm.StatePending,
|
|
Instance: &ec2.Instance{
|
|
LaunchTime: &recentTime,
|
|
},
|
|
}
|
|
|
|
// Run one watchdog tick manually instead of waiting for the ticker
|
|
daemon.Instances.Mu.Lock()
|
|
var stuck []*vm.VM
|
|
for _, instance := range daemon.Instances.VMS {
|
|
if (instance.Status == vm.StatePending || instance.Status == vm.StateProvisioning) &&
|
|
instance.Instance != nil && instance.Instance.LaunchTime != nil &&
|
|
time.Since(*instance.Instance.LaunchTime) > 5*time.Minute {
|
|
stuck = append(stuck, instance)
|
|
}
|
|
}
|
|
daemon.Instances.Mu.Unlock()
|
|
|
|
for _, instance := range stuck {
|
|
daemon.markInstanceFailed(instance, "launch_timeout")
|
|
}
|
|
|
|
// Stuck instance should have transitioned to shutting-down
|
|
stuckInst := daemon.Instances.VMS["i-stuck"]
|
|
assert.Equal(t, vm.StateShuttingDown, stuckInst.Status)
|
|
require.NotNil(t, stuckInst.Instance.StateReason)
|
|
assert.Equal(t, "Server.InternalError", *stuckInst.Instance.StateReason.Code)
|
|
assert.Equal(t, "launch_timeout", *stuckInst.Instance.StateReason.Message)
|
|
|
|
// Fresh instance should still be pending
|
|
freshInst := daemon.Instances.VMS["i-fresh"]
|
|
assert.Equal(t, vm.StatePending, freshInst.Status)
|
|
}
|