Merge remote-tracking branch 'origin/dev' into feat/ddil-scenarios-abc

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

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=

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

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

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

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.<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

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