fix: retry instance recovery when the backing store is not yet ready (#813)

* fix: retry recovery relaunch when the backing store is not yet ready

* fix: treat a missing EBS mount responder as a retryable recovery failure

spinifex/daemon/vm_adapters.go

@@ -175,6 +175,11 @@ func (a *volumeMounterAdapter) Mount(instance *vm.VM) error {
 
 		if err != nil {
 			slog.Error("Failed to request EBS mount", "err", err)
+			// A missing responder means viperblockd has not finished starting;
+			// during recovery that is transient, so let the caller retry.
+			if errors.Is(err, nats.ErrNoResponders) {
+				return rollback(fmt.Errorf("ebs mount had no responder: %w", vm.ErrMountRetryable))
+			}
 			return rollback(err)
 		}
 
@@ -185,7 +190,10 @@ func (a *volumeMounterAdapter) Mount(instance *vm.VM) error {
 		}
 
 		if ebsMountResponse.Error != "" {
-			slog.Error("Failed to mount volume", "error", ebsMountResponse.Error)
+			slog.Error("Failed to mount volume", "error", ebsMountResponse.Error, "retryable", ebsMountResponse.Retryable)
+			if ebsMountResponse.Retryable {
+				return rollback(fmt.Errorf("failed to mount volume: %s: %w", ebsMountResponse.Error, vm.ErrMountRetryable))
+			}
 			return rollback(fmt.Errorf("failed to mount volume: %s", ebsMountResponse.Error))
 		}
 

spinifex/daemon/vm_adapters_test.go

@@ -572,3 +572,65 @@ func TestInstanceCleanerAdapter_ReleaseAttachedENIs_ListInstanceENIsErrorTolerat
 	require.NoError(t, cleaner.DetachAndDeleteENI(instance),
 		"an enumeration failure must not surface as a terminate error")
 }
+
+// TestVolumeMounterAdapter_Mount_WrapsErrMountRetryable verifies that a
+// viperblockd ebs.mount response with Retryable=true is wrapped so
+// vm.Manager's recovery-relaunch retry can match it with errors.Is. A
+// Retryable=false response must NOT match, so relaunchAll still fails fast
+// on permanent mount errors.
+func TestVolumeMounterAdapter_Mount_WrapsErrMountRetryable(t *testing.T) {
+	daemon := createTestDaemon(t, sharedNATSURL)
+	adapter := newVolumeMounterAdapter(daemon.natsConn, daemon.node, nil)
+
+	sub, err := daemon.natsConn.Subscribe("ebs."+daemon.node+".mount", func(msg *nats.Msg) {
+		resp := types.EBSMountResponse{Error: "state not found", Retryable: true}
+		data, _ := json.Marshal(resp)
+		_ = msg.Respond(data)
+	})
+	require.NoError(t, err)
+	defer sub.Unsubscribe()
+
+	instance := &vm.VM{ID: "i-mount-retryable"}
+	instance.EBSRequests.Requests = []types.EBSRequest{{Name: "vol-retryable"}}
+
+	err = adapter.Mount(instance)
+	require.Error(t, err)
+	assert.ErrorIs(t, err, vm.ErrMountRetryable,
+		"a Retryable mount response must wrap vm.ErrMountRetryable so relaunchAll can retry it")
+}
+
+func TestVolumeMounterAdapter_Mount_PermanentErrorNotWrapped(t *testing.T) {
+	daemon := createTestDaemon(t, sharedNATSURL)
+	adapter := newVolumeMounterAdapter(daemon.natsConn, daemon.node, nil)
+
+	sub, err := daemon.natsConn.Subscribe("ebs."+daemon.node+".mount", func(msg *nats.Msg) {
+		resp := types.EBSMountResponse{Error: "volume vol-permanent is already mounted read_only=true on this node"}
+		data, _ := json.Marshal(resp)
+		_ = msg.Respond(data)
+	})
+	require.NoError(t, err)
+	defer sub.Unsubscribe()
+
+	instance := &vm.VM{ID: "i-mount-permanent"}
+	instance.EBSRequests.Requests = []types.EBSRequest{{Name: "vol-permanent"}}
+
+	err = adapter.Mount(instance)
+	require.Error(t, err)
+	assert.NotErrorIs(t, err, vm.ErrMountRetryable,
+		"a non-Retryable mount response must not be mistaken for a transient failure")
+}
+
+func TestVolumeMounterAdapter_Mount_NoResponderIsRetryable(t *testing.T) {
+	daemon := createTestDaemon(t, sharedNATSURL)
+	adapter := newVolumeMounterAdapter(daemon.natsConn, daemon.node, nil)
+
+	// No subscriber on the mount subject: viperblockd is still starting, so
+	// the request returns nats.ErrNoResponders, which must be retryable.
+	instance := &vm.VM{ID: "i-mount-noresponder"}
+	instance.EBSRequests.Requests = []types.EBSRequest{{Name: "vol-noresponder"}}
+
+	err := adapter.Mount(instance)
+	require.Error(t, err)
+	assert.ErrorIs(t, err, vm.ErrMountRetryable,
+		"a mount request with no responder must wrap vm.ErrMountRetryable so relaunchAll can retry it")
+}

spinifex/services/viperblockd/provider_handlers.go

@@ -1322,6 +1322,14 @@ func constructMountedVB(ctx context.Context, cfg *Config, volumeName string) (*v
 	return vb, nbdCacheSize, nil
 }
 
+// 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
+// recovery-relaunch retry in vm.Manager.
+func mountErrRetryable(err error) bool {
+	return errors.Is(err, viperblock.ErrStateNotFound) || errors.Is(err, viperblock.ErrStateBackendUnavailable)
+}
+
 // mountVolume is launchService's ebs.mount body, extracted so the legacy
 // ebs.mount handler and the ebs.provider.v1.*.mount handler share one nbdkit
 // mount path instead of risking two divergent implementations of it. It
@@ -1359,6 +1367,7 @@ func mountVolume(ctx context.Context, cfg *Config, nc *nats.Conn, volumeName str
 	vb, nbdCacheSize, lease, err := cfg.buildVB(ctx, volumeName)
 	if err != nil {
 		ebsResponse.Error = err.Error()
+		ebsResponse.Retryable = mountErrRetryable(err)
 		return ebsResponse, err
 	}
 

spinifex/services/viperblockd/provider_handlers_test.go

@@ -15,6 +15,8 @@ import (
 	"bytes"
 	"context"
 	"encoding/json"
+	"errors"
+	"fmt"
 	"maps"
 	"os"
 	"path/filepath"
@@ -25,11 +27,34 @@ import (
 	"github.com/aws/aws-sdk-go/service/s3"
 	"github.com/mulgadc/spinifex/spinifex/ebsprovider"
 	"github.com/mulgadc/spinifex/spinifex/objectstore"
+	"github.com/mulgadc/viperblock/viperblock"
 	"github.com/nats-io/nats.go"
 	"github.com/stretchr/testify/assert"
 	"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.
+func TestMountErrRetryable(t *testing.T) {
+	tests := []struct {
+		name string
+		err  error
+		want bool
+	}{
+		{"ErrStateNotFound", viperblock.ErrStateNotFound, true},
+		{"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},
+		{"plain error", errors.New("some other mount failure"), false},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			assert.Equal(t, tt.want, mountErrRetryable(tt.err))
+		})
+	}
+}
+
 // versionedErrorResponse decodes the common shape every ebs.provider.v1.*
 // response carries: an embedded Versioned and an optional ProviderError.
 // Every concrete response type marshals a superset of these fields, so

spinifex/types/ebs.go

@@ -54,6 +54,10 @@ type EBSMountResponse struct {
 	URI     string `json:"URI"`
 	Mounted bool   `json:"Mounted"`
 	Error   string `json:"Error"`
+	// Retryable marks a mount failure caused by a backing store that is not
+	// yet ready (e.g. a transient state-load gap), as opposed to a
+	// permanent failure the caller should not retry.
+	Retryable bool `json:"Retryable"`
 }
 
 type EBSUnMountResponse struct {

spinifex/vm/recovery.go

@@ -0,0 +1,9 @@
+package vm
+
+import "errors"
+
+// ErrMountRetryable is returned (wrapped) by VolumeMounter.Mount when the
+// backing store rejected the mount because it is not yet ready rather than
+// permanently unavailable. relaunchAll retries on this sentinel instead of
+// failing the instance immediately.
+var ErrMountRetryable = errors.New("volume mount failed against a not-yet-ready backing store")

spinifex/vm/restore.go

@@ -2,6 +2,7 @@ package vm
 
 import (
 	"context"
+	"errors"
 	"fmt"
 	"log/slog"
 	"net"
@@ -19,6 +20,26 @@ import (
 // maxConcurrentRecovery bounds the recovery fan-out; cold-AMI clones are I/O-heavy.
 const maxConcurrentRecovery = 2
 
+// relaunchMaxAttempts bounds the retry loop a recovery relaunch runs against
+// ErrMountRetryable. relaunchBackoffBase/relaunchBackoffCap are package vars
+// (not consts) so tests can shrink them to avoid sleeping real seconds.
+const relaunchMaxAttempts = 5
+
+var (
+	relaunchBackoffBase = 2 * time.Second
+	relaunchBackoffCap  = 30 * time.Second
+)
+
+// errRelaunchAborted signals relaunchWithRetry stopped because the daemon
+// began shutting down, not because attempts were exhausted or the failure
+// was permanent. The caller must not mark the instance terminal.
+var errRelaunchAborted = errors.New("relaunch retry aborted by shutdown")
+
+// runForRelaunch is a test seam over (*Manager).Run, mirroring
+// attachQMPForReconnect below: it lets tests stub a full launch attempt
+// (success or failure) without spinning up a real QEMU process.
+var runForRelaunch = (*Manager).Run
+
 // Restore loads persisted VM state and re-launches instances that are neither
 // terminated nor user-stopped. Terminated/stopped instances migrate to shared KV;
 // running instances with live QEMU reconnect via QMP; others relaunch via Run.
@@ -313,7 +334,17 @@ func (m *Manager) relaunchAll(toLaunch []*VM) {
 			slog.Info("Launching instance (recovery)",
 				"instance", inst.ID, "managedBy", inst.ManagedBy, "instanceType", inst.InstanceType)
 			// Restore runs at daemon start with no request context.
-			if err := m.Run(context.Background(), inst); err != nil {
+			if err := m.relaunchWithRetry(context.Background(), inst); err != nil {
+				if errors.Is(err, errRelaunchAborted) {
+					slog.Info("Recovery relaunch aborted by shutdown signal", "instanceId", inst.ID)
+					return
+				}
+				if errors.Is(err, ErrMountRetryable) {
+					slog.Error("Recovery relaunch exhausted retries against a not-yet-ready backing store",
+						"instanceId", inst.ID, "managedBy", inst.ManagedBy, "instanceType", inst.InstanceType, "err", err)
+					m.MarkRecoveryFailed(inst, "recovery_mount_state_unavailable")
+					return
+				}
 				slog.Error("Failed to launch instance during recovery",
 					"instanceId", inst.ID, "managedBy", inst.ManagedBy, "instanceType", inst.InstanceType, "err", err)
 				m.MarkRecoveryFailed(inst, "recovery_launch_failed")
@@ -323,6 +354,45 @@ func (m *Manager) relaunchAll(toLaunch []*VM) {
 	wg.Wait()
 }
 
+// relaunchWithRetry calls m.Run, retrying only when the failure is
+// ErrMountRetryable (the backing store is not yet ready). Any other error
+// returns immediately after a single attempt. Bounded by relaunchMaxAttempts
+// with exponential backoff (relaunchBackoffBase, capped at
+// relaunchBackoffCap). Checks the shutdown signal before every attempt and
+// before every sleep so a coordinated shutdown is never delayed by backoff.
+func (m *Manager) relaunchWithRetry(ctx context.Context, inst *VM) error {
+	var err error
+	delay := relaunchBackoffBase
+
+	for attempt := 1; attempt <= relaunchMaxAttempts; attempt++ {
+		if m.deps.ShutdownSignal != nil && m.deps.ShutdownSignal() {
+			return errRelaunchAborted
+		}
+
+		err = runForRelaunch(m, ctx, inst)
+		if err == nil {
+			return nil
+		}
+		if !errors.Is(err, ErrMountRetryable) {
+			return err
+		}
+		if attempt == relaunchMaxAttempts {
+			break
+		}
+
+		slog.Warn("Recovery relaunch mount not ready, retrying",
+			"instanceId", inst.ID, "attempt", attempt, "err", err)
+
+		if m.deps.ShutdownSignal != nil && m.deps.ShutdownSignal() {
+			return errRelaunchAborted
+		}
+		time.Sleep(delay)
+		delay = min(delay*2, relaunchBackoffCap)
+	}
+
+	return err
+}
+
 // attachQMPForReconnect is a test seam over (*Manager).AttachQMP so tests can
 // drive reconnectInstance without spawning the 30s heartbeat goroutine (goleak).
 var attachQMPForReconnect = (*Manager).AttachQMP

spinifex/vm/restore_test.go

@@ -2,6 +2,7 @@ package vm
 
 import (
 	"bytes"
+	"context"
 	"errors"
 	"fmt"
 	"log/slog"
@@ -859,6 +860,133 @@ func TestRelaunchAll(t *testing.T) {
 		assert.Len(t, mounter.mounted, total-1,
 			"every eligible instance must reach Mount: only the flipped one is skipped")
 	})
+
+	t.Run("ErrMountRetryable is retried until Run succeeds", func(t *testing.T) {
+		m, _, _, _ := relaunchTestManager(t)
+
+		origBase, origCap := relaunchBackoffBase, relaunchBackoffCap
+		relaunchBackoffBase, relaunchBackoffCap = time.Millisecond, 4*time.Millisecond
+		t.Cleanup(func() { relaunchBackoffBase, relaunchBackoffCap = origBase, origCap })
+
+		origRun := runForRelaunch
+		t.Cleanup(func() { runForRelaunch = origRun })
+
+		const wantAttempts = 3
+		var calls atomic.Int64
+		runForRelaunch = func(_ *Manager, _ context.Context, _ *VM) error {
+			if calls.Add(1) < wantAttempts {
+				return fmt.Errorf("state not yet loaded: %w", ErrMountRetryable)
+			}
+			return nil
+		}
+
+		instance := &VM{ID: "i-retry-ok", Status: StatePending, InstanceType: "t3.micro", Instance: &ec2.Instance{}}
+		m.Insert(instance)
+
+		m.relaunchAll([]*VM{instance})
+
+		assert.EqualValues(t, wantAttempts, calls.Load(),
+			"Run must be called once per attempt until it stops returning ErrMountRetryable")
+		status := m.Status(instance)
+		assert.NotEqual(t, StateError, status,
+			"a retry that eventually succeeds must never drive the instance into StateError")
+		assert.Equal(t, StatePending, status,
+			"the stub bypasses the real launch/TransitionState, so status stays whatever Run's caller left it at "+
+				"(Pending here) rather than advancing to Running — the point under test is that it is NOT StateError")
+	})
+
+	t.Run("permanent Run error marks recovery_failed after exactly one attempt", func(t *testing.T) {
+		m, mounter, _, rt := relaunchTestManager(t)
+
+		instance := &VM{ID: "i-permanent-fail", Status: StatePending, InstanceType: "t3.micro", Instance: &ec2.Instance{}}
+		m.Insert(instance)
+		errored := rt.waitFor(instance.ID, StateError)
+
+		var calls atomic.Int64
+		mounter.behavior[instance.ID] = func(*VM) error {
+			calls.Add(1)
+			return errors.New("permanent mount failure, not a backing-store readiness issue")
+		}
+
+		m.relaunchAll([]*VM{instance})
+
+		select {
+		case <-errored:
+		case <-time.After(markFailedDeadline):
+			t.Fatalf("MarkRecoveryFailed did not record StateError transition within %s", markFailedDeadline)
+		}
+
+		assert.EqualValues(t, 1, calls.Load(),
+			"a permanent (non-ErrMountRetryable) error must fail fast with no retry")
+		assert.Equal(t, StateError, m.Status(instance))
+		require.NotNil(t, instance.Instance.StateReason)
+		assert.Equal(t, "recovery_launch_failed", *instance.Instance.StateReason.Message)
+	})
+
+	t.Run("exhausted retries against a still-unready backing store marks recovery_mount_state_unavailable", func(t *testing.T) {
+		m, _, _, _ := relaunchTestManager(t)
+
+		origBase, origCap := relaunchBackoffBase, relaunchBackoffCap
+		relaunchBackoffBase, relaunchBackoffCap = time.Millisecond, 4*time.Millisecond
+		t.Cleanup(func() { relaunchBackoffBase, relaunchBackoffCap = origBase, origCap })
+
+		origRun := runForRelaunch
+		t.Cleanup(func() { runForRelaunch = origRun })
+
+		var calls atomic.Int64
+		runForRelaunch = func(_ *Manager, _ context.Context, _ *VM) error {
+			calls.Add(1)
+			return fmt.Errorf("state still not loaded: %w", ErrMountRetryable)
+		}
+
+		instance := &VM{ID: "i-retry-exhausted", Status: StatePending, InstanceType: "t3.micro", Instance: &ec2.Instance{}}
+		m.Insert(instance)
+
+		m.relaunchAll([]*VM{instance})
+
+		assert.EqualValues(t, relaunchMaxAttempts, calls.Load(),
+			"the retry loop must stop at the bounded attempt count, never spin forever")
+		assert.Equal(t, StateError, m.Status(instance))
+		require.NotNil(t, instance.Instance.StateReason)
+		assert.Equal(t, "recovery_mount_state_unavailable", *instance.Instance.StateReason.Message)
+	})
+
+	t.Run("shutdown signal aborts the retry loop without marking terminal", func(t *testing.T) {
+		mounter := &recoveryMounter{behavior: map[string]func(*VM) error{}}
+		cleaner := &recordingInstanceCleaner{}
+		rt := &recordedTransitions{}
+		m := NewManager()
+		rt.bind(m)
+
+		var shuttingDown atomic.Bool
+		m.SetDeps(Deps{
+			NodeID:          "test-node",
+			StateStore:      newFakeStateStore(),
+			VolumeMounter:   mounter,
+			InstanceCleaner: cleaner,
+			TransitionState: rt.apply,
+			ShutdownSignal:  shuttingDown.Load,
+		})
+
+		origRun := runForRelaunch
+		t.Cleanup(func() { runForRelaunch = origRun })
+		var calls atomic.Int64
+		runForRelaunch = func(_ *Manager, _ context.Context, _ *VM) error {
+			calls.Add(1)
+			shuttingDown.Store(true)
+			return fmt.Errorf("state not yet loaded: %w", ErrMountRetryable)
+		}
+
+		instance := &VM{ID: "i-shutdown-abort", Status: StatePending, InstanceType: "t3.micro", Instance: &ec2.Instance{}}
+		m.Insert(instance)
+
+		m.relaunchAll([]*VM{instance})
+
+		assert.EqualValues(t, 1, calls.Load(),
+			"the loop must check the shutdown signal before its next attempt and stop, not keep retrying")
+		assert.NotEqual(t, StateError, m.Status(instance),
+			"an abort for shutdown must never mark the instance recovery-failed")
+	})
 }
 
 // closeTrackingConn wraps net.Conn so a test can assert the connection