From 32bcada2276238fe1beb414dcb1213a062b3164a Mon Sep 17 00:00:00 2001 From: Josh Siviour Date: Sat, 19 Sep 2026 12:47:34 +1000 Subject: [PATCH] fix: make an error-state instance terminable and stop reporting a refusal as terminated (#1089) An instance left in StateError could not be terminated through any path, and the gateway reported success anyway, so a Terraform destroy waited on a record the platform had already claimed was gone. classifyRestoredInstances skipped a StateError instance with the comment that the operator must retry or terminate it, but continued before firing OnInstanceRecovering, which is what binds ec2.cmd.. The topic the terminate travels on therefore had no responder and the stated remedy was unreachable. With no responder there, TerminateInstances fell back to the ec2.terminate topic, whose handler reads the stopped KV bucket only and answered InvalidInstanceID.NotFound for a record held in the active registry. The gateway discarded that payload without reading its code and took the idempotent-absence branch, returning terminated for a record it had not removed. The StateError branch now announces the instance to the recovering hook before skipping relaunch, and the gateway reads the refusal: only a genuine absence falls through to the idempotent branch, and any other code is logged and returned. --- .../ec2/instance/TerminateInstances.go | 17 ++++++- .../ec2/instance/TerminateInstances_test.go | 51 +++++++++++++++++++ spinifex/vm/restore.go | 7 ++- spinifex/vm/restore_test.go | 27 ++++++++++ 4 files changed, 100 insertions(+), 2 deletions(-) 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") +}