diff --git a/go.mod b/go.mod index bc398e870..f74c615c5 100644 --- a/go.mod +++ b/go.mod @@ -13,7 +13,7 @@ require ( github.com/kdomanski/iso9660 v0.4.0 github.com/klauspost/cpuid/v2 v2.3.0 github.com/mulgadc/predastore v1.2.1 - github.com/mulgadc/viperblock v1.2.1 + github.com/mulgadc/viperblock v1.2.2-0.20260511064619-1aa54f0dc1f6 github.com/nats-io/nats-server/v2 v2.14.0 github.com/nats-io/nats.go v1.51.0 github.com/ovn-kubernetes/libovsdb v0.8.1 @@ -120,7 +120,6 @@ require ( go.opentelemetry.io/otel/trace v1.42.0 // indirect go.yaml.in/yaml/v2 v2.4.4 // indirect go.yaml.in/yaml/v3 v3.0.4 // indirect - golang.org/x/exp v0.0.0-20231110203233-9a3e6036ecaa // indirect golang.org/x/mod v0.34.0 // indirect golang.org/x/net v0.53.0 // indirect golang.org/x/sync v0.20.0 // indirect diff --git a/go.sum b/go.sum index fcd641c3f..11a96d3c4 100644 --- a/go.sum +++ b/go.sum @@ -288,8 +288,8 @@ github.com/muesli/termenv v0.16.0 h1:S5AlUN9dENB57rsbnkPyfdGuWIlkmzJjbFf0Tf5FWUc github.com/muesli/termenv v0.16.0/go.mod h1:ZRfOIKPFDYQoDFF4Olj7/QJbW60Ol/kL1pU3VfY/Cnk= github.com/mulgadc/predastore v1.2.1 h1:Os+CTjjs/IneRMWd/+m/Hi24sHI4M+vVxcBSRF2Pr10= github.com/mulgadc/predastore v1.2.1/go.mod h1:WGfbbzk0ExniHXkXsvlmIILSfONakrb7h9yDSxFktaY= -github.com/mulgadc/viperblock v1.2.1 h1:G3cSgh/yspZd7Wdg3C9QGqkqzYbEfYpusd6uMddSDmg= -github.com/mulgadc/viperblock v1.2.1/go.mod h1:CisHy13Uam0gUg5dxV0rKryJF2yGbgfDv09FYApRaNk= +github.com/mulgadc/viperblock v1.2.2-0.20260511064619-1aa54f0dc1f6 h1:PngaxM0orWymcmDMR97JbeedXlgpxoLiykrjSSmPNEU= +github.com/mulgadc/viperblock v1.2.2-0.20260511064619-1aa54f0dc1f6/go.mod h1:WCP9h+u2DvN8/+ccai5oMOa48hOJPUzSbh7Lia0Q8nY= github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 h1:C3w9PqII01/Oq1c1nUAm88MOHcQC9l5mIlSMApZMrHA= github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822/go.mod h1:+n7T8mK8HuQTcFwEeznm/DIxMOiR9yIdICNftLE1DvQ= github.com/mwitkow/go-conntrack v0.0.0-20161129095857-cc309e4a2223/go.mod h1:qRWi+5nqEBWmkhHvq77mSJWrCKwh8bxhgT7d/eI7P4U= diff --git a/spinifex/services/viperblockd/viperblockd.go b/spinifex/services/viperblockd/viperblockd.go index 87375dd2e..15f76ea44 100644 --- a/spinifex/services/viperblockd/viperblockd.go +++ b/spinifex/services/viperblockd/viperblockd.go @@ -2,6 +2,7 @@ package viperblockd import ( "encoding/json" + "errors" "fmt" "log/slog" "os" @@ -22,6 +23,52 @@ import ( "github.com/nats-io/nats.go" ) +// loadStateRetryAttempts and loadStateRetryBaseDelay tune the mount-time +// retry loop for transient backend errors (predastore restarting, NATS +// disconnect, etc.). 5 attempts at 200ms * 1.5^n caps total wait at ~3.7s, +// well under the daemon's 30s ebs.mount NATS timeout. See mulga-siv-25. +const ( + loadStateRetryAttempts = 5 + loadStateRetryBaseDelay = 200 * time.Millisecond +) + +// retryLoadState invokes loadFn up to attempts times, retrying with +// exponential backoff (delay * 3/2 each step) only when the returned error +// is viperblock.ErrStateBackendUnavailable. Success, ErrStateNotFound, and +// any unclassified error return immediately so callers see them. Extracted +// for unit testability — see loadStateWithRetry for the production caller. +func retryLoadState(volume string, attempts int, baseDelay time.Duration, sleep func(time.Duration), loadFn func() error) error { + delay := baseDelay + var err error + for attempt := 1; attempt <= attempts; attempt++ { + err = loadFn() + if err == nil { + if attempt > 1 { + slog.Info("LoadState succeeded after retry", + "volume", volume, "attempt", attempt) + } + return nil + } + if !errors.Is(err, viperblock.ErrStateBackendUnavailable) { + return err + } + if attempt == attempts { + break + } + slog.Warn("LoadState transient failure, retrying", + "volume", volume, "attempt", attempt, "delay", delay, "err", err) + sleep(delay) + delay = delay * 3 / 2 + } + return fmt.Errorf("LoadState exhausted %d retries: %w", attempts, err) +} + +// loadStateWithRetry calls vb.LoadState with the production retry budget +// (loadStateRetryAttempts / loadStateRetryBaseDelay). +func loadStateWithRetry(vb *viperblock.VB, volume string) error { + return retryLoadState(volume, loadStateRetryAttempts, loadStateRetryBaseDelay, time.Sleep, vb.LoadState) +} + var serviceName = "viperblock" type MountedVolume struct { @@ -463,8 +510,10 @@ func launchService(cfg *Config) (err error) { } // Next, connect to the volume and confirm the state - // First, fetch the state from the remote backend - err = vb.LoadState() + // First, fetch the state from the remote backend. Retry on transient + // backend errors (predastore restart races, NATS hiccups) so daemon + // recovery doesn't tip a healthy volume into the cleanup path. + err = loadStateWithRetry(vb, ebsRequest.Name) if err != nil { ebsResponse.Error = err.Error() diff --git a/spinifex/services/viperblockd/viperblockd_test.go b/spinifex/services/viperblockd/viperblockd_test.go index 25edfd9cc..093a084d2 100644 --- a/spinifex/services/viperblockd/viperblockd_test.go +++ b/spinifex/services/viperblockd/viperblockd_test.go @@ -1,12 +1,16 @@ package viperblockd import ( + "errors" + "fmt" "os" "sync" "testing" "time" + "github.com/mulgadc/viperblock/viperblock" "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" ) // TestNew tests the service constructor @@ -439,3 +443,74 @@ func TestMountedVolumePortRange(t *testing.T) { assert.True(t, vol.Port <= 65535) } } + +// TestRetryLoadState pins the retry policy used by ebs.mount: transient +// backend errors retry with backoff, ErrStateNotFound fails fast, and the +// retry budget is honoured. Required by mulga-siv-25 medium-term fix. +func TestRetryLoadState(t *testing.T) { + t.Run("success_first_attempt_no_sleep", func(t *testing.T) { + calls := 0 + sleeps := 0 + err := retryLoadState("vol-x", 5, 10*time.Millisecond, + func(time.Duration) { sleeps++ }, + func() error { calls++; return nil }) + require.NoError(t, err) + assert.Equal(t, 1, calls) + assert.Equal(t, 0, sleeps) + }) + + t.Run("transient_recovers_within_budget", func(t *testing.T) { + calls := 0 + sleeps := 0 + err := retryLoadState("vol-x", 5, 10*time.Millisecond, + func(time.Duration) { sleeps++ }, + func() error { + calls++ + if calls < 3 { + return fmt.Errorf("wrap: %w", viperblock.ErrStateBackendUnavailable) + } + return nil + }) + require.NoError(t, err) + assert.Equal(t, 3, calls) + assert.Equal(t, 2, sleeps) + }) + + t.Run("not_found_fails_fast", func(t *testing.T) { + calls := 0 + sleeps := 0 + err := retryLoadState("vol-x", 5, 10*time.Millisecond, + func(time.Duration) { sleeps++ }, + func() error { calls++; return viperblock.ErrStateNotFound }) + require.Error(t, err) + assert.ErrorIs(t, err, viperblock.ErrStateNotFound) + assert.Equal(t, 1, calls, "ErrStateNotFound must not be retried") + assert.Equal(t, 0, sleeps) + }) + + t.Run("unclassified_fails_fast", func(t *testing.T) { + calls := 0 + sentinel := errors.New("permission denied") + err := retryLoadState("vol-x", 5, 10*time.Millisecond, + func(time.Duration) {}, + func() error { calls++; return sentinel }) + require.Error(t, err) + assert.ErrorIs(t, err, sentinel) + assert.Equal(t, 1, calls) + }) + + t.Run("exhausts_budget_on_persistent_transient", func(t *testing.T) { + calls := 0 + sleeps := 0 + var lastDelay time.Duration + err := retryLoadState("vol-x", 4, 100*time.Millisecond, + func(d time.Duration) { sleeps++; lastDelay = d }, + func() error { calls++; return viperblock.ErrStateBackendUnavailable }) + require.Error(t, err) + assert.ErrorIs(t, err, viperblock.ErrStateBackendUnavailable) + assert.Equal(t, 4, calls) + assert.Equal(t, 3, sleeps, "sleep happens between attempts, not after the final one") + // Backoff multiplier is 1.5: 100ms, 150ms, 225ms. + assert.Equal(t, 225*time.Millisecond, lastDelay) + }) +} diff --git a/spinifex/vm/restore.go b/spinifex/vm/restore.go index 1520773de..633c6bfbd 100644 --- a/spinifex/vm/restore.go +++ b/spinifex/vm/restore.go @@ -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) } diff --git a/spinifex/vm/shutdown.go b/spinifex/vm/shutdown.go index d9fde6068..cf9c70418 100644 --- a/spinifex/vm/shutdown.go +++ b/spinifex/vm/shutdown.go @@ -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. 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 diff --git a/spinifex/vm/shutdown_test.go b/spinifex/vm/shutdown_test.go index d2b467e18..5efb17f2d 100644 --- a/spinifex/vm/shutdown_test.go +++ b/spinifex/vm/shutdown_test.go @@ -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. 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)