fix(ec2): make errored instances renderable and restartable (#325)

Project internal StateError onto AWS stopped (code 80) in DescribeInstances
so SDK/UI clients no longer break on the non-AWS 'error' state label.

Route start-instances through the per-instance ec2.cmd topic first so a
crash-pinned StateError instance is restarted by its live owner node, whose
StartInstance handler now accepts StateError (Manager.Start transitions
error->pending before relaunch). Fall back to the ec2.start queue-group path
on ErrNoResponders for genuinely stopped instances in the shared KV.
This commit is contained in:
Josh Siviour
2026-06-09 16:59:32 +10:00
committed by GitHub
parent 7bcd365549
commit f9b14e5a12
7 changed files with 168 additions and 30 deletions
+4 -3
View File
@@ -273,8 +273,9 @@ func (d *Daemon) handleEC2DescribeInstances(msg *nats.Msg) {
instanceCopy.PublicIpAddress = aws.String(instance.PublicIP)
}
// Map internal status to EC2 state codes using the centralized mapping
if info, ok := vm.EC2StateCodes[instance.Status]; ok {
// Map internal status to AWS state, projecting Spinifex-only states
// (e.g. error -> stopped) so SDK/UI clients see a valid label.
if info, ok := vm.EC2APIState(instance.Status); ok {
instanceCopy.State.SetCode(info.Code)
instanceCopy.State.SetName(info.Name)
} else {
@@ -510,7 +511,7 @@ func (d *Daemon) describeInstancesFromKV(msg *nats.Msg, listFn func() ([]*vm.VM,
instanceCopy := *instance.Instance
instanceCopy.State = &ec2.InstanceState{}
if info, ok := vm.EC2StateCodes[instance.Status]; ok {
if info, ok := vm.EC2APIState(instance.Status); ok {
instanceCopy.State.SetCode(info.Code)
instanceCopy.State.SetName(info.Name)
} else {
+89 -23
View File
@@ -3,11 +3,13 @@ package gateway_ec2_instance
import (
"encoding/json"
"errors"
"fmt"
"log/slog"
"time"
"github.com/aws/aws-sdk-go/service/ec2"
"github.com/mulgadc/spinifex/spinifex/awserrors"
"github.com/mulgadc/spinifex/spinifex/types"
"github.com/mulgadc/spinifex/spinifex/utils"
"github.com/nats-io/nats.go"
)
@@ -27,8 +29,14 @@ func ValidateStartInstancesInput(input *ec2.StartInstancesInput) error {
return nil
}
// StartInstances sends start requests to the ec2.start queue group topic.
// Any available daemon can pick up the request and launch the stopped instance.
// StartInstances starts the requested instances.
//
// A crash/recovery-failed instance stays live in its owner node's vmMgr as
// StateError, with its per-instance ec2.cmd subscription still active — so it
// is restarted via the owner-targeted ec2.cmd path. A genuinely stopped
// instance lives only in the shared stopped-KV with no live owner; for it the
// ec2.cmd attempt returns ErrNoResponders and we fall back to the ec2.start
// queue-group path that rehydrates from KV.
func StartInstances(input *ec2.StartInstancesInput, natsConn *nats.Conn, accountID string) (*ec2.StartInstancesOutput, error) {
if err := ValidateStartInstancesInput(input); err != nil {
return nil, err
@@ -44,33 +52,20 @@ func StartInstances(input *ec2.StartInstancesInput, natsConn *nats.Conn, account
}
instanceID := *instanceIDPtr
req := startStoppedInstanceRequest{InstanceID: instanceID}
jsonData, err := json.Marshal(req)
sc, handled, err := startLiveInstance(natsConn, instanceID, accountID)
if err != nil {
slog.Error("StartInstances: Failed to marshal request", "instance_id", instanceID, "err", err)
return nil, err
}
if handled {
stateChanges = append(stateChanges, sc)
continue
}
slog.Info("StartInstances: Sending NATS request", "subject", "ec2.start", "instance_id", instanceID)
reqMsg := nats.NewMsg("ec2.start")
reqMsg.Data = jsonData
reqMsg.Header.Set(utils.AccountIDHeader, accountID)
msg, err := natsConn.RequestMsg(reqMsg, 30*time.Second)
sc, err = startStoppedInstance(natsConn, instanceID, accountID)
if err != nil {
slog.Error("StartInstances: Failed to send start request", "instance_id", instanceID, "err", err)
stateChanges = append(stateChanges, newStateChange(instanceID, 80, "stopped", 80, "stopped"))
continue
return nil, err
}
// Check if the daemon returned an error response
if responseError, parseErr := utils.ValidateErrorPayload(msg.Data); parseErr != nil {
slog.Error("StartInstances: Daemon returned error", "instance_id", instanceID, "code", *responseError.Code)
return nil, errors.New(*responseError.Code)
}
slog.Info("StartInstances: Command sent successfully", "instance_id", instanceID, "response", string(msg.Data))
stateChanges = append(stateChanges, newStateChange(instanceID, 0, "pending", 80, "stopped"))
stateChanges = append(stateChanges, sc)
}
output := &ec2.StartInstancesOutput{
@@ -80,3 +75,74 @@ func StartInstances(input *ec2.StartInstancesInput, natsConn *nats.Conn, account
slog.Info("StartInstances: Completed", "total_instances", len(stateChanges))
return output, nil
}
// startLiveInstance restarts an instance still live on its owner node (e.g.
// StateError after exhausted auto-restart) via the per-instance ec2.cmd topic,
// reusing the StartInstance handler. handled is false with a nil error when no
// live owner is subscribed (ErrNoResponders), signalling the caller to fall
// back to the stopped-KV path.
func startLiveInstance(natsConn *nats.Conn, instanceID, accountID string) (*ec2.InstanceStateChange, bool, error) {
command := types.EC2InstanceCommand{
ID: instanceID,
Attributes: types.EC2CommandAttributes{StartInstance: true},
}
jsonData, err := json.Marshal(command)
if err != nil {
slog.Error("StartInstances: Failed to marshal cmd", "instance_id", instanceID, "err", err)
return nil, false, errors.New(awserrors.ErrorServerInternal)
}
reqMsg := nats.NewMsg(fmt.Sprintf("ec2.cmd.%s", instanceID))
reqMsg.Data = jsonData
reqMsg.Header.Set(utils.AccountIDHeader, accountID)
msg, err := natsConn.RequestMsg(reqMsg, 30*time.Second)
if err != nil {
// No live owner subscribed — this is a genuinely stopped instance; let
// the caller fall back to the stopped-KV start path.
if errors.Is(err, nats.ErrNoResponders) {
return nil, false, nil
}
slog.Error("StartInstances: cmd request failed", "instance_id", instanceID, "err", err)
return nil, false, errors.New(awserrors.ErrorServerInternal)
}
if responseError, parseErr := utils.ValidateErrorPayload(msg.Data); parseErr != nil {
slog.Error("StartInstances: owner returned error", "instance_id", instanceID, "code", *responseError.Code)
return nil, false, errors.New(*responseError.Code)
}
slog.Info("StartInstances: restarted via owner node", "instance_id", instanceID, "response", string(msg.Data))
return newStateChange(instanceID, 0, "pending", 80, "stopped"), true, nil
}
// startStoppedInstance rehydrates a stopped instance from the shared KV via the
// ec2.start queue-group topic. Any available daemon forwards to the instance's
// original node.
func startStoppedInstance(natsConn *nats.Conn, instanceID, accountID string) (*ec2.InstanceStateChange, error) {
req := startStoppedInstanceRequest{InstanceID: instanceID}
jsonData, err := json.Marshal(req)
if err != nil {
slog.Error("StartInstances: Failed to marshal request", "instance_id", instanceID, "err", err)
return nil, errors.New(awserrors.ErrorServerInternal)
}
slog.Info("StartInstances: Sending NATS request", "subject", "ec2.start", "instance_id", instanceID)
reqMsg := nats.NewMsg("ec2.start")
reqMsg.Data = jsonData
reqMsg.Header.Set(utils.AccountIDHeader, accountID)
msg, err := natsConn.RequestMsg(reqMsg, 30*time.Second)
if err != nil {
slog.Error("StartInstances: Failed to send start request", "instance_id", instanceID, "err", err)
return newStateChange(instanceID, 80, "stopped", 80, "stopped"), nil
}
// Check if the daemon returned an error response
if responseError, parseErr := utils.ValidateErrorPayload(msg.Data); parseErr != nil {
slog.Error("StartInstances: Daemon returned error", "instance_id", instanceID, "code", *responseError.Code)
return nil, errors.New(*responseError.Code)
}
slog.Info("StartInstances: Command sent successfully", "instance_id", instanceID, "response", string(msg.Data))
return newStateChange(instanceID, 0, "pending", 80, "stopped"), nil
}
@@ -1056,8 +1056,10 @@ func (s *InstanceServiceImpl) StartInstance(instance *vm.VM, command spxtypes.EC
slog.Info("StartInstance: starting instance", "id", command.ID)
status := s.vmMgr.Status(instance)
if status != vm.StateStopped {
slog.Error("StartInstance: instance not in stopped state", "instanceId", command.ID, "status", status)
// StateError is startable: a crash/recovery-failed instance can be manually
// recovered. The manager transitions Error->Pending before relaunch.
if status != vm.StateStopped && status != vm.StateError {
slog.Error("StartInstance: instance not in a startable state", "instanceId", command.ID, "status", status)
return errors.New(awserrors.ErrorIncorrectInstanceState)
}
@@ -2260,7 +2262,7 @@ func (s *InstanceServiceImpl) DescribeInstances(input *ec2.DescribeInstancesInpu
instanceCopy.PublicIpAddress = aws.String(instance.PublicIP)
}
if info, ok := vm.EC2StateCodes[instance.Status]; ok {
if info, ok := vm.EC2APIState(instance.Status); ok {
instanceCopy.State.SetCode(info.Code)
instanceCopy.State.SetName(info.Name)
} else {
@@ -2397,7 +2399,7 @@ func (s *InstanceServiceImpl) describeInstancesFromKV(input *ec2.DescribeInstanc
instanceCopy := *instance.Instance
instanceCopy.State = &ec2.InstanceState{}
if info, ok := vm.EC2StateCodes[instance.Status]; ok {
if info, ok := vm.EC2APIState(instance.Status); ok {
instanceCopy.State.SetCode(info.Code)
instanceCopy.State.SetName(info.Name)
} else {
@@ -2993,6 +2993,27 @@ func TestStartInstance_AllocateFails(t *testing.T) {
assert.Equal(t, awserrors.ErrorInsufficientInstanceCapacity, err.Error())
}
// TestStartInstance_ErrorStateStartable verifies a crash/recovery-failed
// instance (StateError) is accepted past the startability guard. Reaching the
// resource Allocate step (here forced to fail) proves the guard did not reject
// it with IncorrectInstanceState — manual recovery of errored instances works.
func TestStartInstance_ErrorStateStartable(t *testing.T) {
id := "i-err"
mgr := mgrWith(map[string]*vm.VM{
id: {ID: id, Status: vm.StateError, InstanceType: "t3.micro"},
})
v, _ := mgr.Get(id)
types, _ := defaultPrepareInstanceTypes()
prov := &fakeResourceCapacityProvider{
instanceTypes: types,
allocateErr: errors.New("no capacity"),
}
svc := &InstanceServiceImpl{vmMgr: mgr, resourceMgr: prov}
err := svc.StartInstance(v, spxtypes.EC2InstanceCommand{ID: id})
require.Error(t, err)
assert.Equal(t, awserrors.ErrorInsufficientInstanceCapacity, err.Error())
}
func TestRebootInstance_NotFound(t *testing.T) {
id := "i-missing"
mgr := mgrWith(nil)
+9
View File
@@ -39,6 +39,15 @@ func (m *Manager) Start(id string) error {
if !ok {
return fmt.Errorf("%w: %s", ErrInstanceNotFound, id)
}
// A crash/recovery-failed instance sits in StateError, which launchStillValid
// rejects. Move it to pending first (the same Error->Pending step the
// auto-restart path takes) so launch proceeds; resource re-allocation is the
// caller's responsibility, mirroring the stopped-start flow.
if m.Status(instance) == StateError && m.deps.TransitionState != nil {
if err := m.deps.TransitionState(instance, StatePending); err != nil {
return err
}
}
return m.launch(instance)
}
+15
View File
@@ -36,6 +36,21 @@ var EC2StateCodes = map[InstanceState]EC2StateInfo{
StateError: {Code: 0, Name: "error"},
}
// EC2APIState returns the AWS-API-facing state code and name for status,
// projecting Spinifex-internal states with no AWS equivalent onto the closest
// valid lifecycle state. AWS InstanceStateName is a closed enum, so leaking the
// internal "error" label breaks SDK/UI clients. A StateError instance (crash or
// recovery-failed: QEMU dead, resources released, volumes unmounted) is, to the
// outside world, stopped — it can be started or terminated. ok is false only for
// a status with no mapping, letting callers apply their own fallback.
func EC2APIState(status InstanceState) (EC2StateInfo, bool) {
if status == StateError {
return EC2StateCodes[StateStopped], true
}
info, ok := EC2StateCodes[status]
return info, ok
}
// ValidTransitions defines the allowed state transitions for an instance.
// StateTerminated is intentionally absent — it is a terminal state with no valid transitions.
var ValidTransitions = map[InstanceState][]InstanceState{
+24
View File
@@ -76,3 +76,27 @@ func TestEC2StateCodes_SpinifexOnlyStatesPresentAsPending(t *testing.T) {
assert.Equal(t, int64(0), EC2StateCodes[StateProvisioning].Code)
assert.Equal(t, int64(0), EC2StateCodes[StateError].Code)
}
func TestEC2APIState_ErrorProjectsToStopped(t *testing.T) {
// The internal "error" label is not a valid AWS InstanceStateName and breaks
// SDK/UI clients. EC2APIState must project StateError onto stopped so the
// instance stays renderable and actionable (start/terminate).
info, ok := EC2APIState(StateError)
assert.True(t, ok)
assert.Equal(t, ec2.InstanceStateNameStopped, info.Name)
assert.Equal(t, int64(80), info.Code)
}
func TestEC2APIState_NeverSurfacesNonAWSName(t *testing.T) {
// Every mapped state must project to a name in the AWS enum. Codes 0/"pending"
// for the provisioning internal state are AWS-understood; "error" must not leak.
for _, s := range []InstanceState{
StateProvisioning, StatePending, StateRunning, StateStopping,
StateStopped, StateShuttingDown, StateTerminated, StateError,
} {
info, ok := EC2APIState(s)
assert.True(t, ok, "EC2APIState(%s) returned no mapping", s)
_, valid := awsEC2StateCodes[info.Name]
assert.True(t, valid, "EC2APIState(%s) surfaced non-AWS name %q", s, info.Name)
}
}