fix: retry a recovery relaunch that lost a race with a starting dependency (#1020)

* fix: retry a recovery relaunch that lost a race with a starting dependency

A guest drain-stopped by a coordinated cluster shutdown could be lost for good.
The relaunch ran before JetStream was serving, the volume lease claim timed out,
and the failure was recorded as permanent.

The backoff this needed already existed. relaunchWithRetry retries five times,
but only on ErrMountRetryable, and mountErrRetryable recognised exactly two
viperblock state-load sentinels. A lease claim returning context deadline
exceeded matched neither, so the mount was reported as permanently failed and
the backoff never engaged once.

- Classify a lease store that cannot answer as ErrLeaseStoreUnavailable, and
  include it in mountErrRetryable. A lease held by another node is deliberately
  excluded: that is an answer, not an outage.
- Probe readiness rather than connectedness. checkViperblockReady tested
  natsConn.IsConnected(), which is true against a server whose JetStream is
  still starting; it now calls AccountInfo with a timeout.
- Reconsider a latched failure. RetryRecoveryFailed relaunches instances parked
  in error by a dependency-unavailable reason, once the backing store is
  confirmed ready, for a bounded number of passes, then leaves them alone.
  Only mount-state and launch failures qualify; a failed reconnect or
  pre-relaunch hook can mean the guest itself needs looking at.

Closes the e2e gap that made this feel covered. The node-failure tests stop the
service units directly so guests keep running and recovery is a QMP re-attach to
a live process. Nothing stopped a guest, so nothing relaunched one from a sealed
volume against dependencies that were still starting.
TestMultinodeClusterRestartResumesGuests does, including the wait for every spx
process to exit that makes it a real restart rather than a stop that returned
early.

Beads mulga-v7kp4, mulga-kyb1d.

* fix: stop every spx process by unit in the cluster-restart helper

The first CI run of TestMultinodeClusterRestartResumesGuests failed at the
process wait with "node1 still has 2 spx process(es)", and the same thing
happened by hand on a prod node.

Every service is the same binary, and the stop was driven from a hard-coded
list. StopNode names seven units and omits spinifex-northstar and
spinifex-qmp-collector, which is exactly the two survivors. The unit name now
comes from the surviving process's cgroup, so the set that stops is whoever is
still running the binary and the list cannot go stale when a service is added.

StopNode itself is left alone: its list is deliberate for the hard outage it
simulates, where the target's drain must not fire.

Also marks recovery_retry_test.go as //test:in-package — the retry policy is
unexported and the tests drive it through the seams restore_test.go uses.

* fix: stop spinifex.target in the cluster-restart helper, not the units

The second CI run reformed NATS to 2 peers of 3: node3's NATS was stopped and
never started again.

StopNode stops the service units directly and leaves spinifex.target active.
That is right for the hard outage it simulates and wrong for a real shutdown —
the target then reads as satisfied while its units are dead, and the later start
raced a stop job that was still queued.

The helper now runs what the operator procedure and update-nodes.sh run: stop
spinifex.target, wait for every spx process to exit, and additionally wait for
systemd to have no spinifex job left queued before starting. That last wait is
the one that closes this race.

* fix: wait for the right NATS peer count in the cluster-restart helper

WaitNATSPeers counts remote peers, so a healthy three-node cluster reports two.
Asking for three could never be satisfied, and the run that hit it had in fact
restarted all three NATS correctly.

spinifex/daemon/cluster_ready_test.go

@@ -62,7 +62,10 @@ func closedPort(t *testing.T) string {
 // TestWaitForClusterReady_ReadyImmediately asserts the loop returns on the first
 // pass without sleeping when both dependencies are already reachable.
 func TestWaitForClusterReady_ReadyImmediately(t *testing.T) {
-	_, nc := testutil.StartTestNATS(t)
+	// JetStream, not a bare NATS server: viperblock readiness means the KV the
+	// mount claims its volume lease from can answer, and a connection to a
+	// server whose JetStream is still starting reports connected regardless.
+	_, nc, _ := testutil.StartTestJetStream(t)
 
 	sleeps := 0
 	stubClusterReadySleep(t, func(time.Duration) { sleeps++ })
@@ -79,7 +82,7 @@ func TestWaitForClusterReady_ReadyImmediately(t *testing.T) {
 // while predastore is unreachable and returns once it starts serving, rather
 // than latching the first not-ready result.
 func TestWaitForClusterReady_PredastoreBecomesReady(t *testing.T) {
-	_, nc := testutil.StartTestNATS(t)
+	_, nc, _ := testutil.StartTestJetStream(t)
 
 	srv := httptest.NewTLSServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
 		w.WriteHeader(http.StatusForbidden)
@@ -108,6 +111,31 @@ func TestWaitForClusterReady_PredastoreBecomesReady(t *testing.T) {
 	assert.NotContains(t, logs.String(), "Cluster readiness timeout")
 }
 
+// TestCheckViperblockReady_ConnectedButJetStreamNotServing is the regression
+// guard for a guest lost across a cluster update. A TCP connection to NATS says
+// nothing about whether JetStream can serve, so readiness used to pass while the
+// volume-lease KV was still unavailable, recovery ran into a lease timeout, and
+// the instance was parked in error permanently.
+func TestCheckViperblockReady_ConnectedButJetStreamNotServing(t *testing.T) {
+	_, nc := testutil.StartTestNATS(t)
+	require.True(t, nc.IsConnected(), "the connection itself must be up, or this proves nothing")
+
+	d := &Daemon{config: &config.Config{}, natsConn: nc}
+
+	assert.False(t, d.checkViperblockReady(),
+		"connected to a server with no JetStream is not ready: the lease KV cannot answer")
+}
+
+// TestCheckViperblockReady_JetStreamServing is the other half: once JetStream
+// answers, readiness must pass rather than holding startup open.
+func TestCheckViperblockReady_JetStreamServing(t *testing.T) {
+	_, nc, _ := testutil.StartTestJetStream(t)
+
+	d := &Daemon{config: &config.Config{}, natsConn: nc}
+
+	assert.True(t, d.checkViperblockReady())
+}
+
 // TestWaitForClusterReady_TimesOut asserts the deadline branch: with viperblock
 // permanently unreachable the loop gives up at maxWait, logs the warning and
 // returns so recovery proceeds anyway, instead of blocking startup forever.

spinifex/daemon/daemon.go

@@ -2350,12 +2350,33 @@ func (d *Daemon) waitForClusterReady() {
 	slog.Warn("Cluster readiness timeout, proceeding with recovery anyway", "max_wait_ms", otelsetup.Millis(maxWait))
 }
 
-// checkViperblockReady reports whether viperblock is reachable via NATS.
+// checkViperblockReady reports whether viperblock can actually serve a mount,
+// which needs more than a live NATS socket: mounting claims a JetStream KV
+// lease before it reads any state, so JetStream must be serving too.
+//
+// Testing only IsConnected was why waitForClusterReady did not prevent the
+// race it exists to prevent. The client reconnects as soon as nats-server
+// accepts TCP, which is well before JetStream has quorum, so readiness passed
+// roughly 30 seconds before the first lease claim could have succeeded — and
+// every guest recovering in that window was failed and latched.
+//
+// AccountInfo is the probe because it is the cheapest round trip that proves
+// JetStream is answering, and it changes nothing.
 func (d *Daemon) checkViperblockReady() bool {
-	if d.natsConn == nil {
+	if d.natsConn == nil || !d.natsConn.IsConnected() {
 		return false
 	}
-	return d.natsConn.IsConnected()
+
+	js, err := jetstream.New(d.natsConn)
+	if err != nil {
+		return false
+	}
+
+	ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
+	defer cancel()
+
+	_, err = js.AccountInfo(ctx)
+	return err == nil
 }
 
 // predastoreReadinessClient is hoisted to package scope so the 2s readiness
@@ -2414,11 +2435,29 @@ func (d *Daemon) LoadState() error {
 
 // restoreInstances delegates to vm.Manager.Restore and syncs the local state
 // file so it matches in-memory state.
+//
+// It then starts the recovery retry loop in the background. Restore is a single
+// pass: any instance whose relaunch lost a race with a still-starting dependency
+// is parked in StateError and, before this, stayed there until an operator
+// noticed. The loop reconsiders exactly those, once the backing store is
+// confirmed ready, and returns as soon as there is nothing left to retry.
+//
+// Backgrounded because it must not hold up the rest of daemon start: the guests
+// it is recovering are already down, and blocking here would keep the API from
+// coming up while it waits.
 func (d *Daemon) restoreInstances() error {
 	d.vmMgr.Restore()
 	if err := d.WriteState(); err != nil {
 		slog.Error("Failed to persist local state after restore", "error", err)
 	}
+
+	go func() {
+		d.vmMgr.RunRecoveryRetryLoop(d.shuttingDown.Load)
+		if err := d.WriteState(); err != nil {
+			slog.Error("Failed to persist local state after recovery retry", "error", err)
+		}
+	}()
+
 	return nil
 }
 

spinifex/services/viperblockd/provider_handlers.go

@@ -1327,10 +1327,17 @@ func constructMountedVB(ctx context.Context, cfg *Config, volumeName string) (*v
 
 // mountErrRetryable reports whether a mount failure was caused by the
 // backing store not yet being ready (a transient state-load gap) rather
-// than a permanent condition. Only these two sentinels qualify for the
+// than a permanent condition. Only these sentinels qualify for the
 // recovery-relaunch retry in vm.Manager.
+//
+// ErrLeaseStoreUnavailable belongs here for the same reason the other two do,
+// and its absence is what made a routine cluster update destructive: the lease
+// is claimed before any state is read, so a node coming up alongside NATS fails
+// there first and never reaches the conditions these sentinels describe.
 func mountErrRetryable(err error) bool {
-	return errors.Is(err, viperblock.ErrStateNotFound) || errors.Is(err, viperblock.ErrStateBackendUnavailable)
+	return errors.Is(err, viperblock.ErrStateNotFound) ||
+		errors.Is(err, viperblock.ErrStateBackendUnavailable) ||
+		errors.Is(err, ErrLeaseStoreUnavailable)
 }
 
 // mountVolume is launchService's ebs.mount body, extracted so the legacy

spinifex/services/viperblockd/provider_handlers_test.go

@@ -33,9 +33,12 @@ import (
 	"github.com/stretchr/testify/require"
 )
 
-// TestMountErrRetryable locks down the classifier relaunchAll's
-// recovery-retry relies on: only the two viperblock state-load sentinels
-// count as a transient, retryable mount failure.
+// TestMountErrRetryable locks down the classifier relaunchAll's recovery-retry
+// relies on: a mount that failed because a dependency was still starting is
+// retryable, and nothing else is. A guest lost across a cluster update because
+// the lease-store cases below were missing — the mount failed on a JetStream
+// that had not finished coming up, was reported as permanent, and the existing
+// backoff never engaged.
 func TestMountErrRetryable(t *testing.T) {
 	tests := []struct {
 		name string
@@ -46,7 +49,14 @@ func TestMountErrRetryable(t *testing.T) {
 		{"wrapped ErrStateNotFound", fmt.Errorf("state present but BlockSize=0: %w", viperblock.ErrStateNotFound), true},
 		{"ErrStateBackendUnavailable", viperblock.ErrStateBackendUnavailable, true},
 		{"wrapped ErrStateBackendUnavailable", fmt.Errorf("LoadState exhausted 5 retries: %w", viperblock.ErrStateBackendUnavailable), true},
+		{"ErrLeaseStoreUnavailable", ErrLeaseStoreUnavailable, true},
+		{
+			"lease claim against a JetStream that is still starting",
+			fmt.Errorf("claim lease for vol-1: %w: %w", context.DeadlineExceeded, ErrLeaseStoreUnavailable),
+			true,
+		},
 		{"plain error", errors.New("some other mount failure"), false},
+		{"lease held by another node is not retryable", errVolumeLeaseHeld, false},
 	}
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {

spinifex/services/viperblockd/volume_lease.go

@@ -20,6 +20,35 @@ import (
 // exclusion rather than a record of it.
 const volumeLeaseBucket = "VIPERBLOCK_VOLUME_LEASES"
 
+// ErrLeaseStoreUnavailable marks a lease claim that failed because JetStream
+// could not be reached, as distinct from one that failed on its merits. The
+// two are opposite conclusions: unreachable means try again shortly, whereas
+// a refused claim means somebody else holds the volume.
+//
+// It exists because they were previously indistinguishable to the caller. A
+// node restarting alongside NATS claims a lease before JetStream is serving,
+// gets a deadline, and the mount is reported permanent — so a guest that only
+// needed a few more seconds is failed and latched out of recovery instead.
+var ErrLeaseStoreUnavailable = errors.New("volume lease store unavailable")
+
+// leaseStoreUnavailable reports whether err means the lease store could not be
+// reached at all. Deliberately a closed list of transport failures: anything
+// unrecognised stays permanent, because retrying a claim that was genuinely
+// refused would keep a volume wedged behind a node that cannot have it.
+func leaseStoreUnavailable(err error) bool {
+	switch {
+	case errors.Is(err, context.DeadlineExceeded),
+		errors.Is(err, nats.ErrNoResponders),
+		errors.Is(err, nats.ErrTimeout),
+		errors.Is(err, nats.ErrConnectionClosed),
+		errors.Is(err, nats.ErrConnectionDraining),
+		errors.Is(err, nats.ErrNoServers),
+		errors.Is(err, jetstream.ErrBucketNotFound):
+		return true
+	}
+	return false
+}
+
 // volumeLeaseTTL is how long a lease outlives the node holding it. A node that
 // dies mid-mount leaves its entry behind and nothing else may open the volume
 // until the entry ages out.
@@ -174,6 +203,9 @@ func (l *volumeLeases) acquire(ctx context.Context, volumeName string) (*volumeL
 			return nil, err
 		}
 	case err != nil:
+		if leaseStoreUnavailable(err) {
+			return nil, fmt.Errorf("claim lease for %s: %w: %w", volumeName, err, ErrLeaseStoreUnavailable)
+		}
 		return nil, fmt.Errorf("claim lease for %s: %w", volumeName, err)
 	}
 

spinifex/services/viperblockd/volume_lease_test.go

@@ -2,6 +2,7 @@ package viperblockd
 
 import (
 	"context"
+	"errors"
 	"fmt"
 	"os"
 	"testing"
@@ -307,3 +308,61 @@ func TestVolumeLease_SurrenderedLeaseIsNotDeletedOnRelease(t *testing.T) {
 	_, err = leases.kv.Get(t.Context(), volumeName)
 	require.NoError(t, err, "releasing a surrendered lease must leave the entry alone: it may be the successor's now")
 }
+
+// TestLeaseStoreUnavailable separates "the store could not answer" from "the
+// store answered no". Only the first is worth retrying, and getting it wrong in
+// either direction is expensive: treating a held lease as transient would retry
+// into a second engine on one encrypted volume, and treating a timeout as
+// permanent strands a guest across a cluster restart, which is what happened.
+func TestLeaseStoreUnavailable(t *testing.T) {
+	tests := []struct {
+		name string
+		err  error
+		want bool
+	}{
+		{"context deadline on a starting JetStream", context.DeadlineExceeded, true},
+		{"wrapped deadline", fmt.Errorf("kv update: %w", context.DeadlineExceeded), true},
+		{"no responders", nats.ErrNoResponders, true},
+		{"request timeout", nats.ErrTimeout, true},
+		{"connection closed", nats.ErrConnectionClosed, true},
+		{"connection draining", nats.ErrConnectionDraining, true},
+		{"no servers", nats.ErrNoServers, true},
+		{"bucket not yet created", jetstream.ErrBucketNotFound, true},
+
+		// The store was reachable and said no. Retrying cannot change it.
+		{"key already exists", jetstream.ErrKeyExists, false},
+		{"lease held by another owner", errVolumeLeaseHeld, false},
+		{"unrelated failure", errors.New("marshal lease record"), false},
+		{"nil", nil, false},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			assert.Equal(t, tt.want, leaseStoreUnavailable(tt.err))
+		})
+	}
+}
+
+// TestVolumeLease_UnreachableStoreIsRetryable is the end of the chain that
+// matters: an acquire against a store that cannot answer must surface as
+// ErrLeaseStoreUnavailable, because that is the sentinel mountErrRetryable
+// looks for and the only reason the relaunch backoff engages.
+func TestVolumeLease_UnreachableStoreIsRetryable(t *testing.T) {
+	_, natsURL := setupEmbeddedNATS(t)
+
+	nc, err := nats.Connect(natsURL)
+	require.NoError(t, err)
+	t.Cleanup(nc.Close)
+
+	leases, err := newVolumeLeases(t.Context(), nc, "node-a")
+	require.NoError(t, err)
+
+	// Close the connection under the bound store, which is the shape of a
+	// daemon that came back before NATS finished starting.
+	nc.Close()
+
+	_, err = leases.acquire(t.Context(), "vol-storedown1")
+	require.Error(t, err)
+	assert.ErrorIs(t, err, ErrLeaseStoreUnavailable,
+		"a store that cannot answer must be reported as retryable, not as a refusal")
+	assert.NotErrorIs(t, err, errVolumeLeaseHeld)
+}

spinifex/vm/recovery_retry.go

@@ -0,0 +1,152 @@
+package vm
+
+import (
+	"log/slog"
+	"time"
+)
+
+// Recovery-failed instances used to stay in StateError until an operator
+// intervened, on the reasoning that a failed relaunch needs a human. That holds
+// when the relaunch failed on its merits. It does not hold when it failed
+// because a dependency was still starting, which is the common case on a
+// coordinated cluster update: the daemon comes back before JetStream is serving,
+// the mount cannot claim its lease, and a guest that needed a few more seconds
+// is parked indefinitely.
+//
+// So the retry below is deliberately narrow. It reconsiders only the failures a
+// not-yet-ready backing store produces, only once the store is confirmed ready,
+// and only a bounded number of times — then it stops and leaves the instance for
+// an operator, which is the right answer for a failure that is actually real.
+
+// retryableRecoveryReasons are the MarkRecoveryFailed reasons that describe a
+// dependency being unavailable rather than an instance being unlaunchable.
+// Reasons absent from this set are never retried automatically: reconnect_failed
+// and pre_relaunch_hook_failed can both indicate a guest whose state needs
+// looking at, and relaunching those on a timer would hide the problem.
+var retryableRecoveryReasons = map[string]bool{
+	"recovery_mount_state_unavailable": true,
+	"recovery_launch_failed":           true,
+}
+
+// recoveryRetryMaxPasses bounds how many times the loop reconsiders a given
+// instance. Five passes at the interval below is a few minutes of trying, which
+// covers a slow cluster start without turning a genuinely broken instance into a
+// relaunch loop that never stops.
+const recoveryRetryMaxPasses = 5
+
+// recoveryRetryInterval is the gap between passes. Package var so tests do not
+// sleep real minutes.
+var recoveryRetryInterval = 30 * time.Second
+
+// RetryRecoveryFailed reconsiders instances parked in StateError by a
+// dependency-unavailable failure and relaunches them. It is a no-op unless the
+// backing store is ready, because retrying against a store that is still down is
+// how they got here.
+//
+// Returns the number of instances it attempted, so a caller can stop polling
+// once there is nothing left to do.
+func (m *Manager) RetryRecoveryFailed() int {
+	if !m.backingStoreReady() {
+		slog.Debug("recovery retry: backing store not ready, deferring")
+		return 0
+	}
+
+	var toRetry []*VM
+
+	for _, instance := range m.Snapshot() {
+		if m.Status(instance) != StateError {
+			continue
+		}
+
+		reason, passes := recoveryFailureReason(instance)
+		if !retryableRecoveryReasons[reason] {
+			continue
+		}
+		if passes >= recoveryRetryMaxPasses {
+			continue
+		}
+
+		// StateError has no outgoing edge in the transition map, so this
+		// mirrors what classifyRestoredInstances already does for a
+		// drain-stopped instance: assign the field directly rather than
+		// widening the state machine for every other caller.
+		m.UpdateState(instance.ID, func(v *VM) {
+			v.Status = StatePending
+			v.recoveryRetries++
+		})
+
+		slog.Info("Retrying recovery-failed instance now the backing store is ready",
+			"instanceId", instance.ID, "reason", reason, "pass", passes+1)
+		toRetry = append(toRetry, instance)
+	}
+
+	if len(toRetry) == 0 {
+		return 0
+	}
+
+	m.relaunchAll(toRetry)
+	return len(toRetry)
+}
+
+// recoveryFailureReason returns the MarkRecoveryFailed reason recorded on the
+// instance and how many automatic retries it has already had.
+// A VM with no Instance has no recorded reason, so it is never retryable: the
+// nil is normal for an entry that failed before its EC2 view was built.
+func recoveryFailureReason(instance *VM) (string, int) {
+	var reason string
+	if instance.Instance != nil && instance.Instance.StateReason != nil &&
+		instance.Instance.StateReason.Message != nil {
+		reason = *instance.Instance.StateReason.Message
+	}
+	return reason, instance.recoveryRetries
+}
+
+// RunRecoveryRetryLoop retries dependency-failed instances until none are left
+// or the passes are exhausted, then returns. It is not a permanent background
+// loop: once the store is up and the backlog is cleared there is nothing for it
+// to do, and a loop that runs forever is one more thing to reason about during
+// an incident.
+//
+// stop is checked before every pass so a coordinated shutdown is never delayed.
+func (m *Manager) RunRecoveryRetryLoop(stop func() bool) {
+	for pass := 1; pass <= recoveryRetryMaxPasses; pass++ {
+		if stop != nil && stop() {
+			slog.Info("recovery retry loop: aborted by shutdown")
+			return
+		}
+
+		if n := m.RetryRecoveryFailed(); n > 0 {
+			slog.Info("recovery retry pass complete", "pass", pass, "attempted", n)
+		}
+
+		if !m.hasRetryableRecoveryFailures() {
+			slog.Debug("recovery retry loop: nothing left to retry")
+			return
+		}
+
+		if stop != nil && stop() {
+			return
+		}
+		time.Sleep(recoveryRetryInterval)
+	}
+
+	if m.hasRetryableRecoveryFailures() {
+		slog.Warn("recovery retry loop: giving up, instances remain in error for operator action",
+			"maxPasses", recoveryRetryMaxPasses)
+	}
+}
+
+// hasRetryableRecoveryFailures reports whether any instance is still parked in
+// StateError with a retryable reason and retries remaining.
+func (m *Manager) hasRetryableRecoveryFailures() bool {
+	for _, instance := range m.Snapshot() {
+		if m.Status(instance) != StateError {
+			continue
+		}
+		reason, passes := recoveryFailureReason(instance)
+		if retryableRecoveryReasons[reason] && passes < recoveryRetryMaxPasses {
+			return true
+		}
+	}
+	return false
+}

spinifex/vm/recovery_retry_test.go

@@ -0,0 +1,219 @@
+//test:in-package — the retry policy is unexported (recoveryRetryMaxPasses,
+//recoveryRetryInterval, hasRetryableRecoveryFailures) and the tests drive it
+//through the same seams restore_test.go uses: relaunchTestManager, the
+//runForRelaunch stub, and VM.recoveryRetries.
+
+package vm
+
+import (
+	"context"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"github.com/aws/aws-sdk-go/aws"
+	"github.com/aws/aws-sdk-go/service/ec2"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+// A guest drain-stopped by a cluster update used to be lost for good if its
+// relaunch lost a race with a still-starting dependency: MarkRecoveryFailed
+// parked it in StateError and nothing ever reconsidered it. These tests cover
+// the retry that now does, and the limits on it — an automatic relaunch that
+// cannot tell "the store was down" from "this guest will not start" would turn
+// one broken instance into a permanent relaunch loop.
+
+// erroredInstance builds a VM parked in StateError by MarkRecoveryFailed with
+// the given reason, which is how the retry decides eligibility.
+func erroredInstance(id, reason string) *VM {
+	return &VM{
+		ID:           id,
+		Status:       StateError,
+		InstanceType: "t3.micro",
+		Instance: &ec2.Instance{
+			StateReason: &ec2.StateReason{
+				Code:    aws.String("Server.RecoveryFailed"),
+				Message: aws.String(reason),
+			},
+		},
+	}
+}
+
+func TestRetryRecoveryFailed(t *testing.T) {
+	t.Run("relaunches an instance failed by an unavailable backing store", func(t *testing.T) {
+		m, _, _, _ := relaunchTestManager(t)
+		m.deps.BackingStoreReady = func() bool { return true }
+
+		origRun := runForRelaunch
+		t.Cleanup(func() { runForRelaunch = origRun })
+		var calls atomic.Int64
+		runForRelaunch = func(_ *Manager, _ context.Context, _ *VM) error {
+			calls.Add(1)
+			return nil
+		}
+
+		inst := erroredInstance("i-retry-me", "recovery_mount_state_unavailable")
+		m.Insert(inst)
+
+		attempted := m.RetryRecoveryFailed()
+
+		assert.Equal(t, 1, attempted, "the instance is eligible and must be retried")
+		assert.EqualValues(t, 1, calls.Load(), "Run must be called for the retried instance")
+		assert.NotEqual(t, StateError, m.Status(inst),
+			"a successful retry must take the instance out of StateError")
+	})
+
+	t.Run("does nothing while the backing store is still down", func(t *testing.T) {
+		m, _, _, _ := relaunchTestManager(t)
+		m.deps.BackingStoreReady = func() bool { return false }
+
+		origRun := runForRelaunch
+		t.Cleanup(func() { runForRelaunch = origRun })
+		var calls atomic.Int64
+		runForRelaunch = func(_ *Manager, _ context.Context, _ *VM) error {
+			calls.Add(1)
+			return nil
+		}
+
+		inst := erroredInstance("i-store-down", "recovery_launch_failed")
+		m.Insert(inst)
+
+		assert.Zero(t, m.RetryRecoveryFailed(),
+			"retrying against a store that is still down is how the instance got here")
+		assert.Zero(t, calls.Load(), "Run must not be called")
+		assert.Equal(t, StateError, m.Status(inst), "the instance must stay parked")
+	})
+
+	t.Run("leaves failures that are not dependency-related alone", func(t *testing.T) {
+		m, _, _, _ := relaunchTestManager(t)
+		m.deps.BackingStoreReady = func() bool { return true }
+
+		origRun := runForRelaunch
+		t.Cleanup(func() { runForRelaunch = origRun })
+		var calls atomic.Int64
+		runForRelaunch = func(_ *Manager, _ context.Context, _ *VM) error {
+			calls.Add(1)
+			return nil
+		}
+
+		// reconnect_failed and pre_relaunch_hook_failed can both mean the
+		// guest itself needs looking at. Relaunching those on a timer would
+		// hide the problem rather than fix it.
+		for _, reason := range []string{"reconnect_failed", "pre_relaunch_hook_failed", ""} {
+			m.Insert(erroredInstance("i-"+reason+"-x", reason))
+		}
+
+		// An entry that failed before its EC2 view was built has no reason at
+		// all. That is ordinary, not a crash.
+		m.Insert(&VM{ID: "i-no-ec2-view", Status: StateError})
+
+		assert.Zero(t, m.RetryRecoveryFailed(),
+			"only dependency-unavailable reasons are retried automatically")
+		assert.Zero(t, calls.Load())
+		assert.False(t, m.hasRetryableRecoveryFailures())
+	})
+
+	t.Run("gives up after the retry budget and leaves the instance for an operator", func(t *testing.T) {
+		m, _, _, _ := relaunchTestManager(t)
+		m.deps.BackingStoreReady = func() bool { return true }
+
+		origRun := runForRelaunch
+		t.Cleanup(func() { runForRelaunch = origRun })
+		// Not ErrMountRetryable, so relaunchWithRetry returns on the first
+		// attempt and every pass re-parks the instance in StateError.
+		runForRelaunch = func(_ *Manager, _ context.Context, _ *VM) error {
+			return assert.AnError
+		}
+
+		inst := erroredInstance("i-never-starts", "recovery_launch_failed")
+		m.Insert(inst)
+
+		for pass := 1; pass <= recoveryRetryMaxPasses; pass++ {
+			require.Equal(t, 1, m.RetryRecoveryFailed(),
+				"pass %d is within budget and must still be attempted", pass)
+			require.Equal(t, StateError, m.Status(inst),
+				"a permanently failing instance must be re-parked each pass")
+		}
+
+		assert.Zero(t, m.RetryRecoveryFailed(),
+			"the budget is exhausted, so the instance is left for an operator rather than relaunched forever")
+		assert.False(t, m.hasRetryableRecoveryFailures(),
+			"an instance past its budget must not keep the retry loop alive")
+	})
+
+	t.Run("running and stopped instances are never touched", func(t *testing.T) {
+		m, _, _, _ := relaunchTestManager(t)
+		m.deps.BackingStoreReady = func() bool { return true }
+
+		origRun := runForRelaunch
+		t.Cleanup(func() { runForRelaunch = origRun })
+		var calls atomic.Int64
+		runForRelaunch = func(_ *Manager, _ context.Context, _ *VM) error {
+			calls.Add(1)
+			return nil
+		}
+
+		for _, st := range []InstanceState{StateRunning, StateStopped, StatePending, StateTerminated} {
+			m.Insert(&VM{ID: "i-" + string(st), Status: st, Instance: &ec2.Instance{}})
+		}
+
+		assert.Zero(t, m.RetryRecoveryFailed(), "only StateError instances are candidates")
+		assert.Zero(t, calls.Load())
+	})
+}
+
+func TestRunRecoveryRetryLoop(t *testing.T) {
+	t.Run("returns as soon as there is nothing left to retry", func(t *testing.T) {
+		m, _, _, _ := relaunchTestManager(t)
+		m.deps.BackingStoreReady = func() bool { return true }
+
+		origInterval := recoveryRetryInterval
+		recoveryRetryInterval = time.Millisecond
+		t.Cleanup(func() { recoveryRetryInterval = origInterval })
+
+		origRun := runForRelaunch
+		t.Cleanup(func() { runForRelaunch = origRun })
+		runForRelaunch = func(_ *Manager, _ context.Context, _ *VM) error { return nil }
+
+		m.Insert(erroredInstance("i-recovers", "recovery_mount_state_unavailable"))
+
+		done := make(chan struct{})
+		go func() { m.RunRecoveryRetryLoop(nil); close(done) }()
+
+		select {
+		case <-done:
+		case <-time.After(5 * time.Second):
+			t.Fatal("loop did not return once the backlog was cleared")
+		}
+	})
+
+	t.Run("aborts promptly on shutdown instead of sleeping out its passes", func(t *testing.T) {
+		m, _, _, _ := relaunchTestManager(t)
+		m.deps.BackingStoreReady = func() bool { return true }
+
+		origInterval := recoveryRetryInterval
+		recoveryRetryInterval = time.Hour // a pass that sleeps would hang the test
+		t.Cleanup(func() { recoveryRetryInterval = origInterval })
+
+		origRun := runForRelaunch
+		t.Cleanup(func() { runForRelaunch = origRun })
+		var calls atomic.Int64
+		runForRelaunch = func(_ *Manager, _ context.Context, _ *VM) error {
+			calls.Add(1)
+			return assert.AnError
+		}
+
+		m.Insert(erroredInstance("i-shutting-down", "recovery_launch_failed"))
+
+		done := make(chan struct{})
+		go func() { m.RunRecoveryRetryLoop(func() bool { return true }); close(done) }()
+
+		select {
+		case <-done:
+		case <-time.After(5 * time.Second):
+			t.Fatal("shutdown must abort the loop before it sleeps")
+		}
+		assert.Zero(t, calls.Load(), "no relaunch may start once shutdown is signalled")
+	})
+}

spinifex/vm/vm.go

@@ -113,6 +113,12 @@ type VM struct {
 	// Set when ownership is released on stop for shared KV storage.
 	LastNode string `json:"last_node,omitempty"`
 
+	// recoveryRetries counts automatic retries of a recovery failure caused by
+	// an unavailable dependency. Unexported and unpersisted on purpose: the
+	// budget is per daemon run, so a node that restarts gets a fresh one, which
+	// is right because a restart is new information about the dependency.
+	recoveryRetries int
+
 	// AZ is the availability zone of the node that last ran this instance,
 	// stamped by that node from its own config. Placement observed by the
 	// owner, not something an operator requests.

tests/e2e/harness/multinode_cluster_restart.go

@@ -0,0 +1,201 @@
+//go:build e2e
+
+package harness
+
+import (
+	"context"
+	"fmt"
+	"strings"
+	"testing"
+	"time"
+)
+
+// A coordinated cluster shutdown followed by a start is the operation every
+// software update performs, and it is the one path that had no e2e coverage:
+// StopNode/StartNode simulate a hard outage, which deliberately leaves guests
+// running, so nothing exercised a guest that was drained to stopped, had its
+// volumes sealed, and then had to relaunch from that sealed state against
+// dependencies that were themselves still starting. A guest was lost that way.
+
+// clusterShutdownPhaseTimeout is passed to `spx admin cluster shutdown` as its
+// per-phase budget. The default is 120s; DRAIN legitimately takes longer when
+// several guests unmount at once, and a drain that times out is reported as
+// success, so allowing the time is cheaper than the ambiguity.
+const clusterShutdownPhaseTimeout = 10 * time.Minute
+
+// spxProcessDrainTimeout bounds the wait for every spx process to exit after
+// the units are stopped. Stopping the target returns once the target is
+// inactive, which is not the same as every service having exited.
+const spxProcessDrainTimeout = 3 * time.Minute
+
+// ClusterShutdownAndRestart performs the full operator shutdown sequence and
+// brings the cluster back: the coordinated phased shutdown, a stop of
+// spinifex.target on every node, a confirmed wait for every spx process to exit
+// and for systemd to have no spinifex job left queued, then a start of the
+// target everywhere. It returns once NATS has reformed and every gateway
+// answers, so a caller can immediately assert on guest state.
+//
+// The spx-process wait is the step that makes this a real restart rather than a
+// systemctl call that returned early. spinifex-shutdown.service is ordered
+// After= the storage services, so its drain runs while they are still up and
+// their stop jobs queue behind it; `systemctl stop` returns as soon as the
+// target is inactive, leaving predastore, viperblock and NATS running.
+func ClusterShutdownAndRestart(t *testing.T, c *Cluster, env *Env) {
+	t.Helper()
+
+	if len(c.Nodes) < 2 {
+		t.Fatalf("cluster restart needs a multi-node cluster, have %d", len(c.Nodes))
+	}
+
+	ClusterShutdown(t, c)
+
+	Step(t, "start spinifex.target on all %d nodes", len(c.Nodes))
+	for _, n := range c.Nodes {
+		StartNode(t, n)
+	}
+
+	// WaitNATSPeers counts remote peers, so a whole cluster is one less than
+	// the node count.
+	Step(t, "wait NATS to reform to %d peers", len(c.Nodes)-1)
+	c.WaitNATSPeers(t, len(c.Nodes)-1, WithTimeout(3*time.Minute), WithPoll(2*time.Second))
+
+	Step(t, "wait every gateway to answer")
+	c.WaitGatewayHealthy(t, WithTimeout(3*time.Minute), WithPoll(2*time.Second))
+
+	if env != nil {
+		Step(t, "wait every daemon ready")
+		c.WaitDaemonReady(t, env, WithTimeout(3*time.Minute), WithPoll(2*time.Second))
+	}
+}
+
+// ClusterShutdown runs the coordinated shutdown and leaves the cluster down.
+// Split out from ClusterShutdownAndRestart so a caller can assert on the
+// stopped state, and so a t.Cleanup can restart unconditionally.
+func ClusterShutdown(t *testing.T, c *Cluster) {
+	t.Helper()
+
+	ssh := NewPeerSSH()
+	leader := c.Nodes[0]
+
+	Step(t, "spx admin cluster shutdown via %s", leader.Name)
+	ctx, cancel := context.WithTimeout(context.Background(), clusterShutdownPhaseTimeout+2*time.Minute)
+	out, err := ssh.Run(ctx, leader.Addr, fmt.Sprintf(
+		"sudo spx admin cluster shutdown --timeout %s --config /etc/spinifex/spinifex.toml",
+		clusterShutdownPhaseTimeout))
+	cancel()
+	Detail(t, "cluster_shutdown", string(out))
+	if err != nil {
+		// The phases can complete and still lose the SSH channel as NATS goes
+		// down under it. The per-unit stop below is the authoritative teardown.
+		t.Logf("cluster shutdown on %s reported %v — proceeding to the per-unit stop", leader.Name, err)
+	}
+
+	// StopNode is not used here. It stops the service units directly and leaves
+	// spinifex.target active, which is right for the hard outage it simulates
+	// and wrong for a real shutdown: the target then reads as satisfied while
+	// its units are dead, and a later start can race a stop job that is still
+	// queued. Exactly that lost node3's NATS on the first run — stopped and
+	// never started again, so the cluster reformed to 2 peers of 3.
+	Step(t, "stop spinifex.target on all %d nodes", len(c.Nodes))
+	for _, n := range c.Nodes {
+		stopTarget(t, n)
+	}
+
+	Step(t, "confirm no spx process survives on any node")
+	for _, n := range c.Nodes {
+		waitNoSpxProcess(t, n)
+	}
+
+	Step(t, "confirm systemd has no spinifex jobs still queued")
+	for _, n := range c.Nodes {
+		waitNoPendingJobs(t, n)
+	}
+}
+
+// stopTarget stops spinifex.target, which is what the operator procedure runs
+// and what update-nodes.sh runs. Non-fatal: the teardown can racily drop the
+// SSH channel, and the waits below are the authoritative gate.
+func stopTarget(t *testing.T, node Node) {
+	t.Helper()
+
+	ssh := NewPeerSSH()
+	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Minute)
+	defer cancel()
+	if _, err := ssh.Run(ctx, node.Addr, "sudo systemctl stop spinifex.target"); err != nil {
+		t.Logf("stop spinifex.target on %s: %v (proceeding — the waits are the gate)", node.Name, err)
+	}
+}
+
+// waitNoPendingJobs blocks until systemd has no queued job for any spinifex
+// unit. Starting the target with a stop still queued is how a service ends up
+// stopped by a job that completes after the start was issued.
+func waitNoPendingJobs(t *testing.T, node Node) {
+	t.Helper()
+
+	ssh := NewPeerSSH()
+	EventuallyErr(t, func() error {
+		ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
+		defer cancel()
+		out, err := ssh.Run(ctx, node.Addr,
+			"systemctl list-jobs --no-legend --no-pager | grep -c spinifex || true")
+		if err != nil {
+			return fmt.Errorf("%s list-jobs: %w", node.Name, err)
+		}
+		if n := strings.TrimSpace(string(out)); n != "0" {
+			return fmt.Errorf("%s still has %s queued spinifex job(s)", node.Name, n)
+		}
+		return nil
+	}, 2*time.Minute, 2*time.Second)
+}
+
+// stopUnitsHoldingSpx stops whatever units are still running the spx binary,
+// found from the processes themselves rather than from a list.
+//
+// Every service is the same binary — `spx service <name> start` — so the set
+// that has to stop is "whoever is still running it", and a hard-coded list goes
+// stale silently the moment a service is added. StopNode's list omits
+// spinifex-northstar and spinifex-qmp-collector, which is harmless for the hard
+// outage it simulates and not harmless here.
+func stopUnitsHoldingSpx(t *testing.T, node Node) {
+	t.Helper()
+
+	ssh := NewPeerSSH()
+	ctx, cancel := context.WithTimeout(context.Background(), 3*time.Minute)
+	defer cancel()
+
+	// The unit name comes from each survivor's cgroup, so this needs no
+	// knowledge of which units exist.
+	const cmd = `for pid in $(pgrep -x spx || true); do
+    unit=$(grep -oE 'spinifex-[a-z0-9-]+\.service' "/proc/$pid/cgroup" 2>/dev/null | head -1)
+    [ -n "$unit" ] && sudo systemctl stop "$unit"
+done; true`
+	if _, err := ssh.Run(ctx, node.Addr, cmd); err != nil {
+		t.Logf("stopUnitsHoldingSpx %s: %v (the wait below is the gate)", node.Name, err)
+	}
+}
+
+// waitNoSpxProcess blocks until `pgrep -x spx` finds nothing on node. A
+// surviving process means the stop returned before the service did, and
+// anything asserted after that is measuring the old process.
+func waitNoSpxProcess(t *testing.T, node Node) {
+	t.Helper()
+
+	ssh := NewPeerSSH()
+	EventuallyErr(t, func() error {
+		ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
+		defer cancel()
+		// pgrep exits 1 when nothing matches, which is the outcome wanted, so
+		// the count is read from stdout rather than from the exit status.
+		out, err := ssh.Run(ctx, node.Addr, "pgrep -x spx | wc -l")
+		if err != nil {
+			return fmt.Errorf("%s pgrep spx: %w", node.Name, err)
+		}
+		if n := strings.TrimSpace(string(out)); n != "0" {
+			// A survivor is a unit that was never asked to stop, or one that
+			// systemd restarted. Ask again rather than only polling.
+			stopUnitsHoldingSpx(t, node)
+			return fmt.Errorf("%s still has %s spx process(es)", node.Name, n)
+		}
+		return nil
+	}, spxProcessDrainTimeout, 3*time.Second)
+}

tests/e2e/multinode/cluster_restart_test.go

@@ -0,0 +1,111 @@
+//go:build e2e
+
+package multinode
+
+import (
+	"fmt"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/aws/aws-sdk-go/aws"
+	"github.com/aws/aws-sdk-go/service/ec2"
+	"github.com/mulgadc/spinifex/tests/e2e/harness"
+	"github.com/stretchr/testify/require"
+)
+
+// runClusterRestartResumesGuests is the software-update path: a coordinated
+// shutdown drains every guest to stopped and seals its volumes, then the
+// cluster comes back and each guest has to relaunch from that sealed state
+// while its dependencies are themselves still starting.
+//
+// NodeFailure/NodeRecovery do not cover this. They stop the service units
+// directly so guests keep running and the target's drain never fires, which
+// makes recovery a QMP re-attach to a live process. Here the process is gone
+// and the volume has to be re-opened, which is where a guest was lost: the
+// relaunch raced a JetStream that had not finished starting, the lease claim
+// timed out, and the failure was recorded as permanent.
+func runClusterRestartResumesGuests(t *testing.T, fix *Fixture) {
+	harness.Phase(t, "Multinode — Cluster Restart Resumes Guests")
+
+	require.GreaterOrEqualf(t, len(fix.Cluster.Nodes), 2,
+		"a cluster restart needs a multi-node cluster, have %d", len(fix.Cluster.Nodes))
+
+	trio := needInstanceTrio(t, fix)
+	require.NotEmpty(t, trio, "the restart is only meaningful with guests to resume")
+
+	harness.Step(t, "confirm all %d guests are running before the shutdown", len(trio))
+	for _, id := range trio {
+		harness.WaitForInstanceState(t, fix.AWS, id, "running")
+	}
+
+	// If an assertion below fails mid-restart the cluster must not be left
+	// down for every later test in the package.
+	t.Cleanup(func() {
+		for _, n := range fix.Cluster.Nodes {
+			harness.StartNode(t, n)
+		}
+	})
+
+	harness.ClusterShutdownAndRestart(t, fix.Cluster, fix.Env)
+
+	harness.Step(t, "every guest resumes to running without operator action")
+	var failures []string
+	for _, id := range trio {
+		if err := waitInstanceRunningNotErrored(t, fix, id, 5*time.Minute); err != nil {
+			failures = append(failures, err.Error())
+		}
+	}
+	require.Emptyf(t, failures,
+		"guests must resume after a coordinated restart; an instance left in error is a guest an operator has to recover by hand:\n%s",
+		strings.Join(failures, "\n"))
+}
+
+// waitInstanceRunningNotErrored polls until id is running, failing early and
+// with the recorded reason if it lands in error instead. Waiting out the full
+// timeout on an instance already parked in error would hide why.
+func waitInstanceRunningNotErrored(t *testing.T, fix *Fixture, id string, timeout time.Duration) error {
+	t.Helper()
+
+	deadline := time.Now().Add(timeout)
+	var last string
+	for time.Now().Before(deadline) {
+		out, err := fix.AWS.EC2.DescribeInstances(&ec2.DescribeInstancesInput{
+			InstanceIds: []*string{aws.String(id)},
+		})
+		if err != nil {
+			// The gateway can still be settling immediately after the restart.
+			last = err.Error()
+			time.Sleep(5 * time.Second)
+			continue
+		}
+		if len(out.Reservations) == 0 || len(out.Reservations[0].Instances) == 0 {
+			last = "not returned by DescribeInstances"
+			time.Sleep(5 * time.Second)
+			continue
+		}
+
+		inst := out.Reservations[0].Instances[0]
+		state := aws.StringValue(inst.State.Name)
+		last = state
+		switch state {
+		case "running":
+			return nil
+		case "error", "terminated":
+			return fmt.Errorf("%s reached %q after the restart instead of running: %s",
+				id, state, stateReason(inst))
+		}
+		time.Sleep(5 * time.Second)
+	}
+	return fmt.Errorf("%s did not resume within %s, last state %q", id, timeout, last)
+}
+
+// stateReason renders an instance's StateReason, which carries the recovery
+// failure that explains a guest parked in error.
+func stateReason(inst *ec2.Instance) string {
+	if inst.StateReason == nil {
+		return "no state reason recorded"
+	}
+	return fmt.Sprintf("%s: %s",
+		aws.StringValue(inst.StateReason.Code), aws.StringValue(inst.StateReason.Message))
+}

tests/e2e/multinode/tests_test.go

@@ -73,6 +73,16 @@ func TestMultinodeNodeRecovery(t *testing.T) {
 	runNodeRecovery(t, requireMultiNodeFixture(t))
 }
 
+// TestMultinodeClusterRestartResumesGuests is sequential and declared after
+// NodeRecovery so the cluster is whole before it takes the whole thing down.
+// It is the software-update path — a coordinated drain and restart — which the
+// node-failure tests deliberately do not cover: they leave guests running, and
+// the failure this catches only appears when a guest has to relaunch from a
+// sealed volume against dependencies that are still starting.
+func TestMultinodeClusterRestartResumesGuests(t *testing.T) {
+	runClusterRestartResumesGuests(t, requireMultiNodeFixture(t))
+}
+
 // TestMultinodeOVNRaft is sequential: stops ovn-central on the NB leader to
 // prove DB failover, restoring it in cleanup before later tests run.
 func TestMultinodeOVNRaft(t *testing.T) {