diff --git a/spinifex/daemon/daemon_handlers_instance.go b/spinifex/daemon/daemon_handlers_instance.go index e4477c0f5..de87ee6b3 100644 --- a/spinifex/daemon/daemon_handlers_instance.go +++ b/spinifex/daemon/daemon_handlers_instance.go @@ -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 { diff --git a/spinifex/gateway/ec2/instance/StartInstances.go b/spinifex/gateway/ec2/instance/StartInstances.go index 2754f2e24..56f67df23 100644 --- a/spinifex/gateway/ec2/instance/StartInstances.go +++ b/spinifex/gateway/ec2/instance/StartInstances.go @@ -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 +} diff --git a/spinifex/handlers/ec2/instance/service_impl.go b/spinifex/handlers/ec2/instance/service_impl.go index 80e881d10..548e8d6a1 100644 --- a/spinifex/handlers/ec2/instance/service_impl.go +++ b/spinifex/handlers/ec2/instance/service_impl.go @@ -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 { diff --git a/spinifex/handlers/ec2/instance/service_impl_test.go b/spinifex/handlers/ec2/instance/service_impl_test.go index fe26bbe19..aeb2df4ee 100644 --- a/spinifex/handlers/ec2/instance/service_impl_test.go +++ b/spinifex/handlers/ec2/instance/service_impl_test.go @@ -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) diff --git a/spinifex/vm/lifecycle.go b/spinifex/vm/lifecycle.go index 37b1a044c..a339f4bc6 100644 --- a/spinifex/vm/lifecycle.go +++ b/spinifex/vm/lifecycle.go @@ -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) } diff --git a/spinifex/vm/state.go b/spinifex/vm/state.go index 35870ca4d..50edf8585 100644 --- a/spinifex/vm/state.go +++ b/spinifex/vm/state.go @@ -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{ diff --git a/spinifex/vm/state_test.go b/spinifex/vm/state_test.go index cd6addc84..b0132c33c 100644 --- a/spinifex/vm/state_test.go +++ b/spinifex/vm/state_test.go @@ -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) + } +}