test(vm): migrate lifecycle coverage from daemon to vm package (#192)

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) {

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;

spinifex/daemon/health.go

@@ -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)
-}

spinifex/daemon/health_test.go

@@ -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
-}

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")
 }
 

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,

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")
+}

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)

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)

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
+}

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")
+	}
+}

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")
+}