diff --git a/spinifex/gateway/ec2/instance/TerminateInstances.go b/spinifex/gateway/ec2/instance/TerminateInstances.go index 5e091638a..3bede49ec 100644 --- a/spinifex/gateway/ec2/instance/TerminateInstances.go +++ b/spinifex/gateway/ec2/instance/TerminateInstances.go @@ -8,6 +8,7 @@ import ( "log/slog" "time" + "github.com/aws/aws-sdk-go/aws" "github.com/aws/aws-sdk-go/service/ec2" "github.com/mulgadc/spinifex/spinifex/awserrors" "github.com/mulgadc/spinifex/spinifex/types" @@ -97,11 +98,25 @@ func TerminateInstances(ctx context.Context, input *ec2.TerminateInstancesInput, utils.InjectTraceContext(ctx, terminateReqMsg.Header) terminateMsg, terminateErr := natsConn.RequestMsg(terminateReqMsg, 30*time.Second) if terminateErr == nil { - if _, parseErr := utils.ValidateErrorPayload(terminateMsg.Data); parseErr == nil { + responseError, parseErr := utils.ValidateErrorPayload(terminateMsg.Data) + if parseErr == nil { slog.InfoContext(ctx, "TerminateInstances: Stopped instance terminated via ec2.terminate", "instance_id", instanceID) stateChanges = append(stateChanges, newStateChange(instanceID, 32, "shutting-down", 80, "stopped")) continue } + + // Only a genuine absence may fall through to the idempotent + // branch below. Any other refusal is a terminate that did not + // happen, so returning success for it strands a record the + // caller has been told is gone. + code := aws.StringValue(responseError.Code) + if code != awserrors.ErrorInvalidInstanceIDNotFound { + slog.ErrorContext(ctx, "TerminateInstances: ec2.terminate refused the instance", + "instance_id", instanceID, "code", code) + return nil, errors.New(code) + } + slog.InfoContext(ctx, "TerminateInstances: instance is not a stopped instance, checking the terminated bucket", + "instance_id", instanceID) } // Idempotent terminate (rule #1): no daemon owns the instance and it diff --git a/spinifex/gateway/ec2/instance/TerminateInstances_test.go b/spinifex/gateway/ec2/instance/TerminateInstances_test.go index 9b1aaf23c..5005863e5 100644 --- a/spinifex/gateway/ec2/instance/TerminateInstances_test.go +++ b/spinifex/gateway/ec2/instance/TerminateInstances_test.go @@ -246,3 +246,54 @@ func TestTerminateInstances_MixedRunningAndStopped(t *testing.T) { assert.Equal(t, int64(80), *output.TerminatingInstances[1].PreviousState.Code) assert.Equal(t, "stopped", *output.TerminatingInstances[1].PreviousState.Name) } + +// A refusal that is not an absence means the terminate did not happen, so +// reporting "terminated" for it would strand a record the caller believes is +// gone and leave a Terraform destroy waiting on it forever. +func TestTerminateInstances_RefusalIsNotReportedAsTerminated(t *testing.T) { + _, nc := startTestNATSServer(t) + + instanceID := "i-refused-term" + + nc.QueueSubscribe("ec2.terminate", "spinifex-workers", func(msg *nats.Msg) { + msg.Respond([]byte(`{"code":"` + awserrors.ErrorIncorrectInstanceState + `","message":"wrong state"}`)) + }) + // A reachable terminated bucket that does not hold the instance is what + // makes the idempotent-absence branch look applicable when it is not. + nc.Subscribe("ec2.DescribeTerminatedInstances", func(msg *nats.Msg) { + msg.Respond([]byte(`{"Reservations":[]}`)) + }) + + output, err := TerminateInstances(context.Background(), &ec2.TerminateInstancesInput{ + InstanceIds: []*string{aws.String(instanceID)}, + }, nc, "123456789012") + + require.Error(t, err) + assert.Equal(t, awserrors.ErrorIncorrectInstanceState, err.Error()) + assert.Nil(t, output) +} + +// A genuine absence still converges: the stopped bucket does not hold the +// instance and neither does the terminated bucket, so the terminate is a no-op +// on something already gone. +func TestTerminateInstances_AbsentInstanceStaysIdempotent(t *testing.T) { + _, nc := startTestNATSServer(t) + + instanceID := "i-absent-term" + + nc.QueueSubscribe("ec2.terminate", "spinifex-workers", func(msg *nats.Msg) { + msg.Respond([]byte(`{"code":"` + awserrors.ErrorInvalidInstanceIDNotFound + `","message":"not found"}`)) + }) + nc.Subscribe("ec2.DescribeTerminatedInstances", func(msg *nats.Msg) { + msg.Respond([]byte(`{"Reservations":[]}`)) + }) + + output, err := TerminateInstances(context.Background(), &ec2.TerminateInstancesInput{ + InstanceIds: []*string{aws.String(instanceID)}, + }, nc, "123456789012") + + require.NoError(t, err) + require.Len(t, output.TerminatingInstances, 1) + assert.Equal(t, int64(48), *output.TerminatingInstances[0].CurrentState.Code) + assert.Equal(t, "terminated", *output.TerminatingInstances[0].CurrentState.Name) +} diff --git a/spinifex/vm/restore.go b/spinifex/vm/restore.go index a986e28ba..069cc8d74 100644 --- a/spinifex/vm/restore.go +++ b/spinifex/vm/restore.go @@ -143,10 +143,15 @@ func (m *Manager) classifyRestoredInstances() []*VM { // Recovery-failed instances stay in StateError until the operator // explicitly retries or terminates. Skip relaunch and resource // re-allocation; resources were already released by stopCleanup - // when MarkRecoveryFailed fired on the previous daemon run. + // when MarkRecoveryFailed fired on the previous daemon run. The + // per-instance command topic is still bound, because an operator who + // cannot reach the instance cannot retry or terminate it either. if instance.Status == StateError { slog.Warn("Instance in error state; skipping recovery relaunch (operator must retry or terminate)", "instance", instance.ID, "managedBy", instance.ManagedBy, "instanceType", instance.InstanceType) + if m.deps.Hooks.OnInstanceRecovering != nil { + m.deps.Hooks.OnInstanceRecovering(instance) + } continue } diff --git a/spinifex/vm/restore_test.go b/spinifex/vm/restore_test.go index 0971f0da1..e565633e9 100644 --- a/spinifex/vm/restore_test.go +++ b/spinifex/vm/restore_test.go @@ -1277,3 +1277,30 @@ func TestReconnectInstance(t *testing.T) { "classifyRestoredInstances logs and continues without re-reconnect") }) } + +// An instance left in StateError is not relaunched, but the operator is told to +// retry or terminate it — which needs the per-instance command topic bound, or +// the terminate never reaches a responder and the record is stranded for good. +func TestClassifyRestoredInstances_ErrorStateStillAnnouncesForCommands(t *testing.T) { + t.Setenv("XDG_RUNTIME_DIR", t.TempDir()) + var announced []string + m := NewManager() + m.SetDeps(Deps{ + NodeID: "test-node", + StateStore: newFakeStateStore(), + Resources: newFakeResourceController(), + InstanceTypes: fakeInstanceTypeResolver{"t3.micro": {VCPUs: 1, MemoryMiB: 1024, Architecture: "x86_64"}}, + Hooks: ManagerHooks{ + OnInstanceRecovering: func(v *VM) { announced = append(announced, v.ID) }, + }, + }) + v := &VM{ID: "i-errored", Status: StateError, InstanceType: "t3.micro", Instance: &ec2.Instance{}} + m.Replace(map[string]*VM{v.ID: v}) + + toLaunch := m.classifyRestoredInstances() + + assert.Empty(t, toLaunch, "an instance in error state must not be relaunched") + assert.Equal(t, StateError, v.Status, "classify must leave the error state alone") + assert.Equal(t, []string{v.ID}, announced, + "ec2.cmd. must be bound for an error-state instance or terminate can never reach it") +}