diff --git a/spinifex/daemon/vm_adapters.go b/spinifex/daemon/vm_adapters.go index 4e217d61e..73b36d32a 100644 --- a/spinifex/daemon/vm_adapters.go +++ b/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)) } diff --git a/spinifex/daemon/vm_adapters_test.go b/spinifex/daemon/vm_adapters_test.go index 53cc7beda..e3cdf2627 100644 --- a/spinifex/daemon/vm_adapters_test.go +++ b/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") +} diff --git a/spinifex/services/viperblockd/provider_handlers.go b/spinifex/services/viperblockd/provider_handlers.go index 2ef0d9133..75c588185 100644 --- a/spinifex/services/viperblockd/provider_handlers.go +++ b/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 } diff --git a/spinifex/services/viperblockd/provider_handlers_test.go b/spinifex/services/viperblockd/provider_handlers_test.go index c6049be9a..9d83a9480 100644 --- a/spinifex/services/viperblockd/provider_handlers_test.go +++ b/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 diff --git a/spinifex/types/ebs.go b/spinifex/types/ebs.go index 842f23d8a..a699713ad 100644 --- a/spinifex/types/ebs.go +++ b/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 { diff --git a/spinifex/vm/recovery.go b/spinifex/vm/recovery.go new file mode 100644 index 000000000..7f9c9d447 --- /dev/null +++ b/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") diff --git a/spinifex/vm/restore.go b/spinifex/vm/restore.go index 55f5e0902..940b388bf 100644 --- a/spinifex/vm/restore.go +++ b/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 diff --git a/spinifex/vm/restore_test.go b/spinifex/vm/restore_test.go index c4deba22c..26c604e70 100644 --- a/spinifex/vm/restore_test.go +++ b/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