refactor(daemon): extract TerminateStoppedInstance to service (1b-pre phase 2b)

Collapses ~140 lines of daemon termination logic into a one-liner
handleNATSRequest wrapper plus an InstanceServiceImpl method, matching
the Pattern A precedent from phase 1 (Describes) and phase 2a
(ModifyInstanceAttribute). Adds VolumeDeleter, ENIDeleter, and
PublicIPReleaser dep interfaces (wired via SetTerminationDeps) so the
service can clean up volumes, ENI, and EIP without depending on daemon
internals.

Hoists publishNATEvent off the daemon into utils.PublishNATEvent so the
service can drive vpc.delete-nat directly. Removes the
terminateStoppedInstanceRequest daemon-local payload in favour of the
exported TerminateStoppedInstanceInput on the service interface.

Tests: 13 service-level cases against an in-memory fake (missing ID,
nil store, load error, not found, not stopped, cross-tenant,
happy path, terminated-write failure aborts, retry-once delete,
user volume deleted, no-deleter graceful skip, internal volumes via
NATS, public IP release, ENI delete). The five JetStream-backed
daemon Terminate tests collapse to one wrapper smoke
(WritesToTerminatedKV); the six error-injection tests in
daemon_handlers_instance_test.go continue to drive the wrapper through
fakeStateStore for wire-format coverage.

Refs mulga-siv-21.

spinifex/daemon/daemon.go

@@ -1096,6 +1096,10 @@ func (d *Daemon) startCluster() error {
 		}
 	}
 
+	// Wire deps needed by InstanceService.TerminateStoppedInstance now that
+	// volumeService / vpcService / externalIPAM are constructed.
+	d.instanceService.SetTerminationDeps(d.volumeService, d.vpcService, d.externalIPAM)
+
 	d.accountService, err = initServiceWithRetry("account settings service", func() (*handlers_ec2_account.AccountSettingsServiceImpl, error) {
 		return handlers_ec2_account.NewAccountSettingsServiceImplWithNATS(d.config, d.natsConn)
 	})

spinifex/daemon/daemon_handlers_instance.go

@@ -6,7 +6,6 @@ import (
 	"fmt"
 	"log/slog"
 	"strings"
-	"time"
 
 	"github.com/aws/aws-sdk-go/aws"
 	"github.com/aws/aws-sdk-go/service/ec2"
@@ -248,7 +247,7 @@ func (d *Daemon) handleEC2RunInstances(msg *nats.Msg) {
 						}
 						// Publish vpc.add-nat for dnat_and_snat rule
 						portName := "port-" + *eni.NetworkInterfaceId
-						d.publishNATEvent("vpc.add-nat", *eni.VpcId, publicIP, *eni.PrivateIpAddress, portName, *eni.MacAddress)
+						utils.PublishNATEvent(d.natsConn, "vpc.add-nat", *eni.VpcId, publicIP, *eni.PrivateIpAddress, portName, *eni.MacAddress)
 						// Set on ec2Instance response
 						ec2Instance.PublicIpAddress = aws.String(publicIP)
 						instance.PublicIP = publicIP
@@ -904,151 +903,8 @@ func (d *Daemon) handleEC2StartStoppedInstance(msg *nats.Msg) {
 	}
 }
 
-// terminateStoppedInstanceRequest is the payload for ec2.terminate topic
-type terminateStoppedInstanceRequest struct {
-	InstanceID string `json:"instance_id"`
-}
-
-// handleEC2TerminateStoppedInstance picks up a stopped instance from shared KV,
-// deletes its volumes, and removes it from shared KV.
 func (d *Daemon) handleEC2TerminateStoppedInstance(msg *nats.Msg) {
-	var req terminateStoppedInstanceRequest
-	if err := json.Unmarshal(msg.Data, &req); err != nil {
-		slog.Error("handleEC2TerminateStoppedInstance: failed to unmarshal request", "err", err)
-		respondWithError(msg, awserrors.ErrorServerInternal)
-		return
-	}
-
-	if req.InstanceID == "" {
-		slog.Error("handleEC2TerminateStoppedInstance: missing instance_id")
-		respondWithError(msg, awserrors.ErrorMissingParameter)
-		return
-	}
-
-	if d.stateStore == nil {
-		slog.Error("handleEC2TerminateStoppedInstance: state store not available")
-		respondWithError(msg, awserrors.ErrorServerInternal)
-		return
-	}
-
-	// Load instance from shared KV
-	instance, err := d.stateStore.LoadStoppedInstance(req.InstanceID)
-	if err != nil {
-		slog.Error("handleEC2TerminateStoppedInstance: failed to load stopped instance", "instanceId", req.InstanceID, "err", err)
-		respondWithError(msg, awserrors.ErrorServerInternal)
-		return
-	}
-	if instance == nil {
-		slog.Warn("handleEC2TerminateStoppedInstance: instance not found in shared KV", "instanceId", req.InstanceID)
-		respondWithError(msg, awserrors.ErrorInvalidInstanceIDNotFound)
-		return
-	}
-
-	if instance.Status != vm.StateStopped {
-		slog.Error("handleEC2TerminateStoppedInstance: instance not in stopped state", "instanceId", req.InstanceID, "status", instance.Status)
-		respondWithError(msg, awserrors.ErrorIncorrectInstanceState)
-		return
-	}
-
-	// Verify the caller owns this instance
-	if !checkInstanceOwnership(msg, req.InstanceID, instance.AccountID) {
-		return
-	}
-
-	// Delete volumes — no QEMU shutdown or unmount needed (already done during stop)
-	instance.EBSRequests.Mu.Lock()
-	for _, ebsRequest := range instance.EBSRequests.Requests {
-		// Internal volumes (EFI, cloud-init) are always cleaned up via ebs.delete
-		if ebsRequest.EFI || ebsRequest.CloudInit {
-			ebsDeleteData, err := json.Marshal(types.EBSDeleteRequest{Volume: ebsRequest.Name})
-			if err != nil {
-				slog.Error("handleEC2TerminateStoppedInstance: failed to marshal ebs.delete request", "name", ebsRequest.Name, "err", err)
-				continue
-			}
-			deleteMsg, err := d.natsConn.Request("ebs.delete", ebsDeleteData, 30*time.Second)
-			if err != nil {
-				slog.Warn("handleEC2TerminateStoppedInstance: ebs.delete failed for internal volume", "name", ebsRequest.Name, "err", err)
-			} else {
-				slog.Info("handleEC2TerminateStoppedInstance: ebs.delete sent for internal volume", "name", ebsRequest.Name, "data", string(deleteMsg.Data))
-			}
-			continue
-		}
-
-		// User-visible volumes: respect DeleteOnTermination flag
-		if !ebsRequest.DeleteOnTermination {
-			slog.Info("handleEC2TerminateStoppedInstance: volume has DeleteOnTermination=false, skipping", "name", ebsRequest.Name)
-			continue
-		}
-
-		slog.Info("handleEC2TerminateStoppedInstance: deleting volume with DeleteOnTermination=true", "name", ebsRequest.Name)
-		_, err := d.volumeService.DeleteVolume(&ec2.DeleteVolumeInput{
-			VolumeId: &ebsRequest.Name,
-		}, instance.AccountID)
-		if err != nil {
-			slog.Error("handleEC2TerminateStoppedInstance: failed to delete volume", "name", ebsRequest.Name, "err", err)
-		}
-	}
-	instance.EBSRequests.Mu.Unlock()
-
-	// Release public IP before termination
-	if instance.PublicIP != "" && instance.PublicIPPool != "" && d.externalIPAM != nil {
-		portName := "port-" + instance.ENIId
-		vpcId := ""
-		logicalIP := ""
-		if instance.Instance != nil {
-			if instance.Instance.VpcId != nil {
-				vpcId = *instance.Instance.VpcId
-			}
-			if instance.Instance.PrivateIpAddress != nil {
-				logicalIP = *instance.Instance.PrivateIpAddress
-			}
-		}
-		d.publishNATEvent("vpc.delete-nat", vpcId, instance.PublicIP, logicalIP, portName, "")
-
-		if err := d.externalIPAM.ReleaseIP(instance.PublicIPPool, instance.PublicIP); err != nil {
-			slog.Warn("handleEC2TerminateStoppedInstance: failed to release public IP", "ip", instance.PublicIP, "pool", instance.PublicIPPool, "err", err)
-		} else {
-			slog.Info("handleEC2TerminateStoppedInstance: released public IP", "ip", instance.PublicIP, "instanceId", req.InstanceID)
-		}
-	}
-
-	// Delete ENI if present
-	if instance.ENIId != "" && d.vpcService != nil {
-		_, eniErr := d.vpcService.DeleteNetworkInterface(&ec2.DeleteNetworkInterfaceInput{
-			NetworkInterfaceId: &instance.ENIId,
-		}, instance.AccountID)
-		if eniErr != nil {
-			slog.Error("handleEC2TerminateStoppedInstance: failed to delete ENI", "eni", instance.ENIId, "err", eniErr)
-		} else {
-			slog.Info("handleEC2TerminateStoppedInstance: deleted ENI", "eni", instance.ENIId, "instanceId", req.InstanceID)
-		}
-	}
-
-	// Write to terminated KV bucket FIRST so the instance is visible in DescribeInstances.
-	// If this fails, the instance remains in the stopped bucket (safe to retry).
-	instance.Status = vm.StateTerminated
-	if err := d.stateStore.WriteTerminatedInstance(req.InstanceID, instance); err != nil {
-		slog.Error("handleEC2TerminateStoppedInstance: failed to write to terminated KV, aborting", "instanceId", req.InstanceID, "err", err)
-		respondWithError(msg, awserrors.ErrorServerInternal)
-		return
-	}
-
-	// Now safe to remove from shared stopped KV — instance already exists in terminated bucket.
-	// Retry once on failure to avoid duplicate entries in DescribeInstances.
-	if err := d.stateStore.DeleteStoppedInstance(req.InstanceID); err != nil {
-		slog.Warn("handleEC2TerminateStoppedInstance: first stopped KV delete failed, retrying",
-			"instanceId", req.InstanceID, "err", err)
-		if retryErr := d.stateStore.DeleteStoppedInstance(req.InstanceID); retryErr != nil {
-			slog.Error("handleEC2TerminateStoppedInstance: stopped KV delete failed after retry, instance may appear in both buckets",
-				"instanceId", req.InstanceID, "err", retryErr)
-		}
-	}
-
-	slog.Info("Terminated stopped instance from shared KV", "instanceId", req.InstanceID)
-
-	if err := msg.Respond(fmt.Appendf(nil, `{"status":"terminated","instanceId":"%s"}`, req.InstanceID)); err != nil {
-		slog.Error("Failed to respond to NATS request", "err", err)
-	}
+	handleNATSRequest(msg, d.instanceService.TerminateStoppedInstance)
 }
 
 // handleEC2DescribeStoppedInstances returns stopped instances from shared KV.
@@ -1277,26 +1133,3 @@ func (d *Daemon) handleEC2DescribeInstanceAttribute(msg *nats.Msg) {
 		"instanceId", instanceID, "attribute", attribute, "accountID", accountID)
 	respondWithJSON(msg, output)
 }
-
-// publishNATEvent sends a NAT lifecycle event (vpc.add-nat or vpc.delete-nat) to NATS.
-// For vpc.add-nat, it uses request-reply to ensure the OVN NAT rule is committed
-// before returning, preventing ARP propagation races. For vpc.delete-nat, it
-// uses fire-and-forget since the caller doesn't need to wait.
-func (d *Daemon) publishNATEvent(topic, vpcId, externalIP, logicalIP, portName, mac string) {
-	evt := struct {
-		VpcId      string `json:"vpc_id"`
-		ExternalIP string `json:"external_ip"`
-		LogicalIP  string `json:"logical_ip"`
-		PortName   string `json:"port_name"`
-		MAC        string `json:"mac"`
-	}{VpcId: vpcId, ExternalIP: externalIP, LogicalIP: logicalIP, PortName: portName, MAC: mac}
-
-	if topic == "vpc.add-nat" {
-		if err := utils.RequestEvent(d.natsConn, topic, evt, 10*time.Second); err != nil {
-			slog.Warn("publishNATEvent: failed to add NAT rule — OVN dnat_and_snat rule not created; restart vpcd or re-associate EIP to recover",
-				"topic", topic, "externalIP", externalIP, "logicalIP", logicalIP, "err", err)
-		}
-		return
-	}
-	utils.PublishEvent(d.natsConn, topic, evt)
-}

spinifex/daemon/daemon_handlers_instance_test.go

@@ -257,7 +257,7 @@ func TestHandleEC2TerminateStoppedInstance_LoadError(t *testing.T) {
 	store.loadStoppedErr = errors.New("kv unavailable")
 	d := daemonWithFakeStateStore(t, store)
 
-	body, _ := json.Marshal(terminateStoppedInstanceRequest{InstanceID: "i-load-fail"})
+	body, _ := json.Marshal(handlers_ec2_instance.TerminateStoppedInstanceInput{InstanceID: "i-load-fail"})
 	reply := requestHandler(t, d.natsConn, "ec2.terminate.test1", d.handleEC2TerminateStoppedInstance, testAccountID, body)
 	assert.Equal(t, awserrors.ErrorServerInternal, decodeError(t, reply.Data)["Code"])
 }
@@ -265,7 +265,7 @@ func TestHandleEC2TerminateStoppedInstance_LoadError(t *testing.T) {
 func TestHandleEC2TerminateStoppedInstance_StateStoreNil(t *testing.T) {
 	d := createTestDaemon(t, sharedNATSURL)
 
-	body, _ := json.Marshal(terminateStoppedInstanceRequest{InstanceID: "i-no-store"})
+	body, _ := json.Marshal(handlers_ec2_instance.TerminateStoppedInstanceInput{InstanceID: "i-no-store"})
 	reply := requestHandler(t, d.natsConn, "ec2.terminate.test2", d.handleEC2TerminateStoppedInstance, testAccountID, body)
 	assert.Equal(t, awserrors.ErrorServerInternal, decodeError(t, reply.Data)["Code"])
 }
@@ -279,7 +279,7 @@ func TestHandleEC2TerminateStoppedInstance_WriteTerminatedFailureAborts(t *testi
 	store.stopped[v.ID] = v
 	d := daemonWithFakeStateStore(t, store)
 
-	body, _ := json.Marshal(terminateStoppedInstanceRequest{InstanceID: v.ID})
+	body, _ := json.Marshal(handlers_ec2_instance.TerminateStoppedInstanceInput{InstanceID: v.ID})
 	reply := requestHandler(t, d.natsConn, "ec2.terminate.test3", d.handleEC2TerminateStoppedInstance, testAccountID, body)
 	assert.Equal(t, awserrors.ErrorServerInternal, decodeError(t, reply.Data)["Code"])
 
@@ -302,7 +302,7 @@ func TestHandleEC2TerminateStoppedInstance_DeleteRetrySucceeds(t *testing.T) {
 	store.stopped[v.ID] = v
 	d := daemonWithFakeStateStore(t, store)
 
-	body, _ := json.Marshal(terminateStoppedInstanceRequest{InstanceID: v.ID})
+	body, _ := json.Marshal(handlers_ec2_instance.TerminateStoppedInstanceInput{InstanceID: v.ID})
 	reply := requestHandler(t, d.natsConn, "ec2.terminate.test4", d.handleEC2TerminateStoppedInstance, testAccountID, body)
 
 	var resp map[string]string
@@ -329,7 +329,7 @@ func TestHandleEC2TerminateStoppedInstance_DeleteAlwaysFailsKeepsTerminated(t *t
 	store.stopped[v.ID] = v
 	d := daemonWithFakeStateStore(t, store)
 
-	body, _ := json.Marshal(terminateStoppedInstanceRequest{InstanceID: v.ID})
+	body, _ := json.Marshal(handlers_ec2_instance.TerminateStoppedInstanceInput{InstanceID: v.ID})
 	reply := requestHandler(t, d.natsConn, "ec2.terminate.test5", d.handleEC2TerminateStoppedInstance, testAccountID, body)
 
 	var resp map[string]string
@@ -347,7 +347,7 @@ func TestHandleEC2TerminateStoppedInstance_CrossTenantRejected(t *testing.T) {
 	store.stopped["i-foreign-term"] = stoppedVMFixture("i-foreign-term", "999988887777")
 	d := daemonWithFakeStateStore(t, store)
 
-	body, _ := json.Marshal(terminateStoppedInstanceRequest{InstanceID: "i-foreign-term"})
+	body, _ := json.Marshal(handlers_ec2_instance.TerminateStoppedInstanceInput{InstanceID: "i-foreign-term"})
 	reply := requestHandler(t, d.natsConn, "ec2.terminate.test6", d.handleEC2TerminateStoppedInstance, testAccountID, body)
 	assert.Equal(t, awserrors.ErrorInvalidInstanceIDNotFound, decodeError(t, reply.Data)["Code"])
 

spinifex/daemon/daemon_handlers_test.go

@@ -1363,121 +1363,13 @@ func TestHandleEC2DescribeStoppedInstances_WithFilter(t *testing.T) {
 	_ = daemon.jsManager.DeleteStoppedInstance("i-filter-002")
 }
 
-// --- handleEC2TerminateStoppedInstance tests ---
-
-func TestHandleEC2TerminateStoppedInstance_MissingInstanceID(t *testing.T) {
-	natsURL := sharedJSNATSURL
-
-	daemon := createFullTestDaemonWithJetStream(t, natsURL)
-
-	sub, err := daemon.natsConn.QueueSubscribe("ec2.terminate", "spinifex-workers", daemon.handleEC2TerminateStoppedInstance)
-	require.NoError(t, err)
-	defer sub.Unsubscribe()
-
-	reqData, _ := json.Marshal(map[string]string{"instance_id": ""})
-	reply, err := daemon.natsConn.Request("ec2.terminate", reqData, 5*time.Second)
-	require.NoError(t, err)
-
-	var errResp map[string]any
-	err = json.Unmarshal(reply.Data, &errResp)
-	require.NoError(t, err)
-	assert.Equal(t, awserrors.ErrorMissingParameter, errResp["Code"])
-}
-
-func TestHandleEC2TerminateStoppedInstance_MissingInstance(t *testing.T) {
-	natsURL := sharedJSNATSURL
-
-	daemon := createFullTestDaemonWithJetStream(t, natsURL)
-
-	sub, err := daemon.natsConn.QueueSubscribe("ec2.terminate", "spinifex-workers", daemon.handleEC2TerminateStoppedInstance)
-	require.NoError(t, err)
-	defer sub.Unsubscribe()
-
-	reqData, _ := json.Marshal(map[string]string{"instance_id": "i-nonexistent"})
-	reply, err := daemon.natsConn.Request("ec2.terminate", reqData, 5*time.Second)
-	require.NoError(t, err)
-
-	var errResp map[string]any
-	err = json.Unmarshal(reply.Data, &errResp)
-	require.NoError(t, err)
-	assert.Equal(t, awserrors.ErrorInvalidInstanceIDNotFound, errResp["Code"])
-}
-
-func TestHandleEC2TerminateStoppedInstance_NotStoppedState(t *testing.T) {
-	natsURL := sharedJSNATSURL
-
-	daemon := createFullTestDaemonWithJetStream(t, natsURL)
-
-	// Write an instance in running state to shared KV
-	runningVM := &vm.VM{
-		ID:           "i-term-running",
-		Status:       vm.StateRunning,
-		InstanceType: getTestInstanceType(t),
-		AccountID:    testAccountID,
-	}
-	err := daemon.jsManager.WriteStoppedInstance(runningVM.ID, runningVM)
-	require.NoError(t, err)
-
-	sub, err := daemon.natsConn.QueueSubscribe("ec2.terminate", "spinifex-workers", daemon.handleEC2TerminateStoppedInstance)
-	require.NoError(t, err)
-	defer sub.Unsubscribe()
-
-	reqData, _ := json.Marshal(map[string]string{"instance_id": "i-term-running"})
-	reply, err := natsRequest(daemon.natsConn, "ec2.terminate", reqData, 5*time.Second)
-	require.NoError(t, err)
-
-	var errResp map[string]any
-	err = json.Unmarshal(reply.Data, &errResp)
-	require.NoError(t, err)
-	assert.Equal(t, awserrors.ErrorIncorrectInstanceState, errResp["Code"])
-
-	// Cleanup
-	_ = daemon.jsManager.DeleteStoppedInstance(runningVM.ID)
-}
-
-func TestHandleEC2TerminateStoppedInstance_Success(t *testing.T) {
-	natsURL := sharedJSNATSURL
-
-	daemon := createFullTestDaemonWithJetStream(t, natsURL)
-
-	// Write a stopped instance to shared KV
-	stoppedVM := &vm.VM{
-		ID:           "i-term-stopped-001",
-		Status:       vm.StateStopped,
-		InstanceType: getTestInstanceType(t),
-		LastNode:     "node-1",
-		AccountID:    testAccountID,
-		Reservation: &ec2.Reservation{
-			ReservationId: aws.String("r-term-001"),
-			OwnerId:       aws.String("123456789012"),
-		},
-		Instance: &ec2.Instance{
-			InstanceId:   aws.String("i-term-stopped-001"),
-			InstanceType: aws.String(getTestInstanceType(t)),
-		},
-	}
-	err := daemon.jsManager.WriteStoppedInstance(stoppedVM.ID, stoppedVM)
-	require.NoError(t, err)
-
-	sub, err := daemon.natsConn.QueueSubscribe("ec2.terminate", "spinifex-workers", daemon.handleEC2TerminateStoppedInstance)
-	require.NoError(t, err)
-	defer sub.Unsubscribe()
-
-	reqData, _ := json.Marshal(map[string]string{"instance_id": "i-term-stopped-001"})
-	reply, err := natsRequest(daemon.natsConn, "ec2.terminate", reqData, 5*time.Second)
-	require.NoError(t, err)
-
-	var resp map[string]string
-	err = json.Unmarshal(reply.Data, &resp)
-	require.NoError(t, err)
-	assert.Equal(t, "terminated", resp["status"])
-	assert.Equal(t, "i-term-stopped-001", resp["instanceId"])
-
-	// Verify instance was removed from shared KV
-	loaded, err := daemon.jsManager.LoadStoppedInstance("i-term-stopped-001")
-	require.NoError(t, err)
-	assert.Nil(t, loaded, "Instance should be removed from shared KV after termination")
-}
+// --- handleEC2TerminateStoppedInstance wrapper smoke test ---
+//
+// Detailed logic coverage lives in
+// handlers/ec2/instance/service_impl_test.go (TestTerminateStoppedInstance_*).
+// The wrapper smoke case is TestHandleEC2TerminateStoppedInstance_WritesToTerminatedKV
+// further below; it confirms the NATS → handleNATSRequest → service round-trip
+// stays intact end-to-end against real JetStream KV.
 
 func TestHandleEC2GetConsoleOutput(t *testing.T) {
 	natsURL := sharedNATSURL
@@ -2833,63 +2725,6 @@ func TestHandleEC2ModifyVolume_Success(t *testing.T) {
 	assert.Equal(t, "completed", *output.VolumeModification.ModificationState)
 }
 
-// --- handleEC2TerminateStoppedInstance with volumes ---
-
-func TestHandleEC2TerminateStoppedInstance_WithVolumes(t *testing.T) {
-	daemon := createFullTestDaemonWithJetStream(t, sharedJSNATSURL)
-
-	// Subscribe a dummy ebs.delete handler
-	ebsDeleteSub, err := daemon.natsConn.Subscribe("ebs.delete", func(msg *nats.Msg) {
-		_ = msg.Respond([]byte(`{"status":"deleted"}`))
-	})
-	require.NoError(t, err)
-	defer ebsDeleteSub.Unsubscribe()
-
-	stoppedVM := &vm.VM{
-		ID:           "i-term-vol-001",
-		Status:       vm.StateStopped,
-		AccountID:    testAccountID,
-		InstanceType: getTestInstanceType(t),
-		LastNode:     "node-1",
-		Reservation: &ec2.Reservation{
-			ReservationId: aws.String("r-term-vol-001"),
-			OwnerId:       aws.String("123456789012"),
-		},
-		Instance: &ec2.Instance{
-			InstanceId:   aws.String("i-term-vol-001"),
-			InstanceType: aws.String(getTestInstanceType(t)),
-		},
-	}
-	// Add EFI, CloudInit, and a user volume with DeleteOnTermination
-	stoppedVM.EBSRequests.Requests = []types.EBSRequest{
-		{Name: "vol-efi-001", EFI: true},
-		{Name: "vol-ci-001", CloudInit: true},
-		{Name: "vol-user-001", DeleteOnTermination: true},
-		{Name: "vol-keep-001", DeleteOnTermination: false},
-	}
-
-	err = daemon.jsManager.WriteStoppedInstance(stoppedVM.ID, stoppedVM)
-	require.NoError(t, err)
-
-	sub, err := daemon.natsConn.QueueSubscribe("ec2.terminate", "spinifex-workers", daemon.handleEC2TerminateStoppedInstance)
-	require.NoError(t, err)
-	defer sub.Unsubscribe()
-
-	reqData, _ := json.Marshal(map[string]string{"instance_id": "i-term-vol-001"})
-	reply, err := natsRequest(daemon.natsConn, "ec2.terminate", reqData, 10*time.Second)
-	require.NoError(t, err)
-
-	var resp map[string]string
-	err = json.Unmarshal(reply.Data, &resp)
-	require.NoError(t, err)
-	assert.Equal(t, "terminated", resp["status"])
-	assert.Equal(t, "i-term-vol-001", resp["instanceId"])
-
-	loaded, err := daemon.jsManager.LoadStoppedInstance("i-term-vol-001")
-	require.NoError(t, err)
-	assert.Nil(t, loaded)
-}
-
 // --- handleEC2DescribeInstanceTypes with capacity filter ---
 
 func TestHandleEC2DescribeInstanceTypes_CapacityFilter(t *testing.T) {
@@ -3638,7 +3473,7 @@ func TestHandleEC2TerminateStoppedInstance_WritesToTerminatedKV(t *testing.T) {
 	require.NoError(t, err)
 	defer sub.Unsubscribe()
 
-	reqData, _ := json.Marshal(terminateStoppedInstanceRequest{InstanceID: stoppedVM.ID})
+	reqData, _ := json.Marshal(handlers_ec2_instance.TerminateStoppedInstanceInput{InstanceID: stoppedVM.ID})
 	reply, err := natsRequest(daemon.natsConn, "ec2.terminate", reqData, 30*time.Second)
 	require.NoError(t, err)
 	assert.Contains(t, string(reply.Data), "terminated")

spinifex/daemon/daemon_system_instance.go

@@ -225,7 +225,7 @@ func (d *Daemon) LaunchSystemInstance(input *handlers_elbv2.SystemInstanceInput)
 				vpcID = *result.NetworkInterfaces[0].VpcId
 			}
 			portName := "port-" + instance.ENIId
-			d.publishNATEvent("vpc.add-nat", vpcID, publicIP, privateIP, portName, instance.ENIMac)
+			utils.PublishNATEvent(d.natsConn, "vpc.add-nat", vpcID, publicIP, privateIP, portName, instance.ENIMac)
 			instance.PublicIP = publicIP
 			instance.PublicIPPool = poolName
 			slog.Info("LaunchSystemInstance: allocated public IP via direct IPAM",

spinifex/daemon/vm_adapters.go

@@ -12,6 +12,7 @@ import (
 	"github.com/mulgadc/spinifex/spinifex/awserrors"
 	handlers_ec2_placementgroup "github.com/mulgadc/spinifex/spinifex/handlers/ec2/placementgroup"
 	"github.com/mulgadc/spinifex/spinifex/types"
+	"github.com/mulgadc/spinifex/spinifex/utils"
 	"github.com/mulgadc/spinifex/spinifex/vm"
 	"github.com/nats-io/nats.go"
 )
@@ -570,7 +571,7 @@ func (a *instanceCleanerAdapter) ReleasePublicIP(instance *vm.VM) {
 			logicalIP = *instance.Instance.PrivateIpAddress
 		}
 	}
-	a.d.publishNATEvent("vpc.delete-nat", vpcId, instance.PublicIP, logicalIP, portName, "")
+	utils.PublishNATEvent(a.d.natsConn, "vpc.delete-nat", vpcId, instance.PublicIP, logicalIP, portName, "")
 
 	if err := a.d.externalIPAM.ReleaseIP(instance.PublicIPPool, instance.PublicIP); err != nil {
 		slog.Warn("Failed to release public IP on termination",

spinifex/handlers/ec2/instance/service.go

@@ -14,6 +14,18 @@ type InstanceService interface {
 	DescribeStoppedInstances(input *ec2.DescribeInstancesInput, accountID string) (*ec2.DescribeInstancesOutput, error)
 	DescribeTerminatedInstances(input *ec2.DescribeInstancesInput, accountID string) (*ec2.DescribeInstancesOutput, error)
 	ModifyInstanceAttribute(input *ec2.ModifyInstanceAttributeInput, accountID string) (*ec2.ModifyInstanceAttributeOutput, error)
+	TerminateStoppedInstance(input *TerminateStoppedInstanceInput, accountID string) (*TerminateStoppedInstanceOutput, error)
+}
+
+// TerminateStoppedInstanceInput is the payload for ec2.TerminateStoppedInstance.
+type TerminateStoppedInstanceInput struct {
+	InstanceID string `json:"instance_id"`
+}
+
+// TerminateStoppedInstanceOutput is the response payload.
+type TerminateStoppedInstanceOutput struct {
+	Status     string `json:"status"`
+	InstanceID string `json:"instanceId"`
 }
 
 // ResourceCapacityProvider exposes per-node instance-type availability used by
@@ -22,11 +34,31 @@ type ResourceCapacityProvider interface {
 	GetAvailableInstanceTypeInfos(showCapacity bool) []*ec2.InstanceTypeInfo
 }
 
-// StoppedInstanceStore provides KV-backed read/write access to stopped and
-// read access to terminated instances. Implemented by daemon.JetStreamManager.
+// StoppedInstanceStore covers KV-backed read/write access for stopped instances
+// and read+write access for terminated instances. Implemented by
+// daemon.JetStreamManager.
 type StoppedInstanceStore interface {
 	LoadStoppedInstance(instanceID string) (*vm.VM, error)
 	ListStoppedInstances() ([]*vm.VM, error)
 	ListTerminatedInstances() ([]*vm.VM, error)
 	WriteStoppedInstance(instanceID string, instance *vm.VM) error
+	DeleteStoppedInstance(instanceID string) error
+	WriteTerminatedInstance(instanceID string, instance *vm.VM) error
+}
+
+// VolumeDeleter deletes EBS volumes. Implemented by handlers/ec2/volume's
+// VolumeServiceImpl (used by the daemon).
+type VolumeDeleter interface {
+	DeleteVolume(input *ec2.DeleteVolumeInput, accountID string) (*ec2.DeleteVolumeOutput, error)
+}
+
+// ENIDeleter deletes ENIs. Implemented by handlers/ec2/vpc's VPCServiceImpl.
+type ENIDeleter interface {
+	DeleteNetworkInterface(input *ec2.DeleteNetworkInterfaceInput, accountID string) (*ec2.DeleteNetworkInterfaceOutput, error)
+}
+
+// PublicIPReleaser releases a previously allocated public IP back to a pool.
+// Implemented by handlers/ec2/vpc.ExternalIPAM.
+type PublicIPReleaser interface {
+	ReleaseIP(pool, ip string) error
 }

spinifex/handlers/ec2/instance/service_impl.go

@@ -3,6 +3,7 @@ package handlers_ec2_instance
 import (
 	"bytes"
 	"encoding/base64"
+	"encoding/json"
 	"errors"
 	"fmt"
 	"io"
@@ -236,6 +237,9 @@ type InstanceServiceImpl struct {
 	vmMgr         *vm.Manager
 	resourceMgr   ResourceCapacityProvider
 	stoppedStore  StoppedInstanceStore
+	volumeDeleter VolumeDeleter
+	eniDeleter    ENIDeleter
+	ipReleaser    PublicIPReleaser
 }
 
 // NewInstanceServiceImpl creates a new instance service implementation for daemon use
@@ -259,6 +263,15 @@ func NewInstanceServiceImpl(
 	}
 }
 
+// SetTerminationDeps wires the dependencies required by TerminateStoppedInstance.
+// Kept separate from the main constructor so handlers needing only read or
+// modify paths can construct a service without dragging the VPC/volume stack in.
+func (s *InstanceServiceImpl) SetTerminationDeps(vd VolumeDeleter, ed ENIDeleter, pr PublicIPReleaser) {
+	s.volumeDeleter = vd
+	s.eniDeleter = ed
+	s.ipReleaser = pr
+}
+
 // RunInstance creates a single EC2 instance (called per-instance by daemon)
 // Returns the VM struct and EC2 instance metadata
 func (s *InstanceServiceImpl) RunInstance(input *ec2.RunInstancesInput) (*vm.VM, *ec2.Instance, error) {
@@ -1237,6 +1250,142 @@ func (s *InstanceServiceImpl) ModifyInstanceAttribute(input *ec2.ModifyInstanceA
 	return &ec2.ModifyInstanceAttributeOutput{}, nil
 }
 
+// TerminateStoppedInstance picks up a stopped instance from shared KV, deletes
+// its volumes, releases its public IP and ENI, writes it to the terminated
+// bucket, then removes it from the stopped bucket. No QEMU shutdown or unmount
+// is needed — the instance was already drained during stop.
+func (s *InstanceServiceImpl) TerminateStoppedInstance(input *TerminateStoppedInstanceInput, accountID string) (*TerminateStoppedInstanceOutput, error) {
+	if input.InstanceID == "" {
+		return nil, errors.New(awserrors.ErrorMissingParameter)
+	}
+	if s.stoppedStore == nil {
+		slog.Error("TerminateStoppedInstance: stopped store not available")
+		return nil, errors.New(awserrors.ErrorServerInternal)
+	}
+
+	instance, err := s.stoppedStore.LoadStoppedInstance(input.InstanceID)
+	if err != nil {
+		slog.Error("TerminateStoppedInstance: failed to load stopped instance", "instanceId", input.InstanceID, "err", err)
+		return nil, errors.New(awserrors.ErrorServerInternal)
+	}
+	if instance == nil {
+		slog.Warn("TerminateStoppedInstance: instance not found in shared KV", "instanceId", input.InstanceID)
+		return nil, errors.New(awserrors.ErrorInvalidInstanceIDNotFound)
+	}
+	if instance.Status != vm.StateStopped {
+		slog.Error("TerminateStoppedInstance: instance not in stopped state", "instanceId", input.InstanceID, "status", instance.Status)
+		return nil, errors.New(awserrors.ErrorIncorrectInstanceState)
+	}
+	if !IsInstanceVisible(accountID, instance.AccountID) {
+		slog.Warn("TerminateStoppedInstance: instance not visible",
+			"instanceId", input.InstanceID, "callerAccount", accountID, "ownerAccount", instance.AccountID)
+		return nil, errors.New(awserrors.ErrorInvalidInstanceIDNotFound)
+	}
+
+	s.deleteInstanceVolumes(instance, input.InstanceID)
+	s.releaseInstancePublicIP(instance, input.InstanceID)
+	s.deleteInstanceENI(instance, input.InstanceID)
+
+	// Write to terminated KV FIRST so the instance is visible in DescribeInstances.
+	// If this fails the instance stays in the stopped bucket — safe to retry.
+	instance.Status = vm.StateTerminated
+	if err := s.stoppedStore.WriteTerminatedInstance(input.InstanceID, instance); err != nil {
+		slog.Error("TerminateStoppedInstance: failed to write to terminated KV, aborting", "instanceId", input.InstanceID, "err", err)
+		return nil, errors.New(awserrors.ErrorServerInternal)
+	}
+
+	// Now safe to remove from stopped KV. Retry once on failure so the instance
+	// doesn't appear in both buckets.
+	if err := s.stoppedStore.DeleteStoppedInstance(input.InstanceID); err != nil {
+		slog.Warn("TerminateStoppedInstance: first stopped KV delete failed, retrying",
+			"instanceId", input.InstanceID, "err", err)
+		if retryErr := s.stoppedStore.DeleteStoppedInstance(input.InstanceID); retryErr != nil {
+			slog.Error("TerminateStoppedInstance: stopped KV delete failed after retry, instance may appear in both buckets",
+				"instanceId", input.InstanceID, "err", retryErr)
+		}
+	}
+
+	slog.Info("Terminated stopped instance from shared KV", "instanceId", input.InstanceID)
+	return &TerminateStoppedInstanceOutput{Status: "terminated", InstanceID: input.InstanceID}, nil
+}
+
+func (s *InstanceServiceImpl) deleteInstanceVolumes(instance *vm.VM, instanceID string) {
+	instance.EBSRequests.Mu.Lock()
+	defer instance.EBSRequests.Mu.Unlock()
+	for _, ebsRequest := range instance.EBSRequests.Requests {
+		// Internal volumes (EFI, cloud-init) are always cleaned up via ebs.delete.
+		if ebsRequest.EFI || ebsRequest.CloudInit {
+			ebsDeleteData, err := json.Marshal(spxtypes.EBSDeleteRequest{Volume: ebsRequest.Name})
+			if err != nil {
+				slog.Error("TerminateStoppedInstance: failed to marshal ebs.delete request", "name", ebsRequest.Name, "err", err)
+				continue
+			}
+			deleteMsg, err := s.natsConn.Request("ebs.delete", ebsDeleteData, 30*time.Second)
+			if err != nil {
+				slog.Warn("TerminateStoppedInstance: ebs.delete failed for internal volume", "name", ebsRequest.Name, "err", err)
+			} else {
+				slog.Info("TerminateStoppedInstance: ebs.delete sent for internal volume", "name", ebsRequest.Name, "data", string(deleteMsg.Data))
+			}
+			continue
+		}
+
+		// User-visible volumes: respect DeleteOnTermination.
+		if !ebsRequest.DeleteOnTermination {
+			slog.Info("TerminateStoppedInstance: volume has DeleteOnTermination=false, skipping", "name", ebsRequest.Name)
+			continue
+		}
+
+		slog.Info("TerminateStoppedInstance: deleting volume with DeleteOnTermination=true", "name", ebsRequest.Name)
+		if s.volumeDeleter == nil {
+			slog.Warn("TerminateStoppedInstance: volume deleter not configured, skipping", "name", ebsRequest.Name)
+			continue
+		}
+		if _, err := s.volumeDeleter.DeleteVolume(&ec2.DeleteVolumeInput{
+			VolumeId: &ebsRequest.Name,
+		}, instance.AccountID); err != nil {
+			slog.Error("TerminateStoppedInstance: failed to delete volume", "name", ebsRequest.Name, "err", err)
+		}
+	}
+	_ = instanceID
+}
+
+func (s *InstanceServiceImpl) releaseInstancePublicIP(instance *vm.VM, instanceID string) {
+	if instance.PublicIP == "" || instance.PublicIPPool == "" || s.ipReleaser == nil {
+		return
+	}
+	portName := "port-" + instance.ENIId
+	vpcID := ""
+	logicalIP := ""
+	if instance.Instance != nil {
+		if instance.Instance.VpcId != nil {
+			vpcID = *instance.Instance.VpcId
+		}
+		if instance.Instance.PrivateIpAddress != nil {
+			logicalIP = *instance.Instance.PrivateIpAddress
+		}
+	}
+	utils.PublishNATEvent(s.natsConn, "vpc.delete-nat", vpcID, instance.PublicIP, logicalIP, portName, "")
+
+	if err := s.ipReleaser.ReleaseIP(instance.PublicIPPool, instance.PublicIP); err != nil {
+		slog.Warn("TerminateStoppedInstance: failed to release public IP", "ip", instance.PublicIP, "pool", instance.PublicIPPool, "err", err)
+	} else {
+		slog.Info("TerminateStoppedInstance: released public IP", "ip", instance.PublicIP, "instanceId", instanceID)
+	}
+}
+
+func (s *InstanceServiceImpl) deleteInstanceENI(instance *vm.VM, instanceID string) {
+	if instance.ENIId == "" || s.eniDeleter == nil {
+		return
+	}
+	if _, err := s.eniDeleter.DeleteNetworkInterface(&ec2.DeleteNetworkInterfaceInput{
+		NetworkInterfaceId: &instance.ENIId,
+	}, instance.AccountID); err != nil {
+		slog.Error("TerminateStoppedInstance: failed to delete ENI", "eni", instance.ENIId, "err", err)
+	} else {
+		slog.Info("TerminateStoppedInstance: deleted ENI", "eni", instance.ENIId, "instanceId", instanceID)
+	}
+}
+
 // DescribeInstanceAttribute returns a single requested attribute for an instance.
 // Checks running instances first (in-memory), then falls back to stopped instances in KV.
 func (s *InstanceServiceImpl) DescribeInstanceAttribute(input *ec2.DescribeInstanceAttributeInput, accountID string) (*ec2.DescribeInstanceAttributeOutput, error) {

spinifex/handlers/ec2/instance/service_impl_test.go

@@ -2,18 +2,24 @@ package handlers_ec2_instance
 
 import (
 	"bytes"
+	"encoding/json"
+	"errors"
 	"fmt"
 	"strings"
 	"testing"
 	"text/template"
+	"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/config"
 	"github.com/mulgadc/spinifex/spinifex/objectstore"
+	spxtypes "github.com/mulgadc/spinifex/spinifex/types"
 	"github.com/mulgadc/spinifex/spinifex/utils"
 	"github.com/mulgadc/spinifex/spinifex/vm"
+	"github.com/nats-io/nats-server/v2/server"
+	"github.com/nats-io/nats.go"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 )
@@ -653,14 +659,20 @@ func (f *fakeResourceCapacityProvider) GetAvailableInstanceTypeInfos(showCapacit
 }
 
 type fakeStoppedStore struct {
-	stopped      []*vm.VM
-	terminated   []*vm.VM
-	loadByID     map[string]*vm.VM
-	wroteStopped map[string]*vm.VM
-	listErr      error
-	listTermErr  error
-	loadErr      error
-	writeErr     error
+	stopped         []*vm.VM
+	terminated      []*vm.VM
+	loadByID        map[string]*vm.VM
+	wroteStopped    map[string]*vm.VM
+	wroteTerminated map[string]*vm.VM
+	deletedStopped  []string
+	listErr         error
+	listTermErr     error
+	loadErr         error
+	writeErr        error
+	writeTermErr    error
+	deleteErr       error
+	deleteFailFirst bool
+	deleteAttempts  int
 }
 
 func (f *fakeStoppedStore) LoadStoppedInstance(id string) (*vm.VM, error) {
@@ -694,6 +706,27 @@ func (f *fakeStoppedStore) WriteStoppedInstance(id string, instance *vm.VM) erro
 	f.wroteStopped[id] = instance
 	return nil
 }
+func (f *fakeStoppedStore) WriteTerminatedInstance(id string, instance *vm.VM) error {
+	if f.writeTermErr != nil {
+		return f.writeTermErr
+	}
+	if f.wroteTerminated == nil {
+		f.wroteTerminated = make(map[string]*vm.VM)
+	}
+	f.wroteTerminated[id] = instance
+	return nil
+}
+func (f *fakeStoppedStore) DeleteStoppedInstance(id string) error {
+	f.deleteAttempts++
+	if f.deleteFailFirst && f.deleteAttempts == 1 {
+		return errors.New("transient delete failure")
+	}
+	if f.deleteErr != nil {
+		return f.deleteErr
+	}
+	f.deletedStopped = append(f.deletedStopped, id)
+	return nil
+}
 
 func TestDescribeInstanceTypes_NilResourceMgr(t *testing.T) {
 	svc := &InstanceServiceImpl{}
@@ -1267,3 +1300,266 @@ func TestModifyInstanceAttribute_WriteError(t *testing.T) {
 	require.Error(t, err)
 	assert.Equal(t, awserrors.ErrorServerInternal, err.Error())
 }
+
+// --- TerminateStoppedInstance tests ---
+
+type fakeVolumeDeleter struct {
+	calls   []string
+	deleted []string
+	err     error
+}
+
+func (f *fakeVolumeDeleter) DeleteVolume(input *ec2.DeleteVolumeInput, _ string) (*ec2.DeleteVolumeOutput, error) {
+	id := aws.StringValue(input.VolumeId)
+	f.calls = append(f.calls, id)
+	if f.err != nil {
+		return nil, f.err
+	}
+	f.deleted = append(f.deleted, id)
+	return &ec2.DeleteVolumeOutput{}, nil
+}
+
+type fakeENIDeleter struct {
+	calls []string
+	err   error
+}
+
+func (f *fakeENIDeleter) DeleteNetworkInterface(input *ec2.DeleteNetworkInterfaceInput, _ string) (*ec2.DeleteNetworkInterfaceOutput, error) {
+	f.calls = append(f.calls, aws.StringValue(input.NetworkInterfaceId))
+	if f.err != nil {
+		return nil, f.err
+	}
+	return &ec2.DeleteNetworkInterfaceOutput{}, nil
+}
+
+type fakePublicIPReleaser struct {
+	pool string
+	ip   string
+	err  error
+}
+
+func (f *fakePublicIPReleaser) ReleaseIP(pool, ip string) error {
+	f.pool = pool
+	f.ip = ip
+	return f.err
+}
+
+// embeddedNATS spins up an in-process NATS server scoped to the test and
+// returns a connected client. Used for service tests that exercise the
+// ebs.delete path inside TerminateStoppedInstance.
+func embeddedNATS(t *testing.T) *nats.Conn {
+	t.Helper()
+	opts := &server.Options{
+		Host:   "127.0.0.1",
+		Port:   -1,
+		NoLog:  true,
+		NoSigs: true,
+	}
+	ns, err := server.NewServer(opts)
+	require.NoError(t, err)
+	go ns.Start()
+	require.True(t, ns.ReadyForConnections(5*time.Second))
+	t.Cleanup(func() { ns.Shutdown() })
+
+	nc, err := nats.Connect(ns.ClientURL())
+	require.NoError(t, err)
+	t.Cleanup(func() { nc.Close() })
+	return nc
+}
+
+func TestTerminateStoppedInstance_MissingInstanceID(t *testing.T) {
+	svc := &InstanceServiceImpl{}
+	_, err := svc.TerminateStoppedInstance(&TerminateStoppedInstanceInput{}, "acc")
+	require.Error(t, err)
+	assert.Equal(t, awserrors.ErrorMissingParameter, err.Error())
+}
+
+func TestTerminateStoppedInstance_NilStore(t *testing.T) {
+	svc := &InstanceServiceImpl{}
+	_, err := svc.TerminateStoppedInstance(&TerminateStoppedInstanceInput{InstanceID: "i-1"}, "acc")
+	require.Error(t, err)
+	assert.Equal(t, awserrors.ErrorServerInternal, err.Error())
+}
+
+func TestTerminateStoppedInstance_LoadError(t *testing.T) {
+	store := &fakeStoppedStore{loadErr: errors.New("kv down")}
+	svc := &InstanceServiceImpl{stoppedStore: store}
+
+	_, err := svc.TerminateStoppedInstance(&TerminateStoppedInstanceInput{InstanceID: "i-1"}, "acc")
+	require.Error(t, err)
+	assert.Equal(t, awserrors.ErrorServerInternal, err.Error())
+}
+
+func TestTerminateStoppedInstance_NotFound(t *testing.T) {
+	store := &fakeStoppedStore{loadByID: map[string]*vm.VM{}}
+	svc := &InstanceServiceImpl{stoppedStore: store}
+
+	_, err := svc.TerminateStoppedInstance(&TerminateStoppedInstanceInput{InstanceID: "i-missing"}, "acc")
+	require.Error(t, err)
+	assert.Equal(t, awserrors.ErrorInvalidInstanceIDNotFound, err.Error())
+}
+
+func TestTerminateStoppedInstance_NotStopped(t *testing.T) {
+	id := "i-running"
+	store := &fakeStoppedStore{loadByID: map[string]*vm.VM{
+		id: {ID: id, Status: vm.StateRunning, AccountID: "acc"},
+	}}
+	svc := &InstanceServiceImpl{stoppedStore: store}
+
+	_, err := svc.TerminateStoppedInstance(&TerminateStoppedInstanceInput{InstanceID: id}, "acc")
+	require.Error(t, err)
+	assert.Equal(t, awserrors.ErrorIncorrectInstanceState, err.Error())
+}
+
+func TestTerminateStoppedInstance_NotVisible(t *testing.T) {
+	id := "i-stopped"
+	store := &fakeStoppedStore{loadByID: map[string]*vm.VM{
+		id: {ID: id, Status: vm.StateStopped, AccountID: "owner-acc"},
+	}}
+	svc := &InstanceServiceImpl{stoppedStore: store}
+
+	_, err := svc.TerminateStoppedInstance(&TerminateStoppedInstanceInput{InstanceID: id}, "other-acc")
+	require.Error(t, err)
+	assert.Equal(t, awserrors.ErrorInvalidInstanceIDNotFound, err.Error())
+}
+
+func TestTerminateStoppedInstance_HappyPath(t *testing.T) {
+	id := "i-term-001"
+	store := &fakeStoppedStore{loadByID: map[string]*vm.VM{
+		id: {ID: id, Status: vm.StateStopped, AccountID: "acc"},
+	}}
+	svc := &InstanceServiceImpl{stoppedStore: store}
+
+	out, err := svc.TerminateStoppedInstance(&TerminateStoppedInstanceInput{InstanceID: id}, "acc")
+	require.NoError(t, err)
+	require.NotNil(t, out)
+	assert.Equal(t, "terminated", out.Status)
+	assert.Equal(t, id, out.InstanceID)
+
+	require.NotNil(t, store.wroteTerminated[id])
+	assert.Equal(t, vm.StateTerminated, store.wroteTerminated[id].Status)
+	assert.Contains(t, store.deletedStopped, id)
+}
+
+func TestTerminateStoppedInstance_WriteTerminatedError_Aborts(t *testing.T) {
+	id := "i-werr"
+	store := &fakeStoppedStore{
+		loadByID:     map[string]*vm.VM{id: {ID: id, Status: vm.StateStopped, AccountID: "acc"}},
+		writeTermErr: errors.New("kv write boom"),
+	}
+	svc := &InstanceServiceImpl{stoppedStore: store}
+
+	_, err := svc.TerminateStoppedInstance(&TerminateStoppedInstanceInput{InstanceID: id}, "acc")
+	require.Error(t, err)
+	assert.Equal(t, awserrors.ErrorServerInternal, err.Error())
+	assert.Empty(t, store.deletedStopped, "stopped delete must not run when terminated write fails")
+}
+
+func TestTerminateStoppedInstance_RetriesStoppedDelete(t *testing.T) {
+	id := "i-retry"
+	store := &fakeStoppedStore{
+		loadByID:        map[string]*vm.VM{id: {ID: id, Status: vm.StateStopped, AccountID: "acc"}},
+		deleteFailFirst: true,
+	}
+	svc := &InstanceServiceImpl{stoppedStore: store}
+
+	_, err := svc.TerminateStoppedInstance(&TerminateStoppedInstanceInput{InstanceID: id}, "acc")
+	require.NoError(t, err)
+	assert.Equal(t, 2, store.deleteAttempts, "first delete fails, retry succeeds")
+	assert.Contains(t, store.deletedStopped, id)
+}
+
+func TestTerminateStoppedInstance_UserVolumeDeleted(t *testing.T) {
+	id := "i-vol"
+	v := &vm.VM{ID: id, Status: vm.StateStopped, AccountID: "acc"}
+	v.EBSRequests.Requests = []spxtypes.EBSRequest{
+		{Name: "vol-user-001", DeleteOnTermination: true},
+		{Name: "vol-keep-001", DeleteOnTermination: false},
+	}
+	store := &fakeStoppedStore{loadByID: map[string]*vm.VM{id: v}}
+	vd := &fakeVolumeDeleter{}
+	svc := &InstanceServiceImpl{stoppedStore: store, volumeDeleter: vd}
+
+	_, err := svc.TerminateStoppedInstance(&TerminateStoppedInstanceInput{InstanceID: id}, "acc")
+	require.NoError(t, err)
+	assert.Equal(t, []string{"vol-user-001"}, vd.calls, "only DeleteOnTermination=true volumes deleted")
+	assert.Equal(t, []string{"vol-user-001"}, vd.deleted)
+}
+
+func TestTerminateStoppedInstance_NoVolumeDeleterSkipsGracefully(t *testing.T) {
+	id := "i-no-vd"
+	v := &vm.VM{ID: id, Status: vm.StateStopped, AccountID: "acc"}
+	v.EBSRequests.Requests = []spxtypes.EBSRequest{
+		{Name: "vol-user-001", DeleteOnTermination: true},
+	}
+	store := &fakeStoppedStore{loadByID: map[string]*vm.VM{id: v}}
+	svc := &InstanceServiceImpl{stoppedStore: store}
+
+	_, err := svc.TerminateStoppedInstance(&TerminateStoppedInstanceInput{InstanceID: id}, "acc")
+	require.NoError(t, err, "missing VolumeDeleter must not abort termination")
+	require.NotNil(t, store.wroteTerminated[id])
+}
+
+func TestTerminateStoppedInstance_InternalVolumesViaNATS(t *testing.T) {
+	id := "i-int-vol"
+	v := &vm.VM{ID: id, Status: vm.StateStopped, AccountID: "acc"}
+	v.EBSRequests.Requests = []spxtypes.EBSRequest{
+		{Name: "vol-efi-001", EFI: true},
+		{Name: "vol-ci-001", CloudInit: true},
+	}
+	store := &fakeStoppedStore{loadByID: map[string]*vm.VM{id: v}}
+
+	nc := embeddedNATS(t)
+	var ebsDeleted []string
+	sub, err := nc.Subscribe("ebs.delete", func(msg *nats.Msg) {
+		var req spxtypes.EBSDeleteRequest
+		_ = json.Unmarshal(msg.Data, &req)
+		ebsDeleted = append(ebsDeleted, req.Volume)
+		_ = msg.Respond([]byte(`{"Success":true}`))
+	})
+	require.NoError(t, err)
+	defer sub.Unsubscribe()
+
+	svc := &InstanceServiceImpl{stoppedStore: store, natsConn: nc}
+
+	_, err = svc.TerminateStoppedInstance(&TerminateStoppedInstanceInput{InstanceID: id}, "acc")
+	require.NoError(t, err)
+	assert.ElementsMatch(t, []string{"vol-efi-001", "vol-ci-001"}, ebsDeleted)
+}
+
+func TestTerminateStoppedInstance_PublicIPReleased(t *testing.T) {
+	id := "i-pubip"
+	v := &vm.VM{
+		ID:           id,
+		Status:       vm.StateStopped,
+		AccountID:    "acc",
+		PublicIP:     "203.0.113.5",
+		PublicIPPool: "pool-a",
+		ENIId:        "eni-xyz",
+		Instance: &ec2.Instance{
+			InstanceId:       aws.String(id),
+			VpcId:            aws.String("vpc-1"),
+			PrivateIpAddress: aws.String("10.0.0.5"),
+		},
+	}
+	store := &fakeStoppedStore{loadByID: map[string]*vm.VM{id: v}}
+	pr := &fakePublicIPReleaser{}
+	svc := &InstanceServiceImpl{stoppedStore: store, ipReleaser: pr}
+
+	_, err := svc.TerminateStoppedInstance(&TerminateStoppedInstanceInput{InstanceID: id}, "acc")
+	require.NoError(t, err)
+	assert.Equal(t, "pool-a", pr.pool)
+	assert.Equal(t, "203.0.113.5", pr.ip)
+}
+
+func TestTerminateStoppedInstance_ENIDeleted(t *testing.T) {
+	id := "i-eni"
+	v := &vm.VM{ID: id, Status: vm.StateStopped, AccountID: "acc", ENIId: "eni-1234"}
+	store := &fakeStoppedStore{loadByID: map[string]*vm.VM{id: v}}
+	ed := &fakeENIDeleter{}
+	svc := &InstanceServiceImpl{stoppedStore: store, eniDeleter: ed}
+
+	_, err := svc.TerminateStoppedInstance(&TerminateStoppedInstanceInput{InstanceID: id}, "acc")
+	require.NoError(t, err)
+	assert.Equal(t, []string{"eni-1234"}, ed.calls)
+}

spinifex/handlers/ec2/instance/service_mock.go

@@ -59,3 +59,7 @@ func (s *MockInstanceService) DescribeTerminatedInstances(_ *ec2.DescribeInstanc
 func (s *MockInstanceService) ModifyInstanceAttribute(_ *ec2.ModifyInstanceAttributeInput, _ string) (*ec2.ModifyInstanceAttributeOutput, error) {
 	return &ec2.ModifyInstanceAttributeOutput{}, nil
 }
+
+func (s *MockInstanceService) TerminateStoppedInstance(input *TerminateStoppedInstanceInput, _ string) (*TerminateStoppedInstanceOutput, error) {
+	return &TerminateStoppedInstanceOutput{Status: "terminated", InstanceID: input.InstanceID}, nil
+}

spinifex/handlers/ec2/instance/service_nats.go

@@ -64,3 +64,7 @@ func (s *NATSInstanceService) DescribeTerminatedInstances(input *ec2.DescribeIns
 func (s *NATSInstanceService) ModifyInstanceAttribute(input *ec2.ModifyInstanceAttributeInput, accountID string) (*ec2.ModifyInstanceAttributeOutput, error) {
 	return utils.NATSRequest[ec2.ModifyInstanceAttributeOutput](s.natsConn, "ec2.ModifyInstanceAttribute", input, 30*time.Second, accountID)
 }
+
+func (s *NATSInstanceService) TerminateStoppedInstance(input *TerminateStoppedInstanceInput, accountID string) (*TerminateStoppedInstanceOutput, error) {
+	return utils.NATSRequest[TerminateStoppedInstanceOutput](s.natsConn, "ec2.TerminateStoppedInstance", input, 30*time.Second, accountID)
+}

spinifex/utils/nats.go

@@ -382,6 +382,28 @@ func PublishEvent(nc *nats.Conn, topic string, event any) {
 	}
 }
 
+// PublishNATEvent sends a NAT lifecycle event (vpc.add-nat or vpc.delete-nat)
+// to NATS. vpc.add-nat uses request-reply to commit the OVN NAT rule before
+// returning (prevents ARP propagation races); vpc.delete-nat is fire-and-forget.
+func PublishNATEvent(nc *nats.Conn, topic, vpcID, externalIP, logicalIP, portName, mac string) {
+	evt := struct {
+		VpcId      string `json:"vpc_id"`
+		ExternalIP string `json:"external_ip"`
+		LogicalIP  string `json:"logical_ip"`
+		PortName   string `json:"port_name"`
+		MAC        string `json:"mac"`
+	}{VpcId: vpcID, ExternalIP: externalIP, LogicalIP: logicalIP, PortName: portName, MAC: mac}
+
+	if topic == "vpc.add-nat" {
+		if err := RequestEvent(nc, topic, evt, 10*time.Second); err != nil {
+			slog.Warn("PublishNATEvent: failed to add NAT rule — OVN dnat_and_snat rule not created; restart vpcd or re-associate EIP to recover",
+				"topic", topic, "externalIP", externalIP, "logicalIP", logicalIP, "err", err)
+		}
+		return
+	}
+	PublishEvent(nc, topic, evt)
+}
+
 // RequestEvent marshals event as JSON and sends a NATS request, waiting for a
 // response. This ensures the subscriber has processed the event before the
 // caller continues. Returns an error if the request times out or the responder