fix(daemon): preserve volumes when recovery fails on restart (siv-25 short-term) (#209)

Stops the P0 data-loss path where `systemctl restart spinifex-daemon` destroys user volumes flagged `DeleteOnTermination=true`. Previously, when daemon recovery couldn't relaunch (Run error) or reconnect (QMP attach failure) to a previously-running instance, the code called `MarkFailed` — which transitions to `terminated`, runs the full destructive cleanup (`terminateCleanup`), and deletes volumes per `DeleteOnTermination`. A benign operator action (daemon restart) was treated identically to a user-initiated `ec2.TerminateInstances`. New `MarkRecoveryFailed` transitions to `StateError` with a `Server.RecoveryFailed` StateReason, runs only the non-destructive `stopCleanup` chain (graceful unmount + tap teardown + GPU release + resource deallocation), and persists the running set. Volumes, public IPs, ENIs, and placement-group membership are preserved so the operator can either retry recovery or issue an explicit terminate (which will then honour DoT as normal). The instance stays in the local map and `OnInstanceDown` is NOT fired, keeping `ec2.cmd.<id>` live so operator commands can still reach this node. `classifyRestoredInstances` now skips relaunch + resource re-allocation for instances already in `StateError`, so a recovery-failed instance stays put across subsequent daemon restarts until the operator acts. Medium-term (viperblock "block size 0" hard-vs-soft classification + mount retry-with-backoff) and long-term (audit of every non-user- initiated cleanup caller) tiers remain open on the siv-25 plan.
2026-10-15 17:11:30 +00:00 · 2026-05-11 16:47:24 +10:00
parent 13d9fb494a
commit f6ddd602b9
3 changed files with 286 additions and 3 deletions
@@ -114,6 +114,16 @@ func (m *Manager) classifyRestoredInstances() []*VM {
 			continue
 		}

+		// Recovery-failed instances stay in StateError until the operator
+		// explicitly retries or terminates. Skip relaunch and resource
+		// re-allocation; resources were already released by stopCleanup
+		// when MarkRecoveryFailed fired on the previous daemon run.
+		if instance.Status == StateError {
+			slog.Warn("Instance in error state; skipping recovery relaunch (operator must retry or terminate)",
+				"instance", instance.ID)
+			continue
+		}
+
 		typeKnown := true
 		if m.deps.InstanceTypes != nil {
 			_, typeKnown = m.deps.InstanceTypes.Resolve(instance.InstanceType)
@@ -149,9 +159,9 @@ func (m *Manager) classifyRestoredInstances() []*VM {
 			} else {
 				slog.Info("Instance QEMU process still alive, reconnecting", "instance", instance.ID)
 				if err := m.reconnectInstance(instance); err != nil {
-					slog.Error("Failed to reconnect to running instance, marking failed to tear down orphaned QEMU",
+					slog.Error("Failed to reconnect to running instance, marking recovery-failed to preserve user data",
 						"instanceId", instance.ID, "err", err)
-					m.MarkFailed(instance, "reconnect_failed")
+					m.MarkRecoveryFailed(instance, "reconnect_failed")
 				}
 				continue
 			}
@@ -293,7 +303,7 @@ func (m *Manager) relaunchAll(toLaunch []*VM) {
 			slog.Info("Launching instance (recovery)", "instance", inst.ID)
 			if err := m.Run(inst); err != nil {
 				slog.Error("Failed to launch instance during recovery", "instanceId", inst.ID, "err", err)
-				m.MarkFailed(inst, "recovery_launch_failed")
+				m.MarkRecoveryFailed(inst, "recovery_launch_failed")
 			}
 		}(instance)
 	}
@@ -173,6 +173,60 @@ func (m *Manager) MarkFailed(instance *VM, reason string) {
 	}()
 }

+// MarkRecoveryFailed handles the case where daemon-restart recovery cannot
+// bring a previously-running instance back online (Run error during
+// relaunch, or QMP reconnect failure on a surviving QEMU). It transitions
+// the instance to StateError with a Server.RecoveryFailed StateReason,
+// then runs the non-destructive cleanup chain (graceful unmount + tap
+// teardown + GPU release + resource deallocation) in a goroutine. Unlike
+// MarkFailed, it does NOT delete volumes, release public IPs, detach
+// ENIs, or remove placement-group membership — those resources are
+// preserved so the operator can either retry recovery or issue an
+// explicit ec2.TerminateInstances (which will then honour
+// DeleteOnTermination as normal). The instance stays in the local map
+// and OnInstanceDown is NOT fired so the ec2.cmd.<id> subscription
+// remains live for operator action.
+func (m *Manager) MarkRecoveryFailed(instance *VM, reason string) {
+	skip := false
+	var observed InstanceState
+	m.Inspect(instance, func(v *VM) {
+		observed = v.Status
+		if v.Status == StateError || v.Status == StateShuttingDown || v.Status == StateTerminated {
+			skip = true
+			return
+		}
+		if v.Instance != nil {
+			v.Instance.StateReason = &ec2.StateReason{
+				Code:    aws.String("Server.RecoveryFailed"),
+				Message: aws.String(reason),
+			}
+		}
+	})
+	if skip {
+		slog.Info("MarkRecoveryFailed: instance already in terminal/cleanup state, skipping",
+			"instanceId", instance.ID, "status", string(observed), "reason", reason)
+		return
+	}
+
+	if err := m.transitionWithPrecheck(instance, StateError); err != nil {
+		slog.Error("MarkRecoveryFailed transition failed", "instanceId", instance.ID, "err", err)
+		if m.Status(instance) != StateError {
+			return
+		}
+	}
+	slog.Error("Instance marked recovery_failed; volumes and ENIs preserved for operator action",
+		"instanceId", instance.ID, "reason", reason)
+
+	go func() {
+		m.stopCleanup(instance)
+		m.Inspect(instance, func(v *VM) { v.LastNode = m.deps.NodeID })
+		if err := m.writeRunningState(); err != nil {
+			slog.Error("Failed to persist state after recovery failure",
+				"instanceId", instance.ID, "err", err)
+		}
+	}()
+}
+
 // finalizeTerminated transitions instance to terminated, writes the
 // terminated KV entry, removes the instance from the local map, fires
 // OnInstanceDown, and persists the running set. Shared by Terminate and
@@ -771,3 +771,222 @@ func TestTerminate_WriteRunningStateFailure_RollbackInsert(t *testing.T) {
 	require.True(t, ok, "writeRunningState failure must re-insert the instance into the local map")
 	assert.Same(t, v, got)
 }
+
+// TestMarkRecoveryFailed_PreservesVolumesAndAWSResources locks down the
+// siv-25 architectural invariant: when daemon recovery cannot bring a
+// previously-running instance back online, the cleanup path must NOT
+// invoke DeleteVolumes, ReleasePublicIP, DetachAndDeleteENI, or
+// RemoveFromPlacementGroup. A regression here re-introduces the P0 data
+// loss: a benign `systemctl restart spinifex-daemon` would destroy user
+// volumes flagged DeleteOnTermination=true.
+func TestMarkRecoveryFailed_PreservesVolumesAndAWSResources(t *testing.T) {
+	defer goleak.VerifyNone(t)
+	m, _, _, cleaner, rt := shutdownTestManager(t)
+	saved := make(chan struct{}, 1)
+	m.SetDeps(Deps{
+		NodeID:          "test-node",
+		StateStore:      &signalingStore{onSave: func() { saved <- struct{}{} }},
+		VolumeMounter:   &fakeVolumeMounter{},
+		InstanceCleaner: cleaner,
+		TransitionState: rt.apply,
+		ShutdownSignal:  func() bool { return false },
+	})
+
+	instance := &VM{
+		ID:        "i-recovery-failed",
+		Status:    StatePending,
+		AccountID: "111122223333",
+		Instance:  &ec2.Instance{},
+	}
+	m.Insert(instance)
+
+	errored := rt.waitFor(instance.ID, StateError)
+	m.MarkRecoveryFailed(instance, "recovery_launch_failed")
+
+	require.NotNil(t, instance.Instance.StateReason,
+		"MarkRecoveryFailed must populate StateReason synchronously")
+	assert.Equal(t, "Server.RecoveryFailed", *instance.Instance.StateReason.Code)
+	assert.Equal(t, "recovery_launch_failed", *instance.Instance.StateReason.Message)
+
+	select {
+	case <-errored:
+	case <-time.After(markFailedDeadline):
+		t.Fatalf("MarkRecoveryFailed did not transition to StateError within %s", markFailedDeadline)
+	}
+
+	assert.Equal(t, StateError, m.Status(instance),
+		"instance must remain in StateError; never transition to terminated")
+
+	// writeRunningState is the goroutine's last step; SaveRunningState
+	// signals on the chan so we can assert on the cleaner without racing.
+	select {
+	case <-saved:
+	case <-time.After(markFailedDeadline):
+		t.Fatalf("cleanup goroutine did not write running state within %s", markFailedDeadline)
+	}
+
+	cleaner.mu.Lock()
+	defer cleaner.mu.Unlock()
+	assert.Empty(t, cleaner.deleteVolumes,
+		"recovery failure must not call DeleteVolumes — preserves user data")
+	assert.Empty(t, cleaner.releasePublicIP,
+		"recovery failure must not call ReleasePublicIP — instance retains its IP")
+	assert.Empty(t, cleaner.detachAndDeleteENI,
+		"recovery failure must not call DetachAndDeleteENI — instance retains its ENI")
+	assert.Empty(t, cleaner.removeFromPlacement,
+		"recovery failure must not call RemoveFromPlacementGroup")
+	assert.NotEmpty(t, cleaner.releaseGPU,
+		"recovery failure must release host-side GPU claim (no QEMU using it)")
+
+	// Local map must retain the instance so a later ec2.TerminateInstances
+	// can find and clean it up via the destructive path.
+	_, ok := m.Get(instance.ID)
+	assert.True(t, ok, "instance must stay in the local map for operator action")
+}
+
+// TestMarkRecoveryFailed_DoesNotFireOnInstanceDown verifies the per-id
+// NATS subscription contract: OnInstanceDown must NOT fire on a
+// recovery-failed instance, because the daemon must keep the
+// ec2.cmd.<id> subscription live so the operator's terminate command
+// can reach this node.
+func TestMarkRecoveryFailed_DoesNotFireOnInstanceDown(t *testing.T) {
+	defer goleak.VerifyNone(t)
+	var down atomic.Int64
+	m, _, _, _, rt := shutdownTestManager(t)
+	m.SetDeps(Deps{
+		NodeID:          "test-node",
+		StateStore:      newFakeStateStore(),
+		VolumeMounter:   &fakeVolumeMounter{},
+		InstanceCleaner: &recordingInstanceCleaner{},
+		TransitionState: rt.apply,
+		ShutdownSignal:  func() bool { return false },
+		Hooks: ManagerHooks{
+			OnInstanceDown: func(string) { down.Add(1) },
+		},
+	})
+
+	instance := &VM{
+		ID:       "i-keep-sub",
+		Status:   StatePending,
+		Instance: &ec2.Instance{},
+	}
+	m.Insert(instance)
+
+	errored := rt.waitFor(instance.ID, StateError)
+	m.MarkRecoveryFailed(instance, "reconnect_failed")
+
+	select {
+	case <-errored:
+	case <-time.After(markFailedDeadline):
+		t.Fatalf("MarkRecoveryFailed did not transition to StateError within %s", markFailedDeadline)
+	}
+
+	// Give the async goroutine a moment to finish; even after completion
+	// OnInstanceDown must remain zero.
+	assert.Eventually(t, func() bool {
+		return m.Status(instance) == StateError
+	}, markFailedDeadline, 10*time.Millisecond)
+	time.Sleep(50 * time.Millisecond)
+
+	assert.Zero(t, down.Load(),
+		"recovery failure must not fire OnInstanceDown — per-id NATS subscription must stay live for operator action")
+}
+
+// TestMarkRecoveryFailed_NoOpOnTerminal verifies idempotency: an
+// instance already in StateError / StateShuttingDown / StateTerminated
+// must not be re-transitioned or have its StateReason overwritten.
+func TestMarkRecoveryFailed_NoOpOnTerminal(t *testing.T) {
+	for _, status := range []InstanceState{StateError, StateShuttingDown, StateTerminated} {
+		t.Run(string(status), func(t *testing.T) {
+			m, _, _, cleaner, rt := shutdownTestManager(t)
+			instance := &VM{
+				ID:       "i-noop-" + string(status),
+				Status:   status,
+				Instance: &ec2.Instance{},
+			}
+			m.Insert(instance)
+
+			m.MarkRecoveryFailed(instance, "should_skip")
+
+			assert.Empty(t, rt.snapshot(),
+				"MarkRecoveryFailed must not transition an already-terminal instance")
+			assert.Nil(t, instance.Instance.StateReason,
+				"MarkRecoveryFailed must not overwrite StateReason on terminal instance")
+			cleaner.mu.Lock()
+			defer cleaner.mu.Unlock()
+			assert.Empty(t, cleaner.releaseGPU,
+				"MarkRecoveryFailed must not run cleanup on terminal instance")
+		})
+	}
+}
+
+// TestClassifyRestoredInstances_StateErrorSkipsRelaunch verifies the
+// restore-side companion fix: an instance already in StateError (from a
+// prior recovery failure) must NOT be queued for relaunch and must NOT
+// have its resources re-allocated. Without this, every daemon restart
+// would re-trigger the failing recovery loop and eventually re-destroy
+// volumes once the loop hit the terminateCleanup path.
+func TestClassifyRestoredInstances_StateErrorSkipsRelaunch(t *testing.T) {
+	m := NewManager()
+	rc := &countingResourceController{}
+	m.SetDeps(Deps{
+		NodeID:        "test-node",
+		StateStore:    newFakeStateStore(),
+		InstanceTypes: fakeInstanceTypeResolver{"t3.micro": {VCPUs: 2, MemoryMiB: 1024}},
+		Resources:     rc,
+	})
+
+	code := "Server.RecoveryFailed"
+	msg := "recovery_launch_failed"
+	v := &VM{
+		ID:           "i-recovery-error",
+		Status:       StateError,
+		InstanceType: "t3.micro",
+		Instance:     &ec2.Instance{StateReason: &ec2.StateReason{Code: &code, Message: &msg}},
+	}
+	m.Insert(v)
+
+	toLaunch := m.classifyRestoredInstances()
+
+	assert.Empty(t, toLaunch, "StateError instance must not be queued for relaunch")
+	assert.Zero(t, rc.allocations,
+		"StateError instance must not re-allocate resources (already released by stopCleanup)")
+	_, ok := m.Get(v.ID)
+	assert.True(t, ok, "StateError instance must stay in the local map for operator action")
+	assert.Equal(t, StateError, m.Status(v), "status must be preserved")
+}
+
+// countingResourceController counts how often Allocate fires so a test
+// can prove the restore path skipped its resource-allocation branch.
+type countingResourceController struct {
+	allocations int
+}
+
+func (c *countingResourceController) Allocate(_ string) error         { c.allocations++; return nil }
+func (c *countingResourceController) Deallocate(_ string)             {}
+func (c *countingResourceController) CanAllocate(_ string, n int) int { return n }
+
+// signalingStore is a no-op StateStore that fires onSave once per
+// SaveRunningState call. Used by recovery-failure tests to wait for the
+// cleanup goroutine's last step without polling a racy shared map.
+type signalingStore struct {
+	onSave func()
+}
+
+func (s *signalingStore) SaveRunningState(string, map[string]*VM) error {
+	if s.onSave != nil {
+		s.onSave()
+	}
+	return nil
+}
+func (s *signalingStore) LoadRunningState(string) (map[string]*VM, error) {
+	return map[string]*VM{}, nil
+}
+func (s *signalingStore) WriteStoppedInstance(string, *VM) error    { return nil }
+func (s *signalingStore) LoadStoppedInstance(string) (*VM, error)   { return nil, nil }
+func (s *signalingStore) DeleteStoppedInstance(string) error        { return nil }
+func (s *signalingStore) ListStoppedInstances() ([]*VM, error)      { return nil, nil }
+func (s *signalingStore) WriteTerminatedInstance(string, *VM) error { return nil }
+func (s *signalingStore) ListTerminatedInstances() ([]*VM, error)   { return nil, nil }
+
+var _ StateStore = (*signalingStore)(nil)