diff --git a/spinifex/daemon/daemon.go b/spinifex/daemon/daemon.go index 7a6b7ad90..9217b54dd 100644 --- a/spinifex/daemon/daemon.go +++ b/spinifex/daemon/daemon.go @@ -1360,13 +1360,6 @@ func (d *Daemon) setupShutdown() { }) } -// pendingWatchdogTimeout aliases the manager's exported watchdog timeout so -// daemon-side tests (currently asserting LaunchTime is within the watchdog -// window after restore) keep working without reaching into the vm package. -// -// Phase 2f cleanup: migrate the assertion into vm/ and delete the alias. -const pendingWatchdogTimeout = vm.PendingWatchdogTimeout - // respondWithVolumeAttachment builds an ec2.VolumeAttachment, marshals it to JSON, and // responds on the NATS message. Used by both AttachVolume and DetachVolume handlers. func (d *Daemon) respondWithVolumeAttachment(msg *nats.Msg, volumeID, instanceID, device, state string) { diff --git a/spinifex/daemon/daemon_test.go b/spinifex/daemon/daemon_test.go index 1924841b0..e0e8e1c70 100644 --- a/spinifex/daemon/daemon_test.go +++ b/spinifex/daemon/daemon_test.go @@ -3095,57 +3095,6 @@ func TestNewDaemon_WalDirPreservedIfSet(t *testing.T) { assert.Equal(t, "/fast-ssd/wal", d.config.WalDir) } -// TestMarkInstanceFailed verifies that MarkFailed sets the StateReason and -// transitions the instance to ShuttingDown synchronously, then completes -// the cleanup chain to Terminated in a goroutine. -func TestMarkInstanceFailed(t *testing.T) { - daemon := createDaemonWithJetStream(t) - - instanceID := "i-test-mark-failed" - ec2Instance := &ec2.Instance{} - ec2Instance.SetInstanceId(instanceID) - - instance := &vm.VM{ - ID: instanceID, - Status: vm.StatePending, - AccountID: testAccountID, - Instance: ec2Instance, - } - daemon.vmMgr.Insert(instance) - - daemon.vmMgr.MarkFailed(instance, "volume_preparation_failed") - - // Synchronous: state reason set, transition to shutting-down done. - require.NotNil(t, instance.Instance.StateReason) - assert.Equal(t, "Server.InternalError", *instance.Instance.StateReason.Code) - assert.Equal(t, "volume_preparation_failed", *instance.Instance.StateReason.Message) - require.Eventually(t, func() bool { - return daemon.vmMgr.Status(instance) == vm.StateTerminated - }, 5*time.Second, 10*time.Millisecond, "cleanup goroutine should reach terminated") -} - -// TestMarkInstanceFailed_NilInstance verifies that MarkFailed handles -// a VM with no ec2.Instance (Instance == nil) gracefully. -func TestMarkInstanceFailed_NilInstance(t *testing.T) { - daemon := createDaemonWithJetStream(t) - - instanceID := "i-test-mark-failed-nil" - instance := &vm.VM{ - ID: instanceID, - Status: vm.StatePending, - AccountID: testAccountID, - Instance: nil, // no ec2.Instance - } - daemon.vmMgr.Insert(instance) - - // Should not panic - daemon.vmMgr.MarkFailed(instance, "test_failure") - - require.Eventually(t, func() bool { - return daemon.vmMgr.Status(instance) == vm.StateTerminated - }, 5*time.Second, 10*time.Millisecond, "cleanup goroutine should reach terminated") -} - // TestVolumeMounterAdapter_UnmountOne_Success verifies that the adapter's // UnmountOne sends an ebs.unmount NATS request and handles a successful // response. UnmountOne is the post-2d successor to Daemon.rollbackEBSMount; diff --git a/spinifex/daemon/health.go b/spinifex/daemon/health.go deleted file mode 100644 index 50909f416..000000000 --- a/spinifex/daemon/health.go +++ /dev/null @@ -1,46 +0,0 @@ -package daemon - -import ( - "time" - - "github.com/mulgadc/spinifex/spinifex/vm" -) - -// classifyCrashReason is preserved as a daemon-side shim so the existing -// classify-crash-reason tests keep their bare-function call shape while -// the implementation moved into the vm package alongside the manager. -// -// Phase 2f cleanup: migrate TestClassifyCrashReason_* into vm/ and delete. -func classifyCrashReason(waitErr error) string { - return vm.ClassifyCrashReason(waitErr) -} - -// Daemon-side aliases for vm package crash-recovery constants. Tests in -// health_test.go reference these by their pre-2e bare names. Phase 2f -// cleanup migrates those tests into vm/ and removes the aliases. -const ( - maxRestartsInWindow = vm.MaxRestartsInWindow - restartWindow = vm.RestartWindow -) - -// restartBackoff is a daemon-side shim mirroring classifyCrashReason; the -// pure implementation lives on vm so the manager's MaybeRestart can use -// it without a daemon import. -func restartBackoff(restartCount int) time.Duration { - return vm.RestartBackoff(restartCount) -} - -// handleInstanceCrash and maybeRestartInstance are daemon-side shims -// around the manager methods that own crash recovery. Pre-2f tests still -// invoke them through the daemon receiver; the bodies live in -// vm/crash_recovery.go. -// -// Phase 2f cleanup: migrate the daemon health_test.go cases into vm/ and -// delete the shims. -func (d *Daemon) handleInstanceCrash(instance *vm.VM, waitErr error) { - d.vmMgr.HandleCrash(instance, waitErr) -} - -func (d *Daemon) maybeRestartInstance(instance *vm.VM) { - d.vmMgr.MaybeRestart(instance) -} diff --git a/spinifex/daemon/health_test.go b/spinifex/daemon/health_test.go deleted file mode 100644 index 856cfa4f0..000000000 --- a/spinifex/daemon/health_test.go +++ /dev/null @@ -1,483 +0,0 @@ -package daemon - -import ( - "fmt" - "os" - "os/exec" - "path/filepath" - "testing" - "time" - - "github.com/mulgadc/spinifex/spinifex/config" - "github.com/mulgadc/spinifex/spinifex/vm" - "github.com/nats-io/nats.go" - "github.com/stretchr/testify/assert" - "github.com/stretchr/testify/require" -) - -func TestClassifyCrashReason_CleanExit(t *testing.T) { - assert.Equal(t, "clean-exit", classifyCrashReason(nil)) -} - -func TestClassifyCrashReason_OOMKill(t *testing.T) { - // Start a process and kill it with SIGKILL to get a real ExitError - cmd := exec.Command("sleep", "60") - require.NoError(t, cmd.Start()) - - require.NoError(t, cmd.Process.Kill()) // sends SIGKILL - waitErr := cmd.Wait() - require.Error(t, waitErr) - - assert.Equal(t, "oom-killed", classifyCrashReason(waitErr)) -} - -func TestClassifyCrashReason_ExitCode(t *testing.T) { - // Run a command that exits with a non-zero code - cmd := exec.Command("sh", "-c", "exit 42") - waitErr := cmd.Run() - require.Error(t, waitErr) - - assert.Equal(t, "exit-42", classifyCrashReason(waitErr)) -} - -func TestClassifyCrashReason_Unknown(t *testing.T) { - // A non-ExitError error - err := fmt.Errorf("some random error") - assert.Equal(t, "unknown", classifyCrashReason(err)) -} - -func TestHandleInstanceCrash_SkipsNonRunning(t *testing.T) { - nc, err := nats.Connect(sharedNATSURL) - require.NoError(t, err) - defer nc.Close() - - rm, err := NewResourceManager() - require.NoError(t, err) - - d := &Daemon{ - natsConn: nc, - resourceMgr: rm, - vmMgr: vm.NewManager(), - } - - instance := &vm.VM{ - ID: "i-test-stopped", - Status: vm.StateStopped, - } - d.vmMgr.Insert(instance) - - // Should return without action (no panic, no state change) - d.handleInstanceCrash(instance, fmt.Errorf("test error")) - assert.Equal(t, vm.StateStopped, instance.Status) -} - -func TestHandleInstanceCrash_SkipsShuttingDown(t *testing.T) { - d, cleanup := newTestDaemon(t) - defer cleanup() - d.shuttingDown.Store(true) - - instance := &vm.VM{ - ID: "i-test-shutdown", - Status: vm.StateRunning, - } - d.vmMgr.Insert(instance) - - d.handleInstanceCrash(instance, fmt.Errorf("test error")) - - // Status unchanged because shuttingDown is true - assert.Equal(t, vm.StateRunning, instance.Status) -} - -func TestMaybeRestart_ExceedsMaxInWindow(t *testing.T) { - nc, err := nats.Connect(sharedNATSURL) - require.NoError(t, err) - defer nc.Close() - - rm, err := NewResourceManager() - require.NoError(t, err) - - d := &Daemon{ - node: "test-node", - natsConn: nc, - config: &config.Config{ - Services: []string{"daemon"}, - }, - resourceMgr: rm, - vmMgr: vm.NewManager(), - } - - // Find an allocatable instance type - var allocType string - for typeName := range d.resourceMgr.instanceTypes { - allocType = typeName - break - } - require.NotEmpty(t, allocType) - - instance := &vm.VM{ - ID: "i-test-maxcrash", - Status: vm.StateError, - InstanceType: allocType, - Health: vm.InstanceHealthState{ - CrashCount: maxRestartsInWindow + 1, // exceeds limit - FirstCrashTime: time.Now(), - RestartCount: maxRestartsInWindow, - }, - } - d.vmMgr.Insert(instance) - - // Should not schedule a restart (no panic, instance stays in error) - d.maybeRestartInstance(instance) - - // Instance should remain in error state - assert.Equal(t, vm.StateError, instance.Status) -} - -func TestMaybeRestart_ResetsAfterWindow(t *testing.T) { - nc, err := nats.Connect(sharedNATSURL) - require.NoError(t, err) - defer nc.Close() - - rm, err := NewResourceManager() - require.NoError(t, err) - - d := &Daemon{ - node: "test-node", - natsConn: nc, - config: &config.Config{ - Services: []string{"daemon"}, - }, - resourceMgr: rm, - vmMgr: vm.NewManager(), - } - - // Find an allocatable instance type - var allocType string - for typeName := range d.resourceMgr.instanceTypes { - allocType = typeName - break - } - require.NotEmpty(t, allocType) - - instance := &vm.VM{ - ID: "i-test-windowreset", - Status: vm.StateError, - InstanceType: allocType, - Health: vm.InstanceHealthState{ - CrashCount: 5, - FirstCrashTime: time.Now().Add(-restartWindow - time.Minute), // outside window - RestartCount: 5, - }, - } - d.vmMgr.Insert(instance) - - // maybeRestartInstance should reset counters and schedule a restart - // (restart will fail because there's no real QEMU, but counters should reset) - d.maybeRestartInstance(instance) - - // Counters should be reset - assert.Equal(t, 1, instance.Health.CrashCount) - assert.Equal(t, 0, instance.Health.RestartCount) -} - -func TestRestartBackoff_Exponential(t *testing.T) { - tests := []struct { - restartCount int - want time.Duration - }{ - {0, 5 * time.Second}, - {1, 10 * time.Second}, - {2, 20 * time.Second}, - {3, 40 * time.Second}, - {4, 80 * time.Second}, - {5, 120 * time.Second}, // capped at 2min - {6, 120 * time.Second}, // stays capped - {100, 120 * time.Second}, // large count stays capped - } - - for _, tc := range tests { - got := restartBackoff(tc.restartCount) - assert.Equal(t, tc.want, got, "Backoff mismatch at restart count %d", tc.restartCount) - } -} - -// newTestDaemon creates a Daemon with shared NATS and a ResourceManager for health tests. -// jsManager is nil — WriteState will error (logged, not fatal). -func newTestDaemon(t *testing.T) (*Daemon, func()) { - t.Helper() - nc, err := nats.Connect(sharedNATSURL) - require.NoError(t, err) - - rm, err := NewResourceManager() - require.NoError(t, err) - - d := &Daemon{ - node: "test-node", - natsConn: nc, - config: &config.Config{ - Services: []string{"daemon"}, - }, - resourceMgr: rm, - vmMgr: vm.NewManager(), - } - // Wire the dependencies the manager-internal crash handler relies on - // (TransitionState, ResourceController, InstanceTypes, ShutdownSignal). - // jsManager is still nil so TransitionState's WriteState will fail — - // matching the pre-2e shape where the in-memory status flip survives a - // write failure. - d.vmMgr.SetDeps(vm.Deps{ - NodeID: d.node, - TransitionState: d.TransitionState, - Resources: newResourceControllerAdapter(rm), - InstanceTypes: newInstanceTypeResolverAdapter(rm), - ShutdownSignal: d.shuttingDown.Load, - }) - return d, func() { nc.Close() } -} - -// smallestAllocType returns the name of the smallest allocatable instance type -// in the resource manager, considering both vCPU and memory to ensure it fits -// on the current host (important for CI runners with limited resources). -func smallestAllocType(t *testing.T, rm *ResourceManager) string { - t.Helper() - var smallest string - var smallestScore int64 = 1<<63 - 1 - for name, it := range rm.instanceTypes { - vcpu := instanceTypeVCPUs(it) - memMiB := instanceTypeMemoryMiB(it) - // Score by total resource footprint so we pick the type smallest in both dimensions - score := vcpu*1024 + memMiB - if score < smallestScore { - smallestScore = score - smallest = name - } - } - if smallest == "" { - t.Fatal("resource manager has no instance types") - } - return smallest -} - -func TestHandleInstanceCrash_CoreFlow(t *testing.T) { - d, cleanup := newTestDaemon(t) - defer cleanup() - - allocType := smallestAllocType(t, d.resourceMgr) - instanceType := d.resourceMgr.instanceTypes[allocType] - require.NoError(t, d.resourceMgr.allocate(instanceType)) - - // Record resource levels before crash - d.resourceMgr.mu.RLock() - allocVCPUBefore := d.resourceMgr.allocatedVCPU - allocMemBefore := d.resourceMgr.allocatedMem - d.resourceMgr.mu.RUnlock() - require.Greater(t, allocVCPUBefore, 0) - - // Create a temp QMP socket file to verify removal - tmpDir := t.TempDir() - qmpPath := filepath.Join(tmpDir, "qmp.sock") - require.NoError(t, os.WriteFile(qmpPath, []byte("dummy"), 0o600)) - - instance := &vm.VM{ - ID: "i-test-core", - Status: vm.StateRunning, - Running: true, - PID: 12345, - InstanceType: allocType, - Config: vm.Config{QMPSocket: qmpPath}, - } - d.vmMgr.Insert(instance) - - d.handleInstanceCrash(instance, fmt.Errorf("test crash")) - - // Status transitioned to error - assert.Equal(t, vm.StateError, instance.Status) - - // Health tracking updated - assert.Equal(t, 1, instance.Health.CrashCount) - assert.False(t, instance.Health.LastCrashTime.IsZero()) - assert.Equal(t, "unknown", instance.Health.LastCrashReason) // fmt.Errorf is not *exec.ExitError - assert.False(t, instance.Health.FirstCrashTime.IsZero()) - - // Running and PID cleared - assert.False(t, instance.Running) - assert.Equal(t, 0, instance.PID) - - // Resources deallocated - d.resourceMgr.mu.RLock() - assert.Less(t, d.resourceMgr.allocatedVCPU, allocVCPUBefore) - assert.Less(t, d.resourceMgr.allocatedMem, allocMemBefore) - d.resourceMgr.mu.RUnlock() - - // QMP socket file removed - _, err := os.Stat(qmpPath) - assert.True(t, os.IsNotExist(err), "QMP socket should be removed") -} - -func TestHandleInstanceCrash_FirstCrashSetsTime(t *testing.T) { - d, cleanup := newTestDaemon(t) - defer cleanup() - - allocType := smallestAllocType(t, d.resourceMgr) - - instance := &vm.VM{ - ID: "i-test-firstcrash", - Status: vm.StateRunning, - Running: true, - PID: 111, - InstanceType: allocType, - } - d.vmMgr.Insert(instance) - - // First crash - d.handleInstanceCrash(instance, fmt.Errorf("crash 1")) - firstTime := instance.Health.FirstCrashTime - assert.False(t, firstTime.IsZero()) - assert.Equal(t, 1, instance.Health.CrashCount) - - // Reset to running for a second crash - d.vmMgr.UpdateState(instance.ID, func(v *vm.VM) { - v.Status = vm.StateRunning - v.Running = true - v.PID = 222 - }) - - time.Sleep(time.Millisecond) // ensure time advances - d.handleInstanceCrash(instance, fmt.Errorf("crash 2")) - - // FirstCrashTime should be unchanged from the first crash - assert.Equal(t, firstTime, instance.Health.FirstCrashTime) - assert.Equal(t, 2, instance.Health.CrashCount) -} - -func TestHandleInstanceCrash_UnknownInstanceType(t *testing.T) { - d, cleanup := newTestDaemon(t) - defer cleanup() - - instance := &vm.VM{ - ID: "i-test-unknown-type", - Status: vm.StateRunning, - Running: true, - PID: 333, - InstanceType: "z99.nonexistent", - } - d.vmMgr.Insert(instance) - - // Should not panic even though instance type is not in resourceMgr - d.handleInstanceCrash(instance, fmt.Errorf("crash")) - - assert.Equal(t, vm.StateError, instance.Status) - assert.Equal(t, 1, instance.Health.CrashCount) - assert.False(t, instance.Running) - assert.Equal(t, 0, instance.PID) -} - -func TestMaybeRestart_SkipsShuttingDown(t *testing.T) { - d, cleanup := newTestDaemon(t) - defer cleanup() - d.shuttingDown.Store(true) - - allocType := smallestAllocType(t, d.resourceMgr) - - instance := &vm.VM{ - ID: "i-test-restart-shutdown", - Status: vm.StateError, - InstanceType: allocType, - Health: vm.InstanceHealthState{ - CrashCount: 1, - FirstCrashTime: time.Now(), - }, - } - d.vmMgr.Insert(instance) - - d.maybeRestartInstance(instance) - - // RestartCount should not have changed — restart was skipped - assert.Equal(t, 0, instance.Health.RestartCount) -} - -func TestMaybeRestart_UnknownInstanceType(t *testing.T) { - d, cleanup := newTestDaemon(t) - defer cleanup() - - instance := &vm.VM{ - ID: "i-test-restart-unknown", - Status: vm.StateError, - InstanceType: "z99.nonexistent", - Health: vm.InstanceHealthState{ - CrashCount: 1, - FirstCrashTime: time.Now(), - }, - } - d.vmMgr.Insert(instance) - - // Should not panic — just log and return - d.maybeRestartInstance(instance) - - // No restart scheduled - assert.Equal(t, 0, instance.Health.RestartCount) -} - -func TestMaybeRestart_InsufficientResources(t *testing.T) { - d, cleanup := newTestDaemon(t) - defer cleanup() - - allocType := smallestAllocType(t, d.resourceMgr) - - // Exhaust all schedulable resources (host - reserved) so canAllocate - // returns 0, matching what's actually achievable at runtime. - d.resourceMgr.mu.Lock() - d.resourceMgr.allocatedVCPU = d.resourceMgr.hostVCPU - d.resourceMgr.reservedVCPU - d.resourceMgr.allocatedMem = d.resourceMgr.hostMemGB - d.resourceMgr.reservedMem - d.resourceMgr.mu.Unlock() - - instance := &vm.VM{ - ID: "i-test-restart-nores", - Status: vm.StateError, - InstanceType: allocType, - Health: vm.InstanceHealthState{ - CrashCount: 1, - FirstCrashTime: time.Now(), - }, - } - d.vmMgr.Insert(instance) - - d.maybeRestartInstance(instance) - - // No restart scheduled - assert.Equal(t, 0, instance.Health.RestartCount) -} - -func TestMaybeRestart_SchedulesRestart(t *testing.T) { - d, cleanup := newTestDaemon(t) - defer cleanup() - - allocType := smallestAllocType(t, d.resourceMgr) - - instance := &vm.VM{ - ID: "i-test-restart-schedule", - Status: vm.StateError, - InstanceType: allocType, - Health: vm.InstanceHealthState{ - CrashCount: 1, - FirstCrashTime: time.Now(), - RestartCount: 0, - }, - } - d.vmMgr.Insert(instance) - - // maybeRestartInstance should pass all guards and call time.AfterFunc. - // The scheduled restartCrashedInstance will increment RestartCount and - // attempt LaunchInstance (which will fail without real infra — that's fine). - d.maybeRestartInstance(instance) - - // The restart is scheduled via time.AfterFunc with 5s delay. - // We verify that the scheduling code path was reached by checking that - // the function didn't bail out early (all early returns leave RestartCount at 0 - // and don't call time.AfterFunc). The counter reset branch does not apply since - // FirstCrashTime is recent. If we got here without panic, the scheduling path ran. - // We can't easily assert the timer fired without waiting 5s, but we can - // verify the counters were not reset (proving we didn't take the window-expired path). - assert.Equal(t, 1, instance.Health.CrashCount) - assert.Equal(t, 0, instance.Health.RestartCount) // incremented by restartCrashedInstance, not maybeRestart -} diff --git a/spinifex/daemon/state_test.go b/spinifex/daemon/state_test.go index 1e366ee1f..1bf93f818 100644 --- a/spinifex/daemon/state_test.go +++ b/spinifex/daemon/state_test.go @@ -736,7 +736,7 @@ func TestRestoreInstances_RunningResetsLaunchTime(t *testing.T) { // 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, + assert.True(t, time.Since(*instance.Instance.LaunchTime) < vm.PendingWatchdogTimeout, "LaunchTime should be within watchdog timeout window") } diff --git a/spinifex/daemon/vm_adapters.go b/spinifex/daemon/vm_adapters.go index 9019be1a0..f1832cc8e 100644 --- a/spinifex/daemon/vm_adapters.go +++ b/spinifex/daemon/vm_adapters.go @@ -525,7 +525,7 @@ func (d *Daemon) buildVMManagerDeps() vm.Deps { OnInstanceRecovering: d.onInstanceRecoveringHook(), }, ShutdownSignal: d.shuttingDown.Load, - CrashHandler: d.handleInstanceCrash, + CrashHandler: d.vmMgr.HandleCrash, TransitionState: d.TransitionState, DevNetworking: d.config.Daemon.DevNetworking, BindHost: d.config.Host, diff --git a/spinifex/vm/crash_recovery_test.go b/spinifex/vm/crash_recovery_test.go new file mode 100644 index 000000000..41dfc278e --- /dev/null +++ b/spinifex/vm/crash_recovery_test.go @@ -0,0 +1,406 @@ +package vm + +import ( + "errors" + "fmt" + "os" + "os/exec" + "path/filepath" + "sync/atomic" + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func TestClassifyCrashReason_CleanExit(t *testing.T) { + assert.Equal(t, "clean-exit", ClassifyCrashReason(nil)) +} + +func TestClassifyCrashReason_OOMKill(t *testing.T) { + cmd := exec.Command("sleep", "60") + require.NoError(t, cmd.Start()) + require.NoError(t, cmd.Process.Kill()) + waitErr := cmd.Wait() + require.Error(t, waitErr) + assert.Equal(t, "oom-killed", ClassifyCrashReason(waitErr)) +} + +func TestClassifyCrashReason_ExitCode(t *testing.T) { + cmd := exec.Command("sh", "-c", "exit 42") + waitErr := cmd.Run() + require.Error(t, waitErr) + assert.Equal(t, "exit-42", ClassifyCrashReason(waitErr)) +} + +func TestClassifyCrashReason_Unknown(t *testing.T) { + err := fmt.Errorf("some random error") + assert.Equal(t, "unknown", ClassifyCrashReason(err)) +} + +func TestRestartBackoff_Exponential(t *testing.T) { + tests := []struct { + restartCount int + want time.Duration + }{ + {0, 5 * time.Second}, + {1, 10 * time.Second}, + {2, 20 * time.Second}, + {3, 40 * time.Second}, + {4, 80 * time.Second}, + {5, 120 * time.Second}, + {6, 120 * time.Second}, + {100, 120 * time.Second}, + } + + for _, tc := range tests { + got := RestartBackoff(tc.restartCount) + assert.Equal(t, tc.want, got, "Backoff mismatch at restart count %d", tc.restartCount) + } +} + +// crashTestManager builds a Manager wired with the dependencies most crash +// and watchdog tests need: a recording resource controller, a static +// instance-type resolver, a recording transition state, and a controllable +// shutdown signal. +func crashTestManager(t *testing.T) (m *Manager, rc *fakeResourceController, rt *recordedTransitions, shuttingDown *atomic.Bool) { + t.Helper() + rc = newFakeResourceController() + rt = &recordedTransitions{} + shuttingDown = &atomic.Bool{} + m = NewManager() + rt.bind(m) + m.SetDeps(Deps{ + NodeID: "test-node", + Resources: rc, + InstanceTypes: fakeInstanceTypeResolver{ + "t3.micro": {VCPUs: 1, MemoryMiB: 1024, Architecture: "x86_64"}, + }, + TransitionState: rt.apply, + ShutdownSignal: shuttingDown.Load, + }) + return m, rc, rt, shuttingDown +} + +func TestHandleCrash_SkipsNonRunning(t *testing.T) { + m, rc, _, _ := crashTestManager(t) + + instance := &VM{ + ID: "i-stopped", + Status: StateStopped, + InstanceType: "t3.micro", + } + m.Insert(instance) + + m.HandleCrash(instance, fmt.Errorf("test error")) + + assert.Equal(t, StateStopped, m.Status(instance), "non-running instance must not transition on crash") + assert.Zero(t, rc.deallocateCount("t3.micro"), "non-running instance must not deallocate") +} + +func TestHandleCrash_SkipsShuttingDown(t *testing.T) { + m, rc, _, shuttingDown := crashTestManager(t) + shuttingDown.Store(true) + + instance := &VM{ + ID: "i-shutdown", + Status: StateRunning, + InstanceType: "t3.micro", + } + m.Insert(instance) + + m.HandleCrash(instance, fmt.Errorf("test error")) + + assert.Equal(t, StateRunning, m.Status(instance), "shutdown skip must leave status untouched") + assert.Zero(t, rc.deallocateCount("t3.micro"), "shutdown skip must not deallocate") +} + +func TestHandleCrash_CoreFlow(t *testing.T) { + m, rc, rt, _ := crashTestManager(t) + require.NoError(t, rc.Allocate("t3.micro")) + require.Equal(t, 1, rc.allocateCount("t3.micro")) + + tmpDir := t.TempDir() + qmpPath := filepath.Join(tmpDir, "qmp.sock") + require.NoError(t, os.WriteFile(qmpPath, []byte("dummy"), 0o600)) + + instance := &VM{ + ID: "i-core", + Status: StateRunning, + Running: true, + PID: 12345, + InstanceType: "t3.micro", + Config: Config{QMPSocket: qmpPath}, + } + m.Insert(instance) + + m.HandleCrash(instance, fmt.Errorf("test crash")) + + assert.Equal(t, StateError, m.Status(instance), "crash must transition to Error") + targets := rt.targets("i-core") + require.NotEmpty(t, targets, "TransitionState must be invoked") + assert.Equal(t, StateError, targets[0]) + + assert.Equal(t, 1, instance.Health.CrashCount) + assert.False(t, instance.Health.LastCrashTime.IsZero()) + assert.Equal(t, "unknown", instance.Health.LastCrashReason) + assert.False(t, instance.Health.FirstCrashTime.IsZero()) + + assert.False(t, instance.Running, "Running must be cleared") + assert.Zero(t, instance.PID, "PID must be cleared") + + assert.Equal(t, 1, rc.deallocateCount("t3.micro"), "crashed instance resources must be deallocated") + + _, err := os.Stat(qmpPath) + assert.True(t, os.IsNotExist(err), "QMP socket must be removed") +} + +func TestHandleCrash_FirstCrashSetsTime(t *testing.T) { + m, _, _, _ := crashTestManager(t) + + instance := &VM{ + ID: "i-firstcrash", + Status: StateRunning, + Running: true, + PID: 111, + InstanceType: "t3.micro", + } + m.Insert(instance) + + m.HandleCrash(instance, fmt.Errorf("crash 1")) + firstTime := instance.Health.FirstCrashTime + assert.False(t, firstTime.IsZero()) + assert.Equal(t, 1, instance.Health.CrashCount) + + m.UpdateState(instance.ID, func(v *VM) { + v.Status = StateRunning + v.Running = true + v.PID = 222 + }) + + time.Sleep(time.Millisecond) + m.HandleCrash(instance, fmt.Errorf("crash 2")) + + assert.Equal(t, firstTime, instance.Health.FirstCrashTime, + "FirstCrashTime must remain set to the first crash") + assert.Equal(t, 2, instance.Health.CrashCount) +} + +// TestHandleCrash_UnknownInstanceType verifies a crash on a VM with an +// instance type not in the resolver still completes the state transition, +// health bookkeeping, and pid/running clear — even though MaybeRestart will +// later refuse to restart it. +func TestHandleCrash_UnknownInstanceType(t *testing.T) { + m, _, _, _ := crashTestManager(t) + + instance := &VM{ + ID: "i-unknown", + Status: StateRunning, + Running: true, + PID: 333, + InstanceType: "z99.nonexistent", + } + m.Insert(instance) + + m.HandleCrash(instance, fmt.Errorf("crash")) + + assert.Equal(t, StateError, m.Status(instance)) + assert.Equal(t, 1, instance.Health.CrashCount) + assert.False(t, instance.Running) + assert.Zero(t, instance.PID) +} + +func TestMaybeRestart_ExceedsMaxInWindow(t *testing.T) { + m, _, _, _ := crashTestManager(t) + + instance := &VM{ + ID: "i-maxcrash", + Status: StateError, + InstanceType: "t3.micro", + Health: InstanceHealthState{ + CrashCount: MaxRestartsInWindow + 1, + FirstCrashTime: time.Now(), + RestartCount: MaxRestartsInWindow, + }, + } + m.Insert(instance) + + m.MaybeRestart(instance) + + assert.Equal(t, StateError, m.Status(instance), "exhausted-window instance must remain in error") + assert.Equal(t, MaxRestartsInWindow, instance.Health.RestartCount, + "RestartCount must not increment when exceeded") +} + +func TestMaybeRestart_ResetsAfterWindow(t *testing.T) { + m, _, _, _ := crashTestManager(t) + + instance := &VM{ + ID: "i-windowreset", + Status: StateError, + InstanceType: "t3.micro", + Health: InstanceHealthState{ + CrashCount: 5, + FirstCrashTime: time.Now().Add(-RestartWindow - time.Minute), + RestartCount: 5, + }, + } + m.Insert(instance) + + m.MaybeRestart(instance) + + assert.Equal(t, 1, instance.Health.CrashCount, + "crash count must reset to 1 once the window expires") + assert.Equal(t, 0, instance.Health.RestartCount, + "restart count must reset to 0 once the window expires") +} + +func TestMaybeRestart_SkipsShuttingDown(t *testing.T) { + m, _, _, shuttingDown := crashTestManager(t) + shuttingDown.Store(true) + + instance := &VM{ + ID: "i-restart-shutdown", + Status: StateError, + InstanceType: "t3.micro", + Health: InstanceHealthState{ + CrashCount: 1, + FirstCrashTime: time.Now(), + }, + } + m.Insert(instance) + + m.MaybeRestart(instance) + + assert.Equal(t, 0, instance.Health.RestartCount, "shutdown skip must not increment restart") +} + +func TestMaybeRestart_UnknownInstanceType(t *testing.T) { + m, _, _, _ := crashTestManager(t) + + instance := &VM{ + ID: "i-restart-unknown", + Status: StateError, + InstanceType: "z99.nonexistent", + Health: InstanceHealthState{ + CrashCount: 1, + FirstCrashTime: time.Now(), + }, + } + m.Insert(instance) + + m.MaybeRestart(instance) + + assert.Equal(t, 0, instance.Health.RestartCount, + "unknown instance type must not schedule a restart") +} + +func TestMaybeRestart_InsufficientResources(t *testing.T) { + m, rc, _, _ := crashTestManager(t) + rc.canAllocateRet = 0 + + instance := &VM{ + ID: "i-restart-nores", + Status: StateError, + InstanceType: "t3.micro", + Health: InstanceHealthState{ + CrashCount: 1, + FirstCrashTime: time.Now(), + }, + } + m.Insert(instance) + + m.MaybeRestart(instance) + + assert.Equal(t, 0, instance.Health.RestartCount, + "insufficient resources must not schedule a restart") +} + +// TestMaybeRestart_SchedulesRestart asserts the all-guards-pass branch +// reaches the time.AfterFunc scheduling path. We can't easily wait the +// full backoff, but the absence of an early-return signal (counter reset +// or RestartCount mutation in MaybeRestart itself) tells us scheduling +// happened. +func TestMaybeRestart_SchedulesRestart(t *testing.T) { + m, _, _, _ := crashTestManager(t) + + instance := &VM{ + ID: "i-restart-schedule", + Status: StateError, + InstanceType: "t3.micro", + Health: InstanceHealthState{ + CrashCount: 1, + FirstCrashTime: time.Now(), + RestartCount: 0, + }, + } + m.Insert(instance) + + m.MaybeRestart(instance) + + assert.Equal(t, 1, instance.Health.CrashCount, + "counter reset path must not run when window is still open") + assert.Equal(t, 0, instance.Health.RestartCount, + "MaybeRestart must not increment RestartCount itself") +} + +// TestRestartCrashedInstance_NotInError verifies the wrong-state guard +// short-circuits before touching the resource controller. +func TestRestartCrashedInstance_NotInError(t *testing.T) { + m, rc, _, _ := crashTestManager(t) + + instance := &VM{ + ID: "i-not-error", + Status: StateRunning, + InstanceType: "t3.micro", + } + m.Insert(instance) + + m.RestartCrashedInstance(instance) + + assert.Equal(t, 0, rc.allocateCount("t3.micro"), + "non-error instance must not be re-allocated") + assert.Equal(t, StateRunning, m.Status(instance)) +} + +// TestRestartCrashedInstance_ShuttingDown verifies the shutdown guard +// inside RestartCrashedInstance — the late check between scheduling and +// relaunch — short-circuits before touching the resource controller. +func TestRestartCrashedInstance_ShuttingDown(t *testing.T) { + m, rc, _, shuttingDown := crashTestManager(t) + shuttingDown.Store(true) + + instance := &VM{ + ID: "i-shutdown-restart", + Status: StateError, + InstanceType: "t3.micro", + } + m.Insert(instance) + + m.RestartCrashedInstance(instance) + + assert.Equal(t, 0, rc.allocateCount("t3.micro"), + "shutdown-guarded restart must not allocate resources") + assert.Equal(t, StateError, m.Status(instance)) +} + +// TestRestartCrashedInstance_AllocateFailure verifies the allocator-error +// branch leaves the instance in StateError without invoking Run. +func TestRestartCrashedInstance_AllocateFailure(t *testing.T) { + m, rc, _, _ := crashTestManager(t) + rc.allocateErr = errors.New("no capacity") + + instance := &VM{ + ID: "i-alloc-fail", + Status: StateError, + InstanceType: "t3.micro", + } + m.Insert(instance) + + m.RestartCrashedInstance(instance) + + assert.Equal(t, StateError, m.Status(instance)) + assert.Equal(t, 1, instance.Health.RestartCount, + "RestartCount is incremented before the allocate attempt") +} diff --git a/spinifex/vm/fakes_test.go b/spinifex/vm/fakes_test.go new file mode 100644 index 000000000..429c40863 --- /dev/null +++ b/spinifex/vm/fakes_test.go @@ -0,0 +1,189 @@ +package vm + +import "sync" + +// fakeResourceController records Allocate/Deallocate calls so lifecycle +// and crash-recovery tests can assert deallocation happened (or didn't) +// without standing up the daemon's real ResourceManager. +type fakeResourceController struct { + mu sync.Mutex + allocated map[string]int + deallocated []string + allocateErr error + canAllocateRet int +} + +func newFakeResourceController() *fakeResourceController { + return &fakeResourceController{allocated: map[string]int{}, canAllocateRet: 100} +} + +func (f *fakeResourceController) Allocate(typeName string) error { + f.mu.Lock() + defer f.mu.Unlock() + if f.allocateErr != nil { + return f.allocateErr + } + f.allocated[typeName]++ + return nil +} + +func (f *fakeResourceController) Deallocate(typeName string) { + f.mu.Lock() + defer f.mu.Unlock() + f.deallocated = append(f.deallocated, typeName) + if f.allocated[typeName] > 0 { + f.allocated[typeName]-- + } +} + +func (f *fakeResourceController) CanAllocate(typeName string, count int) int { + f.mu.Lock() + defer f.mu.Unlock() + if f.canAllocateRet < count { + return f.canAllocateRet + } + return count +} + +func (f *fakeResourceController) deallocateCount(typeName string) int { + f.mu.Lock() + defer f.mu.Unlock() + n := 0 + for _, t := range f.deallocated { + if t == typeName { + n++ + } + } + return n +} + +func (f *fakeResourceController) allocateCount(typeName string) int { + f.mu.Lock() + defer f.mu.Unlock() + return f.allocated[typeName] +} + +var _ ResourceController = (*fakeResourceController)(nil) + +// fakeInstanceTypeResolver is a static map satisfying InstanceTypeResolver. +type fakeInstanceTypeResolver map[string]InstanceTypeSpec + +func (f fakeInstanceTypeResolver) Resolve(name string) (InstanceTypeSpec, bool) { + spec, ok := f[name] + return spec, ok +} + +var _ InstanceTypeResolver = fakeInstanceTypeResolver(nil) + +// recordedTransition is one entry in recordedTransitions.calls. +type recordedTransition struct { + ID string + Target InstanceState +} + +// recordedTransitions is a TransitionState seam that records every call so +// crash and shutdown tests can assert transition order. Writes to v.Status +// go through Manager.Inspect to match production's lock discipline; without +// it the goroutine MarkFailed spawns races with concurrent m.Status reads. +type recordedTransitions struct { + mu sync.Mutex + m *Manager + calls []recordedTransition + err error +} + +// bind associates the recorder with the manager whose lock guards v.Status +// reads. Must be called once before SetDeps installs the apply method. +func (r *recordedTransitions) bind(m *Manager) *recordedTransitions { + r.mu.Lock() + r.m = m + r.mu.Unlock() + return r +} + +func (r *recordedTransitions) apply(v *VM, target InstanceState) error { + r.mu.Lock() + if r.err != nil { + err := r.err + r.mu.Unlock() + return err + } + r.calls = append(r.calls, recordedTransition{ID: v.ID, Target: target}) + m := r.m + r.mu.Unlock() + if m != nil { + m.Inspect(v, func(vv *VM) { vv.Status = target }) + } else { + v.Status = target + } + return nil +} + +func (r *recordedTransitions) snapshot() []recordedTransition { + r.mu.Lock() + defer r.mu.Unlock() + out := make([]recordedTransition, len(r.calls)) + copy(out, r.calls) + return out +} + +func (r *recordedTransitions) targets(id string) []InstanceState { + r.mu.Lock() + defer r.mu.Unlock() + var out []InstanceState + for _, c := range r.calls { + if c.ID == id { + out = append(out, c.Target) + } + } + return out +} + +// recordingInstanceCleaner counts every InstanceCleaner method invocation so +// tests can assert e.g. "Stop must not call DeleteVolumes". +type recordingInstanceCleaner struct { + mu sync.Mutex + deleteVolumes []string + cleanupMgmt []string + releasePublicIP []string + detachAndDeleteENI []string + removeFromPlacement []string +} + +func (c *recordingInstanceCleaner) DeleteVolumes(v *VM) { + c.mu.Lock() + defer c.mu.Unlock() + c.deleteVolumes = append(c.deleteVolumes, v.ID) +} + +func (c *recordingInstanceCleaner) CleanupMgmtNetwork(v *VM) { + c.mu.Lock() + defer c.mu.Unlock() + c.cleanupMgmt = append(c.cleanupMgmt, v.ID) +} + +func (c *recordingInstanceCleaner) ReleasePublicIP(v *VM) { + c.mu.Lock() + defer c.mu.Unlock() + c.releasePublicIP = append(c.releasePublicIP, v.ID) +} + +func (c *recordingInstanceCleaner) DetachAndDeleteENI(v *VM) { + c.mu.Lock() + defer c.mu.Unlock() + c.detachAndDeleteENI = append(c.detachAndDeleteENI, v.ID) +} + +func (c *recordingInstanceCleaner) RemoveFromPlacementGroup(v *VM) { + c.mu.Lock() + defer c.mu.Unlock() + c.removeFromPlacement = append(c.removeFromPlacement, v.ID) +} + +func (c *recordingInstanceCleaner) deleteVolumesCount() int { + c.mu.Lock() + defer c.mu.Unlock() + return len(c.deleteVolumes) +} + +var _ InstanceCleaner = (*recordingInstanceCleaner)(nil) diff --git a/spinifex/vm/manager_deps_test.go b/spinifex/vm/manager_deps_test.go index c593c0367..7e251346b 100644 --- a/spinifex/vm/manager_deps_test.go +++ b/spinifex/vm/manager_deps_test.go @@ -3,6 +3,7 @@ package vm import ( "errors" "maps" + "sync" "testing" "github.com/mulgadc/spinifex/spinifex/types" @@ -136,7 +137,10 @@ func TestManagerHooks_InitiallyNil(t *testing.T) { } // fakeVolumeMounter records every call so lifecycle tests can assert ordering. +// The mutex covers the recording slices so StopAll's per-instance fan-out +// goroutines stay race-free. type fakeVolumeMounter struct { + mu sync.Mutex mounted, unmounted []string mountedOne, unmountedOne []string mountErr error @@ -145,28 +149,39 @@ type fakeVolumeMounter struct { } func (f *fakeVolumeMounter) Mount(v *VM) error { + f.mu.Lock() f.mounted = append(f.mounted, v.ID) - return f.mountErr + err := f.mountErr + f.mu.Unlock() + return err } func (f *fakeVolumeMounter) Unmount(v *VM) error { + f.mu.Lock() f.unmounted = append(f.unmounted, v.ID) + f.mu.Unlock() return nil } func (f *fakeVolumeMounter) MountOne(req *types.EBSRequest) error { + f.mu.Lock() f.mountedOne = append(f.mountedOne, req.Name) - if f.mountOneErr != nil { - return f.mountOneErr + mountOneErr := f.mountOneErr + uri := f.mountOneURI + f.mu.Unlock() + if mountOneErr != nil { + return mountOneErr } - if f.mountOneURI != "" { - req.NBDURI = f.mountOneURI + if uri != "" { + req.NBDURI = uri } return nil } func (f *fakeVolumeMounter) UnmountOne(req types.EBSRequest) { + f.mu.Lock() f.unmountedOne = append(f.unmountedOne, req.Name) + f.mu.Unlock() } var _ VolumeMounter = (*fakeVolumeMounter)(nil) diff --git a/spinifex/vm/shutdown_test.go b/spinifex/vm/shutdown_test.go new file mode 100644 index 000000000..316ab19e5 --- /dev/null +++ b/spinifex/vm/shutdown_test.go @@ -0,0 +1,391 @@ +package vm + +import ( + "errors" + "sync" + "sync/atomic" + "testing" + "time" + + "github.com/aws/aws-sdk-go/service/ec2" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// shutdownTestManager wires the dependencies needed to exercise Stop / +// StopAll / Terminate / MarkFailed without standing up the daemon. The +// returned cleaner records every method invocation so tests can assert +// what cleanup ran (and what didn't). +func shutdownTestManager(t *testing.T) (m *Manager, store *fakeStateStore, mounter *fakeVolumeMounter, cleaner *recordingInstanceCleaner, rt *recordedTransitions) { + t.Helper() + store = newFakeStateStore() + mounter = &fakeVolumeMounter{} + cleaner = &recordingInstanceCleaner{} + rt = &recordedTransitions{} + m = NewManager() + rt.bind(m) + m.SetDeps(Deps{ + NodeID: "test-node", + StateStore: store, + VolumeMounter: mounter, + InstanceCleaner: cleaner, + TransitionState: rt.apply, + ShutdownSignal: func() bool { return false }, + }) + return m, store, mounter, cleaner, rt +} + +// TestMarkFailed_TransitionsToTerminated verifies the synchronous + +// asynchronous parts of MarkFailed: the synchronous transition to +// ShuttingDown plus the StateReason mutation, and the goroutine driving +// the instance through terminateCleanup → Terminated. +func TestMarkFailed_TransitionsToTerminated(t *testing.T) { + m, _, _, _, rt := shutdownTestManager(t) + + instance := &VM{ + ID: "i-mark-failed", + Status: StatePending, + AccountID: "111122223333", + Instance: &ec2.Instance{}, + } + m.Insert(instance) + + m.MarkFailed(instance, "volume_preparation_failed") + + require.NotNil(t, instance.Instance.StateReason, + "MarkFailed must populate StateReason synchronously") + assert.Equal(t, "Server.InternalError", *instance.Instance.StateReason.Code) + assert.Equal(t, "volume_preparation_failed", *instance.Instance.StateReason.Message) + + require.Eventually(t, func() bool { + return m.Status(instance) == StateTerminated + }, 2*time.Second, 5*time.Millisecond, "cleanup goroutine must reach Terminated") + + targets := rt.targets("i-mark-failed") + require.NotEmpty(t, targets) + assert.Equal(t, StateShuttingDown, targets[0], + "first transition must be ShuttingDown") + assert.Contains(t, targets, StateTerminated, + "terminal transition must reach Terminated") +} + +// TestMarkFailed_NilInstance verifies MarkFailed tolerates a VM with no +// embedded ec2.Instance (Instance == nil) without panicking. +func TestMarkFailed_NilInstance(t *testing.T) { + m, _, _, _, _ := shutdownTestManager(t) + + instance := &VM{ + ID: "i-mark-failed-nil", + Status: StatePending, + AccountID: "111122223333", + Instance: nil, + } + m.Insert(instance) + + require.NotPanics(t, func() { + m.MarkFailed(instance, "test_failure") + }) + + require.Eventually(t, func() bool { + return m.Status(instance) == StateTerminated + }, 2*time.Second, 5*time.Millisecond, "cleanup goroutine must reach Terminated") +} + +// TestMarkFailed_AlreadyShuttingDown_NoOp verifies MarkFailed skips its +// work when the instance is already past pending — the existing cleanup +// goroutine owns the transition. +func TestMarkFailed_AlreadyShuttingDown_NoOp(t *testing.T) { + m, _, _, cleaner, rt := shutdownTestManager(t) + + instance := &VM{ + ID: "i-already-down", + Status: StateShuttingDown, + Instance: &ec2.Instance{}, + } + m.Insert(instance) + + m.MarkFailed(instance, "duplicate_call") + + assert.Empty(t, rt.snapshot(), + "MarkFailed must not transition an already-shutting-down instance") + assert.Nil(t, instance.Instance.StateReason, + "MarkFailed must not overwrite StateReason on already-shutting-down instance") + assert.Zero(t, cleaner.deleteVolumesCount(), + "MarkFailed must not run cleanup on already-shutting-down instance") +} + +func TestMarkFailed_AlreadyTerminated_NoOp(t *testing.T) { + m, _, _, _, rt := shutdownTestManager(t) + + instance := &VM{ + ID: "i-terminated", + Status: StateTerminated, + Instance: &ec2.Instance{}, + } + m.Insert(instance) + + m.MarkFailed(instance, "duplicate_call") + + assert.Empty(t, rt.snapshot(), + "MarkFailed must not transition an already-terminated instance") +} + +// TestStop_DoesNotCallDeleteVolumes locks down the architectural +// invariant that Stop must never delete volumes — a regression here +// would silently destroy user data on every stop. Phase 2c removed +// TestStopInstance_NoDelete_OnStop from daemon_test.go without a +// vm-package replacement; this is that replacement. +func TestStop_DoesNotCallDeleteVolumes(t *testing.T) { + m, _, _, cleaner, _ := shutdownTestManager(t) + + instance := &VM{ + ID: "i-stop-nodelete", + Status: StateRunning, + InstanceType: "t3.micro", + Instance: &ec2.Instance{}, + } + m.Insert(instance) + + require.NoError(t, m.Stop(instance.ID)) + + assert.Zero(t, cleaner.deleteVolumesCount(), + "Manager.Stop must never invoke InstanceCleaner.DeleteVolumes") +} + +// TestStopAll_DoesNotCallDeleteVolumes mirrors the Stop invariant for the +// fan-out path used by coordinated shutdown / SIGTERM. +func TestStopAll_DoesNotCallDeleteVolumes(t *testing.T) { + m, _, _, cleaner, _ := shutdownTestManager(t) + + for i := range 3 { + m.Insert(&VM{ + ID: string(rune('a' + i)), + Status: StateRunning, + InstanceType: "t3.micro", + Instance: &ec2.Instance{}, + }) + } + + require.NoError(t, m.StopAll()) + + assert.Zero(t, cleaner.deleteVolumesCount(), + "Manager.StopAll must never invoke InstanceCleaner.DeleteVolumes") +} + +func TestStopAll_EmptyMap_FastPath(t *testing.T) { + m, _, mounter, cleaner, rt := shutdownTestManager(t) + + require.NoError(t, m.StopAll()) + + assert.Empty(t, rt.snapshot(), "empty StopAll must not invoke transitions") + assert.Empty(t, mounter.unmounted, "empty StopAll must not invoke unmount") + assert.Empty(t, cleaner.cleanupMgmt, "empty StopAll must not invoke cleaner") +} + +// TestStopAll_DoesNotFireOnInstanceDown verifies StopAll's fan-out +// shutdown path — used by coordinated DRAIN and SIGTERM — leaves the +// hook contract alone. Per the plan's hook contract, Stop fires +// OnInstanceDown but StopAll does not (it leaves instances in the +// running map for restoreInstances to pick up on next boot). +func TestStopAll_DoesNotFireOnInstanceDown(t *testing.T) { + m, _, _, _, _ := shutdownTestManager(t) + var down atomic.Int64 + rt := (&recordedTransitions{}).bind(m) + m.SetDeps(Deps{ + NodeID: "test-node", + TransitionState: rt.apply, + ShutdownSignal: func() bool { return false }, + Hooks: ManagerHooks{ + OnInstanceDown: func(string) { down.Add(1) }, + }, + }) + + for _, id := range []string{"i-1", "i-2", "i-3"} { + m.Insert(&VM{ + ID: id, + Status: StateRunning, + InstanceType: "t3.micro", + Instance: &ec2.Instance{}, + }) + } + + require.NoError(t, m.StopAll()) + + assert.Zero(t, down.Load(), + "StopAll must not fire OnInstanceDown — instances stay in the running map") + assert.Equal(t, 3, m.Count(), + "StopAll must leave instances in the local map for restoreInstances") +} + +// TestMigrateStoppedToSharedKV_SlotReclaim covers the DeleteIf-mismatch +// branch: another handler reclaimed the slot under the same id while the +// shared write was in flight, so the local delete must not happen and +// the function must report false. +func TestMigrateStoppedToSharedKV_SlotReclaim(t *testing.T) { + m, store, _, _, _ := shutdownTestManager(t) + + original := &VM{ID: "i-reclaim", Status: StateStopped} + reclaimed := &VM{ID: "i-reclaim", Status: StateRunning} + m.Insert(original) + // Replace the slot under the same id — DeleteIf(original) must miss. + m.Insert(reclaimed) + + got := m.MigrateStoppedToSharedKV(original) + require.False(t, got, "MigrateStoppedToSharedKV must return false when the slot was reclaimed") + + // The shared KV write still happened (writeFn fires before DeleteIf). + stored, _ := store.LoadStoppedInstance("i-reclaim") + assert.NotNil(t, stored, "shared KV write must precede the slot check") + + // The reclaimed instance must remain in the local map. + v, ok := m.Get("i-reclaim") + require.True(t, ok, "reclaimed slot must remain in the local map") + assert.Same(t, reclaimed, v, "Get must return the reclaimed pointer, not the original") +} + +func TestMigrateStoppedToSharedKV_KVWriteFailure(t *testing.T) { + mounter := &fakeVolumeMounter{} + cleaner := &recordingInstanceCleaner{} + store := failOnWriteStoppedStore{newFakeStateStore(), errors.New("kv unreachable")} + m := NewManagerWithDeps(Deps{ + NodeID: "test-node", + StateStore: store, + VolumeMounter: mounter, + InstanceCleaner: cleaner, + ShutdownSignal: func() bool { return false }, + }) + + v := &VM{ID: "i-kv-fail", Status: StateStopped} + m.Insert(v) + + got := m.MigrateStoppedToSharedKV(v) + assert.False(t, got, "MigrateStoppedToSharedKV must return false on KV write failure") + + // Local map entry must remain so restoreInstances can retry on boot. + _, ok := m.Get("i-kv-fail") + assert.True(t, ok, "KV write failure must leave the instance in the local map") +} + +// TestMigrateStoppedToSharedKV_NoStateStore covers the no-deps fallback +// where StateStore is nil — used by Manager built via NewManager() with +// no Deps wired (e.g. early-boot daemon construction). +func TestMigrateStoppedToSharedKV_NoStateStore(t *testing.T) { + m := NewManager() + v := &VM{ID: "i-no-store", Status: StateStopped} + m.Insert(v) + + got := m.MigrateStoppedToSharedKV(v) + assert.False(t, got, "missing StateStore must report false (no migration possible)") + _, ok := m.Get("i-no-store") + assert.True(t, ok, "missing StateStore must leave local map untouched") +} + +// failOnWriteStoppedStore wraps fakeStateStore and forces the +// WriteStoppedInstance path to error so the slot-reclaim contract can be +// verified in isolation from the happy path. +type failOnWriteStoppedStore struct { + *fakeStateStore + + err error +} + +func (f failOnWriteStoppedStore) WriteStoppedInstance(string, *VM) error { + return f.err +} + +// TestStop_FiresOnInstanceDownExactlyOnce verifies the success-path hook +// contract: Stop fires OnInstanceDown once after a successful transition +// to Stopped + shared-KV migration. +func TestStop_FiresOnInstanceDownExactlyOnce(t *testing.T) { + m, _, _, _, _ := shutdownTestManager(t) + var down atomic.Int64 + var downIDs []string + var mu sync.Mutex + rt := (&recordedTransitions{}).bind(m) + m.SetDeps(Deps{ + NodeID: "test-node", + StateStore: newFakeStateStore(), + VolumeMounter: &fakeVolumeMounter{}, + InstanceCleaner: &recordingInstanceCleaner{}, + TransitionState: rt.apply, + ShutdownSignal: func() bool { return false }, + Hooks: ManagerHooks{ + OnInstanceDown: func(id string) { + down.Add(1) + mu.Lock() + downIDs = append(downIDs, id) + mu.Unlock() + }, + }, + }) + + v := &VM{ + ID: "i-stop-hook", + Status: StateRunning, + InstanceType: "t3.micro", + Instance: &ec2.Instance{}, + } + m.Insert(v) + + require.NoError(t, m.Stop(v.ID)) + + assert.Equal(t, int64(1), down.Load(), "Stop must fire OnInstanceDown exactly once") + mu.Lock() + defer mu.Unlock() + assert.Equal(t, []string{"i-stop-hook"}, downIDs) +} + +// TestStop_DoesNotFireOnInstanceDown_OnSlotReclaim verifies the slot- +// reclaim branch in Stop: when MigrateStoppedToSharedKV returns false +// because a concurrent handler took the slot, OnInstanceDown must not +// fire (firing it would tear down the new instance's NATS subs). +func TestStop_DoesNotFireOnInstanceDown_OnSlotReclaim(t *testing.T) { + var down atomic.Int64 + store := &reclaimingStateStore{fakeStateStore: newFakeStateStore()} + m := NewManager() + rt := (&recordedTransitions{}).bind(m) + m.SetDeps(Deps{ + NodeID: "test-node", + StateStore: store, + VolumeMounter: &fakeVolumeMounter{}, + InstanceCleaner: &recordingInstanceCleaner{}, + TransitionState: rt.apply, + ShutdownSignal: func() bool { return false }, + Hooks: ManagerHooks{ + OnInstanceDown: func(string) { down.Add(1) }, + }, + }) + store.m = m + + v := &VM{ + ID: "i-reclaim-no-hook", + Status: StateRunning, + InstanceType: "t3.micro", + Instance: &ec2.Instance{}, + } + m.Insert(v) + + require.NoError(t, m.Stop(v.ID)) + + assert.Zero(t, down.Load(), + "slot-reclaim branch must not fire OnInstanceDown") +} + +// reclaimingStateStore reclaims the slot under the same id during the +// shared-KV write callback, so the subsequent DeleteIf in +// migrateInstanceToKV finds a different pointer and returns false. +type reclaimingStateStore struct { + *fakeStateStore + + m *Manager +} + +func (r *reclaimingStateStore) WriteStoppedInstance(id string, v *VM) error { + if err := r.fakeStateStore.WriteStoppedInstance(id, v); err != nil { + return err + } + // Mid-flight slot reclaim: a concurrent start handler installs a new + // pointer under the same id between the write and DeleteIf. + r.m.Insert(&VM{ID: id, Status: StateRunning}) + return nil +} diff --git a/spinifex/vm/watchdog.go b/spinifex/vm/watchdog.go index 1b21b79fe..f54274f17 100644 --- a/spinifex/vm/watchdog.go +++ b/spinifex/vm/watchdog.go @@ -30,19 +30,26 @@ func (m *Manager) StartPendingWatchdog(ctx context.Context) { case <-ctx.Done(): return case <-ticker.C: - stuck := m.Filter(func(v *VM) bool { - return (v.Status == StatePending || v.Status == StateProvisioning) && - v.Instance != nil && v.Instance.LaunchTime != nil && - time.Since(*v.Instance.LaunchTime) > PendingWatchdogTimeout - }) - - for _, instance := range stuck { - slog.Warn("Instance stuck in pending, marking failed", - "instanceId", instance.ID, "status", instance.Status, - "elapsed", time.Since(*instance.Instance.LaunchTime)) - m.MarkFailed(instance, "launch_timeout") - } + m.scanAndMarkStuckPending(time.Now()) } } }() } + +// scanAndMarkStuckPending runs one pass of the watchdog body with a +// caller-supplied "now". Extracted so tests can drive the body +// deterministically without waiting for the production tick interval. +func (m *Manager) scanAndMarkStuckPending(now time.Time) { + stuck := m.Filter(func(v *VM) bool { + return (v.Status == StatePending || v.Status == StateProvisioning) && + v.Instance != nil && v.Instance.LaunchTime != nil && + now.Sub(*v.Instance.LaunchTime) > PendingWatchdogTimeout + }) + + for _, instance := range stuck { + slog.Warn("Instance stuck in pending, marking failed", + "instanceId", instance.ID, "status", instance.Status, + "elapsed", now.Sub(*instance.Instance.LaunchTime)) + m.MarkFailed(instance, "launch_timeout") + } +} diff --git a/spinifex/vm/watchdog_test.go b/spinifex/vm/watchdog_test.go new file mode 100644 index 000000000..c9226144c --- /dev/null +++ b/spinifex/vm/watchdog_test.go @@ -0,0 +1,149 @@ +package vm + +import ( + "context" + "testing" + "time" + + "github.com/aws/aws-sdk-go/service/ec2" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// pendingInstance returns a VM in the supplied state with LaunchTime set +// to launchedAgo before now. now is the synthetic clock used by +// scanAndMarkStuckPending in tests. +func pendingInstance(id string, state InstanceState, launched time.Time) *VM { + return &VM{ + ID: id, + Status: state, + Instance: &ec2.Instance{ + LaunchTime: &launched, + }, + } +} + +func TestScanAndMarkStuckPending_EmptyMap(t *testing.T) { + m, _, rt, _ := crashTestManager(t) + m.scanAndMarkStuckPending(time.Now()) + assert.Empty(t, rt.snapshot(), "empty map must not invoke any transitions") +} + +func TestScanAndMarkStuckPending_FreshPending_NotMarked(t *testing.T) { + m, _, rt, _ := crashTestManager(t) + + now := time.Now() + v := pendingInstance("i-fresh", StatePending, now) + m.Insert(v) + + m.scanAndMarkStuckPending(now) + + assert.Empty(t, rt.snapshot(), + "fresh pending instance (elapsed=0) must not be marked failed") + assert.Equal(t, StatePending, m.Status(v)) +} + +func TestScanAndMarkStuckPending_BoundaryNotStuck(t *testing.T) { + m, _, rt, _ := crashTestManager(t) + + now := time.Now() + // Exactly at the timeout boundary — strict ">" comparison means equal + // is not yet stuck. + v := pendingInstance("i-boundary", StatePending, now.Add(-PendingWatchdogTimeout)) + m.Insert(v) + + m.scanAndMarkStuckPending(now) + + assert.Empty(t, rt.snapshot(), + "instance exactly at the timeout boundary must not be marked stuck") +} + +func TestScanAndMarkStuckPending_StuckPending_MarkedFailed(t *testing.T) { + m, _, rt, _ := crashTestManager(t) + + now := time.Now() + v := pendingInstance("i-stuck", StatePending, now.Add(-PendingWatchdogTimeout-time.Minute)) + m.Insert(v) + + m.scanAndMarkStuckPending(now) + + assertStuckMarkedFailed(t, m, rt, v) +} + +func TestScanAndMarkStuckPending_StuckProvisioning_MarkedFailed(t *testing.T) { + m, _, rt, _ := crashTestManager(t) + + now := time.Now() + v := pendingInstance("i-prov-stuck", StateProvisioning, now.Add(-PendingWatchdogTimeout-time.Second)) + m.Insert(v) + + m.scanAndMarkStuckPending(now) + + assertStuckMarkedFailed(t, m, rt, v) +} + +func TestScanAndMarkStuckPending_NoLaunchTime_NotMarked(t *testing.T) { + m, _, rt, _ := crashTestManager(t) + + v := &VM{ + ID: "i-no-launchtime", + Status: StatePending, + // Instance is nil → predicate must short-circuit safely. + } + m.Insert(v) + + m.scanAndMarkStuckPending(time.Now()) + + assert.Empty(t, rt.snapshot(), + "instance without LaunchTime must not be marked stuck") +} + +func TestScanAndMarkStuckPending_OnlyPendingStatesScanned(t *testing.T) { + m, _, rt, _ := crashTestManager(t) + + now := time.Now() + long := now.Add(-PendingWatchdogTimeout - time.Hour) + + for _, state := range []InstanceState{StateRunning, StateStopped, StateStopping, StateTerminated} { + v := pendingInstance("i-"+string(state), state, long) + m.Insert(v) + } + + m.scanAndMarkStuckPending(now) + + assert.Empty(t, rt.snapshot(), + "non-pending states must not be marked stuck regardless of LaunchTime") +} + +func TestStartPendingWatchdog_CtxCancelStopsGoroutine(t *testing.T) { + m, _, _, _ := crashTestManager(t) + + ctx, cancel := context.WithCancel(t.Context()) + m.StartPendingWatchdog(ctx) + + // Immediately cancel; with PendingWatchdogInterval at 60s, the only + // observable signal that the goroutine respected ctx.Done is that the + // test exits cleanly without waiting for the next tick. + cancel() + // Yield briefly so the goroutine actually returns before the test + // finishes; race detector catches any leaked work. + time.Sleep(10 * time.Millisecond) +} + +func assertStuckMarkedFailed(t *testing.T, m *Manager, rt *recordedTransitions, v *VM) { + t.Helper() + + // MarkFailed transitions Pending/Provisioning → ShuttingDown + // synchronously, then runs terminateCleanup + finalizeTerminated in a + // goroutine. Wait for the terminal transition to land. + require.Eventually(t, func() bool { + return m.Status(v) == StateTerminated + }, 2*time.Second, 5*time.Millisecond, "stuck instance must reach Terminated") + + targets := rt.targets(v.ID) + require.NotEmpty(t, targets) + assert.Equal(t, StateShuttingDown, targets[0], + "first transition must be ShuttingDown (set by MarkFailed)") + assert.Contains(t, targets, StateTerminated, + "terminal transition must land in Terminated") +}