From addf2bc86e591499803b58d6b14190f326475cc9 Mon Sep 17 00:00:00 2001 From: Julian Sommer <115516918+juliansommer@users.noreply.github.com> Date: Tue, 5 May 2026 16:37:21 +1000 Subject: [PATCH] refactor(vm): typed Status accessor, sentinel-aware Stop/Terminate (#191) --- spinifex/daemon/daemon_handlers_instance.go | 25 +++++++++++------- spinifex/daemon/daemon_handlers_volume.go | 3 +-- spinifex/daemon/daemon_system_instance.go | 2 +- .../daemon/daemon_system_instance_test.go | 2 +- spinifex/daemon/daemon_test.go | 26 +++++++++---------- spinifex/daemon/health_test.go | 2 +- spinifex/daemon/state.go | 9 +++++-- spinifex/daemon/state_test.go | 7 +++-- spinifex/vm/crash_recovery.go | 11 +++----- spinifex/vm/device_map.go | 2 +- spinifex/vm/lifecycle.go | 10 +++---- spinifex/vm/manager.go | 14 +++++++++- spinifex/vm/manager_test.go | 19 ++++++++++++++ spinifex/vm/restore.go | 3 +-- spinifex/vm/shutdown.go | 21 +++++++-------- spinifex/vm/volumes.go | 14 ++++------ spinifex/vm/volumes_test.go | 4 +-- 17 files changed, 97 insertions(+), 77 deletions(-) diff --git a/spinifex/daemon/daemon_handlers_instance.go b/spinifex/daemon/daemon_handlers_instance.go index deafbd180..371672049 100644 --- a/spinifex/daemon/daemon_handlers_instance.go +++ b/spinifex/daemon/daemon_handlers_instance.go @@ -347,8 +347,7 @@ func (d *Daemon) handleEC2RunInstances(msg *nats.Msg) { var successCount int for _, instance := range instances { // Skip if instance was terminated by a concurrent request - var status vm.InstanceState - d.vmMgr.Inspect(instance, func(v *vm.VM) { status = v.Status }) + status := d.vmMgr.Status(instance) if status != vm.StatePending && status != vm.StateProvisioning { slog.Info("Instance state changed during provisioning, skipping launch", "instanceId", instance.ID, "status", string(status)) @@ -429,8 +428,7 @@ func (d *Daemon) handleStartInstance(msg *nats.Msg, command types.EC2InstanceCom slog.Info("Starting instance", "id", command.ID) // Validate instance is in stopped state - var status vm.InstanceState - d.vmMgr.Inspect(instance, func(v *vm.VM) { status = v.Status }) + status := d.vmMgr.Status(instance) if status != vm.StateStopped { slog.Error("StartInstance: instance not in stopped state", "instanceId", command.ID, "status", status) @@ -451,7 +449,7 @@ func (d *Daemon) handleStartInstance(msg *nats.Msg, command types.EC2InstanceCom // Clear stop attribute before launch so WriteState inside the manager // persists the correct attributes. Without this, a daemon restart after // a stop→start cycle would see StopInstance=true and skip reconnecting QEMU. - d.vmMgr.Inspect(instance, func(v *vm.VM) { v.Attributes = command.Attributes }) + d.vmMgr.UpdateState(instance.ID, func(v *vm.VM) { v.Attributes = command.Attributes }) // Launch the instance infrastructure (QEMU, QMP, NATS subscriptions) if err := d.vmMgr.Start(instance.ID); err != nil { @@ -484,11 +482,7 @@ func (d *Daemon) handleStopOrTerminateInstance(msg *nats.Msg, command types.EC2I slog.Info(action+" instance", "id", command.ID) - // Check state validity before dispatching so we can return the correct - // AWS error code (IncorrectInstanceState) when the instance is already - // stopped/terminated/etc. Manager.Stop/Terminate re-validate internally. - var currentState vm.InstanceState - d.vmMgr.Inspect(instance, func(v *vm.VM) { currentState = v.Status }) + currentState := d.vmMgr.Status(instance) // Idempotent: a concurrent terminate goroutine is already cleaning up. if isTerminate && currentState == vm.StateShuttingDown { @@ -499,6 +493,10 @@ func (d *Daemon) handleStopOrTerminateInstance(msg *nats.Msg, command types.EC2I return } + // Validate the transition synchronously before dispatching so the AWS + // SDK sees IncorrectInstanceState (400) instead of a stale 200. The + // async Stop/Terminate path re-validates and surfaces vm.ErrInvalidTransition + // on a racing transition; we map that into the same AWS error below. if !vm.IsValidTransition(currentState, initialState) { slog.Warn("Instance in incorrect state for "+strings.ToLower(action), "instanceId", instance.ID, "currentState", string(currentState)) @@ -524,6 +522,13 @@ func (d *Daemon) handleStopOrTerminateInstance(msg *nats.Msg, command types.EC2I err = d.vmMgr.Stop(id) } if err != nil { + // Race with another transition: log at the right level so it + // shows up in dashboards but doesn't trigger paging. + if errors.Is(err, vm.ErrInvalidTransition) { + slog.Warn("Lifecycle transition raced; ack already sent", + "id", id, "action", strings.ToLower(action), "err", err) + return + } slog.Error("Failed to "+strings.ToLower(action)+" instance", "err", err, "id", id) } }(instance.ID) diff --git a/spinifex/daemon/daemon_handlers_volume.go b/spinifex/daemon/daemon_handlers_volume.go index e3656c290..2ca7911b1 100644 --- a/spinifex/daemon/daemon_handlers_volume.go +++ b/spinifex/daemon/daemon_handlers_volume.go @@ -32,8 +32,7 @@ func (d *Daemon) handleAttachVolume(msg *nats.Msg, command types.EC2InstanceComm // Surface IncorrectInstanceState before any volume-side lookups so the // AWS-SDK error matches the pre-2d ordering. - var status vm.InstanceState - d.vmMgr.Inspect(instance, func(v *vm.VM) { status = v.Status }) + status := d.vmMgr.Status(instance) if status != vm.StateRunning { slog.Error("AttachVolume: instance not running", "instanceId", command.ID, "status", status) diff --git a/spinifex/daemon/daemon_system_instance.go b/spinifex/daemon/daemon_system_instance.go index b637e053f..d796c91f0 100644 --- a/spinifex/daemon/daemon_system_instance.go +++ b/spinifex/daemon/daemon_system_instance.go @@ -370,7 +370,7 @@ func (d *Daemon) TerminateSystemInstance(instanceID string) error { } else { slog.Info("TerminateSystemInstance: released EIP", "instanceId", instanceID, "ip", instance.PublicIP, "allocationId", instance.PublicIPAllocID) } - d.vmMgr.Inspect(instance, func(v *vm.VM) { + d.vmMgr.UpdateState(instance.ID, func(v *vm.VM) { v.PublicIP = "" v.PublicIPPool = "" v.PublicIPAllocID = "" diff --git a/spinifex/daemon/daemon_system_instance_test.go b/spinifex/daemon/daemon_system_instance_test.go index 2038bd731..18ddc9db6 100644 --- a/spinifex/daemon/daemon_system_instance_test.go +++ b/spinifex/daemon/daemon_system_instance_test.go @@ -49,7 +49,7 @@ func TestWaitForSystemInstance_TransitionsToRunning(t *testing.T) { // Transition to running after a short delay go func() { time.Sleep(200 * time.Millisecond) - d.vmMgr.Inspect(inst, func(v *vm.VM) { v.Status = vm.StateRunning }) + d.vmMgr.UpdateState(inst.ID, func(v *vm.VM) { v.Status = vm.StateRunning }) }() err := d.WaitForSystemInstance("i-pending", 2*time.Second) diff --git a/spinifex/daemon/daemon_test.go b/spinifex/daemon/daemon_test.go index 831f0a32d..1924841b0 100644 --- a/spinifex/daemon/daemon_test.go +++ b/spinifex/daemon/daemon_test.go @@ -1954,8 +1954,9 @@ func TestHandleEC2Events_AttachVolume(t *testing.T) { }) t.Run("InstanceNotRunning", func(t *testing.T) { - // Temporarily set status to stopped - instance.Status = vm.StateStopped + // Temporarily set status to stopped under the manager lock so -race + // reflects production discipline. + daemon.vmMgr.UpdateState(instance.ID, func(v *vm.VM) { v.Status = vm.StateStopped }) command := types.EC2InstanceCommand{ ID: instanceID, Attributes: types.EC2CommandAttributes{ @@ -1976,7 +1977,7 @@ func TestHandleEC2Events_AttachVolume(t *testing.T) { assert.Contains(t, string(resp.Data), "IncorrectInstanceState") // Restore running state - instance.Status = vm.StateRunning + daemon.vmMgr.UpdateState(instance.ID, func(v *vm.VM) { v.Status = vm.StateRunning }) }) t.Run("VolumeNotFound", func(t *testing.T) { @@ -2093,8 +2094,9 @@ func TestHandleEC2Events_DetachVolume(t *testing.T) { }) t.Run("InstanceNotRunning", func(t *testing.T) { - // Temporarily set status to stopped - instance.Status = vm.StateStopped + // Temporarily set status to stopped under the manager lock so -race + // reflects production discipline. + daemon.vmMgr.UpdateState(instance.ID, func(v *vm.VM) { v.Status = vm.StateStopped }) command := types.EC2InstanceCommand{ ID: instanceID, Attributes: types.EC2CommandAttributes{ @@ -2115,7 +2117,7 @@ func TestHandleEC2Events_DetachVolume(t *testing.T) { assert.Contains(t, string(resp.Data), "IncorrectInstanceState") // Restore running state - instance.Status = vm.StateRunning + daemon.vmMgr.UpdateState(instance.ID, func(v *vm.VM) { v.Status = vm.StateRunning }) }) t.Run("VolumeNotAttached", func(t *testing.T) { @@ -3118,9 +3120,7 @@ func TestMarkInstanceFailed(t *testing.T) { assert.Equal(t, "Server.InternalError", *instance.Instance.StateReason.Code) assert.Equal(t, "volume_preparation_failed", *instance.Instance.StateReason.Message) require.Eventually(t, func() bool { - var status vm.InstanceState - daemon.vmMgr.Inspect(instance, func(v *vm.VM) { status = v.Status }) - return status == vm.StateTerminated + return daemon.vmMgr.Status(instance) == vm.StateTerminated }, 5*time.Second, 10*time.Millisecond, "cleanup goroutine should reach terminated") } @@ -3142,9 +3142,7 @@ func TestMarkInstanceFailed_NilInstance(t *testing.T) { daemon.vmMgr.MarkFailed(instance, "test_failure") require.Eventually(t, func() bool { - var status vm.InstanceState - daemon.vmMgr.Inspect(instance, func(v *vm.VM) { status = v.Status }) - return status == vm.StateTerminated + return daemon.vmMgr.Status(instance) == vm.StateTerminated }, 5*time.Second, 10*time.Millisecond, "cleanup goroutine should reach terminated") } @@ -3292,7 +3290,7 @@ func TestStopTerminate_IncorrectInstanceState(t *testing.T) { defer sub.Unsubscribe() t.Run("StopAlreadyStoppedInstance", func(t *testing.T) { - instance.Status = vm.StateStopped + daemon.vmMgr.UpdateState(instance.ID, func(v *vm.VM) { v.Status = vm.StateStopped }) command := types.EC2InstanceCommand{ ID: instanceID, Attributes: types.EC2CommandAttributes{ @@ -3312,7 +3310,7 @@ func TestStopTerminate_IncorrectInstanceState(t *testing.T) { }) t.Run("TerminateAlreadyTerminatedInstance", func(t *testing.T) { - instance.Status = vm.StateTerminated + daemon.vmMgr.UpdateState(instance.ID, func(v *vm.VM) { v.Status = vm.StateTerminated }) command := types.EC2InstanceCommand{ ID: instanceID, Attributes: types.EC2CommandAttributes{ diff --git a/spinifex/daemon/health_test.go b/spinifex/daemon/health_test.go index a945d6ec0..856cfa4f0 100644 --- a/spinifex/daemon/health_test.go +++ b/spinifex/daemon/health_test.go @@ -336,7 +336,7 @@ func TestHandleInstanceCrash_FirstCrashSetsTime(t *testing.T) { assert.Equal(t, 1, instance.Health.CrashCount) // Reset to running for a second crash - d.vmMgr.Inspect(instance, func(v *vm.VM) { + d.vmMgr.UpdateState(instance.ID, func(v *vm.VM) { v.Status = vm.StateRunning v.Running = true v.PID = 222 diff --git a/spinifex/daemon/state.go b/spinifex/daemon/state.go index e2529bc4c..1d40cf30b 100644 --- a/spinifex/daemon/state.go +++ b/spinifex/daemon/state.go @@ -9,7 +9,9 @@ import ( // TransitionState validates and applies a state transition on the given instance. // It sets instance.Status, persists via WriteState, and logs. -// On validation failure, the instance status is unchanged and an error is returned. +// On validation failure, the instance status is unchanged and a wrapped +// vm.ErrInvalidTransition is returned so callers can map it to the AWS +// IncorrectInstanceState error code via errors.Is. // If WriteState fails, the in-memory status retains the new value (the VM has // physically changed state regardless) and an error is returned. func (d *Daemon) TransitionState(instance *vm.VM, target vm.InstanceState) error { @@ -17,6 +19,8 @@ func (d *Daemon) TransitionState(instance *vm.VM, target vm.InstanceState) error current vm.InstanceState invalid bool ) + // Inspect (not UpdateState): MarkFailed may invoke this for an instance + // that is no longer in the running map. d.vmMgr.Inspect(instance, func(v *vm.VM) { current = v.Status if !vm.IsValidTransition(current, target) { @@ -26,7 +30,8 @@ func (d *Daemon) TransitionState(instance *vm.VM, target vm.InstanceState) error v.Status = target }) if invalid { - return fmt.Errorf("invalid state transition: %s -> %s for instance %s", current, target, instance.ID) + return fmt.Errorf("%w: %s -> %s for instance %s", + vm.ErrInvalidTransition, current, target, instance.ID) } slog.Info("Instance state transition", "instanceId", instance.ID, "from", string(current), "to", string(target)) diff --git a/spinifex/daemon/state_test.go b/spinifex/daemon/state_test.go index 983639ce8..1e366ee1f 100644 --- a/spinifex/daemon/state_test.go +++ b/spinifex/daemon/state_test.go @@ -1006,7 +1006,8 @@ func TestPendingWatchdog_MarksStuckInstanceFailed(t *testing.T) { // Run one watchdog tick manually instead of waiting for the ticker var stuck []*vm.VM for _, instance := range daemon.vmMgr.Snapshot() { - if (instance.Status == vm.StatePending || instance.Status == vm.StateProvisioning) && + status := daemon.vmMgr.Status(instance) + if (status == vm.StatePending || status == vm.StateProvisioning) && instance.Instance != nil && instance.Instance.LaunchTime != nil && time.Since(*instance.Instance.LaunchTime) > 5*time.Minute { stuck = append(stuck, instance) @@ -1027,9 +1028,7 @@ func TestPendingWatchdog_MarksStuckInstanceFailed(t *testing.T) { assert.Equal(t, "Server.InternalError", *stuckBefore.Instance.StateReason.Code) assert.Equal(t, "launch_timeout", *stuckBefore.Instance.StateReason.Message) require.Eventually(t, func() bool { - var status vm.InstanceState - daemon.vmMgr.Inspect(stuckBefore, func(v *vm.VM) { status = v.Status }) - return status == vm.StateTerminated + return daemon.vmMgr.Status(stuckBefore) == vm.StateTerminated }, 5*time.Second, 10*time.Millisecond, "MarkFailed cleanup goroutine should reach terminated") // Fresh instance should still be pending diff --git a/spinifex/vm/crash_recovery.go b/spinifex/vm/crash_recovery.go index b3f79cf9f..5db3059e9 100644 --- a/spinifex/vm/crash_recovery.go +++ b/spinifex/vm/crash_recovery.go @@ -80,10 +80,7 @@ func ClassifyCrashReason(waitErr error) string { // Wired as Deps.CrashHandler so the launch goroutine in lifecycle.go can // invoke it without importing the manager into a separate goroutine spawner. func (m *Manager) HandleCrash(instance *VM, waitErr error) { - var status InstanceState - m.Inspect(instance, func(v *VM) { status = v.Status }) - - if status != StateRunning { + if status := m.Status(instance); status != StateRunning { slog.Debug("QEMU exited but instance not in running state, skipping crash handler", "instance", instance.ID, "status", status) return @@ -106,7 +103,7 @@ func (m *Manager) HandleCrash(instance *VM, waitErr error) { } now := time.Now() - m.Inspect(instance, func(v *VM) { + m.UpdateState(instance.ID, func(v *VM) { v.Health.CrashCount++ v.Health.LastCrashTime = now v.Health.LastCrashReason = reason @@ -158,7 +155,7 @@ func (m *Manager) MaybeRestart(instance *VM) { restartCount int exceeded bool ) - m.Inspect(instance, func(v *VM) { + m.UpdateState(instance.ID, func(v *VM) { health := &v.Health if !health.FirstCrashTime.IsZero() && now.Sub(health.FirstCrashTime) > RestartWindow { slog.Info("Crash window expired, resetting counters", "instance", v.ID) @@ -217,7 +214,7 @@ func (m *Manager) MaybeRestart(instance *VM) { func (m *Manager) RestartCrashedInstance(instance *VM) { var skipReason string var restartCount int - m.Inspect(instance, func(v *VM) { + m.UpdateState(instance.ID, func(v *VM) { if v.Status != StateError { skipReason = fmt.Sprintf("not in error state (%s)", v.Status) return diff --git a/spinifex/vm/device_map.go b/spinifex/vm/device_map.go index 9f19fec9d..5751b05f2 100644 --- a/spinifex/vm/device_map.go +++ b/spinifex/vm/device_map.go @@ -177,7 +177,7 @@ func (m *Manager) UpdateGuestDeviceNames(instance *VM) { } instance.EBSRequests.Mu.Unlock() - m.Inspect(instance, func(v *VM) { + m.UpdateState(instance.ID, func(v *VM) { if v.Instance == nil { return } diff --git a/spinifex/vm/lifecycle.go b/spinifex/vm/lifecycle.go index 2030af105..540a70a16 100644 --- a/spinifex/vm/lifecycle.go +++ b/spinifex/vm/lifecycle.go @@ -47,9 +47,7 @@ func (m *Manager) Reboot(id string) error { if !ok { return ErrInstanceNotFound } - var status InstanceState - m.Inspect(instance, func(v *VM) { status = v.Status }) - if status != StateRunning { + if status := m.Status(instance); status != StateRunning { return fmt.Errorf("%w: cannot reboot instance %s in state %s", ErrInvalidTransition, id, status) } @@ -65,8 +63,7 @@ func (m *Manager) Reboot(id string) error { // terminate goroutine owns cleanup and launch must bail without further side // effects. func (m *Manager) launchStillValid(instance *VM) bool { - var status InstanceState - m.Inspect(instance, func(v *VM) { status = v.Status }) + status := m.Status(instance) if status == StatePending || status == StateStopped || status == StateProvisioning { return true } @@ -446,8 +443,7 @@ func (m *Manager) qmpHeartbeat(instance *VM) { for { time.Sleep(30 * time.Second) - var status InstanceState - m.Inspect(instance, func(v *VM) { status = v.Status }) + status := m.Status(instance) if status == StateStopping || status == StateStopped || status == StateShuttingDown || status == StateTerminated || status == StateError { diff --git a/spinifex/vm/manager.go b/spinifex/vm/manager.go index 8b1c6c980..73f2e9f22 100644 --- a/spinifex/vm/manager.go +++ b/spinifex/vm/manager.go @@ -220,10 +220,22 @@ func (m *Manager) UpdateState(id string, fn func(*VM)) bool { return true } +// Status returns v.Status under the manager lock. Replaces the dominant +// "Inspect to read Status" pattern with a typed accessor. Read-only — no +// membership check, so callers may pass a pointer to an instance no longer +// in the map (e.g. mid-cleanup) and still get a consistent read. +func (m *Manager) Status(v *VM) InstanceState { + m.mu.Lock() + defer m.mu.Unlock() + return v.Status +} + // Inspect runs fn(v) under the manager lock for an already-resolved VM // pointer. Used by call sites that hold a *VM (e.g. from a NATS handler // dispatch) and need to read or mutate its fields with the same memory- -// ordering guarantee as map-keyed access. +// ordering guarantee as map-keyed access. Prefer UpdateState (membership- +// checked) for mutation and Status for the read-Status pattern; Inspect +// remains for closures that mutate fields on a possibly-orphaned VM. func (m *Manager) Inspect(v *VM, fn func(*VM)) { m.mu.Lock() defer m.mu.Unlock() diff --git a/spinifex/vm/manager_test.go b/spinifex/vm/manager_test.go index c98db5acb..6e5940231 100644 --- a/spinifex/vm/manager_test.go +++ b/spinifex/vm/manager_test.go @@ -119,6 +119,25 @@ func TestManager_Inspect(t *testing.T) { } } +func TestManager_Status(t *testing.T) { + m := NewManager() + a := mkVM("i-a") + a.Status = StatePending + m.Insert(a) + + if got := m.Status(a); got != StatePending { + t.Fatalf("Status: got %s, want %s", got, StatePending) + } + + // Status reads under lock without verifying membership — orphaned VMs + // (mid-cleanup, never-inserted) still return a consistent value. + loose := mkVM("i-loose") + loose.Status = StateRunning + if got := m.Status(loose); got != StateRunning { + t.Fatalf("Status on loose VM: got %s, want %s", got, StateRunning) + } +} + func TestManager_ForEachAndSnapshot(t *testing.T) { m := NewManager() want := map[string]bool{"i-1": true, "i-2": true, "i-3": true} diff --git a/spinifex/vm/restore.go b/spinifex/vm/restore.go index b97dda4ff..6c866af42 100644 --- a/spinifex/vm/restore.go +++ b/spinifex/vm/restore.go @@ -275,8 +275,7 @@ func (m *Manager) relaunchAll(toLaunch []*VM) { } }() - var status InstanceState - m.Inspect(inst, func(v *VM) { status = v.Status }) + status := m.Status(inst) if status != StatePending && status != StateProvisioning { slog.Info("Instance state changed during recovery, skipping launch", "instanceId", inst.ID, "status", string(status)) diff --git a/spinifex/vm/shutdown.go b/spinifex/vm/shutdown.go index dc8eb1626..9f0399507 100644 --- a/spinifex/vm/shutdown.go +++ b/spinifex/vm/shutdown.go @@ -38,7 +38,7 @@ func (m *Manager) Stop(id string) error { m.stopCleanup(instance) - m.Inspect(instance, func(v *VM) { v.LastNode = m.deps.NodeID }) + m.UpdateState(instance.ID, func(v *VM) { v.LastNode = m.deps.NodeID }) if err := m.transitionWithPrecheck(instance, StateStopped); err != nil { slog.Error("Failed to transition to stopped", "instanceId", instance.ID, "err", err) @@ -110,9 +110,7 @@ func (m *Manager) Terminate(id string) error { return ErrInstanceNotFound } - var current InstanceState - m.Inspect(instance, func(v *VM) { current = v.Status }) - if current == StateShuttingDown { + if current := m.Status(instance); current == StateShuttingDown { // Concurrent failed-launch goroutine already owns cleanup. return nil } @@ -161,9 +159,7 @@ func (m *Manager) MarkFailed(instance *VM, reason string) { slog.Error("MarkFailed transition failed", "instanceId", instance.ID, "err", err) // If this was a persistence-only failure, in-memory state is now // shutting-down and we still want to finalize. Otherwise bail. - var postErr InstanceState - m.Inspect(instance, func(v *VM) { postErr = v.Status }) - if postErr != StateShuttingDown { + if m.Status(instance) != StateShuttingDown { return } } @@ -182,6 +178,8 @@ func (m *Manager) MarkFailed(instance *VM, reason string) { // OnInstanceDown, and persists the running set. Shared by Terminate and // MarkFailed. func (m *Manager) finalizeTerminated(instance *VM) error { + // Inspect (not UpdateState): MarkFailed may invoke this for an + // instance that was never inserted into the local map. m.Inspect(instance, func(v *VM) { v.LastNode = m.deps.NodeID }) if err := m.transitionWithPrecheck(instance, StateTerminated); err != nil { @@ -323,13 +321,14 @@ func (m *Manager) deallocateResources(instance *VM) { // as a persistence failure on a transition whose memory mutation // already succeeded. func (m *Manager) transitionWithPrecheck(instance *VM, target InstanceState) error { - var current InstanceState - m.Inspect(instance, func(v *VM) { current = v.Status }) + current := m.Status(instance) if !IsValidTransition(current, target) { return fmt.Errorf("%w: %s -> %s for instance %s", ErrInvalidTransition, current, target, instance.ID) } if m.deps.TransitionState == nil { + // Inspect (not UpdateState): MarkFailed may run this on an instance + // that was never inserted into the local map. m.Inspect(instance, func(v *VM) { v.Status = target }) return nil } @@ -337,9 +336,7 @@ func (m *Manager) transitionWithPrecheck(instance *VM, target InstanceState) err // Could be persistence failure (memory state already updated) or a // racing transition that invalidated the precheck. Re-inspect to // distinguish. - var observed InstanceState - m.Inspect(instance, func(v *VM) { observed = v.Status }) - if observed != target { + if m.Status(instance) != target { return fmt.Errorf("%w: %s -> %s for instance %s (raced)", ErrInvalidTransition, current, target, instance.ID) } diff --git a/spinifex/vm/volumes.go b/spinifex/vm/volumes.go index 8d4948b07..b86c917bf 100644 --- a/spinifex/vm/volumes.go +++ b/spinifex/vm/volumes.go @@ -44,15 +44,13 @@ func (m *Manager) AttachVolume(id, volumeID, device string) (AttachVolumeResult, return AttachVolumeResult{}, ErrInstanceNotFound } - var status InstanceState - m.Inspect(instance, func(v *VM) { status = v.Status }) - if status != StateRunning { + if status := m.Status(instance); status != StateRunning { return AttachVolumeResult{}, fmt.Errorf("%w: cannot attach to instance %s in state %s", ErrInvalidTransition, id, status) } if device == "" { - m.Inspect(instance, func(v *VM) { device = nextAvailableDevice(v) }) + m.UpdateState(id, func(v *VM) { device = nextAvailableDevice(v) }) if device == "" { return AttachVolumeResult{}, ErrAttachmentLimitExceeded } @@ -190,7 +188,7 @@ func (m *Manager) AttachVolume(id, volumeID, device string) (AttachVolumeResult, } instance.EBSRequests.Mu.Unlock() - m.Inspect(instance, func(v *VM) { + m.UpdateState(id, func(v *VM) { if v.Instance == nil { return } @@ -238,9 +236,7 @@ func (m *Manager) DetachVolume(id, volumeID, device string, force bool) (string, return "", ErrInstanceNotFound } - var status InstanceState - m.Inspect(instance, func(v *VM) { status = v.Status }) - if status != StateRunning { + if status := m.Status(instance); status != StateRunning { return "", fmt.Errorf("%w: cannot detach from instance %s in state %s", ErrInvalidTransition, id, status) } @@ -330,7 +326,7 @@ func (m *Manager) DetachVolume(id, volumeID, device string, force bool) (string, } instance.EBSRequests.Mu.Unlock() - m.Inspect(instance, func(v *VM) { + m.UpdateState(id, func(v *VM) { if v.Instance == nil { return } diff --git a/spinifex/vm/volumes_test.go b/spinifex/vm/volumes_test.go index 829c82176..d9524ad54 100644 --- a/spinifex/vm/volumes_test.go +++ b/spinifex/vm/volumes_test.go @@ -374,9 +374,7 @@ func TestReboot_DoesNotChangeStatus(t *testing.T) { v, ok := m.Get("i-1") require.True(t, ok) - var status InstanceState - m.Inspect(v, func(v *VM) { status = v.Status }) - assert.Equal(t, StateRunning, status) + assert.Equal(t, StateRunning, m.Status(v)) } func TestReboot_QMPFailureSurfacesError(t *testing.T) {