feat(lambda): boot test-link invocations under Firecracker (#1648)

* feat(lambda): deliver guest configuration on a disk the agent erases

* feat(lambda): boot test-link attempts under Firecracker with a kernel hard kill

* fix(lambda): detect the configuration disk by its missing fw_cfg entry

* test(lambda): cover the configuration-disk seal from the agent and state in-package reasons

* fix(lambda): release a Firecracker attempt whose supervisor never started

architecture/layering/allowlist_test.go

@@ -17,7 +17,6 @@ var allowlist = []allowance{
 	{From: "operator/accountlifecycle", To: "domains/ec2/awsapi/instance", Debt: "accountteardown (dependency)", Reason: "EC2 reapers call the awsapi action adapters; needs domain-owned cleanup capabilities"},
 	{From: "operator/accountlifecycle", To: "domains/ec2/awsapi/routetable", Debt: "accountteardown (dependency)", Reason: "EC2 reapers call the awsapi action adapters; needs domain-owned cleanup capabilities"},
 	{From: "operator/accountlifecycle", To: "domains/ec2/awsapi/snapshot", Debt: "accountteardown (dependency)", Reason: "EC2 reapers call the awsapi action adapters; needs domain-owned cleanup capabilities"},
-	{From: "operator/accountlifecycle", To: "domains/ec2/awsapi/volume", Debt: "accountteardown (dependency)", Reason: "EC2 reapers call the awsapi action adapters; needs domain-owned cleanup capabilities"},
 	{From: "operator/accountlifecycle", To: "domains/ec2/awsapi/vpc", Debt: "accountteardown (dependency)", Reason: "EC2 reapers call the awsapi action adapters; needs domain-owned cleanup capabilities"},
 
 	{From: "agents/ecs/agent", To: "domains/ecs", Debt: "agents/ecs/agent", Reason: "assignment/state wire types from the ECS implementation; needs a guest/controller contract"},

build/microvm/lambda-runtime/init.sh

@@ -16,8 +16,8 @@ load_mod() {
     done
     return 1
 }
-# lambda-agent reads its identity from fw_cfg and unloads qemu_fw_cfg before
-# anything else runs. virtio_net needs net_failover and failover, and the
+# lambda-agent reads its identity from fw_cfg, or the configuration disk, and
+# seals it before anything else runs. virtio_net needs net_failover and failover, and the
 # insmod fallback resolves no dependencies, so load the chain bottom-up.
 load_mod qemu_fw_cfg || echo "[init] qemu_fw_cfg did not load" >/dev/console
 load_mod failover || true
@@ -26,8 +26,30 @@ load_mod virtio_net || echo "[init] virtio_net did not load" >/dev/console
 
 ip link set lo up
 
+# Without fw_cfg data the executor attaches the same entries as a tar archive
+# on the second disk; lambda-agent reads them here and erases the disk before
+# the function starts. qemu_fw_cfg loads without a device, so test the entry.
+CFG_ROOT=/spinifex-cfg
+FWCFG_ROOT=/sys/firmware/qemu_fw_cfg
+if [ ! -f "$FWCFG_ROOT/by_name/opt/spinifex/netcfg/raw" ]; then
+    load_mod virtio_blk || echo "[init] virtio_blk did not load" >/dev/console
+    i=0
+    while [ "$i" -lt 25 ] && [ ! -b /dev/vdb ]; do
+        sleep 0.04
+        i=$((i + 1))
+    done
+    if [ -b /dev/vdb ]; then
+        mkdir -m 0700 "$CFG_ROOT"
+        if ! mount -t tmpfs -o mode=0700,size=1m tmpfs "$CFG_ROOT" ||
+            ! tar -xf /dev/vdb -C "$CFG_ROOT"; then
+            echo "[init] configuration disk not readable" >/dev/console
+        fi
+        FWCFG_ROOT=$CFG_ROOT
+    fi
+fi
+
 # The guest's addresses come from the same netcfg entry as other system VMs.
-NETCFG=/sys/firmware/qemu_fw_cfg/by_name/opt/spinifex/netcfg/raw
+NETCFG=$FWCFG_ROOT/by_name/opt/spinifex/netcfg/raw
 i=0
 while [ "$i" -lt 25 ] && [ ! -f "$NETCFG" ]; do
     sleep 0.04

cmd/lambda-agent/connect.go

@@ -5,6 +5,7 @@ import (
 	"crypto/tls"
 	"errors"
 	"fmt"
+	"os"
 
 	lambdav1 "github.com/mulgadc/spinifex/contracts/lambda/v1"
 	"github.com/mulgadc/spinifex/spinifex/agents/lambda/hostlink"
@@ -25,9 +26,22 @@ type startup struct {
 }
 
 func defaultStartup() startup {
+	if _, err := os.Stat(hostlink.ConfigDiskRoot); err == nil {
+		return configDiskStartup(hostlink.ConfigDiskRoot, hostlink.ConfigDiskDevice)
+	}
 	return startup{fwCfgDir: hostlink.FwCfgDir, fwCfgRoot: hostlink.FwCfgRoot, unload: hostlink.UnloadFwCfg}
 }
 
+// configDiskStartup reads the entries init extracted from the configuration
+// disk, and seals by erasing the disk and removing the extracted tree.
+func configDiskStartup(root, device string) startup {
+	return startup{
+		fwCfgDir:  root + "/by_name/" + lambdav1.FwCfgPrefix,
+		fwCfgRoot: root,
+		unload:    func() error { return hostlink.SealConfigDisk(device, root) },
+	}
+}
+
 // connect loads the identity, seals fw_cfg whether or not the load succeeded,
 // then dials the host. Under TLS 1.3 a host refusal is seen only on first read.
 func (s startup) connect(ctx context.Context) (*tls.Conn, lambdav1.Scope, int, error) {

cmd/lambda-agent/connect_configdisk_test.go

@@ -0,0 +1,43 @@
+//test:in-package — package main cannot be imported by an external test package.
+
+package main
+
+import (
+	"bytes"
+	"os"
+	"path/filepath"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+)
+
+// An identity that does not load still seals the configuration disk: the
+// guest key on it is erased before the agent refuses to start.
+func TestConfigDiskStartup_RefusedLoadErasesDisk(t *testing.T) {
+	dir := t.TempDir()
+	root := filepath.Join(dir, "cfg")
+	require.NoError(t, os.MkdirAll(filepath.Join(root, "by_name"), 0o700))
+	device := filepath.Join(dir, "vdb")
+	require.NoError(t, os.WriteFile(device, bytes.Repeat([]byte("guest-key"), 512), 0o600))
+
+	_, _, code, err := configDiskStartup(root, device).connect(t.Context())
+	require.Error(t, err)
+	require.Equal(t, exitRefused, code)
+	got, rerr := os.ReadFile(device)
+	require.NoError(t, rerr)
+	require.Len(t, got, 9*512)
+	require.Equal(t, make([]byte, len(got)), got, "configuration disk erased")
+}
+
+// A disk that cannot be erased fails closed: the agent refuses, and the
+// extracted tree is kept so the seal check sees it.
+func TestConfigDiskStartup_UnerasableDiskRefuses(t *testing.T) {
+	dir := t.TempDir()
+	root := filepath.Join(dir, "cfg")
+	require.NoError(t, os.MkdirAll(filepath.Join(root, "by_name"), 0o700))
+
+	_, _, code, err := configDiskStartup(root, filepath.Join(dir, "missing")).connect(t.Context())
+	require.ErrorContains(t, err, "erase configuration disk")
+	require.Equal(t, exitRefused, code)
+	require.DirExists(t, root)
+}

contracts/ec2/v1/commands.go

@@ -12,6 +12,11 @@ const (
 	// this contract family. It is the NATS permission/subscription pattern for
 	// the per-instance command route, rather than the broader descendant match.
 	InstanceCommandSubjectWildcard = InstanceCommandSubjectPrefix + "*"
+
+	// HandledHeader is set to "1" on every reply an owner sends to a
+	// modify-instance-metadata-options or set-disable-api-termination command.
+	// An owner that predates those commands cannot set it.
+	HandledHeader = "Spinifex-Ec2-Cmd-Handled"
 )
 
 // InstanceCommandSubject returns the targeted command subject for instanceID.
@@ -23,7 +28,9 @@ func InstanceCommandSubject(instanceID string) string {
 }
 
 // EC2InstanceCommand is the NATS wire format for EC2 instance commands
-// (stop, terminate, start, attach/detach-volume, attach/detach-eni).
+// (stop, terminate, start, reboot, attach/detach-volume, attach/detach-eni,
+// IAM profile, spot lineage, tags, monitoring, metadata options and
+// termination protection).
 type EC2InstanceCommand struct {
 	ID                        string                     `json:"id"`
 	Attributes                EC2CommandAttributes       `json:"attributes"`
@@ -35,6 +42,9 @@ type EC2InstanceCommand struct {
 	SpotLineageData           *SpotLineageData           `json:"spot_lineage_data,omitempty"`
 	InstanceTagsData          *InstanceTagsData          `json:"instance_tags_data,omitempty"`
 	InstanceMonitoringData    *InstanceMonitoringData    `json:"instance_monitoring_data,omitempty"`
+
+	InstanceMetadataOptionsData *InstanceMetadataOptionsData `json:"instance_metadata_options_data,omitempty"`
+	DisableApiTerminationData   *DisableApiTerminationData   `json:"disable_api_termination_data,omitempty"`
 }
 
 // EC2CommandAttributes indicates which action the daemon should perform.
@@ -52,6 +62,9 @@ type EC2CommandAttributes struct {
 	SetInstanceTags             bool `json:"set_instance_tags,omitempty"`
 	RemoveInstanceTags          bool `json:"remove_instance_tags,omitempty"`
 	SetInstanceMonitoring       bool `json:"set_instance_monitoring,omitempty"`
+
+	ModifyInstanceMetadataOptions bool `json:"modify_instance_metadata_options,omitempty"`
+	SetDisableApiTermination      bool `json:"set_disable_api_termination,omitempty"`
 }
 
 // AttachVolumeData carries parameters for an attach-volume command.
@@ -103,6 +116,23 @@ type InstanceMonitoringData struct {
 	Enabled bool `json:"enabled"`
 }
 
+// InstanceMetadataOptionsData carries the requested metadata options for a
+// modify-instance-metadata-options command. Unset fields are left unchanged.
+type InstanceMetadataOptionsData struct {
+	HttpTokens              string `json:"http_tokens,omitempty"`
+	HttpEndpoint            string `json:"http_endpoint,omitempty"`
+	HttpProtocolIpv6        string `json:"http_protocol_ipv6,omitempty"`
+	InstanceMetadataTags    string `json:"instance_metadata_tags,omitempty"`
+	HttpPutResponseHopLimit *int64 `json:"http_put_response_hop_limit,omitempty"`
+}
+
+// DisableApiTerminationData carries the desired termination-protection value
+// for a set-disable-api-termination command. Carried as data so "false" is an
+// explicit value, not the absence of one.
+type DisableApiTerminationData struct {
+	Value bool `json:"value"`
+}
+
 // SpotLineageData carries the SIR id stamped onto a spot-launched instance.
 // Lifecycle is always "spot" for this command, so only the SIR id travels.
 type SpotLineageData struct {

contracts/ec2/v1/commands_test.go

@@ -24,6 +24,9 @@ func TestEC2InstanceCommandJSONContract(t *testing.T) {
 			SetInstanceTags:             true,
 			RemoveInstanceTags:          true,
 			SetInstanceMonitoring:       true,
+
+			ModifyInstanceMetadataOptions: true,
+			SetDisableApiTermination:      true,
 		},
 		AttachVolumeData:          &AttachVolumeData{VolumeID: "vol-1", Device: "/dev/sdf"},
 		DetachVolumeData:          &DetachVolumeData{VolumeID: "vol-2", Device: "/dev/sdg", Force: true},
@@ -33,6 +36,15 @@ func TestEC2InstanceCommandJSONContract(t *testing.T) {
 		SpotLineageData:           &SpotLineageData{SpotInstanceRequestId: "sir-1"},
 		InstanceTagsData:          &InstanceTagsData{Tags: map[string]string{"Name": "example"}, TagKeys: []string{"old"}},
 		InstanceMonitoringData:    &InstanceMonitoringData{Enabled: true},
+
+		InstanceMetadataOptionsData: &InstanceMetadataOptionsData{
+			HttpTokens:              "required",
+			HttpEndpoint:            "enabled",
+			HttpProtocolIpv6:        "disabled",
+			InstanceMetadataTags:    "enabled",
+			HttpPutResponseHopLimit: int64Ptr(3),
+		},
+		DisableApiTerminationData: &DisableApiTerminationData{Value: false},
 	}
 
 	data, err := json.Marshal(command)
@@ -52,7 +64,9 @@ func TestEC2InstanceCommandJSONContract(t *testing.T) {
 			"set_spot_lineage": true,
 			"set_instance_tags": true,
 			"remove_instance_tags": true,
-			"set_instance_monitoring": true
+			"set_instance_monitoring": true,
+			"modify_instance_metadata_options": true,
+			"set_disable_api_termination": true
 		},
 		"attach_volume_data": {"volume_id": "vol-1", "device": "/dev/sdf"},
 		"detach_volume_data": {"volume_id": "vol-2", "device": "/dev/sdg", "force": true},
@@ -61,10 +75,64 @@ func TestEC2InstanceCommandJSONContract(t *testing.T) {
 		"iam_profile_association_data": {"instance_profile_arn": "arn:aws:iam::123456789012:instance-profile/example"},
 		"spot_lineage_data": {"spot_instance_request_id": "sir-1"},
 		"instance_tags_data": {"tags": {"Name": "example"}, "tag_keys": ["old"]},
-		"instance_monitoring_data": {"enabled": true}
+		"instance_monitoring_data": {"enabled": true},
+		"instance_metadata_options_data": {
+			"http_tokens": "required",
+			"http_endpoint": "enabled",
+			"http_protocol_ipv6": "disabled",
+			"instance_metadata_tags": "enabled",
+			"http_put_response_hop_limit": 3
+		},
+		"disable_api_termination_data": {"value": false}
 	}`, string(data))
 }
 
+func int64Ptr(v int64) *int64 { return &v }
+
+// An owner that predates the metadata and termination-protection commands
+// sends and receives payloads without the new fields.
+func TestEC2InstanceCommandNewFieldsAreOptionalOnTheWire(t *testing.T) {
+	var n1 EC2InstanceCommand
+	require.NoError(t, json.Unmarshal([]byte(`{
+		"id": "i-0123456789abcdef0",
+		"attributes": {"stop_instance": true},
+		"instance_monitoring_data": {"enabled": true}
+	}`), &n1))
+	require.Equal(t, "i-0123456789abcdef0", n1.ID)
+	require.True(t, n1.Attributes.StopInstance)
+	require.False(t, n1.Attributes.ModifyInstanceMetadataOptions)
+	require.False(t, n1.Attributes.SetDisableApiTermination)
+	require.Nil(t, n1.InstanceMetadataOptionsData)
+	require.Nil(t, n1.DisableApiTerminationData)
+
+	unset, err := json.Marshal(EC2InstanceCommand{ID: "i-1", Attributes: EC2CommandAttributes{StopInstance: true}})
+	require.NoError(t, err)
+	require.NotContains(t, string(unset), "modify_instance_metadata_options")
+	require.NotContains(t, string(unset), "set_disable_api_termination")
+	require.NotContains(t, string(unset), "instance_metadata_options_data")
+	require.NotContains(t, string(unset), "disable_api_termination_data")
+}
+
+func TestInstanceMetadataOptionsDataOmitsUnsetFields(t *testing.T) {
+	data, err := json.Marshal(InstanceMetadataOptionsData{HttpPutResponseHopLimit: int64Ptr(1)})
+	require.NoError(t, err)
+	require.JSONEq(t, `{"http_put_response_hop_limit": 1}`, string(data))
+
+	data, err = json.Marshal(InstanceMetadataOptionsData{})
+	require.NoError(t, err)
+	require.JSONEq(t, `{}`, string(data))
+}
+
+func TestDisableApiTerminationDataKeepsExplicitFalse(t *testing.T) {
+	data, err := json.Marshal(DisableApiTerminationData{Value: false})
+	require.NoError(t, err)
+	require.JSONEq(t, `{"value": false}`, string(data))
+}
+
+func TestHandledHeaderName(t *testing.T) {
+	require.Equal(t, "Spinifex-Ec2-Cmd-Handled", HandledHeader)
+}
+
 func TestInstanceCommandSubjects(t *testing.T) {
 	require.Equal(t, "ec2.cmd.i-0123456789abcdef0", InstanceCommandSubject("i-0123456789abcdef0"))
 	require.Equal(t, "ec2.cmd.*", InstanceCommandSubjectWildcard)

contracts/lambda/v1/README.md

@@ -28,6 +28,9 @@ Its encoded frame is at most the base64 length of 6 MiB plus 64 KiB of metadata
 The credential travels as its own frame of at most 8 KiB, and the reply has the same payload and frame bounds as the request.
 The daemon holds at most two connections at once, from accept to close, and leaves the rest unaccepted in the listen backlog.
 Under TLS 1.3 the guest's handshake can complete before the host refuses its certificate, so the guest treats the connection as accepted only after the host's first message.
+In the same builds, the function owner moves a created function with stored code from `Pending` to `Active` once this node's daemon serves the pinned image and the cluster configuration has exactly one node; a function with environment variables or a code copy over the test-link bound stays `Pending`, and `StateReason` names the unmet condition.
+Release builds never leave `Pending`.
+The Mulga deploy role installs a test-link build with `deploy_lambda_testlink=true`; this profile does not meet Q-199 fail-closed revocation, and its guests take management addresses whose record can expire (mulga-0hsub).
 
 **Bootstrap material:** the executor passes these fw_cfg entries under `opt/spinifex/lambda/`, and the agent reads them as root and unloads `qemu_fw_cfg` before user code starts.
 
@@ -41,6 +44,8 @@ Under TLS 1.3 the guest's handshake can complete before the host refuses its cer
 
 The guest's own interface settings come from `opt/spinifex/netcfg`, the entry other system VMs use.
 The agent unloads `qemu_fw_cfg` even when an entry is missing, and refuses to start if the fw_cfg tree is still readable afterwards.
+On a VMM without fw_cfg (the test-link Firecracker executor), the executor attaches the same entries as a tar archive on the second, writable virtio-blk disk, laid out as `by_name/<entry>/raw`.
+The runtime image's init extracts it to a root-only tmpfs at `/spinifex-cfg`, and the agent reads the entries there, then overwrites the whole disk with zeros and removes the tree; it refuses to start if the tree is still present.
 Until the message format exists, the agent authenticates and holds the connection but accepts no host message and starts no user code.
 
 **Authority:** one environment's attempt only.

spinifex/agents/lambda/hostlink/configdisk.go

@@ -0,0 +1,71 @@
+package hostlink
+
+import (
+	"errors"
+	"fmt"
+	"io"
+	"os"
+
+	lambdav1 "github.com/mulgadc/spinifex/contracts/lambda/v1"
+	"golang.org/x/sys/unix"
+)
+
+// On a VMM without fw_cfg the executor attaches the same entries as a tar
+// archive on a second disk, which the runtime image's init extracts here in
+// the fw_cfg sysfs layout, so the fw_cfg readers work unchanged.
+const (
+	ConfigDiskRoot   = "/spinifex-cfg"
+	ConfigDiskDir    = ConfigDiskRoot + "/by_name/" + lambdav1.FwCfgPrefix
+	ConfigDiskDevice = "/dev/vdb"
+)
+
+// zeroChunk bounds each write while erasing the configuration disk.
+const zeroChunk = 64 << 10
+
+// unmount is a var so tests without privilege can stand in for the tmpfs.
+var unmount = unix.Unmount
+
+// SealConfigDisk overwrites the whole configuration disk with zeros, then
+// removes the extracted tree. The tree stays if the disk cannot be erased, so
+// SealFwCfg's check on root fails closed.
+func SealConfigDisk(device, root string) error {
+	if err := zeroDevice(device); err != nil {
+		return fmt.Errorf("hostlink: erase configuration disk: %w", err)
+	}
+	if err := unmount(root, 0); err != nil && !errors.Is(err, unix.EINVAL) {
+		return fmt.Errorf("hostlink: unmount %s: %w", root, err)
+	}
+	if err := os.RemoveAll(root); err != nil {
+		return fmt.Errorf("hostlink: remove %s: %w", root, err)
+	}
+	return nil
+}
+
+func zeroDevice(device string) error {
+	f, err := os.OpenFile(device, os.O_WRONLY, 0)
+	if err != nil {
+		return err
+	}
+	size, err := f.Seek(0, io.SeekEnd)
+	if err == nil {
+		_, err = f.Seek(0, io.SeekStart)
+	}
+	if err != nil {
+		_ = f.Close()
+		return err
+	}
+	zeros := make([]byte, zeroChunk)
+	for left := size; left > 0; {
+		n := min(left, int64(len(zeros)))
+		if _, err := f.Write(zeros[:n]); err != nil {
+			_ = f.Close()
+			return err
+		}
+		left -= n
+	}
+	if err := f.Sync(); err != nil {
+		_ = f.Close()
+		return err
+	}
+	return f.Close()
+}

spinifex/agents/lambda/hostlink/configdisk_test.go

@@ -0,0 +1,63 @@
+//test:in-package reaches the unexported configDisk and fwCfgTree readers to seal and read a real archive
+
+package hostlink
+
+import (
+	"bytes"
+	"os"
+	"path/filepath"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"golang.org/x/sys/unix"
+)
+
+// configDisk writes a fake disk holding secret and an extracted tree; the
+// stand-in unmount reports the tree is not a mount, as on a plain directory.
+func configDisk(t *testing.T, secret []byte) (device, root string) {
+	t.Helper()
+	dir := t.TempDir()
+	device = filepath.Join(dir, "vdb")
+	require.NoError(t, os.WriteFile(device, append(bytes.Repeat([]byte{0xa5}, 3*zeroChunk+7), secret...), 0o600))
+	root = fwCfgTree(t)
+	require.NoError(t, os.WriteFile(filepath.Join(root, "by_name", "key"), secret, 0o600))
+	old := unmount
+	unmount = func(string, int) error { return unix.EINVAL }
+	t.Cleanup(func() { unmount = old })
+	return device, root
+}
+
+func TestSealConfigDisk(t *testing.T) {
+	secret := []byte("guest-key-material")
+
+	t.Run("erases the disk and removes the tree", func(t *testing.T) {
+		device, root := configDisk(t, secret)
+		before, err := os.Stat(device)
+		require.NoError(t, err)
+		require.NoError(t, SealFwCfg(root, func() error { return SealConfigDisk(device, root) }))
+
+		disk, err := os.ReadFile(device)
+		require.NoError(t, err)
+		assert.Len(t, disk, int(before.Size()))
+		assert.Equal(t, make([]byte, len(disk)), disk, "configuration disk not fully erased")
+		_, err = os.Stat(root)
+		assert.ErrorIs(t, err, os.ErrNotExist)
+	})
+	t.Run("an unwritable disk keeps the tree and fails closed", func(t *testing.T) {
+		device, root := configDisk(t, secret)
+		require.NoError(t, os.Chmod(device, 0o400))
+		err := SealFwCfg(root, func() error { return SealConfigDisk(device, root) })
+		require.Error(t, err)
+		assert.Contains(t, err.Error(), "still readable")
+		_, err = os.Stat(root)
+		assert.NoError(t, err)
+	})
+	t.Run("a failed unmount keeps the tree", func(t *testing.T) {
+		device, root := configDisk(t, secret)
+		unmount = func(string, int) error { return unix.EBUSY }
+		require.Error(t, SealFwCfg(root, func() error { return SealConfigDisk(device, root) }))
+		_, err := os.Stat(root)
+		assert.NoError(t, err)
+	})
+}

spinifex/domains/ec2/accountcleanup/remainder.go

@@ -0,0 +1,155 @@
+package accountcleanup
+
+import (
+	"context"
+	"log/slog"
+	"slices"
+
+	"github.com/aws/aws-sdk-go/aws"
+	"github.com/aws/aws-sdk-go/service/ec2"
+	awserrors "github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
+	awsidentifiers "github.com/mulgadc/spinifex/spinifex/foundation/aws/identifiers"
+)
+
+// ReasonRecordRemains marks a resource whose record the owner still holds.
+const ReasonRecordRemains = "record-remains"
+
+// ComputeRemainder reads what compute the account still holds, changing nothing.
+type ComputeRemainder interface {
+	ComputeRemaining(ctx context.Context, accountID string) (*Outcome, error)
+}
+
+// StorageRemainder reads what volumes and tag residue the account still holds,
+// changing nothing.
+type StorageRemainder interface {
+	StorageRemaining(ctx context.Context, accountID string) (*Outcome, error)
+}
+
+var (
+	_ ComputeRemainder = (*ComputeCleanup)(nil)
+	_ StorageRemainder = (*StorageCleanup)(nil)
+)
+
+func validRemainderAccount(accountID string) error {
+	if accountID == "" {
+		return awserrors.Errorf(awserrors.ErrorMissingParameter, "The request must specify an account ID.")
+	}
+	if !awsidentifiers.IsAccountID(accountID) {
+		return awserrors.Errorf(awserrors.ErrorInvalidParameterValue, "The account ID must be 12 digits.")
+	}
+	return nil
+}
+
+// ComputeRemaining lists every instance the account still holds and every spot
+// request that can still launch. A read that does not complete is an incomplete
+// reason, never a shorter list. The result is a snapshot, not evidence of absence.
+func (c *ComputeCleanup) ComputeRemaining(ctx context.Context, accountID string) (*Outcome, error) {
+	if err := validRemainderAccount(accountID); err != nil {
+		return nil, err
+	}
+	if c == nil || c.spot == nil || c.instances == nil || c.teardown == nil {
+		return nil, awserrors.Errorf(awserrors.ErrorServerInternal, "Compute cleanup is not fully wired.")
+	}
+	res := &Outcome{AccountID: accountID}
+	defer res.finish()
+
+	scan, err := c.spot.ScanAccount(ctx, accountID)
+	if err != nil {
+		slog.WarnContext(ctx, "ComputeRemaining: spot scan failed", "account_id", accountID, "err", err)
+		res.incomplete(ReasonSpotInventoryIncomplete)
+	} else {
+		for _, r := range slices.Concat(scan.Active, scan.Terminal) {
+			if !terminalSpotState(r.State) {
+				res.Items = append(res.Items, Item{ResourceID: r.RequestID, Kind: KindSpotRequest, State: r.State, Disposition: DispositionResidue, Reason: ReasonRecordRemains})
+			}
+		}
+	}
+
+	inventory, err := c.instances.AccountInstances(ctx, accountID)
+	if err != nil || !inventoryUsable(inventory) {
+		slog.WarnContext(ctx, "ComputeRemaining: instance inventory incomplete", "account_id", accountID, "err", err)
+		res.incomplete(ReasonInstanceInventoryIncomplete)
+	} else {
+		for _, reservation := range inventory.Reservations {
+			for _, in := range reservation.Instances {
+				state := ""
+				if in.State != nil {
+					state = aws.StringValue(in.State.Name)
+				}
+				res.Items = append(res.Items, Item{ResourceID: aws.StringValue(in.InstanceId), Kind: KindInstance, State: state, Disposition: DispositionResidue, Reason: ReasonRecordRemains})
+			}
+		}
+	}
+
+	outstanding, err := c.teardown.AccountTeardownOutstanding(accountID)
+	if err != nil {
+		slog.WarnContext(ctx, "ComputeRemaining: teardown unread", "account_id", accountID, "err", err)
+		res.incomplete(ReasonTeardownUnread)
+	} else {
+		for _, o := range outstanding {
+			if !res.hasInstance(o.InstanceID) {
+				res.Items = append(res.Items, Item{ResourceID: o.InstanceID, Kind: KindInstance, State: o.State, Disposition: DispositionResidue, Reason: ReasonTerminating})
+			}
+		}
+	}
+	return res, nil
+}
+
+// StorageRemaining lists every volume record the account still holds, deleting
+// ones included, and the tag entries of resources whose owner holds no record.
+func (c *StorageCleanup) StorageRemaining(ctx context.Context, accountID string) (*Outcome, error) {
+	if err := validRemainderAccount(accountID); err != nil {
+		return nil, err
+	}
+	if c == nil || c.volumes == nil || c.instances == nil || c.teardown == nil || c.spot == nil || c.tags == nil {
+		return nil, awserrors.Errorf(awserrors.ErrorServerInternal, "Storage cleanup is not fully wired.")
+	}
+	res := &Outcome{AccountID: accountID}
+	defer res.finish()
+
+	volumes, err := c.volumes.VolumeInventory(ctx, accountID)
+	if err != nil {
+		slog.WarnContext(ctx, "StorageRemaining: volume inventory incomplete", "account_id", accountID, "err", err)
+		res.incomplete(ReasonVolumeInventoryIncomplete)
+	}
+	for _, v := range volumes {
+		reason := ReasonRecordRemains
+		if v.State == ec2.VolumeStateDeleting {
+			reason = ReasonVolumeDeleting
+		}
+		res.Items = append(res.Items, Item{ResourceID: v.VolumeID, Kind: KindVolume, State: v.State, Disposition: DispositionResidue, Reason: reason})
+	}
+
+	live, err := c.instances.AccountInstances(ctx, accountID)
+	if err != nil || !inventoryUsable(live) {
+		slog.WarnContext(ctx, "StorageRemaining: no complete instance view", "account_id", accountID, "err", err)
+		res.incomplete(ReasonHolderProofIncomplete)
+		return res, nil
+	}
+	outstanding, err := c.teardown.AccountTeardownOutstanding(accountID)
+	if err != nil {
+		slog.WarnContext(ctx, "StorageRemaining: teardown unread", "account_id", accountID, "err", err)
+		res.incomplete(ReasonHolderProofIncomplete)
+		return res, nil
+	}
+
+	tagged, err := c.tags.ScanAccountResourceIDs(ctx, accountID, storageTagPrefixes...)
+	if err != nil {
+		slog.WarnContext(ctx, "StorageRemaining: tag index read failed", "account_id", accountID, "err", err)
+		res.incomplete(ReasonTagIndexUnread)
+		return res, nil
+	}
+	owned, resolved := c.ownedResourceIDs(ctx, accountID, live, outstanding)
+	for _, id := range tagged {
+		switch {
+		case !resolved[tagPrefix(id)]:
+			res.incomplete(ReasonTagOwnerUnresolved)
+			res.Items = append(res.Items, Item{ResourceID: id, Kind: KindTag, Disposition: DispositionResidue, Reason: ReasonOwnerRecordsUnresolved})
+		case owned[id]:
+			// The resource itself is reported; its tags are not residue.
+		default:
+			res.Items = append(res.Items, Item{ResourceID: id, Kind: KindTag, Disposition: DispositionResidue, Reason: ReasonRecordRemains})
+		}
+	}
+	return res, nil
+}

spinifex/domains/ec2/accountcleanup/remainder_test.go

@@ -0,0 +1,97 @@
+package accountcleanup
+
+import (
+	"errors"
+	"testing"
+
+	"github.com/aws/aws-sdk-go/service/ec2"
+	"github.com/mulgadc/spinifex/spinifex/domains/ec2/accountinventory"
+	ec2instance "github.com/mulgadc/spinifex/spinifex/domains/ec2/instance"
+	ec2spotinstance "github.com/mulgadc/spinifex/spinifex/domains/ec2/spotinstance"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func TestComputeRemainingReportsEveryHeldInstanceAndOpenSpot(t *testing.T) {
+	r := newComputeRig()
+	r.spot.scans = []*ec2spotinstance.AccountScan{{
+		Active:   []ec2spotinstance.AccountSpotRequest{{RequestID: "sir-0aaaaaaaaaaaaaaa1", State: ec2.SpotInstanceStateOpen}},
+		Terminal: []ec2spotinstance.AccountSpotRequest{{RequestID: "sir-0aaaaaaaaaaaaaaa2", State: ec2.SpotInstanceStateClosed}},
+	}}
+	r.teardown.outstanding = []ec2instance.OutstandingInstance{{InstanceID: "i-0aaaaaaaaaaaaaa99", State: ec2.InstanceStateNameTerminated}}
+
+	res, err := r.cleanup().ComputeRemaining(t.Context(), testAccount)
+	require.NoError(t, err)
+
+	assert.Equal(t, StatusIncomplete, res.Status)
+	assert.NotNil(t, itemFor(res, runningID))
+	assert.NotNil(t, itemFor(res, stoppedID))
+	assert.NotNil(t, itemFor(res, "i-0aaaaaaaaaaaaaa99"))
+	assert.NotNil(t, itemFor(res, "sir-0aaaaaaaaaaaaaaa1"))
+	assert.Nil(t, itemFor(res, "sir-0aaaaaaaaaaaaaaa2"), "a closed request launches nothing")
+	assert.Empty(t, r.terminator.terminated, "a re-read changes nothing")
+}
+
+func TestComputeRemainingCleanWhenNothingHeld(t *testing.T) {
+	r := newComputeRig()
+	r.inventory.out = listing(nil)
+	res, err := r.cleanup().ComputeRemaining(t.Context(), testAccount)
+	require.NoError(t, err)
+	assert.Equal(t, StatusClean, res.Status)
+}
+
+func TestComputeRemainingUnreadStoreIsIncompleteNeverClean(t *testing.T) {
+	for name, mutate := range map[string]func(*computeRig){
+		"instances": func(r *computeRig) { r.inventory.out, r.inventory.err = nil, errors.New("node silent") },
+		"spot":      func(r *computeRig) { r.spot.scanErr = errors.New("spot unreadable") },
+		"teardown":  func(r *computeRig) { r.teardown.err = errors.New("no responder") },
+	} {
+		t.Run(name, func(t *testing.T) {
+			r := newComputeRig()
+			r.inventory.out = listing(nil)
+			mutate(r)
+			res, err := r.cleanup().ComputeRemaining(t.Context(), testAccount)
+			require.NoError(t, err)
+			assert.Equal(t, StatusIncomplete, res.Status)
+			assert.NotEmpty(t, res.IncompleteReasons)
+		})
+	}
+}
+
+func TestStorageRemainingReportsVolumeRecordsAndTagResidue(t *testing.T) {
+	r := newStorageRig()
+	r.volumes.vols = []accountinventory.VolumeItem{
+		volume(freeVol, ec2.VolumeStateAvailable, ""),
+		volume("vol-0aaaaaaaaaaaaaaa2", ec2.VolumeStateDeleting, ""),
+	}
+	r.tags.ids = []string{"i-0aaaaaaaaaaaaaaaf", freeVol}
+
+	res, err := r.cleanup().StorageRemaining(t.Context(), testAccount)
+	require.NoError(t, err)
+
+	assert.Equal(t, StatusIncomplete, res.Status)
+	assert.Equal(t, ReasonRecordRemains, itemFor(res, freeVol).Reason)
+	assert.Equal(t, ReasonVolumeDeleting, itemFor(res, "vol-0aaaaaaaaaaaaaaa2").Reason)
+	assert.Equal(t, KindTag, itemFor(res, "i-0aaaaaaaaaaaaaaaf").Kind, "tags of a resource with no record are residue")
+	assert.Empty(t, r.deleter.deleted, "a re-read deletes nothing")
+	assert.Empty(t, r.tags.deleted, "a re-read clears no tag")
+}
+
+func TestStorageRemainingUnreadStoreIsIncompleteNeverClean(t *testing.T) {
+	for name, mutate := range map[string]func(*storageRig){
+		"volumes":   func(r *storageRig) { r.volumes.failAt = -1 },
+		"instances": func(r *storageRig) { r.instances.out, r.instances.err = nil, errors.New("node silent") },
+		"teardown":  func(r *storageRig) { r.teardown.err = errors.New("no responder") },
+		"tags":      func(r *storageRig) { r.tags.scanErr = errors.New("tag index unreadable") },
+	} {
+		t.Run(name, func(t *testing.T) {
+			r := newStorageRig()
+			r.volumes.vols = nil
+			mutate(r)
+			res, err := r.cleanup().StorageRemaining(t.Context(), testAccount)
+			require.NoError(t, err)
+			assert.Equal(t, StatusIncomplete, res.Status)
+			assert.NotEmpty(t, res.IncompleteReasons)
+		})
+	}
+}

spinifex/domains/ec2/awsapi/instance/AccountTeardownOutstanding.go

@@ -0,0 +1,28 @@
+package instance
+
+import (
+	"context"
+	"fmt"
+	"time"
+
+	ec2instance "github.com/mulgadc/spinifex/spinifex/domains/ec2/instance"
+	natsmsg "github.com/mulgadc/spinifex/spinifex/foundation/messaging/nats"
+	"github.com/nats-io/nats.go"
+)
+
+// AccountTeardownOutstandingSubject serves the instances of an account whose
+// teardown is not complete, from the shared record store.
+const AccountTeardownOutstandingSubject = "ec2.AccountTeardownOutstanding"
+
+var accountTeardownOutstandingTimeout = 15 * time.Second
+
+// AccountTeardownOutstanding reads the account's unfinished instance teardowns.
+// No responder, a timeout or an error reply is an error, never an empty list.
+func AccountTeardownOutstanding(ctx context.Context, natsConn *nats.Conn, accountID string) ([]ec2instance.OutstandingInstance, error) {
+	out, err := natsmsg.NATSRequest[ec2instance.TeardownOutstandingOutput](ctx, natsConn, AccountTeardownOutstandingSubject,
+		&ec2instance.TeardownOutstandingInput{}, accountTeardownOutstandingTimeout, accountID)
+	if err != nil {
+		return nil, fmt.Errorf("read outstanding teardown: %w", err)
+	}
+	return out.Instances, nil
+}

spinifex/domains/ec2/awsapi/instance/AccountTeardownOutstanding_test.go

@@ -0,0 +1,88 @@
+package instance
+
+import (
+	"context"
+	"encoding/json"
+	"testing"
+	"time"
+
+	ec2instance "github.com/mulgadc/spinifex/spinifex/domains/ec2/instance"
+	awserrors "github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
+	"github.com/nats-io/nats.go"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func serveTeardownOutstanding(t *testing.T, nc *nats.Conn, reply []byte) {
+	t.Helper()
+	_, err := nc.Subscribe(AccountTeardownOutstandingSubject, func(msg *nats.Msg) { _ = msg.Respond(reply) })
+	require.NoError(t, err)
+	require.NoError(t, nc.Flush())
+}
+
+func TestAccountTeardownOutstanding_ReturnsTheServedInstances(t *testing.T) {
+	t.Parallel()
+	_, nc := startTestNATSServer(t)
+	reply, err := json.Marshal(ec2instance.TeardownOutstandingOutput{Instances: []ec2instance.OutstandingInstance{{InstanceID: "i-1"}}})
+	require.NoError(t, err)
+	serveTeardownOutstanding(t, nc, reply)
+
+	got, err := AccountTeardownOutstanding(context.Background(), nc, "123456789012")
+
+	require.NoError(t, err)
+	require.Len(t, got, 1)
+	assert.Equal(t, "i-1", got[0].InstanceID)
+}
+
+// A read that cannot be answered must never look like an empty list, because an
+// empty list lets the storage cleanup believe no QEMU still holds a volume.
+func TestAccountTeardownOutstanding_UnansweredReadIsAnError(t *testing.T) {
+	t.Parallel()
+
+	t.Run("no responder", func(t *testing.T) {
+		t.Parallel()
+		_, nc := startTestNATSServer(t)
+
+		got, err := AccountTeardownOutstanding(context.Background(), nc, "123456789012")
+
+		require.Error(t, err)
+		assert.Nil(t, got)
+		assert.ErrorIs(t, err, nats.ErrNoResponders)
+	})
+	t.Run("error reply", func(t *testing.T) {
+		t.Parallel()
+		_, nc := startTestNATSServer(t)
+		serveTeardownOutstanding(t, nc, awserrors.GenerateErrorPayload(awserrors.ErrorServerInternal))
+
+		got, err := AccountTeardownOutstanding(context.Background(), nc, "123456789012")
+
+		require.Error(t, err)
+		assert.Nil(t, got)
+	})
+	t.Run("unreadable reply", func(t *testing.T) {
+		t.Parallel()
+		_, nc := startTestNATSServer(t)
+		serveTeardownOutstanding(t, nc, []byte("not json"))
+
+		got, err := AccountTeardownOutstanding(context.Background(), nc, "123456789012")
+
+		require.Error(t, err)
+		assert.Nil(t, got)
+	})
+}
+
+// Not parallel: it shortens the package timeout.
+func TestAccountTeardownOutstanding_TimeoutIsAnError(t *testing.T) {
+	_, nc := startTestNATSServer(t)
+	_, err := nc.Subscribe(AccountTeardownOutstandingSubject, func(*nats.Msg) {})
+	require.NoError(t, err)
+	require.NoError(t, nc.Flush())
+	previous := accountTeardownOutstandingTimeout
+	accountTeardownOutstandingTimeout = 200 * time.Millisecond
+	t.Cleanup(func() { accountTeardownOutstandingTimeout = previous })
+
+	got, err := AccountTeardownOutstanding(context.Background(), nc, "123456789012")
+
+	require.Error(t, err)
+	assert.Nil(t, got)
+}

spinifex/domains/ec2/awsapi/instance/ModifyInstanceAttribute.go

@@ -11,6 +11,7 @@ import (
 	"time"
 
 	"github.com/aws/aws-sdk-go/service/ec2"
+	"github.com/mulgadc/spinifex/contracts/ec2/v1"
 	"github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
 	"github.com/nats-io/nats.go"
 )
@@ -65,6 +66,20 @@ func ModifyInstanceAttribute(ctx context.Context, input *ec2.ModifyInstanceAttri
 
 	slog.InfoContext(ctx, "ModifyInstanceAttribute: Processing request", "instance_id", *input.InstanceId)
 
+	// Termination protection on a running instance lives on its owner, so it
+	// goes there first. A nil value is not a value and keeps the queue path.
+	if input.DisableApiTermination != nil && input.DisableApiTermination.Value != nil {
+		cmd := ec2v1.EC2InstanceCommand{
+			ID:                        *input.InstanceId,
+			Attributes:                ec2v1.EC2CommandAttributes{SetDisableApiTermination: true},
+			DisableApiTerminationData: &ec2v1.DisableApiTerminationData{Value: *input.DisableApiTermination.Value},
+		}
+		if _, err := routeInstanceMutation[ec2.ModifyInstanceAttributeOutput](ctx, natsConn, *input.InstanceId, accountID, cmd, "ec2.ModifyInstanceAttribute", input); err != nil {
+			return ec2.ModifyInstanceAttributeOutput{}, err
+		}
+		return ec2.ModifyInstanceAttributeOutput{}, nil
+	}
+
 	jsonData, err := json.Marshal(input)
 	if err != nil {
 		slog.ErrorContext(ctx, "ModifyInstanceAttribute: Failed to marshal request", "instance_id", *input.InstanceId, "err", err)

spinifex/domains/ec2/awsapi/instance/ModifyInstanceAttribute_test.go

@@ -3,10 +3,13 @@ package instance
 import (
 	"context"
 	"encoding/json"
+	"sync"
+	"sync/atomic"
 	"testing"
 
 	"github.com/aws/aws-sdk-go/aws"
 	"github.com/aws/aws-sdk-go/service/ec2"
+	"github.com/mulgadc/spinifex/contracts/ec2/v1"
 	"github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
 	"github.com/nats-io/nats.go"
 	"github.com/stretchr/testify/assert"
@@ -226,3 +229,53 @@ func TestModifyInstanceAttribute_ValidationFailure(t *testing.T) {
 	require.Error(t, err)
 	assert.Equal(t, awserrors.ErrorInvalidParameterValue, err.Error())
 }
+
+func TestModifyInstanceAttribute_DisableApiTerminationRouting(t *testing.T) {
+	const id = "i-0123456789abcdef0"
+	_, nc := startTestNATSServer(t)
+
+	var gotValues []bool
+	var mu sync.Mutex
+	_, err := nc.Subscribe(ec2v1.InstanceCommandSubject(id), func(msg *nats.Msg) {
+		var cmd ec2v1.EC2InstanceCommand
+		if json.Unmarshal(msg.Data, &cmd) == nil && cmd.Attributes.SetDisableApiTermination && cmd.DisableApiTerminationData != nil {
+			mu.Lock()
+			gotValues = append(gotValues, cmd.DisableApiTerminationData.Value)
+			mu.Unlock()
+		}
+		reply := nats.NewMsg(msg.Reply)
+		reply.Data = []byte(`{}`)
+		reply.Header.Set(ec2v1.HandledHeader, "1")
+		_ = msg.RespondMsg(reply)
+	})
+	require.NoError(t, err)
+	var queue atomic.Int32
+	_, err = nc.Subscribe("ec2.ModifyInstanceAttribute", func(msg *nats.Msg) {
+		queue.Add(1)
+		_ = msg.Respond([]byte(`{}`))
+	})
+	require.NoError(t, err)
+
+	for _, v := range []bool{true, false} {
+		_, err := ModifyInstanceAttribute(context.Background(), &ec2.ModifyInstanceAttributeInput{
+			InstanceId:            aws.String(id),
+			DisableApiTermination: &ec2.AttributeBooleanValue{Value: aws.Bool(v)},
+		}, nc, "acct")
+		require.NoError(t, err)
+	}
+	mu.Lock()
+	assert.Equal(t, []bool{true, false}, gotValues, "explicit true and false both reach the owner")
+	mu.Unlock()
+	assert.Zero(t, queue.Load())
+
+	// A nil value is not a value: queue path, no owner command.
+	_, err = ModifyInstanceAttribute(context.Background(), &ec2.ModifyInstanceAttributeInput{
+		InstanceId:            aws.String(id),
+		DisableApiTermination: &ec2.AttributeBooleanValue{},
+	}, nc, "acct")
+	require.NoError(t, err)
+	assert.Equal(t, int32(1), queue.Load())
+	mu.Lock()
+	assert.Len(t, gotValues, 2, "no owner command for a nil value")
+	mu.Unlock()
+}

spinifex/domains/ec2/awsapi/instance/ModifyInstanceMetadataOptions.go

@@ -3,11 +3,11 @@ package instance
 import (
 	"context"
 	"errors"
-	"github.com/mulgadc/spinifex/spinifex/foundation/messaging/nats"
 	"strings"
-	"time"
 
+	"github.com/aws/aws-sdk-go/aws"
 	"github.com/aws/aws-sdk-go/service/ec2"
+	"github.com/mulgadc/spinifex/contracts/ec2/v1"
 	"github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
 	"github.com/nats-io/nats.go"
 )
@@ -22,5 +22,16 @@ func ModifyInstanceMetadataOptions(ctx context.Context, input *ec2.ModifyInstanc
 	if !strings.HasPrefix(*input.InstanceId, "i-") {
 		return nil, errors.New(awserrors.ErrorInvalidInstanceIDMalformed)
 	}
-	return natsmsg.NATSRequest[ec2.ModifyInstanceMetadataOptionsOutput](ctx, natsConn, "ec2.ModifyInstanceMetadataOptions", input, 30*time.Second, accountID)
+	cmd := ec2v1.EC2InstanceCommand{
+		ID:         *input.InstanceId,
+		Attributes: ec2v1.EC2CommandAttributes{ModifyInstanceMetadataOptions: true},
+		InstanceMetadataOptionsData: &ec2v1.InstanceMetadataOptionsData{
+			HttpTokens:              aws.StringValue(input.HttpTokens),
+			HttpEndpoint:            aws.StringValue(input.HttpEndpoint),
+			HttpProtocolIpv6:        aws.StringValue(input.HttpProtocolIpv6),
+			InstanceMetadataTags:    aws.StringValue(input.InstanceMetadataTags),
+			HttpPutResponseHopLimit: input.HttpPutResponseHopLimit,
+		},
+	}
+	return routeInstanceMutation[ec2.ModifyInstanceMetadataOptionsOutput](ctx, natsConn, *input.InstanceId, accountID, cmd, "ec2.ModifyInstanceMetadataOptions", input)
 }

spinifex/domains/ec2/awsapi/instance/ModifyInstanceMetadataOptions_test.go

@@ -2,11 +2,16 @@ package instance
 
 import (
 	"context"
+	"encoding/json"
+	"sync/atomic"
 	"testing"
+	"time"
 
 	"github.com/aws/aws-sdk-go/aws"
 	"github.com/aws/aws-sdk-go/service/ec2"
+	"github.com/mulgadc/spinifex/contracts/ec2/v1"
 	"github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
+	"github.com/nats-io/nats.go"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 )
@@ -30,3 +35,179 @@ func TestModifyInstanceMetadataOptions_RejectsMalformedID(t *testing.T) {
 		})
 	}
 }
+
+const metaTestID = "i-0123456789abcdef0"
+
+func metaInput(hop int64) *ec2.ModifyInstanceMetadataOptionsInput {
+	return &ec2.ModifyInstanceMetadataOptionsInput{
+		InstanceId:              aws.String(metaTestID),
+		HttpPutResponseHopLimit: aws.Int64(hop),
+	}
+}
+
+// ownerReply answers one ec2.cmd request, optionally carrying the handled marker.
+func ownerReply(t *testing.T, nc *nats.Conn, handled bool, body []byte, calls *atomic.Int32) {
+	t.Helper()
+	_, err := nc.Subscribe(ec2v1.InstanceCommandSubject(metaTestID), func(msg *nats.Msg) {
+		calls.Add(1)
+		reply := nats.NewMsg(msg.Reply)
+		reply.Data = body
+		if handled {
+			reply.Header.Set(ec2v1.HandledHeader, "1")
+		}
+		_ = msg.RespondMsg(reply)
+	})
+	require.NoError(t, err)
+}
+
+func queueReply(t *testing.T, nc *nats.Conn, body []byte, calls *atomic.Int32) {
+	t.Helper()
+	_, err := nc.QueueSubscribe("ec2.ModifyInstanceMetadataOptions", "spinifex-workers", func(msg *nats.Msg) {
+		calls.Add(1)
+		_ = msg.Respond(body)
+	})
+	require.NoError(t, err)
+}
+
+func metaOutputJSON(t *testing.T, hop int64) []byte {
+	t.Helper()
+	b, err := json.Marshal(&ec2.ModifyInstanceMetadataOptionsOutput{
+		InstanceId:              aws.String(metaTestID),
+		InstanceMetadataOptions: &ec2.InstanceMetadataOptionsResponse{HttpPutResponseHopLimit: aws.Int64(hop)},
+	})
+	require.NoError(t, err)
+	return b
+}
+
+func TestModifyInstanceMetadataOptions_OwnerAnswerReturnedAsIs(t *testing.T) {
+	_, nc := startTestNATSServer(t)
+	var owner, queue atomic.Int32
+	ownerReply(t, nc, true, metaOutputJSON(t, 3), &owner)
+	queueReply(t, nc, []byte(`{}`), &queue)
+
+	out, err := ModifyInstanceMetadataOptions(context.Background(), metaInput(3), nc, "acct")
+	require.NoError(t, err)
+	assert.Equal(t, int64(3), aws.Int64Value(out.InstanceMetadataOptions.HttpPutResponseHopLimit))
+	assert.Equal(t, int32(1), owner.Load())
+	assert.Zero(t, queue.Load())
+}
+
+func TestModifyInstanceMetadataOptions_OwnerErrorWithHeaderIsNotRetried(t *testing.T) {
+	_, nc := startTestNATSServer(t)
+	var owner, queue atomic.Int32
+	ownerReply(t, nc, true, awserrors.GenerateErrorPayload(awserrors.ErrorServerInternal), &owner)
+	queueReply(t, nc, metaOutputJSON(t, 3), &queue)
+
+	_, err := ModifyInstanceMetadataOptions(context.Background(), metaInput(3), nc, "acct")
+	require.Error(t, err)
+	assert.Equal(t, awserrors.ErrorServerInternal, err.Error())
+	assert.Zero(t, queue.Load())
+}
+
+func TestModifyInstanceMetadataOptions_OwnerValidationErrorRelayed(t *testing.T) {
+	_, nc := startTestNATSServer(t)
+	var owner, queue atomic.Int32
+	ownerReply(t, nc, true, awserrors.GenerateErrorPayload(awserrors.ErrorUnsupportedOperation), &owner)
+	queueReply(t, nc, metaOutputJSON(t, 3), &queue)
+
+	_, err := ModifyInstanceMetadataOptions(context.Background(), metaInput(3), nc, "acct")
+	require.Error(t, err)
+	assert.Equal(t, awserrors.ErrorUnsupportedOperation, err.Error())
+	assert.Zero(t, queue.Load())
+}
+
+func TestModifyInstanceMetadataOptions_HeaderlessOtherErrorReturnedAsIs(t *testing.T) {
+	_, nc := startTestNATSServer(t)
+	var owner, queue atomic.Int32
+	ownerReply(t, nc, false, awserrors.GenerateErrorPayload(awserrors.ErrorInvalidInstanceIDNotFound), &owner)
+	queueReply(t, nc, metaOutputJSON(t, 3), &queue)
+
+	_, err := ModifyInstanceMetadataOptions(context.Background(), metaInput(3), nc, "acct")
+	require.Error(t, err)
+	assert.Equal(t, awserrors.ErrorInvalidInstanceIDNotFound, err.Error())
+	assert.Zero(t, queue.Load())
+}
+
+func TestS9_N1OwnerRejectionFallsBackOnceToQueueGroup(t *testing.T) {
+	_, nc := startTestNATSServer(t)
+	var owner, queue atomic.Int32
+	ownerReply(t, nc, false, awserrors.GenerateErrorPayload(awserrors.ErrorServerInternal), &owner)
+	queueReply(t, nc, metaOutputJSON(t, 3), &queue)
+
+	out, err := ModifyInstanceMetadataOptions(context.Background(), metaInput(3), nc, "acct")
+	require.NoError(t, err)
+	assert.Equal(t, int64(3), aws.Int64Value(out.InstanceMetadataOptions.HttpPutResponseHopLimit))
+	assert.Equal(t, int32(1), owner.Load(), "exactly one owner request")
+	assert.Equal(t, int32(1), queue.Load(), "exactly one queue-group request")
+}
+
+func TestModifyInstanceMetadataOptions_NoOwnerUsesQueueGroupAfterThreeAttempts(t *testing.T) {
+	_, nc := startTestNATSServer(t)
+	var sleeps atomic.Int32
+	prev := cmdRetrySleep
+	cmdRetrySleep = func(time.Duration) { sleeps.Add(1) }
+	t.Cleanup(func() { cmdRetrySleep = prev })
+
+	var queue atomic.Int32
+	queueReply(t, nc, metaOutputJSON(t, 2), &queue)
+
+	out, err := ModifyInstanceMetadataOptions(context.Background(), metaInput(2), nc, "acct")
+	require.NoError(t, err)
+	assert.Equal(t, int64(2), aws.Int64Value(out.InstanceMetadataOptions.HttpPutResponseHopLimit))
+	assert.Equal(t, int32(2), sleeps.Load(), "three attempts means two backoffs")
+	assert.Equal(t, int32(1), queue.Load())
+}
+
+func TestModifyInstanceMetadataOptions_AbsentIsNotFound(t *testing.T) {
+	_, nc := startTestNATSServer(t)
+	var queue atomic.Int32
+	queueReply(t, nc, awserrors.GenerateErrorPayload(awserrors.ErrorInvalidInstanceIDNotFound), &queue)
+
+	_, err := ModifyInstanceMetadataOptions(context.Background(), metaInput(2), nc, "acct")
+	require.Error(t, err)
+	assert.Equal(t, awserrors.ErrorInvalidInstanceIDNotFound, err.Error())
+}
+
+func TestModifyInstanceMetadataOptions_OwnerTimeoutIsServerInternalWithoutFallback(t *testing.T) {
+	_, nc := startTestNATSServer(t)
+	prev := instanceCmdTimeout
+	instanceCmdTimeout = 50 * time.Millisecond
+	t.Cleanup(func() { instanceCmdTimeout = prev })
+
+	// The owner applies the change, then withholds its reply.
+	var applied, queue atomic.Int32
+	_, err := nc.Subscribe(ec2v1.InstanceCommandSubject(metaTestID), func(*nats.Msg) { applied.Add(1) })
+	require.NoError(t, err)
+	queueReply(t, nc, metaOutputJSON(t, 3), &queue)
+
+	_, err = ModifyInstanceMetadataOptions(context.Background(), metaInput(3), nc, "acct")
+	require.Error(t, err)
+	assert.Equal(t, awserrors.ErrorServerInternal, err.Error())
+	assert.Equal(t, int32(1), applied.Load(), "the owner saw the command, so the outcome is unknown")
+	assert.Zero(t, queue.Load(), "a timeout never falls back to the queue group")
+}
+
+func TestModifyInstanceMetadataOptions_TimeoutAfterApplyIsUnknownNotRolledBack(t *testing.T) {
+	_, nc := startTestNATSServer(t)
+	prev := instanceCmdTimeout
+	instanceCmdTimeout = 50 * time.Millisecond
+	t.Cleanup(func() { instanceCmdTimeout = prev })
+
+	// The owner persists the change, then the reply is lost.
+	var persisted atomic.Int64
+	_, err := nc.Subscribe(ec2v1.InstanceCommandSubject(metaTestID), func(msg *nats.Msg) {
+		var cmd ec2v1.EC2InstanceCommand
+		if json.Unmarshal(msg.Data, &cmd) == nil && cmd.InstanceMetadataOptionsData != nil {
+			persisted.Store(aws.Int64Value(cmd.InstanceMetadataOptionsData.HttpPutResponseHopLimit))
+		}
+	})
+	require.NoError(t, err)
+	var queue atomic.Int32
+	queueReply(t, nc, metaOutputJSON(t, 1), &queue)
+
+	_, err = ModifyInstanceMetadataOptions(context.Background(), metaInput(3), nc, "acct")
+	require.Error(t, err)
+	assert.Equal(t, awserrors.ErrorServerInternal, err.Error())
+	assert.Zero(t, queue.Load(), "no fallback that could race a live owner")
+	assert.Equal(t, int64(3), persisted.Load(), "the gateway error does not undo what the owner stored")
+}

spinifex/domains/ec2/awsapi/instance/instanceid.go

@@ -99,3 +99,46 @@ func classifyUnownedInstance(ctx context.Context, natsConn *nats.Conn, instanceI
 	}
 	return bucketStateAbsent, allOK
 }
+
+// routeInstanceMutation sends cmd to the instance's owner first and falls back
+// to the queue-group subject only when that is safe: no daemon runs the
+// instance (ErrNoResponders after every retry), or the owner is N-1 and
+// rejected the command without writing (ServerInternal with no HandledHeader).
+// A timeout or other transport error is an unknown outcome and never falls back.
+func routeInstanceMutation[Out any](ctx context.Context, natsConn *nats.Conn, instanceID, accountID string, cmd ec2v1.EC2InstanceCommand, queueSubject string, queueInput any) (*Out, error) {
+	queue := func() (*Out, error) {
+		return natsmsg.NATSRequest[Out](ctx, natsConn, queueSubject, queueInput, 30*time.Second, accountID)
+	}
+
+	data, err := json.Marshal(cmd)
+	if err != nil {
+		return nil, errors.New(awserrors.ErrorServerInternal)
+	}
+	msg, err := requestInstanceCmd(ctx, natsConn, instanceID, accountID, data)
+	if err != nil {
+		if errors.Is(err, nats.ErrNoResponders) {
+			return queue()
+		}
+		slog.ErrorContext(ctx, "Owner request failed, outcome unknown", "instance_id", instanceID, "subject", queueSubject, "err", err)
+		return nil, errors.New(awserrors.ErrorServerInternal)
+	}
+
+	responseError, parseErr := awserrors.ValidateErrorPayload(msg.Data)
+	if parseErr != nil {
+		handled := msg.Header.Get(ec2v1.HandledHeader) == "1"
+		if !handled && *responseError.Code == awserrors.ErrorServerInternal {
+			slog.WarnContext(ctx, "Owner did not recognise the command, using the queue group", "instance_id", instanceID, "subject", queueSubject)
+			return queue()
+		}
+		if responseError.Message != nil && *responseError.Message != "" {
+			return nil, awserrors.Errorf(*responseError.Code, "%s", *responseError.Message)
+		}
+		return nil, errors.New(*responseError.Code)
+	}
+
+	var out Out
+	if err := json.Unmarshal(msg.Data, &out); err != nil {
+		return nil, errors.New(awserrors.ErrorServerInternal)
+	}
+	return &out, nil
+}

spinifex/domains/ec2/instance/teardown_outstanding.go

@@ -1,12 +1,33 @@
 package instance
 
+import "context"
+
 // OutstandingInstance is an instance of an account that is not yet fully torn
 // down, with the state its record carries.
 type OutstandingInstance struct {
-	InstanceID string
+	InstanceID string `json:"instance_id"`
 	// State is the record's lifecycle state, such as running or terminated.
-	State string
+	State string `json:"state"`
 	// Teardown is the per-dependent progress (done, pending or failed) the
 	// record carries. It is empty when no dependent has been recorded.
-	Teardown map[string]string
+	Teardown map[string]string `json:"teardown,omitempty"`
+}
+
+// TeardownOutstandingInput is the empty request of the outstanding-teardown
+// read; the account comes from the message header.
+type TeardownOutstandingInput struct{}
+
+// TeardownOutstandingOutput carries every instance not yet fully torn down.
+type TeardownOutstandingOutput struct {
+	Instances []OutstandingInstance `json:"instances"`
+}
+
+// AccountTeardownOutstandingRPC serves AccountTeardownOutstanding over the
+// message bus, so a process without the instance record store can read it.
+func (s *InstanceServiceImpl) AccountTeardownOutstandingRPC(_ context.Context, _ *TeardownOutstandingInput, accountID string) (*TeardownOutstandingOutput, error) {
+	outstanding, err := s.AccountTeardownOutstanding(accountID)
+	if err != nil {
+		return nil, err
+	}
+	return &TeardownOutstandingOutput{Instances: outstanding}, nil
 }

spinifex/domains/lambda/function/activate.go

@@ -0,0 +1,155 @@
+package function
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"log/slog"
+	"strings"
+	"time"
+
+	"github.com/nats-io/nats.go/jetstream"
+)
+
+// Readiness reports why a function whose code is stored cannot run yet, or nil
+// when it can. Its error text becomes the function's StateReason.
+type Readiness func(ctx context.Context, rec Record) error
+
+// SetReadiness enables activation. Without it no function ever leaves Pending.
+func (s *MetaService) SetReadiness(r Readiness) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	s.ready = r
+	if s.nudge == nil {
+		s.nudge = make(chan struct{}, 1)
+	}
+}
+
+func (s *MetaService) readiness() (Readiness, chan struct{}) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	return s.ready, s.nudge
+}
+
+// wake asks the activation loop for an early pass; it never blocks.
+func (s *MetaService) wake() {
+	if _, nudge := s.readiness(); nudge != nil {
+		select {
+		case nudge <- struct{}{}:
+		default:
+		}
+	}
+}
+
+// RunActivation activates pending functions until ctx ends: once at start, on
+// every code-stored completion, and every interval, so a crash between storing
+// the code and activating it is converged by the next pass.
+func (s *MetaService) RunActivation(ctx context.Context, interval time.Duration) {
+	ready, nudge := s.readiness()
+	if ready == nil {
+		return
+	}
+	t := time.NewTicker(interval)
+	defer t.Stop()
+	for {
+		if err := s.ActivatePending(ctx); err != nil && ctx.Err() == nil {
+			slog.Warn("Lambda: activation pass incomplete; retrying", "err", err)
+		}
+		select {
+		case <-ctx.Done():
+			return
+		case <-nudge:
+		case <-t.C:
+		}
+	}
+}
+
+// ActivatePending makes one pass over every account's functions, moving each
+// Pending function with stored code to Active when readiness allows it.
+func (s *MetaService) ActivatePending(ctx context.Context) error {
+	ready, _ := s.readiness()
+	if ready == nil {
+		return nil
+	}
+	var accounts []string
+	names := s.js.KeyValueStoreNames(ctx)
+	for name := range names.Name() {
+		if id, ok := strings.CutPrefix(name, KVBucketAccountPrefix); ok {
+			accounts = append(accounts, id)
+		}
+	}
+	if err := names.Error(); err != nil {
+		return fmt.Errorf("lambda: list account buckets: %w", err)
+	}
+	var errs []error
+	for _, accountID := range accounts {
+		if err := s.activateAccount(ctx, accountID, ready); err != nil {
+			errs = append(errs, err)
+		}
+	}
+	return errors.Join(errs...)
+}
+
+func (s *MetaService) activateAccount(ctx context.Context, accountID string, ready Readiness) error {
+	kv, err := s.bucket(ctx, accountID)
+	if err != nil {
+		return err
+	}
+	keys, err := kv.ListKeysFiltered(ctx, kvFunctionKeyPrefix+">")
+	if errors.Is(err, jetstream.ErrNoKeysFound) {
+		return nil
+	}
+	if err != nil {
+		return fmt.Errorf("lambda: list functions of account %s: %w", accountID, err)
+	}
+	var errs []error
+	for key := range keys.Keys() {
+		if err := s.activateOne(ctx, kv, key, ready); err != nil {
+			errs = append(errs, err)
+		}
+	}
+	return errors.Join(errs...)
+}
+
+// activateOne writes the readiness outcome by compare-and-swap at the revision
+// it read, so a concurrent update or delete wins and is never overwritten.
+func (s *MetaService) activateOne(ctx context.Context, kv jetstream.KeyValue, key string, ready Readiness) error {
+	entry, err := kv.Get(ctx, key)
+	if errors.Is(err, jetstream.ErrKeyNotFound) || errors.Is(err, jetstream.ErrKeyDeleted) {
+		return nil
+	}
+	if err != nil {
+		return fmt.Errorf("lambda: read %s: %w", key, err)
+	}
+	rec, err := decode(entry.Value())
+	if err != nil {
+		return err
+	}
+	if rec.State != StatePending || rec.Code.State != CodeStored {
+		return nil
+	}
+	if notReady := ready(ctx, rec); notReady != nil {
+		if rec.StateReason == notReady.Error() {
+			return nil
+		}
+		rec.StateReasonCode, rec.StateReason = ReasonCodeCreating, notReady.Error()
+	} else {
+		rec.State, rec.StateReasonCode, rec.StateReason = StateActive, "", ""
+	}
+	data, err := json.Marshal(rec)
+	if err != nil {
+		return err
+	}
+	if _, err := kv.Update(ctx, key, data, entry.Revision()); err != nil {
+		if errors.Is(err, jetstream.ErrKeyExists) {
+			return nil
+		}
+		return fmt.Errorf("lambda: write readiness of %s: %w", key, err)
+	}
+	if rec.State == StateActive {
+		slog.Info("Lambda: function active", "account_id", rec.AccountID, "region", rec.Region,
+			"function_name", rec.FunctionName, "function_id", rec.FunctionID)
+	}
+	return nil
+}

spinifex/domains/lambda/function/activate_test.go

@@ -0,0 +1,120 @@
+package function_test
+
+import (
+	"context"
+	"errors"
+	"testing"
+	"time"
+
+	"github.com/mulgadc/spinifex/internal/testkit"
+	"github.com/mulgadc/spinifex/spinifex/domains/lambda/function"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+const (
+	actAccount = "000000000001"
+	actRegion  = "ap-southeast-2"
+	actDigest  = "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"
+)
+
+func newOwner(t *testing.T) *function.MetaService {
+	t.Helper()
+	_, nc, _ := testutil.StartTestJetStream(t)
+	return function.NewMetaService(testutil.NewJetStream(t, nc))
+}
+
+func create(t *testing.T, svc *function.MetaService, name string) uint64 {
+	t.Helper()
+	resp, err := svc.Create(t.Context(), &function.CreateRequest{Region: actRegion, Record: function.Record{
+		FunctionID: "id-" + name, FunctionName: name, Role: "arn:aws:iam::000000000001:role/r",
+		Runtime: "provided.al2023", Handler: "bootstrap", Timeout: 3, MemorySize: 128,
+		Code: function.Code{RevisionID: "rev-" + name},
+	}}, actAccount)
+	require.NoError(t, err)
+	return resp.Revision
+}
+
+func get(t *testing.T, svc *function.MetaService, name string) *function.GetResponse {
+	t.Helper()
+	resp, err := svc.Get(t.Context(), &function.GetRequest{Region: actRegion, FunctionName: name}, actAccount)
+	require.NoError(t, err)
+	return resp
+}
+
+func storeCode(t *testing.T, svc *function.MetaService, name string, rev uint64) {
+	t.Helper()
+	_, err := svc.CodeStored(t.Context(), &function.CodeStoredRequest{Region: actRegion, FunctionName: name, Revision: rev, SHA256: actDigest, Size: 9}, actAccount)
+	require.NoError(t, err)
+}
+
+func TestActivation_WithoutReadinessNothingLeavesPending(t *testing.T) {
+	svc := newOwner(t)
+	storeCode(t, svc, "hello", create(t, svc, "hello"))
+
+	require.NoError(t, svc.ActivatePending(t.Context()))
+	ctx, cancel := context.WithCancel(t.Context())
+	cancel()
+	svc.RunActivation(ctx, time.Millisecond)
+
+	got := get(t, svc, "hello").Record
+	assert.Equal(t, function.StatePending, got.State)
+	assert.Equal(t, function.ReasonCreating, got.StateReason)
+}
+
+func TestActivation_UnmetConditionStaysPendingWithItsReason(t *testing.T) {
+	svc := newOwner(t)
+	storeCode(t, svc, "hello", create(t, svc, "hello"))
+	svc.SetReadiness(func(context.Context, function.Record) error { return errors.New("The node is not serving.") })
+
+	require.NoError(t, svc.ActivatePending(t.Context()))
+	first := get(t, svc, "hello")
+	assert.Equal(t, function.StatePending, first.Record.State)
+	assert.Equal(t, function.ReasonCodeCreating, first.Record.StateReasonCode)
+	assert.Equal(t, "The node is not serving.", first.Record.StateReason)
+
+	require.NoError(t, svc.ActivatePending(t.Context()))
+	assert.Equal(t, first.Revision, get(t, svc, "hello").Revision, "an unchanged reason is not rewritten")
+}
+
+func TestActivation_ReadyStoredFunctionBecomesActive(t *testing.T) {
+	svc := newOwner(t)
+	storeCode(t, svc, "hello", create(t, svc, "hello"))
+	storingRev := create(t, svc, "storing")
+	var seen []string
+	svc.SetReadiness(func(_ context.Context, rec function.Record) error {
+		seen = append(seen, rec.FunctionName)
+		return nil
+	})
+
+	require.NoError(t, svc.ActivatePending(t.Context()))
+
+	got := get(t, svc, "hello").Record
+	assert.Equal(t, function.StateActive, got.State)
+	assert.Empty(t, got.StateReason)
+	assert.Empty(t, got.StateReasonCode)
+	assert.Equal(t, function.CodeStored, got.Code.State)
+	storing := get(t, svc, "storing")
+	assert.Equal(t, function.StatePending, storing.Record.State, "a function still storing its code is never activated")
+	assert.Equal(t, storingRev, storing.Revision)
+	assert.Equal(t, []string{"hello"}, seen)
+}
+
+func TestActivation_CodeStoredWakesTheLoop(t *testing.T) {
+	svc := newOwner(t)
+	storeCode(t, svc, "first", create(t, svc, "first"))
+	// Readiness is enabled after first is stored, so no wake is buffered and
+	// only the start pass can activate first.
+	svc.SetReadiness(func(context.Context, function.Record) error { return nil })
+	ctx, cancel := context.WithCancel(t.Context())
+	done := make(chan struct{})
+	go func() { defer close(done); svc.RunActivation(ctx, time.Hour) }()
+	t.Cleanup(func() { cancel(); <-done })
+	require.Eventually(t, func() bool { return get(t, svc, "first").Record.State == function.StateActive },
+		10*time.Second, 20*time.Millisecond, "the start pass activates what was already stored")
+
+	storeCode(t, svc, "hello", create(t, svc, "hello"))
+
+	require.Eventually(t, func() bool { return get(t, svc, "hello").Record.State == function.StateActive },
+		10*time.Second, 20*time.Millisecond, "after the start pass, only the code-stored wake can activate within the hour")
+}

spinifex/domains/lambda/function/meta.go

@@ -44,6 +44,8 @@ type MetaService struct {
 
 	mu      sync.Mutex
 	buckets map[string]jetstream.KeyValue
+	ready   Readiness
+	nudge   chan struct{}
 }
 
 // NewMetaService builds the service over a JetStream handle.
@@ -306,6 +308,7 @@ func (s *MetaService) CodeStored(ctx context.Context, req *CodeStoredRequest, ac
 		return nil, fmt.Errorf("lambda: update function record: %w", err)
 	}
 	attach(ctx, kv, up, upRev, rev)
+	s.wake()
 	return &CodeStoredResponse{Record: rec, Revision: rev}, nil
 }
 

spinifex/domains/rds/instance/delete.go

@@ -66,6 +66,12 @@ func (s *Service) DeleteDBInstance(ctx context.Context, input *rds.DeleteDBInsta
 			"DB instance %s is already being deleted with %s; a retry must repeat that choice", id, snapshotChoice(rec.FinalSnapshotIdentifier))
 	}
 
+	// The handler that accepted the delete tears it down, and the reconciler
+	// resumes one whose handler is gone, so a repeat only reports the state.
+	if rec.Status == StatusDeleting {
+		return &rds.DeleteDBInstanceOutput{DBInstance: s.projectDBInstance(rec)}, nil
+	}
+
 	// Reserve the final snapshot before the instance enters deleting. A collision
 	// therefore fails while the VM and data volume are still intact, and a crash
 	// afterwards leaves a creating record the reconciler can resolve.
@@ -74,29 +80,29 @@ func (s *Service) DeleteDBInstance(ctx context.Context, input *rds.DeleteDBInsta
 		return nil, err
 	}
 
-	// A repeat of a delete already under way re-runs the teardown rather than
-	// being rejected: the first call may have died partway through it.
-	if rec.Status != StatusDeleting {
-		now := time.Now().UTC()
-		rec.Status = StatusDeleting
-		rec.FailureReason = ""
-		rec.FinalSnapshotIdentifier = finalSnapshot
-		rec.TransitionStartedAt = &now
-		rec.UpdatedAt = now
-		if err := updateJSON(ctx, kv, DBInstanceKey(id), rev, rec); err != nil {
-			if reservation.creator {
-				s.rollbackFinalSnapshotReservation(ctx, kv, finalSnapshot, reservation.revision)
-			}
-			if errors.Is(err, kvstore.ErrConflict) {
-				return nil, awserrors.Errorf(awserrors.ErrorDBInstanceInvalidState,
-					"DB instance %s changed state concurrently; retry the delete", id)
-			}
-			return nil, err
+	now := time.Now().UTC()
+	rec.Status = StatusDeleting
+	rec.FailureReason = ""
+	rec.FinalSnapshotIdentifier = finalSnapshot
+	rec.TransitionStartedAt = &now
+	rec.UpdatedAt = now
+	if err := updateJSON(ctx, kv, DBInstanceKey(id), rev, rec); err != nil {
+		if reservation.creator {
+			s.rollbackFinalSnapshotReservation(ctx, kv, finalSnapshot, reservation.revision)
 		}
-		s.RecordEvent(ctx, accountID, EventSourceTypeDBInstance, id, "DB instance deleted.", EventCategoryDeletion)
+		if errors.Is(err, kvstore.ErrConflict) {
+			return nil, awserrors.Errorf(awserrors.ErrorDBInstanceInvalidState,
+				"DB instance %s changed state concurrently; retry the delete", id)
+		}
+		return nil, err
 	}
+	s.RecordEvent(ctx, accountID, EventSourceTypeDBInstance, id, "DB instance deleted.", EventCategoryDeletion)
 
-	if err := s.teardownDBInstance(ctx, kv, accountID, rec, reservation.creator); err != nil {
+	// Ends exactly when the reconciler's guard does, so the teardown is this
+	// handler's alone until the reconciler may resume it.
+	teardownCtx, cancel := context.WithDeadline(ctx, deleteClaimEnds(rec))
+	defer cancel()
+	if err := s.teardownDBInstance(teardownCtx, kv, accountID, rec, reservation.creator); err != nil {
 		// The reconciler finishes the delete once the modify lets go.
 		if errors.Is(err, errModifyInProgress) {
 			return &rds.DeleteDBInstanceOutput{DBInstance: s.projectDBInstance(rec)}, nil
@@ -202,6 +208,11 @@ func (s *Service) teardownDBInstance(ctx context.Context, kv *kvstore.Bucket, ac
 		s.stopEngineOrRecordFallback(ctx, accountID, rec, "deleting")
 	}
 
+	// The engine stop swallows a deadline, and a handler past its claim must not
+	// start a terminate the reconciler may already be running.
+	if err := ctx.Err(); err != nil {
+		return err
+	}
 	if err := s.terminateInstanceVM(ctx, rec.InstanceID, rec.DataVolumeID); err != nil {
 		return fmt.Errorf("rds: terminate the VM behind %s: %w", rec.DBInstanceIdentifier, err)
 	}

spinifex/domains/rds/instance/delete_test.go

@@ -1,9 +1,11 @@
 package instance
 
 import (
+	"context"
 	"errors"
 	"fmt"
 	"testing"
+	"time"
 
 	"github.com/aws/aws-sdk-go/aws"
 	"github.com/aws/aws-sdk-go/service/rds"
@@ -176,13 +178,12 @@ func TestDeleteDBInstance_ReservesTheFinalSnapshotBeforeCuttingIt(t *testing.T)
 }
 
 // A dead worker can leave the EC2 snapshot cut while its RDS record still says
-// creating. A retry adopts that exact account-scoped snapshot instead of
-// cutting another one and permanently pinning the source volume twice.
-func TestDeleteDBInstance_AdoptsAnInterruptedFinalSnapshot(t *testing.T) {
+// creating. The resumed teardown adopts that exact account-scoped snapshot
+// instead of cutting another one and permanently pinning the source volume twice.
+func TestReconciler_AdoptsAnInterruptedFinalSnapshot(t *testing.T) {
 	t.Parallel()
 	h := newLifecycleHarness(t, false)
-	rec := availableRecord()
-	rec.Status = StatusDeleting
+	rec := abandonedDelete(availableRecord())
 	rec.FinalSnapshotIdentifier = "orders-db-final"
 	seedInstance(t, h.svc, rec)
 
@@ -201,13 +202,9 @@ func TestDeleteDBInstance_AdoptsAnInterruptedFinalSnapshot(t *testing.T) {
 		},
 	}
 
-	_, err = h.svc.DeleteDBInstance(t.Context(), &rds.DeleteDBInstanceInput{
-		DBInstanceIdentifier:      aws.String(testDBID),
-		FinalDBSnapshotIdentifier: aws.String(rec.FinalSnapshotIdentifier),
-	}, testAccountID)
-	require.NoError(t, err)
+	require.NoError(t, onePass(t, NewReconciler(h.svc, "node-a")))
 
-	assert.Empty(t, h.snaps.created, "the retry must adopt the cut snapshot, not create another")
+	assert.Empty(t, h.snaps.created, "the resumed teardown must adopt the cut snapshot, not create another")
 	stored, found := h.snapshotRecord(t, rec.FinalSnapshotIdentifier)
 	require.True(t, found)
 	assert.Equal(t, SnapshotStatusAvailable, stored.Status)
@@ -325,26 +322,18 @@ func TestDeleteDBInstance_RetainsTheVolumeWhenTheVolumeStoreReportsItInUse(t *te
 
 // A teardown that died partway through is replayed, so every step has to treat
 // a resource that is already gone as work it has done.
-func TestDeleteDBInstance_IsIdempotentAcrossARetriedTeardown(t *testing.T) {
+func TestReconciler_IsIdempotentAcrossARetriedTeardown(t *testing.T) {
 	t.Parallel()
 	h := newLifecycleHarness(t, false)
-	rec := availableRecord()
-	// The shape a delete leaves behind when its caller died mid-teardown.
-	rec.Status = StatusDeleting
+	rec := abandonedDelete(availableRecord())
 	rec.FinalSnapshotIdentifier = "orders-db-final"
 	seedInstance(t, h.svc, rec)
-
-	input := &rds.DeleteDBInstanceInput{
-		DBInstanceIdentifier:      aws.String(testDBID),
-		FinalDBSnapshotIdentifier: aws.String("orders-db-final"),
-	}
-	_, err := h.svc.DeleteDBInstance(t.Context(), input, testAccountID)
-	require.NoError(t, err)
+	reconciler := NewReconciler(h.svc, "node-a")
+	require.NoError(t, onePass(t, reconciler))
 
 	// The record left behind by the interrupted attempt, replayed.
 	seedInstance(t, h.svc, rec)
-	_, err = h.svc.DeleteDBInstance(t.Context(), input, testAccountID)
-	require.NoError(t, err)
+	require.NoError(t, onePass(t, reconciler))
 
 	assert.Len(t, h.snaps.created, 1, "a replayed teardown must not take a second final snapshot")
 	assert.False(t, h.recordExists(t, testDBID))
@@ -355,9 +344,7 @@ func TestDeleteDBInstance_IsIdempotentAcrossARetriedTeardown(t *testing.T) {
 func TestReconciler_ResumesAnInterruptedDelete(t *testing.T) {
 	t.Parallel()
 	h := newLifecycleHarness(t, false)
-	rec := availableRecord()
-	rec.Status = StatusDeleting
-	seedInstance(t, h.svc, rec)
+	seedInstance(t, h.svc, abandonedDelete(availableRecord()))
 
 	reconciler := NewReconciler(h.svc, "node-a")
 	require.NoError(t, onePass(t, reconciler))
@@ -366,6 +353,87 @@ func TestReconciler_ResumesAnInterruptedDelete(t *testing.T) {
 	assert.False(t, h.recordExists(t, testDBID))
 }
 
+// A second teardown racing the handler's sees the VM mid-shutdown as gone and
+// fails on the still-attached volume, so the handler's delete is left to it.
+func TestReconciler_LeavesADeleteItsHandlerMayStillBeRunning(t *testing.T) {
+	t.Parallel()
+	h := newLifecycleHarness(t, false)
+	rec := availableRecord()
+	started := time.Now().UTC()
+	rec.Status = StatusDeleting
+	rec.TransitionStartedAt = &started
+	seedInstance(t, h.svc, rec)
+
+	require.NoError(t, onePass(t, NewReconciler(h.svc, "node-a")))
+
+	assert.Empty(t, h.launcher.terminated)
+	assert.True(t, h.recordExists(t, testDBID))
+}
+
+// A repeat must not start a teardown of its own alongside the one under way;
+// the reconciler is what resumes an abandoned one.
+func TestDeleteDBInstance_ARepeatStartsNoTeardown(t *testing.T) {
+	t.Parallel()
+	h := newLifecycleHarness(t, false)
+	seedInstance(t, h.svc, abandonedDelete(availableRecord()))
+
+	out, err := h.svc.DeleteDBInstance(t.Context(), skipFinalSnapshot(), testAccountID)
+	require.NoError(t, err)
+
+	assert.Equal(t, string(StatusDeleting), aws.StringValue(out.DBInstance.DBInstanceStatus))
+	assert.Empty(t, h.launcher.terminated)
+	assert.True(t, h.recordExists(t, testDBID))
+}
+
+// The reconciler resumes a delete from the persisted stamp, so the handler's
+// teardown must stop at that same instant, however long the acceptance took.
+func TestDeleteDBInstance_TeardownEndsWhenTheClaimDoes(t *testing.T) {
+	t.Parallel()
+	h := newLifecycleHarness(t, false)
+	seedInstance(t, h.svc, availableRecord())
+	var claimEnds time.Time
+	h.snaps.beforeCreate = func() { stored := h.record(t); claimEnds = deleteClaimEnds(&stored) }
+
+	_, err := h.svc.DeleteDBInstance(t.Context(), &rds.DeleteDBInstanceInput{
+		DBInstanceIdentifier:      aws.String(testDBID),
+		FinalDBSnapshotIdentifier: aws.String("orders-db-final"),
+	}, testAccountID)
+	require.NoError(t, err)
+
+	require.False(t, claimEnds.IsZero())
+	assert.True(t, h.snaps.deadline.Equal(claimEnds), "deadline %s, claim ends %s", h.snaps.deadline, claimEnds)
+}
+
+// A graceful stop that runs into the claim's end is swallowed, so the teardown
+// itself has to notice and leave the terminate to the reconciler.
+func TestTeardownDBInstance_StartsNoTerminatePastItsDeadline(t *testing.T) {
+	t.Parallel()
+	h := newLifecycleHarness(t, false)
+	h.agent.silenceType(CommandStopEngine)
+	rec := availableRecord()
+	rec.Status = StatusDeleting
+	rec.FinalSnapshotIdentifier = "orders-db-final"
+	seedInstance(t, h.svc, rec)
+	kv, err := h.svc.bucket(t.Context(), testAccountID)
+	require.NoError(t, err)
+
+	ctx, cancel := context.WithTimeout(t.Context(), 200*time.Millisecond)
+	defer cancel()
+	err = h.svc.teardownDBInstance(ctx, kv, testAccountID, &rec, true)
+
+	require.ErrorIs(t, err, context.DeadlineExceeded)
+	assert.Empty(t, h.launcher.terminated)
+	assert.True(t, h.recordExists(t, testDBID))
+}
+
+// The shape a delete leaves once its handler's request budget has run out.
+func abandonedDelete(rec DBInstanceRecord) DBInstanceRecord {
+	started := time.Now().UTC().Add(-deleteTimeout - time.Minute)
+	rec.Status = StatusDeleting
+	rec.TransitionStartedAt = &started
+	return rec
+}
+
 // A stop whose caller died leaves the VM possibly still running, so the stop is
 // re-issued rather than assumed.
 func TestReconciler_ResumesAnInterruptedStop(t *testing.T) {

spinifex/domains/rds/instance/lifecycle_test.go

@@ -117,7 +117,9 @@ func (f *fakeInstanceCommander) DetachVolumeFromStopped(_ context.Context, insta
 // DescribeSnapshots reports against the data volume, which is what decides
 // between deleting the volume and retaining it.
 type fakeSnapshots struct {
-	created       []*ec2.CreateSnapshotInput
+	created []*ec2.CreateSnapshotInput
+	// The context deadline the last CreateSnapshot ran under.
+	deadline      time.Time
 	holding       []string
 	deleted       []string
 	createErr     error
@@ -132,7 +134,8 @@ type fakeSnapshots struct {
 
 var _ snapshotProvider = (*fakeSnapshots)(nil)
 
-func (f *fakeSnapshots) CreateSnapshot(_ context.Context, input *ec2.CreateSnapshotInput, _ string) (*ec2.Snapshot, error) {
+func (f *fakeSnapshots) CreateSnapshot(ctx context.Context, input *ec2.CreateSnapshotInput, _ string) (*ec2.Snapshot, error) {
+	f.deadline, _ = ctx.Deadline()
 	if f.beforeCreate != nil {
 		f.beforeCreate()
 	}

spinifex/domains/rds/instance/reconciler.go

@@ -610,6 +610,11 @@ func (r *Reconciler) reconcileStopping(ctx context.Context, kv *kvstore.Bucket,
 // did; only a teardown still stuck past the bound is called failed, which is
 // what lets the customer retry the delete.
 func (r *Reconciler) reconcileDeleting(ctx context.Context, kv *kvstore.Bucket, rev uint64, accountID string, rec *DBInstanceRecord) error {
+	// The handler that accepted the delete tears it down until its claim ends; a
+	// second teardown alongside it sees the VM mid-shutdown as gone.
+	if time.Now().Before(deleteClaimEnds(rec)) {
+		return nil
+	}
 	err := r.svc.teardownDBInstance(ctx, kv, accountID, rec, false)
 	if err == nil || errors.Is(err, errFinalSnapshotInProgress) || errors.Is(err, errModifyInProgress) {
 		return nil
@@ -795,6 +800,12 @@ func (s *Service) findEC2SnapshotFor(ctx context.Context, accountID, dbSnapshotI
 // When the transition began. A record written by an older control plane carries
 // no stamp, so its last write stands in — it is never earlier than the
 // transition, so the bound cannot be under-counted.
+// When the handler that accepted a delete gives up its teardown, measured from
+// the persisted stamp so the handler and the reconciler agree on it.
+func deleteClaimEnds(rec *DBInstanceRecord) time.Time {
+	return transitionStarted(rec).Add(deleteTimeout)
+}
+
 func transitionStarted(rec *DBInstanceRecord) time.Time {
 	if rec.TransitionStartedAt != nil {
 		return *rec.TransitionStartedAt

spinifex/operator/accountlifecycle/absence_proof_silent_node_test.go

@@ -1,89 +0,0 @@
-package accountlifecycle
-
-//test:in-package — the volume reaper and its fake cluster are unexported, and
-// these drive its force path against a node that never answers.
-
-// Every test here waits out the fixed describe fan-out deadline a silent node
-// costs, two or three times over. This file sorts first so they take parallel
-// slots before the package's short tests queue ahead of them.
-
-import (
-	"testing"
-
-	"github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
-	"github.com/stretchr/testify/assert"
-	"github.com/stretchr/testify/require"
-)
-
-// A lone node's NotFound says nothing about the node that stayed silent, so
-// the holder is not proven gone and the volume keeps its attachment.
-func TestVolumeReaperForceRetainsAVolumeWhenALoneNodeAnswersNotFoundWhileAnotherIsSilent(t *testing.T) {
-	t.Parallel()
-	assertForceRetainsTheHeldVolume(t, func(cluster *fakeCluster) {
-		cluster.fail("ec2.DescribeInstances", awserrors.ErrorInvalidInstanceIDNotFound)
-	}, []string{fakeClusterNode, "node-2"})
-}
-
-// A terminated copy is not proof either while the node that may still run the
-// holder has not answered.
-func TestVolumeReaperForceRetainsAVolumeWhenATerminatedCopyIsFoundWhileTheHolderNodeIsSilent(t *testing.T) {
-	t.Parallel()
-	assertForceRetainsTheHeldVolume(t, func(cluster *fakeCluster) {
-		cluster.reply("ec2.DescribeTerminatedInstances", instancesIn("terminated", "i-held"))
-	}, []string{fakeClusterNode, "node-2"})
-}
-
-// The holder lives on node-2, which never answers, while node-1 answers every
-// fan-out twice. A count of replies reads that as complete and the holder as
-// gone; only judging by responder identity sees that node-2 is missing.
-func TestVolumeReaperForceRetainsAVolumeWhenOneNodeAnswersTwiceAndTheHolderNodeIsSilent(t *testing.T) {
-	t.Parallel()
-	cluster := newFakeCluster(t)
-	volume := newHeldVolume(cluster, "vol-held", "i-on-silent-node")
-	cluster.echoAsNode(t, "ec2.DescribeInstances", fakeClusterNode)
-
-	reaper := newVolumeReaper(cluster.nc, fakeClusterNode, "node-2")
-	err := reaper.Delete(silentClusterCtx(t), "000000000042", Resource{Kind: "volume", ID: "vol-held"}, true)
-
-	require.Error(t, err)
-	assert.Contains(t, err.Error(), "vol-held")
-	assert.Contains(t, err.Error(), "i-on-silent-node")
-	deleted, holder, detaches := volume.state()
-	assert.False(t, deleted, "a holder on a silent node is not proven gone")
-	assert.Equal(t, "i-on-silent-node", holder)
-	assert.Zero(t, detaches)
-	assert.NotContains(t, cluster.called(), forceDetachSubject)
-}
-
-func TestForcedTeardownKeepsTheAccountWhenOneNodeAnswersTwiceAndTheHolderNodeIsSilent(t *testing.T) {
-	t.Parallel()
-	cluster := newFakeCluster(t)
-	volume := newHeldVolume(cluster, "vol-held", "i-on-silent-node")
-	cluster.echoAsNode(t, "ec2.DescribeInstances", fakeClusterNode)
-
-	accounts := newFakeAccounts()
-	engine := quietEngine(accounts, newVolumeReaper(cluster.nc, fakeClusterNode, "node-2"))
-	req := request()
-	req.Force = true
-
-	result, err := engine.Teardown(silentClusterCtx(t), req)
-
-	require.ErrorIs(t, err, ErrResourcesStuck)
-	require.Equal(t, 1, result.StuckCount())
-	var stuck Stuck
-	for _, stage := range result.Stages {
-		if len(stage.Stuck) > 0 {
-			stuck = stage.Stuck[0]
-		}
-	}
-	assert.Equal(t, "vol-held", stuck.Resource.ID)
-	assert.Contains(t, stuck.Reason, "vol-held")
-	assert.Contains(t, stuck.Reason, "i-on-silent-node")
-	assert.False(t, result.AccountDeleted)
-	assert.False(t, accounts.deleted)
-	assert.Equal(t, AccountStatusTerminating, accounts.account.Status)
-	deleted, holder, detaches := volume.state()
-	assert.False(t, deleted)
-	assert.Equal(t, "i-on-silent-node", holder)
-	assert.Zero(t, detaches)
-}

spinifex/operator/accountlifecycle/cluster.go

@@ -35,6 +35,7 @@ func NewClusterEngine(
 	svc IAMBackend,
 	buckets BucketStore,
 	cleaner PlatformCleaner,
+	options ...ClusterOption,
 ) (*Engine, error) {
 	// Refused rather than skipped. A teardown missing its bucket reaper still
 	// reports every stage drained and still deletes the account record, leaving
@@ -95,5 +96,20 @@ func NewClusterEngine(
 	// May be nil: a dry run needs no cleanup, and Teardown and Exercise refuse
 	// to start without it.
 	engine.Cleanup = cleaner
+	for _, option := range options {
+		option(engine)
+	}
 	return engine, nil
 }
+
+// ClusterOption binds an optional capability to the engine NewClusterEngine builds.
+type ClusterOption func(*Engine)
+
+// WithComputeCleanup binds the EC2 compute and storage cleanup. Teardown
+// refuses to start without both, so an engine built without them can still
+// answer a dry run but never deletes an account.
+func WithComputeCleanup(compute ComputeStage, storage StorageStage) ClusterOption {
+	return func(e *Engine) {
+		e.Compute, e.Storage = compute, storage
+	}
+}

spinifex/operator/accountlifecycle/converge.go

@@ -0,0 +1,317 @@
+package accountlifecycle
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"slices"
+	"strings"
+	"time"
+
+	operatorv1 "github.com/mulgadc/spinifex/contracts/operator/v1"
+	"github.com/mulgadc/spinifex/spinifex/domains/ec2/accountcleanup"
+	otelsetup "github.com/mulgadc/spinifex/spinifex/foundation/telemetry"
+)
+
+// ComputeStage is the EC2 compute cleanup and its read-only remainder.
+type ComputeStage interface {
+	accountcleanup.ComputeCleaner
+	accountcleanup.ComputeRemainder
+}
+
+// StorageStage is the EC2 storage cleanup and its read-only remainder.
+type StorageStage interface {
+	accountcleanup.StorageCleaner
+	accountcleanup.StorageRemainder
+}
+
+var errComputeCleanupMissing = errors.New("teardown needs the EC2 compute and storage cleanup")
+
+// Converge bounds the loop that repeats the compute and storage cleanup inside
+// one teardown. Each cleanup call returns without waiting for teardown, so the
+// loop is what turns a terminate into a deleted volume.
+type Converge struct {
+	// Budget is the longest the loop runs, well under the job's own budget.
+	Budget time.Duration
+	// MinBackoff and MaxBackoff bound the wait between passes, which doubles.
+	MinBackoff time.Duration
+	MaxBackoff time.Duration
+	// IncompletePasses is how many passes in a row may end with a read that did
+	// not complete before the loop gives up.
+	IncompletePasses int
+}
+
+// DefaultConverge is sized under the 2 h job budget and over the 10 min
+// stuck-terminate backstop, so a wedged terminate still ends the loop clean.
+func DefaultConverge() Converge {
+	return Converge{Budget: 30 * time.Minute, MinBackoff: 5 * time.Second, MaxBackoff: time.Minute, IncompletePasses: 3}
+}
+
+func (e *Engine) convergeConfig() Converge {
+	cfg := DefaultConverge()
+	if e.Converge.Budget > 0 {
+		cfg.Budget = e.Converge.Budget
+	}
+	if e.Converge.MinBackoff > 0 {
+		cfg.MinBackoff = e.Converge.MinBackoff
+	}
+	if e.Converge.MaxBackoff > 0 {
+		cfg.MaxBackoff = e.Converge.MaxBackoff
+	}
+	if e.Converge.IncompletePasses > 0 {
+		cfg.IncompletePasses = e.Converge.IncompletePasses
+	}
+	return cfg
+}
+
+// holdsFence stops a holder that has been superseded or cancelled before it
+// issues another terminate or delete. It reads the account back and requires
+// it to still be TERMINATING under this operation and fence.
+func (e *Engine) holdsFence(ctx context.Context, req Request) error {
+	if err := ctx.Err(); err != nil {
+		return err
+	}
+	account, err := e.Accounts.GetAccount(req.AccountID)
+	if err != nil {
+		return fmt.Errorf("read account to confirm the fence: %w", err)
+	}
+	op := account.TerminationOp
+	if account.Status != AccountStatusTerminating || op == nil || op.OperationID != req.OperationID || op.Fence != req.Fence {
+		return errFenceLost
+	}
+	return nil
+}
+
+var errFenceLost = errors.New("the account is no longer held under this operation and fence")
+
+func waitContext(ctx context.Context, d time.Duration) error {
+	timer := time.NewTimer(d)
+	defer timer.Stop()
+	select {
+	case <-ctx.Done():
+		return ctx.Err()
+	case <-timer.C:
+		return nil
+	}
+}
+
+// convergent reports whether time alone can resolve an item.
+func convergent(it accountcleanup.Item) bool {
+	if it.Disposition == accountcleanup.DispositionPending {
+		return true
+	}
+	switch it.Reason {
+	case accountcleanup.ReasonTerminating, accountcleanup.ReasonVolumeDeleting, accountcleanup.ReasonTeardownUnconfirmed:
+		return true
+	}
+	return false
+}
+
+// passSignature identifies what a pass left, so two identical passes in a row
+// with nothing convergent show the loop is not going anywhere.
+func passSignature(legacy []Stuck, outcomes ...*accountcleanup.Outcome) string {
+	var lines []string
+	for _, s := range legacy {
+		lines = append(lines, "legacy/"+s.Resource.Kind+"/"+s.Resource.ID+"/"+s.Reason)
+	}
+	for _, out := range outcomes {
+		for _, it := range out.Items {
+			lines = append(lines, it.Kind+"/"+it.ResourceID+"/"+it.Disposition+"/"+it.Reason+"/"+it.State)
+		}
+		lines = append(lines, "incomplete:"+strings.Join(out.IncompleteReasons, ","))
+	}
+	slices.Sort(lines)
+	return strings.Join(lines, "\n")
+}
+
+// convergeCompute repeats the compute cleanup, the storage-position legacy
+// reapers, the storage cleanup and the remainder read until the account holds
+// no instance, spot request or volume, or the loop proves it cannot make
+// progress. It ends the teardown with the stuck items named, never with a
+// shorter list: a lost fence or a cancelled context is an error.
+func (e *Engine) convergeCompute(ctx context.Context, req Request, result *Result) error {
+	cfg := e.convergeConfig()
+	started := time.Now()
+	deadline := started.Add(cfg.Budget)
+	backoff := cfg.MinBackoff
+	wait := e.wait
+	if wait == nil {
+		wait = waitContext
+	}
+
+	attachments := StageResult{Stage: StageAttachments}
+	storage := StageResult{Stage: StageStorage}
+	var lastSignature string
+	var idlePasses, incompletePasses int
+
+	for pass := 1; ; pass++ {
+		if err := e.holdsFence(ctx, req); err != nil {
+			return err
+		}
+		computeOut, err := e.Compute.CleanupCompute(ctx, &accountcleanup.ComputeRequest{
+			AccountID: req.AccountID, CorrelationRef: req.OperationID, Force: req.Force,
+		})
+		if err != nil {
+			return fmt.Errorf("compute cleanup: %w", err)
+		}
+		legacy, err := e.runStage(ctx, StageStorage, req)
+		if err != nil {
+			return err
+		}
+		storageOut, err := e.Storage.CleanupStorage(ctx, &accountcleanup.StorageRequest{
+			AccountID: req.AccountID, CorrelationRef: req.OperationID, Force: req.Force,
+		})
+		if err != nil {
+			return fmt.Errorf("storage cleanup: %w", err)
+		}
+		computeLeft, err := e.Compute.ComputeRemaining(ctx, req.AccountID)
+		if err != nil {
+			return fmt.Errorf("compute remainder: %w", err)
+		}
+		storageLeft, err := e.Storage.StorageRemaining(ctx, req.AccountID)
+		if err != nil {
+			return fmt.Errorf("storage remainder: %w", err)
+		}
+
+		attachments.Deleted = appendDeleted(attachments.Deleted, computeOut)
+		storage.Deleted = appendDeleted(storage.Deleted, storageOut)
+		for _, r := range legacy.Deleted {
+			if !containsResource(storage.Deleted, r) {
+				storage.Deleted = append(storage.Deleted, r)
+			}
+		}
+		storage.Stuck = legacy.Stuck
+		if e.OnPass != nil {
+			e.OnPass(ctx, pass)
+		}
+		e.logger().InfoContext(ctx, "Teardown convergence pass",
+			"account_id", req.AccountID, "pass", pass,
+			"compute_remaining", len(computeLeft.Items), "storage_remaining", len(storageLeft.Items),
+			"legacy_stuck", len(legacy.Stuck), "elapsed_ms", otelsetup.Millis(time.Since(started)))
+
+		if len(computeLeft.Items) == 0 && len(storageLeft.Items) == 0 &&
+			computeLeft.Status == accountcleanup.StatusClean && storageLeft.Status == accountcleanup.StatusClean &&
+			len(legacy.Stuck) == 0 {
+			return e.finishConverge(ctx, result, attachments, storage, started)
+		}
+
+		var moving bool
+		for _, out := range []*accountcleanup.Outcome{computeOut, storageOut} {
+			moving = moving || slices.ContainsFunc(out.Items, convergent)
+		}
+		if len(computeOut.IncompleteReasons)+len(storageOut.IncompleteReasons)+len(computeLeft.IncompleteReasons)+len(storageLeft.IncompleteReasons) > 0 {
+			incompletePasses++
+		} else {
+			incompletePasses = 0
+		}
+		signature := passSignature(legacy.Stuck, computeOut, storageOut, computeLeft, storageLeft)
+		if signature == lastSignature && !moving {
+			idlePasses++
+		} else {
+			idlePasses = 0
+		}
+		lastSignature = signature
+
+		reason := ""
+		switch {
+		case idlePasses >= 1:
+			reason = "only items time cannot resolve remain"
+		case incompletePasses >= cfg.IncompletePasses:
+			reason = "a read stayed incomplete"
+		case time.Now().Add(backoff).After(deadline):
+			reason = "the convergence budget ended"
+		}
+		if reason != "" {
+			attachments.Stuck = stuckFrom(attachments.Stuck[:0], computeOut, computeLeft)
+			storage.Stuck = append(legacy.Stuck, stuckFrom(nil, storageOut, storageLeft)...)
+			if err := e.finishConverge(ctx, result, attachments, storage, started); err != nil {
+				return err
+			}
+			e.logger().WarnContext(ctx, "Account teardown stopped converging",
+				"accountID", req.AccountID, "reason", reason, "passes", pass)
+			return fmt.Errorf("%d resources left (%s): %w", result.StuckCount(), reason, ErrResourcesStuck)
+		}
+
+		if err := wait(ctx, backoff); err != nil {
+			return err
+		}
+		backoff = min(backoff*2, cfg.MaxBackoff)
+	}
+}
+
+func (e *Engine) finishConverge(ctx context.Context, result *Result, attachments, storage StageResult, started time.Time) error {
+	elapsed := time.Since(started).String()
+	attachments.Elapsed, storage.Elapsed = elapsed, elapsed
+	for _, stage := range []StageResult{attachments, storage} {
+		result.Stages = append(result.Stages, stage)
+		if e.OnStage != nil {
+			e.OnStage(ctx, stage)
+		}
+	}
+	return nil
+}
+
+func appendDeleted(into []Resource, out *accountcleanup.Outcome) []Resource {
+	for _, it := range out.Items {
+		if it.Disposition == accountcleanup.DispositionDeleted {
+			r := Resource{Kind: it.Kind, ID: it.ResourceID}
+			if !containsResource(into, r) {
+				into = append(into, r)
+			}
+		}
+	}
+	return into
+}
+
+// stuckFrom names every item the account still holds with the reason the
+// cleanup gave for keeping it, or that the record remains. An incomplete read
+// is named too, because it proves nothing about what is left.
+func stuckFrom(into []Stuck, cleanup, remainder *accountcleanup.Outcome) []Stuck {
+	reasons := map[string]string{}
+	for _, it := range cleanup.Items {
+		if it.Reason != "" && it.Disposition != accountcleanup.DispositionDeleted {
+			reasons[it.Kind+"/"+it.ResourceID] = it.Reason
+		}
+	}
+	for _, it := range remainder.Items {
+		reason := reasons[it.Kind+"/"+it.ResourceID]
+		if reason == "" {
+			reason = it.Reason
+		}
+		into = append(into, Stuck{Resource: Resource{Kind: it.Kind, ID: it.ResourceID, Detail: it.State}, Reason: reason})
+	}
+	for _, reason := range slices.Concat(cleanup.IncompleteReasons, remainder.IncompleteReasons) {
+		into = append(into, Stuck{Resource: Resource{Kind: "inventory", ID: reason}, Reason: reason})
+	}
+	return into
+}
+
+// rereadComputeStorage reads the instances, spot requests and volumes the
+// account still holds immediately before the record is removed, so anything
+// that appeared after the convergence loop ended keeps the account. A read
+// that fails or is incomplete is named, never taken as empty.
+func (e *Engine) rereadComputeStorage(ctx context.Context, accountID string, result *Result) {
+	compute, cerr := e.Compute.ComputeRemaining(ctx, accountID)
+	storage, serr := e.Storage.StorageRemaining(ctx, accountID)
+	e.addFinalStuck(result, StageAttachments, accountID, compute, cerr)
+	e.addFinalStuck(result, StageStorage, accountID, storage, serr)
+}
+
+func (e *Engine) addFinalStuck(result *Result, stage Stage, accountID string, out *accountcleanup.Outcome, err error) {
+	var stuck []Stuck
+	if err != nil || out == nil {
+		stuck = []Stuck{{Resource: Resource{Kind: "inventory", ID: stage.String()}, Reason: operatorv1.AccountingReasonRereadIncomplete}}
+	} else {
+		stuck = stuckFrom(nil, &accountcleanup.Outcome{}, out)
+	}
+	if len(stuck) == 0 {
+		return
+	}
+	for i := range result.Stages {
+		if result.Stages[i].Stage == stage {
+			result.Stages[i].Stuck = append(result.Stages[i].Stuck, stuck...)
+			return
+		}
+	}
+	result.Stages = append(result.Stages, StageResult{Stage: stage, Elapsed: "0s", Stuck: stuck})
+}

spinifex/operator/accountlifecycle/converge_test.go

@@ -0,0 +1,256 @@
+package accountlifecycle
+
+//test:in-package — the convergence loop drives unexported engine fields and
+// the unexported backoff hook.
+
+import (
+	"context"
+	"errors"
+	"testing"
+	"time"
+
+	"github.com/mulgadc/spinifex/spinifex/domains/ec2/accountcleanup"
+)
+
+// scriptedStage answers each pass from a script, repeating the last answer.
+type scriptedStage struct {
+	cleanups   []*accountcleanup.Outcome
+	remainders []*accountcleanup.Outcome
+	cleanupN   int
+	remainN    int
+	forces     []bool
+}
+
+func pick(list []*accountcleanup.Outcome, n int) *accountcleanup.Outcome {
+	if n >= len(list) {
+		n = len(list) - 1
+	}
+	return list[n]
+}
+
+func (s *scriptedStage) run(forceIn bool) *accountcleanup.Outcome {
+	s.forces = append(s.forces, forceIn)
+	out := pick(s.cleanups, s.cleanupN)
+	s.cleanupN++
+	return out
+}
+
+func (s *scriptedStage) remaining() *accountcleanup.Outcome {
+	out := pick(s.remainders, s.remainN)
+	s.remainN++
+	return out
+}
+
+type scriptedCompute struct{ scriptedStage }
+
+func (s *scriptedCompute) CleanupCompute(_ context.Context, req *accountcleanup.ComputeRequest) (*accountcleanup.ComputeResult, error) {
+	return s.run(req.Force), nil
+}
+
+func (s *scriptedCompute) ComputeRemaining(context.Context, string) (*accountcleanup.Outcome, error) {
+	return s.remaining(), nil
+}
+
+type scriptedStorage struct{ scriptedStage }
+
+func (s *scriptedStorage) CleanupStorage(_ context.Context, req *accountcleanup.StorageRequest) (*accountcleanup.StorageResult, error) {
+	return s.run(req.Force), nil
+}
+
+func (s *scriptedStorage) StorageRemaining(context.Context, string) (*accountcleanup.Outcome, error) {
+	return s.remaining(), nil
+}
+
+func outcome(items ...accountcleanup.Item) *accountcleanup.Outcome {
+	out := &accountcleanup.Outcome{AccountID: testAccountID, Status: accountcleanup.StatusClean, Items: items}
+	return out
+}
+
+func incomplete(reason string) *accountcleanup.Outcome {
+	return &accountcleanup.Outcome{AccountID: testAccountID, Status: accountcleanup.StatusIncomplete, IncompleteReasons: []string{reason}}
+}
+
+func item(kind, id, state, disposition, reason string) accountcleanup.Item {
+	return accountcleanup.Item{Kind: kind, ResourceID: id, State: state, Disposition: disposition, Reason: reason}
+}
+
+// convergeEngine runs a teardown with no real delay between passes.
+func convergeEngine(accounts AccountStore, compute *scriptedCompute, storage *scriptedStorage) (*Engine, *int) {
+	engine := quietEngine(accounts)
+	engine.Compute, engine.Storage = compute, storage
+	waits := 0
+	engine.wait = func(ctx context.Context, _ time.Duration) error {
+		waits++
+		return ctx.Err()
+	}
+	return engine, &waits
+}
+
+var (
+	runningInstance     = item(accountcleanup.KindInstance, "i-1", "running", accountcleanup.DispositionPending, accountcleanup.ReasonTerminating)
+	termInstance        = item(accountcleanup.KindInstance, "i-1", "terminated", accountcleanup.DispositionDeleted, "")
+	blockedHolderVolume = item(accountcleanup.KindVolume, "vol-1", "in-use", accountcleanup.DispositionBlocked, accountcleanup.ReasonHolderPresent)
+)
+
+func TestConvergeRunningInstanceFinishesInOneJob(t *testing.T) {
+	compute := &scriptedCompute{scriptedStage{
+		cleanups:   []*accountcleanup.Outcome{outcome(runningInstance), outcome(termInstance)},
+		remainders: []*accountcleanup.Outcome{outcome(item(accountcleanup.KindInstance, "i-1", "shutting-down", accountcleanup.DispositionResidue, accountcleanup.ReasonRecordRemains)), outcome()},
+	}}
+	storage := &scriptedStorage{scriptedStage{
+		cleanups:   []*accountcleanup.Outcome{outcome(blockedHolderVolume), outcome(item(accountcleanup.KindVolume, "vol-1", "available", accountcleanup.DispositionDeleted, ""))},
+		remainders: []*accountcleanup.Outcome{outcome(item(accountcleanup.KindVolume, "vol-1", "in-use", accountcleanup.DispositionResidue, accountcleanup.ReasonRecordRemains)), outcome()},
+	}}
+	accounts := newFakeAccounts()
+	engine, waits := convergeEngine(accounts, compute, storage)
+
+	result, err := engine.Teardown(context.Background(), request())
+	if err != nil {
+		t.Fatalf("a running instance must converge in one job, got %v", err)
+	}
+	if !result.AccountDeleted || !accounts.deleted {
+		t.Fatalf("account must be deleted once the loop ends clean: %+v", result)
+	}
+	if *waits != 1 || compute.cleanupN != 2 || storage.cleanupN != 2 {
+		t.Fatalf("expected two passes and one backoff, got passes=%d/%d waits=%d", compute.cleanupN, storage.cleanupN, *waits)
+	}
+}
+
+func TestConvergeStopsOnBlockerTimeCannotResolve(t *testing.T) {
+	managed := item(accountcleanup.KindInstance, "i-managed", "running", accountcleanup.DispositionBlocked, accountcleanup.ReasonManagedServiceOwner)
+	remain := item(accountcleanup.KindInstance, "i-managed", "running", accountcleanup.DispositionResidue, accountcleanup.ReasonRecordRemains)
+	compute := &scriptedCompute{scriptedStage{cleanups: []*accountcleanup.Outcome{outcome(managed)}, remainders: []*accountcleanup.Outcome{outcome(remain)}}}
+	storage := &scriptedStorage{scriptedStage{cleanups: []*accountcleanup.Outcome{outcome()}, remainders: []*accountcleanup.Outcome{outcome()}}}
+	accounts := newFakeAccounts()
+	engine, _ := convergeEngine(accounts, compute, storage)
+	engine.Converge = Converge{Budget: time.Hour}
+
+	result, err := engine.Teardown(context.Background(), request())
+	if !errors.Is(err, ErrResourcesStuck) {
+		t.Fatalf("a non-converging blocker must end the job stuck, got %v", err)
+	}
+	if accounts.deleted {
+		t.Fatal("account must not be deleted while an instance remains")
+	}
+	if compute.cleanupN > 2 {
+		t.Fatalf("loop must end before the budget on a blocker time cannot resolve, ran %d passes", compute.cleanupN)
+	}
+	var named bool
+	for _, stage := range result.Stages {
+		for _, s := range stage.Stuck {
+			named = named || (s.Resource.ID == "i-managed" && s.Reason == accountcleanup.ReasonManagedServiceOwner)
+		}
+	}
+	if !named {
+		t.Fatalf("stuck item must be named with the cleanup's reason: %+v", result.Stages)
+	}
+}
+
+// supersededAccounts reports a different holder once the first cleanup ran.
+type supersededAccounts struct {
+	*fakeAccounts
+
+	after *scriptedCompute
+}
+
+func (s *supersededAccounts) GetAccount(id string) (*Account, error) {
+	account, err := s.fakeAccounts.GetAccount(id)
+	if err != nil || s.after.cleanupN == 0 {
+		return account, err
+	}
+	taken := *account
+	taken.TerminationOp = &TerminationOp{OperationID: "other-op", Fence: account.TerminationOp.Fence + 1}
+	return &taken, nil
+}
+
+func TestConvergeLostFenceStopsBeforeNextPass(t *testing.T) {
+	compute := &scriptedCompute{scriptedStage{
+		cleanups:   []*accountcleanup.Outcome{outcome(runningInstance)},
+		remainders: []*accountcleanup.Outcome{outcome(item(accountcleanup.KindInstance, "i-1", "shutting-down", accountcleanup.DispositionResidue, accountcleanup.ReasonRecordRemains))},
+	}}
+	storage := &scriptedStorage{scriptedStage{cleanups: []*accountcleanup.Outcome{outcome()}, remainders: []*accountcleanup.Outcome{outcome()}}}
+	accounts := &supersededAccounts{fakeAccounts: newFakeAccounts(), after: compute}
+	engine, _ := convergeEngine(accounts, compute, storage)
+
+	// The deadline lets a loop that ignores the lost fence end, so it fails
+	// the pass count below instead of hanging.
+	ctx, cancel := context.WithTimeout(context.Background(), 500*time.Millisecond)
+	defer cancel()
+	_, err := engine.Teardown(ctx, request())
+	if err == nil || errors.Is(err, ErrResourcesStuck) {
+		t.Fatalf("a lost fence must stop the loop with an error, got %v", err)
+	}
+	if compute.cleanupN != 1 {
+		t.Fatalf("no terminate may be issued after the fence is lost, ran %d passes", compute.cleanupN)
+	}
+	if accounts.deleted {
+		t.Fatal("a superseded holder must not delete the account")
+	}
+}
+
+func TestConvergeCancelledContextStopsLoop(t *testing.T) {
+	compute := &scriptedCompute{scriptedStage{
+		cleanups:   []*accountcleanup.Outcome{outcome(runningInstance)},
+		remainders: []*accountcleanup.Outcome{outcome(item(accountcleanup.KindInstance, "i-1", "shutting-down", accountcleanup.DispositionResidue, accountcleanup.ReasonRecordRemains))},
+	}}
+	storage := &scriptedStorage{scriptedStage{cleanups: []*accountcleanup.Outcome{outcome()}, remainders: []*accountcleanup.Outcome{outcome()}}}
+	engine, _ := convergeEngine(newFakeAccounts(), compute, storage)
+	ctx, cancel := context.WithCancel(context.Background())
+	engine.OnPass = func(context.Context, int) { cancel() }
+
+	if _, err := engine.Teardown(ctx, request()); !errors.Is(err, context.Canceled) {
+		t.Fatalf("a cancelled job must stop the loop, got %v", err)
+	}
+	if compute.cleanupN != 1 {
+		t.Fatalf("no pass may start after cancel, ran %d", compute.cleanupN)
+	}
+}
+
+func TestConvergeIncompleteReadsEndStuckNotClean(t *testing.T) {
+	compute := &scriptedCompute{scriptedStage{cleanups: []*accountcleanup.Outcome{incomplete(accountcleanup.ReasonInstanceInventoryIncomplete)}, remainders: []*accountcleanup.Outcome{incomplete(accountcleanup.ReasonInstanceInventoryIncomplete)}}}
+	storage := &scriptedStorage{scriptedStage{cleanups: []*accountcleanup.Outcome{outcome()}, remainders: []*accountcleanup.Outcome{outcome()}}}
+	accounts := newFakeAccounts()
+	engine, _ := convergeEngine(accounts, compute, storage)
+
+	_, err := engine.Teardown(context.Background(), request())
+	if !errors.Is(err, ErrResourcesStuck) || accounts.deleted {
+		t.Fatalf("an unreadable inventory must never read as an empty account: err=%v deleted=%v", err, accounts.deleted)
+	}
+}
+
+// The loop keeps the account while the storage owner reports a volume still
+// recorded attached. That the owner refuses such a volume is the owner's own
+// invariant test in accountcleanup, against its real cleanup.
+func TestConvergeKeepsTheAccountWhileAnAttachedVolumeRemains(t *testing.T) {
+	compute := &scriptedCompute{scriptedStage{cleanups: []*accountcleanup.Outcome{outcome()}, remainders: []*accountcleanup.Outcome{outcome()}}}
+	storage := &scriptedStorage{scriptedStage{
+		cleanups:   []*accountcleanup.Outcome{outcome(blockedHolderVolume)},
+		remainders: []*accountcleanup.Outcome{outcome(item(accountcleanup.KindVolume, "vol-1", "in-use", accountcleanup.DispositionResidue, accountcleanup.ReasonRecordRemains))},
+	}}
+	for _, force := range []bool{false, true} {
+		accounts := newFakeAccounts()
+		engine, _ := convergeEngine(accounts, compute, storage)
+		req := request()
+		req.Force = force
+		storage.cleanupN, storage.remainN = 0, 0
+
+		_, err := engine.Teardown(context.Background(), req)
+		if !errors.Is(err, ErrResourcesStuck) || accounts.deleted {
+			t.Fatalf("force=%v: a volume still recorded attached must keep the account, err=%v deleted=%v", force, err, accounts.deleted)
+		}
+	}
+}
+
+func TestConvergeMissingCleanerFailsClosed(t *testing.T) {
+	engine := quietEngine(newFakeAccounts())
+	engine.Storage = nil
+	if _, err := engine.Teardown(context.Background(), request()); !errors.Is(err, errComputeCleanupMissing) {
+		t.Fatalf("compute without storage must refuse to start, got %v", err)
+	}
+}
+
+// cleanStages are compute and storage cleanups of an account that holds nothing.
+func cleanStages() (*scriptedCompute, *scriptedStorage) {
+	return &scriptedCompute{scriptedStage{cleanups: []*accountcleanup.Outcome{outcome()}, remainders: []*accountcleanup.Outcome{outcome()}}},
+		&scriptedStorage{scriptedStage{cleanups: []*accountcleanup.Outcome{outcome()}, remainders: []*accountcleanup.Outcome{outcome()}}}
+}

spinifex/operator/accountlifecycle/cutover_test.go

@@ -0,0 +1,169 @@
+package accountlifecycle
+
+//test:in-package — these drive the engine's unexported stage wiring with
+// scripted compute and storage owners, so each can count what it was asked.
+
+import (
+	"context"
+	"errors"
+	"testing"
+
+	"github.com/mulgadc/spinifex/spinifex/domains/ec2/accountcleanup"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+// failingRemainder answers the compute remainder read until its nth call.
+type failingRemainder struct {
+	ComputeStage
+
+	failOn int
+	calls  int
+}
+
+func (f *failingRemainder) ComputeRemaining(ctx context.Context, accountID string) (*accountcleanup.Outcome, error) {
+	f.calls++
+	if f.calls == f.failOn {
+		return nil, errors.New("owner unreachable")
+	}
+	return f.ComputeStage.ComputeRemaining(ctx, accountID)
+}
+
+func stuckIDs(result *Result) []string {
+	var ids []string
+	for _, stage := range result.Stages {
+		for _, s := range stage.Stuck {
+			ids = append(ids, s.Resource.ID)
+		}
+	}
+	return ids
+}
+
+// An account deletion that was never accepted must reach no terminate and no
+// volume delete: the quiesce refuses first, whatever Force says.
+func TestS7_NoTerminateOrVolumeDeleteWithoutAnAcceptedDeletion(t *testing.T) {
+	for _, force := range []bool{false, true} {
+		accounts := newFakeAccounts()
+		accounts.statusErr = errors.New("account is owned by a newer fence")
+		// The record already reads as ours, so only the quiesce itself stands
+		// between this holder and the terminate.
+		accounts.account.Status = AccountStatusTerminating
+		accounts.account.TerminationOp = &TerminationOp{OperationID: "op-1", Fence: 1}
+		engine := quietEngine(accounts)
+		compute, storage := cleanStages()
+		engine.Compute, engine.Storage = compute, storage
+
+		req := request()
+		req.OperationID, req.Fence, req.Force = "op-1", 1, force
+		_, err := engine.Teardown(context.Background(), req)
+
+		require.Error(t, err)
+		assert.Zero(t, compute.cleanupN, "force=%v: no terminate", force)
+		assert.Zero(t, storage.cleanupN, "force=%v: no volume delete", force)
+		assert.False(t, accounts.deleted)
+	}
+}
+
+// With the compute and storage owners missing the teardown refuses before it
+// quiesces anything, so a mis-wired gateway cannot half-delete an account.
+func TestS7_MissingComputeOwnerRefusesBeforeAnyStageRuns(t *testing.T) {
+	accounts := newFakeAccounts()
+	engine := quietEngine(accounts)
+	engine.Compute = nil
+
+	_, err := engine.Teardown(context.Background(), request())
+
+	require.ErrorIs(t, err, errComputeCleanupMissing)
+	assert.Empty(t, accounts.quiesces)
+	assert.False(t, accounts.deleted)
+}
+
+// A resource that appears after the loop ended, while the later stages ran,
+// still keeps the account, and it is named.
+func TestFinalRereadKeepsTheAccountWhenAnInstanceArrivesLate(t *testing.T) {
+	accounts := newFakeAccounts()
+	engine := quietEngine(accounts)
+	engine.Compute = &scriptedCompute{scriptedStage{
+		cleanups:   []*accountcleanup.Outcome{outcome()},
+		remainders: []*accountcleanup.Outcome{outcome(), outcome(item(accountcleanup.KindInstance, "i-late", "running", accountcleanup.DispositionResidue, accountcleanup.ReasonRecordRemains))},
+	}}
+
+	result, err := engine.Teardown(context.Background(), request())
+
+	require.ErrorIs(t, err, ErrResourcesStuck)
+	assert.False(t, accounts.deleted)
+	assert.Contains(t, stuckIDs(result), "i-late")
+}
+
+// A final read that cannot be answered proves nothing, so the account stays.
+func TestFinalRereadKeepsTheAccountWhenTheReadFails(t *testing.T) {
+	accounts := newFakeAccounts()
+	engine := quietEngine(accounts)
+	compute, _ := cleanStages()
+	engine.Compute = &failingRemainder{ComputeStage: compute, failOn: 2}
+
+	result, err := engine.Teardown(context.Background(), request())
+
+	require.ErrorIs(t, err, ErrResourcesStuck)
+	assert.False(t, accounts.deleted)
+	assert.Contains(t, stuckIDs(result), StageAttachments.String())
+}
+
+// An ECS, EKS or RDS owner that cannot finish deleting may be running on these
+// instances, so none is terminated under it.
+func TestStuckComputeOwnerLeavesTheInstancesAlone(t *testing.T) {
+	accounts := newFakeAccounts()
+	cluster := &fakeReaper{
+		kind: "ecs-cluster", stage: StageCompute, blockedUntilForce: true,
+		resources: []Resource{{Kind: "ecs-cluster", ID: "web"}},
+	}
+	engine := quietEngine(accounts, cluster)
+	compute, _ := cleanStages()
+	engine.Compute = compute
+
+	result, err := engine.Teardown(context.Background(), request())
+
+	require.ErrorIs(t, err, ErrResourcesStuck)
+	assert.Zero(t, compute.cleanupN, "no instance is terminated while the owner holds resources")
+	assert.False(t, accounts.deleted)
+	assert.Contains(t, stuckIDs(result), "web")
+}
+
+// The dry run reports what the actual deletion would destroy.
+func TestDryRunListsInstancesSpotRequestsAndVolumes(t *testing.T) {
+	accounts := newFakeAccounts()
+	engine := quietEngine(accounts)
+	engine.Compute = &scriptedCompute{scriptedStage{
+		cleanups: []*accountcleanup.Outcome{outcome()},
+		remainders: []*accountcleanup.Outcome{outcome(
+			item(accountcleanup.KindInstance, "i-1", "running", accountcleanup.DispositionPending, ""),
+			item(accountcleanup.KindSpotRequest, "sir-1", "open", accountcleanup.DispositionPending, ""),
+		)},
+	}}
+	engine.Storage = &scriptedStorage{scriptedStage{
+		cleanups:   []*accountcleanup.Outcome{outcome()},
+		remainders: []*accountcleanup.Outcome{outcome(item(accountcleanup.KindVolume, "vol-1", "available", accountcleanup.DispositionPending, ""))},
+	}}
+
+	result, err := engine.DryRun(context.Background(), testAccountID)
+
+	require.NoError(t, err)
+	var ids []string
+	for _, stage := range result.Stages {
+		for _, r := range stage.Deleted {
+			ids = append(ids, r.ID)
+		}
+	}
+	assert.ElementsMatch(t, []string{"i-1", "sir-1", "vol-1"}, ids)
+}
+
+// A dry run that cannot read the owners fails rather than listing less.
+func TestDryRunFailsWhenTheComputeOwnerCannotBeRead(t *testing.T) {
+	engine := quietEngine(newFakeAccounts())
+	compute, _ := cleanStages()
+	engine.Compute = &failingRemainder{ComputeStage: compute, failOn: 1}
+
+	_, err := engine.DryRun(context.Background(), testAccountID)
+
+	require.Error(t, err)
+}

spinifex/operator/accountlifecycle/ec2_platform_inventory_test.go

@@ -16,6 +16,7 @@ import (
 	"github.com/aws/aws-sdk-go/service/s3"
 	ec2v1 "github.com/mulgadc/spinifex/contracts/ec2/v1"
 	"github.com/mulgadc/spinifex/internal/testkit"
+	"github.com/mulgadc/spinifex/spinifex/domains/ec2/accountcleanup"
 	"github.com/mulgadc/spinifex/spinifex/domains/ec2/accountinventory"
 	iamdomain "github.com/mulgadc/spinifex/spinifex/domains/iam"
 	awserrors "github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
@@ -209,6 +210,21 @@ func TestNoLegacyReaperOwnsAnL09Kind(t *testing.T) {
 	}
 }
 
+// Instances, spot requests and volumes are the EC2 compute and storage
+// owners', so no legacy reaper may list or delete them either.
+func TestNoLegacyReaperOwnsAnL03L05Kind(t *testing.T) {
+	var reapers []Reaper
+	reapers = append(reapers, ECSReapers(nil)...)
+	reapers = append(reapers, EKSReapers(nil)...)
+	reapers = append(reapers, RDSReapers(nil)...)
+	reapers = append(reapers, ELBv2Reapers(nil)...)
+	reapers = append(reapers, EC2ExtraReapers(nil)...)
+	reapers = append(reapers, EC2Reapers(nil, 1, nil)...)
+	for _, reaper := range reapers {
+		assert.NotContains(t, []string{accountcleanup.KindInstance, accountcleanup.KindSpotRequest, accountcleanup.KindVolume}, reaper.Kind())
+	}
+}
+
 // A caller deadline shorter than the adapter's own timeout ends the request
 // at that deadline, and a context already done sends nothing.
 func TestEC2PlatformInventoryClientHonoursTheCallersDeadline(t *testing.T) {

spinifex/operator/accountlifecycle/engine.go

@@ -7,9 +7,11 @@ import (
 	"log/slog"
 	"slices"
 	"sort"
+	"strings"
 	"time"
 
 	operatorv1 "github.com/mulgadc/spinifex/contracts/operator/v1"
+	"github.com/mulgadc/spinifex/spinifex/domains/ec2/accountcleanup"
 )
 
 // AccountStatusTerminating mirrors iamdomain.AccountStatusTerminating.
@@ -41,6 +43,21 @@ type Engine struct {
 	// OnExercised receives what a non-finalizing run did after quiescence.
 	OnExercised func(ctx context.Context, result operatorv1.ExerciseResult)
 
+	// Compute and Storage are EC2's instance, spot request and volume cleanup.
+	// When set, Teardown repeats them within Converge's bounds in place of the
+	// legacy attachments and storage stages.
+	Compute ComputeStage
+	Storage StorageStage
+
+	// Converge bounds the repeat loop. The zero value uses DefaultConverge.
+	Converge Converge
+
+	// OnPass reports each finished pass of the loop so the caller can persist progress.
+	OnPass func(ctx context.Context, pass int)
+
+	// wait replaces the backoff delay in tests.
+	wait func(ctx context.Context, d time.Duration) error
+
 	// ec2Inventory is EC2's read-only platform inventory: Inventory lists the
 	// L09 kinds from it, and the quiescence evaluation and final re-read use it.
 	ec2Inventory ec2PlatformInventory
@@ -104,11 +121,44 @@ func (e *Engine) Inventory(ctx context.Context, accountID string) (*Result, erro
 			}
 			stageResult.Deleted = append(stageResult.Deleted, listed...)
 		}
+		if stage == StageAttachments || stage == StageStorage {
+			listed, err := e.listComputeStorage(ctx, stage, accountID)
+			if err != nil {
+				return nil, err
+			}
+			stageResult.Deleted = append(stageResult.Deleted, listed...)
+		}
 		result.Stages = append(result.Stages, stageResult)
 	}
 	return result, nil
 }
 
+// listComputeStorage reads what the EC2 compute and storage owners still hold.
+// An unanswered or incomplete read is an error, never a shorter list.
+func (e *Engine) listComputeStorage(ctx context.Context, stage Stage, accountID string) ([]Resource, error) {
+	if e.Compute == nil || e.Storage == nil {
+		return nil, errComputeCleanupMissing
+	}
+	var out *accountcleanup.Outcome
+	var err error
+	if stage == StageAttachments {
+		out, err = e.Compute.ComputeRemaining(ctx, accountID)
+	} else {
+		out, err = e.Storage.StorageRemaining(ctx, accountID)
+	}
+	if err != nil {
+		return nil, fmt.Errorf("inventory %s: %w", stage, err)
+	}
+	if len(out.IncompleteReasons) > 0 {
+		return nil, fmt.Errorf("inventory %s incomplete: %s", stage, strings.Join(out.IncompleteReasons, ","))
+	}
+	var found []Resource
+	for _, it := range out.Items {
+		found = append(found, Resource{Kind: it.Kind, ID: it.ResourceID, Detail: it.State})
+	}
+	return found, nil
+}
+
 // DryRun is the explicit operator dry run: exactly what Inventory returns.
 func (e *Engine) DryRun(ctx context.Context, accountID string) (*Result, error) {
 	return e.Inventory(ctx, accountID)
@@ -130,6 +180,9 @@ func (e *Engine) Teardown(ctx context.Context, req Request) (*Result, error) {
 	if e.Cleanup == nil || e.ec2Inventory == nil {
 		return nil, errPlatformCleanupMissing
 	}
+	if e.Compute == nil || e.Storage == nil {
+		return nil, errComputeCleanupMissing
+	}
 
 	generation, err := e.Accounts.QuiesceAccount(req.AccountID, req.OperationID, req.Fence)
 	if err != nil {
@@ -148,6 +201,9 @@ func (e *Engine) Teardown(ctx context.Context, req Request) (*Result, error) {
 	}
 	var platform *platformRun
 	for _, stage := range Stages() {
+		if stage == StageAttachments || stage == StageStorage {
+			continue
+		}
 		if stage == StagePlatform {
 			var err error
 			if platform, err = e.cleanPlatform(ctx, req, generation); err != nil {
@@ -165,8 +221,21 @@ func (e *Engine) Teardown(ctx context.Context, req Request) (*Result, error) {
 		if e.OnStage != nil {
 			e.OnStage(ctx, stageResult)
 		}
+		if stage == StageCompute {
+			if len(stageResult.Stuck) > 0 {
+				// An ECS, EKS or RDS owner that still holds resources may be
+				// running on these instances, so none is terminated under it.
+				e.logger().WarnContext(ctx, "Account teardown: compute owners still hold resources, instances left alone",
+					"accountID", account.AccountID, "stuck", len(stageResult.Stuck))
+				continue
+			}
+			if err := e.convergeCompute(ctx, req, result); err != nil {
+				return result, err
+			}
+		}
 	}
 	e.rereadPlatform(ctx, req, platform, result)
+	e.rereadComputeStorage(ctx, req.AccountID, result)
 
 	if stuck := result.StuckCount(); stuck > 0 {
 		// The account stays TERMINATING on purpose. Removing the record now

spinifex/operator/accountlifecycle/engine_test.go

@@ -130,6 +130,7 @@ func quietEngine(accounts AccountStore, reapers ...Reaper) *Engine {
 	// stage need not stand up the EC2 owners.
 	engine.Cleanup = &stubCleaner{result: cleanResult()}
 	engine.ec2Inventory = &stubInventory{reply: *completeInventory()}
+	engine.Compute, engine.Storage = cleanStages()
 	return engine
 }
 

spinifex/operator/accountlifecycle/exercise.go

@@ -157,6 +157,10 @@ type InstanceReferenceReader struct {
 
 var _ accountcleanup.InstanceReferenceReader = (*InstanceReferenceReader)(nil)
 
+// terminalInstanceStates are the states an instance no longer holds anything
+// in. Anything else is still occupying a volume, an address and a host.
+var terminalInstanceStates = map[string]bool{"terminated": true, "shutting-down": true}
+
 // NewInstanceReferenceReader builds the reader over the configured node set.
 func NewInstanceReferenceReader(nc *nats.Conn, nodeIDs []string) *InstanceReferenceReader {
 	return &InstanceReferenceReader{nc: nc, nodeIDs: nodeIDs, budget: ec2PlatformInventoryBudget}
@@ -165,7 +169,7 @@ func NewInstanceReferenceReader(nc *nats.Conn, nodeIDs []string) *InstanceRefere
 func (r *InstanceReferenceReader) AccountInstanceReferences(ctx context.Context, accountID string) (*accountcleanup.References, error) {
 	ctx, cancel := context.WithTimeout(ctx, r.budget)
 	defer cancel()
-	out, err := ec2instanceapi.DescribeAccountInstancesComplete(ctx, r.nc, r.nodeIDs, accountID)
+	out, err := ec2instanceapi.DescribeAccountInstancesStrict(ctx, r.nc, r.nodeIDs, accountID)
 	if err != nil {
 		return nil, err
 	}

spinifex/operator/accountlifecycle/quiescence_test.go

@@ -241,9 +241,9 @@ func TestTeardownReportsQuiescenceAfterTheQuiesceAndBeforeTheFirstStage(t *testi
 	assert.True(t, accounts.deleted)
 }
 
-// A fence the read-back does not confirm must not spend an inventory call, and a
-// failed read-back must not stop the teardown: this step records and gates nothing.
-func TestTeardownRecordsAFailedReadbackAndStillRunsEveryStage(t *testing.T) {
+// A fence the read-back does not confirm must not spend an inventory call, and
+// the convergence loop must not run a pass behind a fence it cannot read.
+func TestTeardownRecordsAFailedReadbackAndRunsNoPassBehindIt(t *testing.T) {
 	accounts := newFakeAccounts()
 	engine := quietEngine(accounts)
 	inventory := &stubInventory{reply: *completeInventory()}
@@ -257,11 +257,11 @@ func TestTeardownRecordsAFailedReadbackAndStillRunsEveryStage(t *testing.T) {
 	req.OperationID, req.Fence = "op-1", 1
 	_, err := engine.Teardown(context.Background(), req)
 
-	require.NoError(t, err)
+	require.Error(t, err)
 	require.Len(t, got.Unresolved, 1)
 	assert.Equal(t, operatorv1.ReasonIAMReadbackFailed, got.Unresolved[0].Reason)
-	assert.Equal(t, 1, inventory.calls, "only the final re-read: no evaluation reads inventory behind an unconfirmed fence")
-	assert.True(t, accounts.deleted)
+	assert.Equal(t, 0, inventory.calls, "no evaluation reads inventory behind an unconfirmed fence")
+	assert.False(t, accounts.deleted)
 }
 
 // readbackFails serves the pre-quiesce read and fails the one after it.

spinifex/operator/accountlifecycle/realcleanup_test.go

@@ -0,0 +1,104 @@
+package accountlifecycle
+
+import (
+	"context"
+	"errors"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"github.com/aws/aws-sdk-go/service/ec2"
+	"github.com/mulgadc/spinifex/spinifex/domains/ec2/accountcleanup"
+	"github.com/mulgadc/spinifex/spinifex/domains/ec2/accountinventory"
+	ec2instance "github.com/mulgadc/spinifex/spinifex/domains/ec2/instance"
+	ec2spotinstance "github.com/mulgadc/spinifex/spinifex/domains/ec2/spotinstance"
+)
+
+// Stores that answer as an account does once its instance record has expired:
+// no instance, no outstanding teardown, no spot request, no tag residue.
+type emptyComputeStores struct{}
+
+func (emptyComputeStores) ScanAccount(context.Context, string) (*ec2spotinstance.AccountScan, error) {
+	return &ec2spotinstance.AccountScan{}, nil
+}
+
+func (emptyComputeStores) CancelSpotInstanceRequests(context.Context, *ec2.CancelSpotInstanceRequestsInput, string) (*ec2.CancelSpotInstanceRequestsOutput, error) {
+	return &ec2.CancelSpotInstanceRequestsOutput{}, nil
+}
+
+func (emptyComputeStores) AccountInstances(context.Context, string) (*ec2.DescribeInstancesOutput, error) {
+	return &ec2.DescribeInstancesOutput{}, nil
+}
+
+func (emptyComputeStores) AccountTeardownOutstanding(string) ([]ec2instance.OutstandingInstance, error) {
+	return nil, nil
+}
+
+func (emptyComputeStores) AccountManagedInstances(context.Context, string) (map[string]bool, error) {
+	return map[string]bool{}, nil
+}
+
+func (emptyComputeStores) ScanAccountResourceIDs(context.Context, string, ...string) ([]string, error) {
+	return nil, nil
+}
+
+func (emptyComputeStores) DeleteAllTags(context.Context, string, string) error { return nil }
+
+type heldVolumes struct{ items []accountinventory.VolumeItem }
+
+func (h heldVolumes) VolumeInventory(context.Context, string) ([]accountinventory.VolumeItem, error) {
+	return h.items, nil
+}
+
+type countingEC2 struct{ terminates, deletes atomic.Int32 }
+
+func (c *countingEC2) TerminateInstance(context.Context, string, string) error {
+	c.terminates.Add(1)
+	return nil
+}
+
+func (c *countingEC2) DeleteVolume(context.Context, string, string) error {
+	c.deletes.Add(1)
+	return nil
+}
+
+// ADR-0003:S7. The instance's terminated record has expired and the instance is
+// absent from the inventory, but its volume is still recorded in-use and
+// attached. With the real compute and storage cleanup bound to these stores,
+// nothing is deleted whatever Force says, the account is not removed, and the
+// volume is named.
+func TestS7_ExpiredTerminatedRecordAndAttachedVolumeDeleteNothingAndKeepTheAccount(t *testing.T) {
+	for _, force := range []bool{false, true} {
+		stores, ec2ops := emptyComputeStores{}, &countingEC2{}
+		accounts := newFakeAccounts()
+		engine := quietEngine(accounts)
+		engine.Compute = accountcleanup.NewCompute(stores, stores, ec2ops, stores, stores)
+		engine.Storage = accountcleanup.NewStorage(
+			heldVolumes{items: []accountinventory.VolumeItem{{VolumeID: "vol-held", State: "in-use", AttachedInstance: "i-gone", DeleteOnTermination: true}}},
+			ec2ops, stores, stores, stores, stores)
+		engine.wait = func(ctx context.Context, _ time.Duration) error { return ctx.Err() }
+
+		req := request()
+		req.Force = force
+		result, err := engine.Teardown(context.Background(), req)
+
+		if !errors.Is(err, ErrResourcesStuck) {
+			t.Fatalf("force=%v: a held volume must end the teardown stuck, got %v", force, err)
+		}
+		if ec2ops.deletes.Load() != 0 || ec2ops.terminates.Load() != 0 {
+			t.Fatalf("force=%v: nothing may be deleted or terminated, deletes=%d terminates=%d", force, ec2ops.deletes.Load(), ec2ops.terminates.Load())
+		}
+		if accounts.deleted {
+			t.Fatalf("force=%v: the account must not be removed", force)
+		}
+		named := false
+		for _, stage := range result.Stages {
+			for _, s := range stage.Stuck {
+				named = named || s.Resource.ID == "vol-held"
+			}
+		}
+		if !named {
+			t.Fatalf("force=%v: the stuck volume must be named in the result", force)
+		}
+	}
+}

spinifex/operator/accountlifecycle/reapers_ec2.go

@@ -2,11 +2,8 @@ package accountlifecycle
 
 import (
 	"context"
-	"fmt"
-	"github.com/mulgadc/spinifex/spinifex/foundation/messaging/nats"
 	"slices"
 	"strings"
-	"time"
 
 	"github.com/aws/aws-sdk-go/aws"
 	"github.com/aws/aws-sdk-go/service/ec2"
@@ -14,10 +11,8 @@ import (
 	ec2eipapi "github.com/mulgadc/spinifex/spinifex/domains/ec2/awsapi/eip"
 	ec2igwapi "github.com/mulgadc/spinifex/spinifex/domains/ec2/awsapi/igw"
 	ec2imageapi "github.com/mulgadc/spinifex/spinifex/domains/ec2/awsapi/image"
-	ec2instanceapi "github.com/mulgadc/spinifex/spinifex/domains/ec2/awsapi/instance"
 	ec2routetableapi "github.com/mulgadc/spinifex/spinifex/domains/ec2/awsapi/routetable"
 	ec2snapshotapi "github.com/mulgadc/spinifex/spinifex/domains/ec2/awsapi/snapshot"
-	ec2volumeapi "github.com/mulgadc/spinifex/spinifex/domains/ec2/awsapi/volume"
 	ec2vpcapi "github.com/mulgadc/spinifex/spinifex/domains/ec2/awsapi/vpc"
 	"github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
 	"github.com/nats-io/nats.go"
@@ -30,13 +25,10 @@ import (
 // accounting correct without teardown having to reimplement any of it.
 func EC2Reapers(nc *nats.Conn, expectedNodes int, nodeIDs []string) []Reaper {
 	return []Reaper{
-		&instanceReaper{nc: nc, nodeIDs: nodeIDs},
-
 		// Snapshots before volumes: a volume with a live snapshot refuses to
 		// delete, because snapshot-backed clones read from its chunk prefix.
 		&snapshotReaper{nc: nc},
 		&imageReaper{nc: nc},
-		&volumeReaper{nc: nc, nodeIDs: nodeIDs},
 
 		// Reverse of creation order. The VPC cannot go until everything
 		// inside it has, and the gateway enforces that.
@@ -55,107 +47,6 @@ func EC2Reapers(nc *nats.Conn, expectedNodes int, nodeIDs []string) []Reaper {
 	}
 }
 
-// terminalInstanceStates are the states an instance no longer holds anything
-// in. Anything else is still occupying a volume, an address and a host.
-var terminalInstanceStates = map[string]bool{"terminated": true, "shutting-down": true}
-
-type instanceReaper struct {
-	nc *nats.Conn
-	// nodeIDs is the configured node set; all must answer before an empty
-	// listing counts as drained.
-	nodeIDs []string
-
-	// describeOpts tunes the inventory fan-out; production leaves it empty.
-	describeOpts []ec2instanceapi.DescribeOption
-}
-
-func (r *instanceReaper) Kind() string { return "instance" }
-func (r *instanceReaper) Stage() Stage { return StageCompute }
-
-func (r *instanceReaper) List(ctx context.Context, accountID string) ([]Resource, error) {
-	// Complete or nothing: a partial answer would read as "no instances left"
-	// and let storage deletion start while a guest still holds a volume.
-	out, err := ec2instanceapi.DescribeAccountInstancesComplete(ctx, r.nc, r.nodeIDs, accountID, r.describeOpts...)
-	if err != nil {
-		return nil, err
-	}
-
-	var found []Resource
-	for _, reservation := range out.Reservations {
-		for _, instance := range reservation.Instances {
-			if instance == nil || instance.InstanceId == nil {
-				continue
-			}
-			state := ""
-			if instance.State != nil && instance.State.Name != nil {
-				state = *instance.State.Name
-			}
-			if terminalInstanceStates[state] {
-				continue
-			}
-			found = append(found, Resource{Kind: r.Kind(), ID: *instance.InstanceId, Detail: state})
-		}
-	}
-	return found, nil
-}
-
-func (r *instanceReaper) Delete(ctx context.Context, accountID string, resource Resource, _ bool) error {
-	_, err := ec2instanceapi.TerminateInstances(ctx, &ec2.TerminateInstancesInput{
-		InstanceIds: []*string{aws.String(resource.ID)},
-	}, r.nc, accountID)
-	return ignoreAlreadyGone(err)
-}
-
-type volumeReaper struct {
-	nc *nats.Conn
-	// nodeIDs is the configured node set; every one must answer, by identity,
-	// before a volume's holder can be proven gone or the delete guard passes.
-	nodeIDs []string
-}
-
-func (r *volumeReaper) Kind() string { return "volume" }
-func (r *volumeReaper) Stage() Stage { return StageStorage }
-
-func (r *volumeReaper) List(ctx context.Context, accountID string) ([]Resource, error) {
-	out, err := ec2volumeapi.DescribeVolumes(ctx, &ec2.DescribeVolumesInput{}, r.nc, accountID)
-	if err != nil {
-		return nil, err
-	}
-
-	var found []Resource
-	for _, volume := range out.Volumes {
-		if volume == nil || volume.VolumeId == nil {
-			continue
-		}
-		resource := Resource{Kind: r.Kind(), ID: *volume.VolumeId}
-		// The attachment is the reason a volume will not go, so carry it: it
-		// is the difference between "stuck" and "stuck on i-abc".
-		for _, attachment := range volume.Attachments {
-			if attachment != nil && attachment.InstanceId != nil {
-				resource.Detail = "attached to " + *attachment.InstanceId
-				break
-			}
-		}
-		found = append(found, resource)
-	}
-	return found, nil
-}
-
-func (r *volumeReaper) Delete(ctx context.Context, accountID string, resource Resource, force bool) error {
-	_, err := ec2volumeapi.DeleteVolume(ctx, &ec2.DeleteVolumeInput{
-		VolumeId: aws.String(resource.ID),
-	}, r.nc, r.nodeIDs, accountID)
-	if isAlreadyGone(err) {
-		return nil
-	}
-	if !force {
-		return err
-	}
-	// Force path: the ordinary delete refuses an attached volume, and a holder
-	// that is gone but still on the record would otherwise strand it forever.
-	return forceDeleteVolume(ctx, r.nc, r.nodeIDs, accountID, resource.ID, err)
-}
-
 type snapshotReaper struct{ nc *nats.Conn }
 
 func (r *snapshotReaper) Kind() string { return "snapshot" }
@@ -599,131 +490,3 @@ func ignoreAlreadyGone(err error) error {
 	}
 	return err
 }
-
-// forceDetachSubject clears a volume's attachment in the control plane only.
-// It is answered by any node, unlike the ordinary detach, which routes to the
-// host running the instance and so cannot help when that host is the problem.
-const forceDetachSubject = "ec2.ForceDetachVolume"
-
-// forceDetachTimeout bounds the metadata-only detach. It touches one document
-// and needs nothing from the guest, so a slow answer means the cluster is
-// unwell rather than that the work is large.
-const forceDetachTimeout = 30 * time.Second
-
-// forceDeleteVolume clears an attachment the ordinary delete refuses to
-// override, then deletes, but only once every instance the volume names is
-// proven absent. Anything short of proof leaves the volume and its record alone.
-func forceDeleteVolume(ctx context.Context, nc *nats.Conn, nodeIDs []string, accountID, volumeID string, refusal error) error {
-	holders, err := volumeHolders(ctx, nc, accountID, volumeID)
-	if err != nil {
-		if isAlreadyGone(err) {
-			return nil
-		}
-		return fmt.Errorf("volume %s retained: cannot read its attachment: %w", volumeID, err)
-	}
-	if len(holders) == 0 {
-		// No holder on record, so the refusal is not an attachment and
-		// clearing the record would change nothing but the evidence.
-		return refusal
-	}
-	for _, instanceID := range holders {
-		if err := proveInstanceAbsent(ctx, nc, nodeIDs, accountID, instanceID); err != nil {
-			return fmt.Errorf("volume %s retained: %w", volumeID, err)
-		}
-	}
-
-	_, err = natsmsg.NATSRequest[ec2.VolumeAttachment](ctx, nc, forceDetachSubject,
-		&ec2.DetachVolumeInput{VolumeId: aws.String(volumeID), Force: aws.Bool(true)},
-		forceDetachTimeout, accountID)
-	if err != nil && !isAlreadyGone(err) {
-		return fmt.Errorf("force detach %s: %w", volumeID, err)
-	}
-
-	// The ordinary delete re-checks that the record is unattached and that no
-	// live instance maps the volume, so a re-attach in between still refuses.
-	_, err = ec2volumeapi.DeleteVolume(ctx, &ec2.DeleteVolumeInput{
-		VolumeId: aws.String(volumeID),
-	}, nc, nodeIDs, accountID)
-	return ignoreAlreadyGone(err)
-}
-
-// volumeHolders returns the instances the volume's own record says hold it,
-// read fresh rather than from the listing that queued the delete.
-func volumeHolders(ctx context.Context, nc *nats.Conn, accountID, volumeID string) ([]string, error) {
-	out, err := ec2volumeapi.DescribeVolumes(ctx, &ec2.DescribeVolumesInput{
-		VolumeIds: []*string{aws.String(volumeID)},
-	}, nc, accountID)
-	if err != nil {
-		return nil, err
-	}
-	var holders []string
-	for _, volume := range out.Volumes {
-		if volume == nil || aws.StringValue(volume.VolumeId) != volumeID {
-			continue
-		}
-		for _, attachment := range volume.Attachments {
-			if attachment != nil && aws.StringValue(attachment.InstanceId) != "" {
-				holders = append(holders, aws.StringValue(attachment.InstanceId))
-			}
-		}
-	}
-	return holders, nil
-}
-
-// proveInstanceAbsent accepts only a definitive InvalidInstanceID.NotFound or
-// a terminated copy from a describe naming the instance, then confirms both on
-// a listing every configured node answered by identity. Anything else is not proof.
-func proveInstanceAbsent(ctx context.Context, nc *nats.Conn, nodeIDs []string, accountID, instanceID string) error {
-	if len(nodeIDs) == 0 {
-		return fmt.Errorf("cannot prove holding instance %s is gone: the configured node set is unknown", instanceID)
-	}
-	named, err := ec2instanceapi.DescribeInstancesChecked(ctx, &ec2.DescribeInstancesInput{
-		InstanceIds: []*string{aws.String(instanceID)},
-	}, nc, 0, nodeIDs, accountID)
-	if err != nil {
-		if code, ok := awserrors.ResolveErrorCode(err); !ok || code != awserrors.ErrorInvalidInstanceIDNotFound {
-			return fmt.Errorf("cannot prove holding instance %s is gone: %w", instanceID, err)
-		}
-	} else {
-		found, err := requireTerminated(named.Reservations, instanceID)
-		if err != nil {
-			return err
-		}
-		if !found {
-			return fmt.Errorf("cannot prove holding instance %s is gone: the describe naming it returned neither the instance nor NotFound", instanceID)
-		}
-	}
-
-	// The named describe stops collecting once the id turns up, and a lone
-	// node's 4xx can surface as NotFound, so neither answer is complete alone.
-	listing, err := ec2instanceapi.DescribeAccountInstancesComplete(ctx, nc, nodeIDs, accountID)
-	if err != nil {
-		return fmt.Errorf("cannot prove holding instance %s is gone: %w", instanceID, err)
-	}
-	_, err = requireTerminated(listing.Reservations, instanceID)
-	return err
-}
-
-// requireTerminated reports whether instanceID appears in reservations, and
-// fails if any copy of it is in a state other than terminated.
-func requireTerminated(reservations []*ec2.Reservation, instanceID string) (found bool, err error) {
-	for _, reservation := range reservations {
-		if reservation == nil {
-			continue
-		}
-		for _, instance := range reservation.Instances {
-			if instance == nil || aws.StringValue(instance.InstanceId) != instanceID {
-				continue
-			}
-			found = true
-			state := "in an unknown state"
-			if instance.State != nil && aws.StringValue(instance.State.Name) != "" {
-				state = aws.StringValue(instance.State.Name)
-			}
-			if state != ec2.InstanceStateNameTerminated {
-				return true, fmt.Errorf("holding instance %s is still %s", instanceID, state)
-			}
-		}
-	}
-	return found, nil
-}

spinifex/operator/accountlifecycle/reapers_ec2_extra.go

@@ -7,66 +7,21 @@ import (
 	"github.com/aws/aws-sdk-go/service/ec2"
 	ec2eigw "github.com/mulgadc/spinifex/spinifex/domains/ec2/eigw"
 	ec2natgw "github.com/mulgadc/spinifex/spinifex/domains/ec2/natgw"
-	ec2spotinstance "github.com/mulgadc/spinifex/spinifex/domains/ec2/spotinstance"
 	"github.com/nats-io/nats.go"
 )
 
 // EC2ExtraReapers returns the EC2 reapers that hang off the VPC rather than off
 // an instance.
 //
-// Spot requests come first and in the compute stage: a live request can launch
-// a replacement after the instance reaper has drained, so leaving one running
-// makes teardown race its own fulfilment rather than merely leak.
+// Spot requests are not here: the EC2 compute cleanup cancels them before it
+// terminates anything, so none can launch a replacement.
 func EC2ExtraReapers(nc *nats.Conn) []Reaper {
 	return []Reaper{
-		&spotRequestReaper{svc: ec2spotinstance.NewNATSSpotInstanceService(nc)},
 		&natGatewayReaper{svc: ec2natgw.NewNATSNatGatewayService(nc)},
 		&egressOnlyIGWReaper{svc: ec2eigw.NewNATSEgressOnlyIGWService(nc)},
 	}
 }
 
-// terminalSpotStates are the states a request can no longer launch from.
-// Anything else may still produce an instance.
-var terminalSpotStates = map[string]bool{"cancelled": true, "closed": true, "failed": true}
-
-type spotRequestReaper struct {
-	svc ec2spotinstance.SpotInstanceService
-}
-
-func (r *spotRequestReaper) Kind() string { return "spot-request" }
-func (r *spotRequestReaper) Stage() Stage { return StageCompute }
-
-func (r *spotRequestReaper) List(ctx context.Context, accountID string) ([]Resource, error) {
-	out, err := r.svc.DescribeSpotInstanceRequests(ctx, &ec2.DescribeSpotInstanceRequestsInput{}, accountID)
-	if err != nil {
-		return nil, err
-	}
-
-	var found []Resource
-	for _, request := range out.SpotInstanceRequests {
-		if request == nil || request.SpotInstanceRequestId == nil {
-			continue
-		}
-		state := aws.StringValue(request.State)
-		if terminalSpotStates[state] {
-			continue
-		}
-		found = append(found, Resource{
-			Kind:   r.Kind(),
-			ID:     aws.StringValue(request.SpotInstanceRequestId),
-			Detail: state,
-		})
-	}
-	return found, nil
-}
-
-func (r *spotRequestReaper) Delete(ctx context.Context, accountID string, resource Resource, _ bool) error {
-	_, err := r.svc.CancelSpotInstanceRequests(ctx, &ec2.CancelSpotInstanceRequestsInput{
-		SpotInstanceRequestIds: []*string{aws.String(resource.ID)},
-	}, accountID)
-	return ignoreAlreadyGone(err)
-}
-
 // deletedNatGatewayStates are the states a NAT gateway has already released its
 // address and its subnet placement in.
 var deletedNatGatewayStates = map[string]bool{"deleted": true}

spinifex/operator/accountlifecycle/reapers_ec2_extra_test.go

@@ -12,36 +12,11 @@ import (
 	"github.com/aws/aws-sdk-go/service/ec2"
 	ec2eigw "github.com/mulgadc/spinifex/spinifex/domains/ec2/eigw"
 	ec2natgw "github.com/mulgadc/spinifex/spinifex/domains/ec2/natgw"
-	ec2spotinstance "github.com/mulgadc/spinifex/spinifex/domains/ec2/spotinstance"
 	awserrors "github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 )
 
-// The embedded interfaces are deliberately nil: a reaper that calls anything
-// beyond the two methods it is supposed to use panics rather than passing.
-type fakeSpot struct {
-	ec2spotinstance.SpotInstanceService
-
-	requests  []*ec2.SpotInstanceRequest
-	cancelled []string
-	err       error
-}
-
-func (f *fakeSpot) DescribeSpotInstanceRequests(_ context.Context, _ *ec2.DescribeSpotInstanceRequestsInput, _ string) (*ec2.DescribeSpotInstanceRequestsOutput, error) {
-	return &ec2.DescribeSpotInstanceRequestsOutput{SpotInstanceRequests: f.requests}, nil
-}
-
-func (f *fakeSpot) CancelSpotInstanceRequests(_ context.Context, in *ec2.CancelSpotInstanceRequestsInput, _ string) (*ec2.CancelSpotInstanceRequestsOutput, error) {
-	if f.err != nil {
-		return nil, f.err
-	}
-	for _, id := range in.SpotInstanceRequestIds {
-		f.cancelled = append(f.cancelled, aws.StringValue(id))
-	}
-	return &ec2.CancelSpotInstanceRequestsOutput{}, nil
-}
-
 type fakeNatGateways struct {
 	ec2natgw.NatGatewayService
 
@@ -82,42 +57,6 @@ func (f *fakeEgressGateways) DeleteEgressOnlyInternetGateway(_ context.Context,
 	return &ec2.DeleteEgressOnlyInternetGatewayOutput{}, nil
 }
 
-// A cancelled request cannot launch anything, so listing it would leave the
-// stage waiting on a row that never clears — spot requests are never removed
-// from the account, only moved to a terminal state.
-func TestSpotRequestReaperSkipsTerminalRequests(t *testing.T) {
-	svc := &fakeSpot{requests: []*ec2.SpotInstanceRequest{
-		{SpotInstanceRequestId: aws.String("sir-open"), State: aws.String("open")},
-		{SpotInstanceRequestId: aws.String("sir-active"), State: aws.String("active")},
-		{SpotInstanceRequestId: aws.String("sir-cancelled"), State: aws.String("cancelled")},
-		{SpotInstanceRequestId: aws.String("sir-closed"), State: aws.String("closed")},
-		{SpotInstanceRequestId: aws.String("sir-failed"), State: aws.String("failed")},
-	}}
-	reaper := &spotRequestReaper{svc: svc}
-
-	assert.Equal(t, StageCompute, reaper.Stage())
-
-	found, err := reaper.List(t.Context(), "000000000002")
-	require.NoError(t, err)
-	require.Len(t, found, 2)
-	assert.Equal(t, "sir-open", found[0].ID)
-	assert.Equal(t, "open", found[0].Detail)
-	assert.Equal(t, "sir-active", found[1].ID)
-
-	require.NoError(t, reaper.Delete(t.Context(), "000000000002", found[0], false))
-	assert.Equal(t, []string{"sir-open"}, svc.cancelled)
-}
-
-// A request cancelled by someone else between the listing and the delete is
-// the ordinary second-pass case, not a failure.
-func TestSpotRequestReaperTreatsAMissingRequestAsCancelled(t *testing.T) {
-	svc := &fakeSpot{err: errors.New("InvalidSpotInstanceRequestID.NotFound: no such request")}
-	reaper := &spotRequestReaper{svc: svc}
-
-	err := reaper.Delete(t.Context(), "000000000002", Resource{ID: "sir-gone"}, false)
-	assert.NoError(t, err)
-}
-
 // A deleting gateway still holds its address and its subnet, so it has to keep
 // listing until it is actually gone or the address reaper runs too early.
 func TestNatGatewayReaperKeepsDeletingGateways(t *testing.T) {

spinifex/operator/accountlifecycle/reapers_ec2_inventory_test.go

@@ -1,198 +0,0 @@
-package accountlifecycle
-
-//test:in-package — the instance reaper is unexported, and whether its listing
-// can be mistaken for an empty account is the safety property under test.
-
-import (
-	"context"
-	"encoding/json"
-	"testing"
-	"time"
-
-	"github.com/aws/aws-sdk-go/aws"
-	"github.com/aws/aws-sdk-go/service/ec2"
-	"github.com/mulgadc/spinifex/internal/testkit"
-	ec2instanceapi "github.com/mulgadc/spinifex/spinifex/domains/ec2/awsapi/instance"
-	"github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
-	"github.com/mulgadc/spinifex/spinifex/foundation/messaging/nats"
-	"github.com/nats-io/nats.go"
-	"github.com/stretchr/testify/assert"
-	"github.com/stretchr/testify/require"
-)
-
-var inventoryNodes = []string{"node-1", "node-2"}
-
-// testInventoryReaper builds the instance reaper for a two-node cluster, with
-// a short fan-out deadline so a silent node costs the test little.
-func testInventoryReaper(nc *nats.Conn) *instanceReaper {
-	return &instanceReaper{nc: nc, nodeIDs: inventoryNodes, describeOpts: []ec2instanceapi.DescribeOption{
-		ec2instanceapi.WithFanoutTimeout(300 * time.Millisecond),
-	}}
-}
-
-// answerAsNode replies on subject the way a daemon does, carrying the node ID
-// header the fan-out attributes frames by.
-func answerAsNode(t *testing.T, nc *nats.Conn, subject, nodeID string, payload []byte) {
-	t.Helper()
-	sub, err := nc.Subscribe(subject, func(msg *nats.Msg) {
-		reply := nats.NewMsg(msg.Reply)
-		reply.Data = payload
-		if nodeID != "" {
-			reply.Header.Set(natsmsg.NodeIDHeader, nodeID)
-		}
-		_ = msg.RespondMsg(reply)
-	})
-	require.NoError(t, err)
-	t.Cleanup(func() { _ = sub.Unsubscribe() })
-}
-
-func describeOutput(t *testing.T, instanceIDs ...string) []byte {
-	t.Helper()
-	out := ec2.DescribeInstancesOutput{Reservations: []*ec2.Reservation{}}
-	if len(instanceIDs) > 0 {
-		reservation := &ec2.Reservation{ReservationId: aws.String("r-" + instanceIDs[0])}
-		for _, id := range instanceIDs {
-			reservation.Instances = append(reservation.Instances, &ec2.Instance{
-				InstanceId: aws.String(id),
-				State:      &ec2.InstanceState{Name: aws.String("running")},
-			})
-		}
-		out.Reservations = append(out.Reservations, reservation)
-	}
-	payload, err := json.Marshal(out)
-	require.NoError(t, err)
-	return payload
-}
-
-// emptyInstanceBuckets answers the stopped and terminated bucket queries with
-// nothing, so only the running-instance fan-out decides completeness.
-func emptyInstanceBuckets(t *testing.T, nc *nats.Conn) {
-	t.Helper()
-	answerAsNode(t, nc, "ec2.DescribeStoppedInstances", "", describeOutput(t))
-	answerAsNode(t, nc, "ec2.DescribeTerminatedInstances", "", describeOutput(t))
-}
-
-// A node that never answers is the timeout case: its running guests are absent
-// from the aggregate, so an empty answer from the rest is not evidence the
-// account has no instances.
-func TestInstanceReaperFailsClosedWhenANodeDoesNotAnswer(t *testing.T) {
-	t.Parallel()
-	_, nc := testutil.StartTestNATS(t)
-	emptyInstanceBuckets(t, nc)
-	answerAsNode(t, nc, "ec2.DescribeInstances", "node-1", describeOutput(t))
-	require.NoError(t, nc.Flush())
-
-	found, err := testInventoryReaper(nc).List(testCtx(t), testAccountID)
-
-	require.Error(t, err, "a listing missing a node must not read as an empty account, got %v", found)
-	code, _, ok := awserrors.ResolveErrorDetail(err)
-	require.True(t, ok)
-	assert.Equal(t, awserrors.ErrorServiceUnavailable, code)
-}
-
-// The partial list is the dangerous shape: the reaper would terminate what it
-// can see and then report the stage drained on the next, equally partial pass.
-func TestInstanceReaperFailsClosedOnAPartialList(t *testing.T) {
-	t.Parallel()
-	_, nc := testutil.StartTestNATS(t)
-	emptyInstanceBuckets(t, nc)
-	answerAsNode(t, nc, "ec2.DescribeInstances", "node-1", describeOutput(t, "i-on-node-1"))
-	require.NoError(t, nc.Flush())
-
-	found, err := testInventoryReaper(nc).List(testCtx(t), testAccountID)
-
-	require.Error(t, err, "a partial listing must not be returned as the inventory, got %v", found)
-}
-
-// An error frame from a node is no better than silence: it says nothing about
-// what that node is running.
-func TestInstanceReaperFailsClosedWhenANodeAnswersWithAnError(t *testing.T) {
-	_, nc := testutil.StartTestNATS(t)
-	emptyInstanceBuckets(t, nc)
-	answerAsNode(t, nc, "ec2.DescribeInstances", "node-1", describeOutput(t))
-	answerAsNode(t, nc, "ec2.DescribeInstances", "node-2", awserrors.GenerateErrorPayload(awserrors.ErrorServerInternal))
-	require.NoError(t, nc.Flush())
-
-	_, err := testInventoryReaper(nc).List(testCtx(t), testAccountID)
-
-	require.Error(t, err)
-}
-
-// Completeness is judged against the configured node set, not a count of
-// replies: two frames from one node while another is silent is still missing
-// a node.
-func TestInstanceReaperFailsClosedWhenOneNodeAnswersTwice(t *testing.T) {
-	t.Parallel()
-	_, nc := testutil.StartTestNATS(t)
-	emptyInstanceBuckets(t, nc)
-	answerAsNode(t, nc, "ec2.DescribeInstances", "node-1", describeOutput(t))
-	answerAsNode(t, nc, "ec2.DescribeInstances", "node-1", describeOutput(t))
-	require.NoError(t, nc.Flush())
-
-	_, err := testInventoryReaper(nc).List(testCtx(t), testAccountID)
-
-	require.Error(t, err)
-}
-
-// Stopped instances live in a bucket, not on a node. A failed bucket query
-// hides them as surely as a silent node hides running ones.
-func TestInstanceReaperFailsClosedWhenTheStoppedBucketFails(t *testing.T) {
-	_, nc := testutil.StartTestNATS(t)
-	answerAsNode(t, nc, "ec2.DescribeStoppedInstances", "", awserrors.GenerateErrorPayload(awserrors.ErrorServerInternal))
-	answerAsNode(t, nc, "ec2.DescribeTerminatedInstances", "", describeOutput(t))
-	answerAsNode(t, nc, "ec2.DescribeInstances", "node-1", describeOutput(t))
-	answerAsNode(t, nc, "ec2.DescribeInstances", "node-2", describeOutput(t))
-	require.NoError(t, nc.Flush())
-
-	_, err := testInventoryReaper(nc).List(testCtx(t), testAccountID)
-
-	require.Error(t, err)
-}
-
-// The healthy cluster still lists: every configured node answered, so the
-// union is the account's inventory.
-func TestInstanceReaperListsEveryNodeWhenAllAnswer(t *testing.T) {
-	_, nc := testutil.StartTestNATS(t)
-	emptyInstanceBuckets(t, nc)
-	answerAsNode(t, nc, "ec2.DescribeInstances", "node-1", describeOutput(t, "i-on-node-1"))
-	answerAsNode(t, nc, "ec2.DescribeInstances", "node-2", describeOutput(t, "i-on-node-2"))
-	require.NoError(t, nc.Flush())
-
-	started := time.Now()
-	found, err := testInventoryReaper(nc).List(testCtx(t), testAccountID)
-	require.NoError(t, err)
-
-	var ids []string
-	for _, resource := range found {
-		ids = append(ids, resource.ID)
-	}
-	assert.ElementsMatch(t, []string{"i-on-node-1", "i-on-node-2"}, ids)
-	assert.Less(t, time.Since(started), 2*time.Second, "a fully answered fan-out must not wait out its deadline")
-}
-
-// End to end through the engine: with a node silent, the compute stage must
-// not read as drained and the account record must survive.
-func TestTeardownDoesNotReportComputeDrainedWhileANodeIsSilent(t *testing.T) {
-	t.Parallel()
-	_, nc := testutil.StartTestNATS(t)
-	emptyInstanceBuckets(t, nc)
-	answerAsNode(t, nc, "ec2.DescribeInstances", "node-1", describeOutput(t))
-	require.NoError(t, nc.Flush())
-
-	accounts := newFakeAccounts()
-	volumes := &fakeReaper{kind: "volume", stage: StageStorage,
-		resources: []Resource{{Kind: "volume", ID: "vol-held-by-node-2"}}}
-	engine := quietEngine(accounts, testInventoryReaper(nc), volumes)
-
-	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
-	defer cancel()
-	result, err := engine.Teardown(ctx, request())
-
-	require.Error(t, err, "teardown must not succeed on an incomplete instance inventory")
-	assert.ErrorContains(t, err, "list instance")
-	assert.False(t, accounts.deleted, "the account record must survive an incomplete inventory")
-	for _, stage := range result.Stages {
-		assert.NotEqual(t, StageCompute, stage.Stage, "the compute stage must not be reported finished")
-	}
-	assert.Equal(t, 0, volumes.deletesAttempted, "storage must not be touched before compute is proven drained")
-}

spinifex/operator/accountlifecycle/reapers_ec2_test.go

@@ -106,29 +106,6 @@ func (c *fakeCluster) fail(subject, code string) {
 	c.errorCode[subject] = code
 }
 
-// handle answers subject from state the test owns, for doubles whose answer
-// must change as the reaper acts — an attachment a force detach clears.
-func (c *fakeCluster) handle(subject string, handler func(data []byte) (any, string)) {
-	c.mu.Lock()
-	defer c.mu.Unlock()
-	c.handlers[subject] = handler
-}
-
-// echoAsNode adds a second reply to every request on subject, attributed to
-// node, so one node can answer twice while another stays silent.
-func (c *fakeCluster) echoAsNode(t *testing.T, subject, node string) {
-	t.Helper()
-	sub, err := c.nc.Subscribe(subject, func(msg *nats.Msg) {
-		out := nats.NewMsg(msg.Reply)
-		out.Data = []byte(`{}`)
-		out.Header.Set(natsmsg.NodeIDHeader, node)
-		_ = msg.RespondMsg(out)
-	})
-	require.NoError(t, err)
-	t.Cleanup(func() { _ = sub.Unsubscribe() })
-	require.NoError(t, c.nc.Flush())
-}
-
 func (c *fakeCluster) called() []string {
 	c.mu.Lock()
 	defer c.mu.Unlock()
@@ -150,13 +127,6 @@ func testCtx(t *testing.T) context.Context {
 	return ctx
 }
 
-// silentClusterCtx outlasts the full fan-out deadline a silent node costs.
-func silentClusterCtx(t *testing.T) context.Context {
-	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
-	t.Cleanup(cancel)
-	return ctx
-}
-
 // The stage graph is what keeps teardown converging, so it is asserted rather
 // than left to the order the constructor happens to list.
 func TestEC2ReapersAreOrderedByDependency(t *testing.T) {
@@ -168,16 +138,10 @@ func TestEC2ReapersAreOrderedByDependency(t *testing.T) {
 	}
 
 	assert.Equal(t, []string{
-		"instance",
-		"snapshot", "image", "volume",
+		"snapshot", "image",
 		"address", "network-interface", "internet-gateway", "route-table", "subnet", "security-group", "vpc",
 	}, kinds)
 
-	// A volume with a live snapshot refuses to delete, so the reverse order
-	// would strand every snapshot-backed volume in the account.
-	assert.Less(t, indexOfKind(reapers, "snapshot"), indexOfKind(reapers, "volume"))
-	assert.Equal(t, StageCompute, reapers[indexOfKind(reapers, "instance")].Stage())
-	assert.Equal(t, StageStorage, reapers[indexOfKind(reapers, "volume")].Stage())
 	assert.Equal(t, StageNetwork, reapers[indexOfKind(reapers, "vpc")].Stage())
 }
 
@@ -190,333 +154,6 @@ func indexOfKind(reapers []Reaper, kind string) int {
 	return -1
 }
 
-// A terminated instance holds nothing. Listing it would keep the compute stage
-// polling until its timeout and then report it stuck, for no reason.
-func TestInstanceReaperSkipsTerminalStates(t *testing.T) {
-	cluster := newFakeCluster(t)
-	cluster.reply("ec2.DescribeInstances", ec2.DescribeInstancesOutput{
-		Reservations: []*ec2.Reservation{{Instances: []*ec2.Instance{
-			{InstanceId: aws.String("i-running"), State: &ec2.InstanceState{Name: aws.String("running")}},
-			{InstanceId: aws.String("i-gone"), State: &ec2.InstanceState{Name: aws.String("terminated")}},
-			{InstanceId: aws.String("i-going"), State: &ec2.InstanceState{Name: aws.String("shutting-down")}},
-			{State: &ec2.InstanceState{Name: aws.String("running")}},
-		}}},
-	})
-
-	reaper := &instanceReaper{nc: cluster.nc, nodeIDs: []string{fakeClusterNode}}
-	found, err := reaper.List(testCtx(t), "000000000042")
-	require.NoError(t, err)
-
-	require.Len(t, found, 1)
-	assert.Equal(t, "i-running", found[0].ID)
-	assert.Equal(t, "running", found[0].Detail)
-}
-
-// Termination is addressed to the node hosting the instance, which is what
-// makes a stuck host the case --force exists for.
-func TestInstanceReaperTerminates(t *testing.T) {
-	cluster := newFakeCluster(t)
-	reaper := &instanceReaper{nc: cluster.nc, nodeIDs: []string{fakeClusterNode}}
-
-	require.NoError(t, reaper.Delete(testCtx(t), "000000000042",
-		Resource{Kind: "instance", ID: "i-0abc0123456789def"}, false))
-
-	assert.Contains(t, cluster.called(), "ec2.cmd.i-0abc0123456789def")
-}
-
-// The attachment is the reason a volume will not go, so it has to reach the
-// stuck report — "stuck" and "stuck on i-abc" are different bug reports.
-func TestVolumeReaperCarriesItsAttachment(t *testing.T) {
-	cluster := newFakeCluster(t)
-	cluster.reply("ec2.DescribeVolumes", ec2.DescribeVolumesOutput{
-		Volumes: []*ec2.Volume{
-			{VolumeId: aws.String("vol-free")},
-			{VolumeId: aws.String("vol-held"), Attachments: []*ec2.VolumeAttachment{
-				{InstanceId: aws.String("i-abc")},
-			}},
-		},
-	})
-
-	reaper := newVolumeReaper(cluster.nc, fakeClusterNode)
-	found, err := reaper.List(testCtx(t), "000000000042")
-	require.NoError(t, err)
-
-	require.Len(t, found, 2)
-	assert.Empty(t, found[0].Detail)
-	assert.Equal(t, "attached to i-abc", found[1].Detail)
-}
-
-// Without --force an attached volume must surface the refusal, not swallow it:
-// a stuck resource is a bug we want reported rather than papered over.
-func TestVolumeReaperReportsRefusalWithoutForce(t *testing.T) {
-	cluster := newFakeCluster(t)
-	cluster.fail("ec2.DeleteVolume", "VolumeInUse")
-
-	reaper := newVolumeReaper(cluster.nc, fakeClusterNode)
-	err := reaper.Delete(testCtx(t), "000000000042", Resource{Kind: "volume", ID: "vol-held"}, false)
-
-	require.Error(t, err)
-	assert.NotContains(t, cluster.called(), forceDetachSubject)
-}
-
-// heldVolume is one volume record as its owner keeps it: the ordinary delete
-// refuses while the record names an instance, and a force detach clears it.
-type heldVolume struct {
-	mu       sync.Mutex
-	id       string
-	holder   string
-	deleted  bool
-	detaches int
-}
-
-func newHeldVolume(cluster *fakeCluster, id, holder string) *heldVolume {
-	volume := &heldVolume{id: id, holder: holder}
-	cluster.handle("ec2.DescribeVolumes", func(data []byte) (any, string) {
-		var input ec2.DescribeVolumesInput
-		_ = json.Unmarshal(data, &input)
-		volume.mu.Lock()
-		defer volume.mu.Unlock()
-		if volume.deleted {
-			if len(input.VolumeIds) > 0 {
-				return nil, awserrors.ErrorInvalidVolumeNotFound
-			}
-			return ec2.DescribeVolumesOutput{}, ""
-		}
-		described := &ec2.Volume{VolumeId: aws.String(volume.id), State: aws.String("available")}
-		if volume.holder != "" {
-			described.State = aws.String("in-use")
-			described.Attachments = []*ec2.VolumeAttachment{
-				{VolumeId: aws.String(volume.id), InstanceId: aws.String(volume.holder), State: aws.String("attached")},
-			}
-		}
-		return ec2.DescribeVolumesOutput{Volumes: []*ec2.Volume{described}}, ""
-	})
-	cluster.handle(forceDetachSubject, func([]byte) (any, string) {
-		volume.mu.Lock()
-		defer volume.mu.Unlock()
-		volume.detaches++
-		previous := volume.holder
-		volume.holder = ""
-		return ec2.VolumeAttachment{VolumeId: aws.String(volume.id), InstanceId: aws.String(previous)}, ""
-	})
-	cluster.handle("ec2.DeleteVolume", func([]byte) (any, string) {
-		volume.mu.Lock()
-		defer volume.mu.Unlock()
-		switch {
-		case volume.deleted:
-			return nil, awserrors.ErrorInvalidVolumeNotFound
-		case volume.holder != "":
-			return nil, awserrors.ErrorVolumeInUse
-		}
-		volume.deleted = true
-		return ec2.DeleteVolumeOutput{}, ""
-	})
-	return volume
-}
-
-func (v *heldVolume) state() (deleted bool, holder string, detaches int) {
-	v.mu.Lock()
-	defer v.mu.Unlock()
-	return v.deleted, v.holder, v.detaches
-}
-
-func instancesIn(state string, instanceIDs ...string) ec2.DescribeInstancesOutput {
-	reservation := &ec2.Reservation{ReservationId: aws.String("r-" + state)}
-	for _, id := range instanceIDs {
-		reservation.Instances = append(reservation.Instances, &ec2.Instance{
-			InstanceId: aws.String(id), State: &ec2.InstanceState{Name: aws.String(state)},
-		})
-	}
-	return ec2.DescribeInstancesOutput{Reservations: []*ec2.Reservation{reservation}}
-}
-
-// Force must not clear an attachment a live guest still holds. The instance
-// carries no block device mapping for the volume, so the attachment record is
-// the only thing standing between the guest and a deleted disk.
-func TestVolumeReaperForceRetainsAVolumeALiveInstanceHolds(t *testing.T) {
-	t.Parallel()
-	for _, state := range []string{"running", "stopped", "shutting-down", "pending"} {
-		t.Run(state, func(t *testing.T) {
-			t.Parallel()
-			cluster := newFakeCluster(t)
-			volume := newHeldVolume(cluster, "vol-held", "i-live")
-			subject := "ec2.DescribeInstances"
-			if state == "stopped" {
-				subject = "ec2.DescribeStoppedInstances"
-			}
-			cluster.reply(subject, instancesIn(state, "i-live"))
-
-			reaper := newVolumeReaper(cluster.nc, fakeClusterNode)
-			err := reaper.Delete(testCtx(t), "000000000042", Resource{Kind: "volume", ID: "vol-held"}, true)
-
-			require.Error(t, err)
-			assert.Contains(t, err.Error(), "vol-held")
-			assert.Contains(t, err.Error(), "i-live")
-			assert.Contains(t, err.Error(), state)
-			deleted, holder, detaches := volume.state()
-			assert.False(t, deleted, "a volume a live guest holds must survive teardown")
-			assert.Equal(t, "i-live", holder, "the attachment must not be cleared under a live guest")
-			assert.Zero(t, detaches)
-		})
-	}
-}
-
-// "Cannot prove gone" is not "gone": every way the proof can fall short keeps
-// the volume and its attachment exactly as they were.
-func TestVolumeReaperForceRetainsAVolumeWhenTheHolderCannotBeProvenGone(t *testing.T) {
-	t.Parallel()
-	cases := []struct {
-		name  string
-		setup func(cluster *fakeCluster)
-		nodes []string
-	}{
-		{"configured node set unknown", func(*fakeCluster) {}, []string{}},
-		{"node describe fails", func(cluster *fakeCluster) {
-			cluster.fail("ec2.DescribeInstances", awserrors.ErrorServerInternal)
-		}, nil},
-		{"terminated bucket unreachable", func(cluster *fakeCluster) {
-			cluster.fail("ec2.DescribeTerminatedInstances", awserrors.ErrorServerInternal)
-		}, nil},
-		{"stopped bucket unreachable", func(cluster *fakeCluster) {
-			cluster.fail("ec2.DescribeStoppedInstances", awserrors.ErrorServerInternal)
-		}, nil},
-		{"sources disagree", func(cluster *fakeCluster) {
-			cluster.reply("ec2.DescribeTerminatedInstances", instancesIn("terminated", "i-held"))
-			cluster.reply("ec2.DescribeInstances", instancesIn("running", "i-held"))
-		}, nil},
-		{"volume describe fails", func(cluster *fakeCluster) {
-			cluster.fail("ec2.DescribeVolumes", awserrors.ErrorServerInternal)
-		}, nil},
-	}
-	for _, tc := range cases {
-		t.Run(tc.name, func(t *testing.T) {
-			t.Parallel()
-			nodes := tc.nodes
-			if nodes == nil {
-				nodes = []string{fakeClusterNode}
-			}
-			assertForceRetainsTheHeldVolume(t, tc.setup, nodes)
-		})
-	}
-}
-
-// assertForceRetainsTheHeldVolume force-deletes a volume i-held holds and
-// asserts the volume and its attachment survive untouched.
-func assertForceRetainsTheHeldVolume(t *testing.T, setup func(cluster *fakeCluster), nodes []string) {
-	t.Helper()
-	cluster := newFakeCluster(t)
-	volume := newHeldVolume(cluster, "vol-held", "i-held")
-	setup(cluster)
-
-	reaper := newVolumeReaper(cluster.nc, nodes...)
-	err := reaper.Delete(silentClusterCtx(t), "000000000042", Resource{Kind: "volume", ID: "vol-held"}, true)
-
-	require.Error(t, err)
-	assert.Contains(t, err.Error(), "vol-held")
-	deleted, holder, detaches := volume.state()
-	assert.False(t, deleted)
-	assert.Equal(t, "i-held", holder)
-	assert.Zero(t, detaches)
-}
-
-// The deadlock force exists for: the instance is gone but its volume record
-// still names it, so the ordinary delete refuses forever. Once the holder is
-// proven absent the attachment is cleared and the volume deleted.
-func TestVolumeReaperForceDeletesOnceTheHolderIsProvenGone(t *testing.T) {
-	t.Parallel()
-	cases := map[string]func(cluster *fakeCluster){
-		"terminated": func(cluster *fakeCluster) {
-			cluster.reply("ec2.DescribeTerminatedInstances", instancesIn("terminated", "i-gone"))
-		},
-		"not found": func(*fakeCluster) {},
-	}
-	for name, setup := range cases {
-		t.Run(name, func(t *testing.T) {
-			t.Parallel()
-			cluster := newFakeCluster(t)
-			volume := newHeldVolume(cluster, "vol-held", "i-gone")
-			setup(cluster)
-
-			reaper := newVolumeReaper(cluster.nc, fakeClusterNode)
-			require.NoError(t, reaper.Delete(testCtx(t), "000000000042",
-				Resource{Kind: "volume", ID: "vol-held"}, true))
-
-			deleted, holder, detaches := volume.state()
-			assert.True(t, deleted)
-			assert.Empty(t, holder)
-			assert.Equal(t, 1, detaches)
-
-			var input ec2.DetachVolumeInput
-			cluster.request(t, forceDetachSubject, &input)
-			assert.Equal(t, "vol-held", aws.StringValue(input.VolumeId))
-		})
-	}
-}
-
-// With no holder on record, the refusal is about something else — a snapshot,
-// a mapping the delete guard found — and clearing the record cannot help.
-func TestVolumeReaperForceLeavesAnUnattachedRefusalAlone(t *testing.T) {
-	cluster := newFakeCluster(t)
-	cluster.reply("ec2.DescribeVolumes", ec2.DescribeVolumesOutput{
-		Volumes: []*ec2.Volume{{VolumeId: aws.String("vol-snapped"), State: aws.String("available")}},
-	})
-	cluster.fail("ec2.DeleteVolume", awserrors.ErrorVolumeInUse)
-
-	reaper := newVolumeReaper(cluster.nc, fakeClusterNode)
-	err := reaper.Delete(testCtx(t), "000000000042", Resource{Kind: "volume", ID: "vol-snapped"}, true)
-
-	require.Error(t, err)
-	assert.Contains(t, err.Error(), awserrors.ErrorVolumeInUse)
-	assert.NotContains(t, cluster.called(), forceDetachSubject)
-}
-
-// End to end through the engine: a forced teardown whose guest is still up
-// leaves the volume, names why, and keeps the account TERMINATING.
-func TestForcedTeardownKeepsTheAccountWhileALiveGuestHoldsAVolume(t *testing.T) {
-	t.Parallel()
-	cluster := newFakeCluster(t)
-	volume := newHeldVolume(cluster, "vol-held", "i-live")
-	cluster.reply("ec2.DescribeInstances", instancesIn("running", "i-live"))
-
-	accounts := newFakeAccounts()
-	engine := quietEngine(accounts, newVolumeReaper(cluster.nc, fakeClusterNode))
-	req := request()
-	req.Force = true
-
-	result, err := engine.Teardown(testCtx(t), req)
-
-	require.ErrorIs(t, err, ErrResourcesStuck)
-	require.Equal(t, 1, result.StuckCount())
-	var stuck Stuck
-	for _, stage := range result.Stages {
-		if len(stage.Stuck) > 0 {
-			stuck = stage.Stuck[0]
-		}
-	}
-	assert.Equal(t, "vol-held", stuck.Resource.ID)
-	assert.Contains(t, stuck.Reason, "vol-held")
-	assert.Contains(t, stuck.Reason, "i-live")
-	assert.Contains(t, stuck.Reason, "running")
-	assert.False(t, result.AccountDeleted)
-	assert.False(t, accounts.deleted)
-	assert.Equal(t, AccountStatusTerminating, accounts.account.Status)
-	deleted, holder, _ := volume.state()
-	assert.False(t, deleted)
-	assert.Equal(t, "i-live", holder)
-}
-
-// A volume that is already gone is a success. Teardown re-runs after a crash
-// and would otherwise never finish what the first pass started.
-func TestVolumeReaperTreatsAMissingVolumeAsDeleted(t *testing.T) {
-	cluster := newFakeCluster(t)
-	cluster.fail("ec2.DeleteVolume", "InvalidVolume.NotFound")
-
-	reaper := newVolumeReaper(cluster.nc, fakeClusterNode)
-
-	assert.NoError(t, reaper.Delete(testCtx(t), "000000000042",
-		Resource{Kind: "volume", ID: "vol-gone"}, false))
-}
-
 // Owner-scoped: an image the account can merely see is not its property and
 // must survive its deletion.
 func TestImageReaperListsOnlyOwnedImages(t *testing.T) {
@@ -781,9 +418,9 @@ func TestSubnetAndVPCReapersReportTheirVPC(t *testing.T) {
 // empty account is how a stage reports itself drained while resources remain.
 func TestAListingFailureIsReported(t *testing.T) {
 	cluster := newFakeCluster(t)
-	cluster.fail("ec2.DescribeVolumes", "InternalError")
+	cluster.fail("ec2.DescribeSnapshots", "InternalError")
 
-	_, err := (newVolumeReaper(cluster.nc, fakeClusterNode)).List(testCtx(t), "000000000042")
+	_, err := (&snapshotReaper{nc: cluster.nc}).List(testCtx(t), "000000000042")
 
 	assert.Error(t, err)
 }
@@ -931,7 +568,3 @@ func TestEC2ReapersRemoveInterfacesBeforeSubnets(t *testing.T) {
 	require.NotEqual(t, -1, eni, "there is no network-interface reaper")
 	require.Less(t, eni, subnet, "interfaces must be reaped before the subnets holding them")
 }
-
-func newVolumeReaper(nc *nats.Conn, nodeIDs ...string) *volumeReaper {
-	return &volumeReaper{nc: nc, nodeIDs: nodeIDs}
-}

spinifex/operator/accountlifecycle/reapers_ecs_test.go

@@ -5,6 +5,7 @@ package accountlifecycle
 // registered in.
 
 import (
+	"context"
 	"encoding/json"
 	"github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
 	"sync"
@@ -13,6 +14,7 @@ import (
 	"github.com/aws/aws-sdk-go/aws"
 	"github.com/aws/aws-sdk-go/service/ecs"
 	"github.com/mulgadc/spinifex/internal/testkit"
+	"github.com/mulgadc/spinifex/spinifex/domains/ec2/accountcleanup"
 	ecsdomain "github.com/mulgadc/spinifex/spinifex/domains/ecs"
 	"github.com/nats-io/nats.go"
 	"github.com/stretchr/testify/assert"
@@ -173,32 +175,37 @@ func TestECSClusterNameFromARN(t *testing.T) {
 	assert.Empty(t, ecsClusterName(""))
 }
 
+// orderedCompute records when the engine reaches the instance cleanup.
+type orderedCompute struct {
+	ComputeStage
+
+	order *[]string
+}
+
+func (o *orderedCompute) CleanupCompute(ctx context.Context, req *accountcleanup.ComputeRequest) (*accountcleanup.ComputeResult, error) {
+	*o.order = append(*o.order, "instances")
+	return o.ComputeStage.CleanupCompute(ctx, req)
+}
+
 // Container instances are ordinary EC2 instances in the tenant's account, so
-// the instance reaper takes them — but the tasks riding on them have to be
-// stopped first, which is what puts ECS ahead of it in the compute stage.
-func TestECSReapersRunBeforeTheInstanceReaper(t *testing.T) {
-	reapers := ECSReapers(nil)
-	reapers = append(reapers, EC2Reapers(nil, 1, []string{"node-1"})...)
+// the compute cleanup terminates them — but the tasks riding on them have to be
+// stopped first, so the compute-stage reapers must run before that cleanup.
+func TestECSClustersAreDeletedBeforeTheInstancesAreTerminated(t *testing.T) {
+	var order []string
+	cluster := &fakeReaper{
+		kind: "ecs-cluster", stage: StageCompute, order: &order,
+		resources: []Resource{{Kind: "ecs-cluster", ID: "web"}},
+	}
+	accounts := newFakeAccounts()
+	engine := quietEngine(accounts, cluster)
+	compute, _ := cleanStages()
+	engine.Compute = &orderedCompute{ComputeStage: compute, order: &order}
 
-	ecsIndex := indexOfKind(reapers, "ecs-cluster")
-	instanceIndex := indexOfKind(reapers, "instance")
+	_, err := engine.Teardown(context.Background(), request())
 
-	require.NotEqual(t, -1, ecsIndex)
-	require.NotEqual(t, -1, instanceIndex)
-	assert.Less(t, ecsIndex, instanceIndex)
-	assert.Equal(t, StageCompute, reapers[ecsIndex].Stage())
-
-	// Both are in the same stage, so the engine's within-stage ordering is
-	// what enforces this rather than the stage graph.
-	assert.Equal(t, reapers[instanceIndex].Stage(), reapers[ecsIndex].Stage())
-}
-
-// SortReapers is what the engine applies, and a stable sort is the only reason
-// the registration order above survives it.
-func TestSortReapersKeepsECSAheadOfInstances(t *testing.T) {
-	reapers := ECSReapers(nil)
-	reapers = append(reapers, EC2Reapers(nil, 1, []string{"node-1"})...)
-	SortReapers(reapers)
-
-	assert.Less(t, indexOfKind(reapers, "ecs-cluster"), indexOfKind(reapers, "instance"))
+	require.NoError(t, err)
+	require.Equal(t, []string{"ecs-cluster/web", "instances"}, order)
+	for _, reaper := range ECSReapers(nil) {
+		assert.Equal(t, StageCompute, reaper.Stage())
+	}
 }

spinifex/operator/admin/account_delete.go

@@ -124,12 +124,12 @@ func (d Deps) deleteAccount(ctx context.Context, body []byte) (any, error) {
 		return nil, errors.New(awserrors.ErrorInternalError)
 	}
 
-	if !req.DryRun && d.PlatformCleaner == nil {
+	if !req.DryRun && (d.PlatformCleaner == nil || d.ComputeCleaner == nil || d.StorageCleaner == nil) {
 		slog.Error("DeleteAccount: platform cleanup not available")
 		return nil, errors.New(awserrors.ErrorServiceUnavailable)
 	}
 
-	engine, err := accountteardown.NewClusterEngine(ctx, d.NATS, accountteardown.ClusterNodes{Expected: d.ExpectedNodes, DaemonIDs: d.NodeIDs, GatewayIDs: d.GatewayNodeIDs}, d.Termination, d.BucketStore, d.PlatformCleaner)
+	engine, err := accountteardown.NewClusterEngine(ctx, d.NATS, accountteardown.ClusterNodes{Expected: d.ExpectedNodes, DaemonIDs: d.NodeIDs, GatewayIDs: d.GatewayNodeIDs}, d.Termination, d.BucketStore, d.PlatformCleaner, accountteardown.WithComputeCleanup(d.ComputeCleaner, d.StorageCleaner))
 	if err != nil {
 		slog.Error("DeleteAccount: failed to build teardown engine", "err", err)
 		return nil, errors.New(awserrors.ErrorServiceUnavailable)

spinifex/operator/admin/account_delete_dryrun_test.go

@@ -26,7 +26,7 @@ func TestAdminDeleteAccountDryRunAnswersTheInventoryAndCleansNothing(t *testing.
 	gotJSON, err := json.Marshal(got)
 	require.NoError(t, err)
 
-	engine, err := accountteardown.NewClusterEngine(t.Context(), gw.NATS, accountteardown.ClusterNodes{Expected: gw.ExpectedNodes, DaemonIDs: gw.NodeIDs, GatewayIDs: gw.GatewayNodeIDs}, gw.Termination, gw.BucketStore, gw.PlatformCleaner)
+	engine, err := accountteardown.NewClusterEngine(t.Context(), gw.NATS, accountteardown.ClusterNodes{Expected: gw.ExpectedNodes, DaemonIDs: gw.NodeIDs, GatewayIDs: gw.GatewayNodeIDs}, gw.Termination, gw.BucketStore, gw.PlatformCleaner, accountteardown.WithComputeCleanup(gw.ComputeCleaner, gw.StorageCleaner))
 	require.NoError(t, err)
 	inventory, err := engine.Inventory(t.Context(), account.AccountID)
 	require.NoError(t, err)

spinifex/operator/admin/account_delete_job_test.go

@@ -8,7 +8,9 @@ import (
 	"context"
 	"encoding/json"
 	operatorv1 "github.com/mulgadc/spinifex/contracts/operator/v1"
+	"github.com/mulgadc/spinifex/spinifex/domains/ec2/accountcleanup"
 	"strings"
+	"sync/atomic"
 	"testing"
 	"time"
 
@@ -148,6 +150,8 @@ func newDeleteAccountGateway(t *testing.T) Deps {
 		GatewayNodeIDs:  []string{"node-1"},
 		BucketStore:     emptyBucketStore{},
 		PlatformCleaner: &panicCleaner{},
+		ComputeCleaner:  emptyCompute{},
+		StorageCleaner:  emptyStorage{},
 	}
 }
 
@@ -607,3 +611,85 @@ func TestDeletionTrackerHeartbeatStops(t *testing.T) {
 	tracker.persist(t.Context())
 	assert.NotEmpty(t, describeDeletion(t, gw, "000000000042").UpdatedAt)
 }
+
+// emptyCompute and emptyStorage stand in for the EC2 cleanup of an account that
+// holds no instance, spot request or volume.
+type (
+	emptyCompute struct{}
+	emptyStorage struct{}
+)
+
+func cleanOutcome(accountID string) *accountcleanup.Outcome {
+	return &accountcleanup.Outcome{AccountID: accountID, Status: accountcleanup.StatusClean}
+}
+
+func (emptyCompute) CleanupCompute(_ context.Context, req *accountcleanup.ComputeRequest) (*accountcleanup.ComputeResult, error) {
+	return cleanOutcome(req.AccountID), nil
+}
+
+func (emptyCompute) ComputeRemaining(_ context.Context, accountID string) (*accountcleanup.Outcome, error) {
+	return cleanOutcome(accountID), nil
+}
+
+func (emptyStorage) CleanupStorage(_ context.Context, req *accountcleanup.StorageRequest) (*accountcleanup.StorageResult, error) {
+	return cleanOutcome(req.AccountID), nil
+}
+
+func (emptyStorage) StorageRemaining(_ context.Context, accountID string) (*accountcleanup.Outcome, error) {
+	return cleanOutcome(accountID), nil
+}
+
+// blockedCompute holds an instance the cleanup cannot release until unblocked,
+// the way a managed service still owning a worker does.
+type blockedCompute struct {
+	blocked atomic.Bool
+}
+
+func (b *blockedCompute) outcome(accountID string) *accountcleanup.Outcome {
+	out := cleanOutcome(accountID)
+	if b.blocked.Load() {
+		out.Items = []accountcleanup.Item{{
+			Kind: accountcleanup.KindInstance, ResourceID: "i-held", State: "running",
+			Disposition: accountcleanup.DispositionBlocked, Reason: accountcleanup.ReasonManagedServiceOwner,
+		}}
+	}
+	return out
+}
+
+func (b *blockedCompute) CleanupCompute(_ context.Context, req *accountcleanup.ComputeRequest) (*accountcleanup.ComputeResult, error) {
+	return b.outcome(req.AccountID), nil
+}
+
+func (b *blockedCompute) ComputeRemaining(_ context.Context, accountID string) (*accountcleanup.Outcome, error) {
+	return b.outcome(accountID), nil
+}
+
+// A job that ends FAILED naming a blocker the loop cannot clear keeps the
+// account, and an operator rerun after the blocker clears converges it.
+func TestAdminDeleteAccountFailedJobConvergesOnOperatorRerun(t *testing.T) {
+	t.Parallel()
+	gw := newDeleteAccountGateway(t)
+	compute := &blockedCompute{}
+	compute.blocked.Store(true)
+	gw.ComputeCleaner = compute
+	account := newTenantAccount(t, gw, "tenant@example.com")
+
+	_, err := gw.deleteAccount(t.Context(),
+		deleteAccountBody(t, account.AccountID, "tenant@example.com", strings.Repeat("a", 32)))
+	require.NoError(t, err)
+	require.Eventually(t, func() bool {
+		return describeDeletion(t, gw, account.AccountID).State == operatorv1.DeletionStateFailed
+	}, 60*time.Second, 50*time.Millisecond)
+	failed := describeDeletion(t, gw, account.AccountID)
+	assert.NotEmpty(t, failed.Error)
+	_, stillThere := gw.IAM.GetAccount(account.AccountID)
+	require.NoError(t, stillThere, "a FAILED job must keep the account")
+
+	compute.blocked.Store(false)
+	_, err = gw.deleteAccount(t.Context(),
+		deleteAccountBody(t, account.AccountID, "tenant@example.com", strings.Repeat("b", 32)))
+	require.NoError(t, err)
+	require.Eventually(t, func() bool {
+		return describeDeletion(t, gw, account.AccountID).State == operatorv1.DeletionStateCompleted
+	}, 60*time.Second, 50*time.Millisecond)
+}

spinifex/operator/admin/account_delete_resumer.go

@@ -99,7 +99,7 @@ func (d Deps) resumeDeletion(
 	revision uint64,
 	now time.Time,
 ) bool {
-	engine, err := accountteardown.NewClusterEngine(ctx, d.NATS, accountteardown.ClusterNodes{Expected: d.ExpectedNodes, DaemonIDs: d.NodeIDs, GatewayIDs: d.GatewayNodeIDs}, d.Termination, d.BucketStore, d.PlatformCleaner)
+	engine, err := accountteardown.NewClusterEngine(ctx, d.NATS, accountteardown.ClusterNodes{Expected: d.ExpectedNodes, DaemonIDs: d.NodeIDs, GatewayIDs: d.GatewayNodeIDs}, d.Termination, d.BucketStore, d.PlatformCleaner, accountteardown.WithComputeCleanup(d.ComputeCleaner, d.StorageCleaner))
 	if err != nil {
 		// Nothing was written, so a later scan retries.
 		slog.Error("DeletionResumer: failed to build teardown engine", "account_id", existing.AccountID, "operation_id", existing.OperationID, "err", err)

spinifex/operator/admin/registration.go

@@ -57,7 +57,11 @@ type Deps struct {
 	Termination accountteardown.IAMBackend
 	// PlatformCleaner is the in-process EC2 platform cleanup every DeleteAccount
 	// runs. Nil refuses a request that is not a dry run.
-	PlatformCleaner   accountteardown.PlatformCleaner
+	PlatformCleaner accountteardown.PlatformCleaner
+	// ComputeCleaner and StorageCleaner are the EC2 instance, spot request and
+	// volume cleanup. Nil refuses a request that is not a dry run.
+	ComputeCleaner    accountteardown.ComputeStage
+	StorageCleaner    accountteardown.StorageStage
 	Quota             *quota.Service
 	ExpectedNodes     int
 	NodeIDs           []string

spinifex/runtime/compute/lambda/testlink/control.go

@@ -151,6 +151,9 @@ type ControlServer struct {
 	// IOTimeout bounds reading the request and writing the reply.
 	IOTimeout time.Duration
 	Handle    func(ctx context.Context, req []byte, cred Credential) Reply
+	// AcceptStopped, if set, runs once Serve stops accepting, before it waits
+	// for the connections it started.
+	AcceptStopped func()
 }
 
 // Serve accepts until ctx ends or the listener fails, then waits for every
@@ -164,6 +167,9 @@ func (s *ControlServer) Serve(ctx context.Context, ln *net.UnixListener) error {
 	defer func() { _ = ln.Close() }()
 	var wg sync.WaitGroup
 	defer wg.Wait()
+	if s.AcceptStopped != nil {
+		defer s.AcceptStopped()
+	}
 	slots := make(chan struct{}, MaxAdmitted)
 	for {
 		select {

spinifex/runtime/compute/lambda/testlink/control_test.go

@@ -436,18 +436,23 @@ func TestControl_WriteErrorAfterDeliveryIsUnknown(t *testing.T) {
 	require.Equal(t, testCred, s.cred.Load())
 }
 
-// legacyControlName is the protocol name before the request carried the image
-// digest; an old daemon still serves it and ignores the field.
-const legacyControlName = "spinifex-lambda-testlink-control"
+// Earlier control protocol names: unversioned, before the request carried the
+// image digest, and /2, before the 6 MiB payload and frame bounds.
+const (
+	legacyControlName = "spinifex-lambda-testlink-control"
+	v2ControlName     = "spinifex-lambda-testlink-control/2"
+)
 
 // A gateway and daemon on different control protocol names never connect, in
 // either direction. The scratch prefix isolates the binds; the names after it
-// are the production constant and its predecessor.
+// are the production constant and its predecessors.
 func TestControl_MixedProtocolNamesNeverConnect(t *testing.T) {
 	prefix := controlName(t) + "/"
 	for name, tc := range map[string]struct{ daemon, gateway string }{
-		"old daemon, new gateway": {prefix + legacyControlName, prefix + ControlName},
-		"new daemon, old gateway": {prefix + ControlName, prefix + legacyControlName},
+		"unversioned daemon, new gateway": {prefix + legacyControlName, prefix + ControlName},
+		"new daemon, unversioned gateway": {prefix + ControlName, prefix + legacyControlName},
+		"/2 daemon, new gateway":          {prefix + v2ControlName, prefix + ControlName},
+		"new daemon, /2 gateway":          {prefix + ControlName, prefix + v2ControlName},
 	} {
 		t.Run(name, func(t *testing.T) {
 			s, _ := serveControl(t, tc.daemon, os.Getuid(), nil)
@@ -457,3 +462,45 @@ func TestControl_MixedProtocolNamesNeverConnect(t *testing.T) {
 		})
 	}
 }
+
+func TestControl_AcceptStoppedRunsBeforeTheDrain(t *testing.T) {
+	name := controlName(t)
+	ln, err := ListenControl(name)
+	require.NoError(t, err)
+	inHandle, release := make(chan struct{}), make(chan struct{})
+	stopped := make(chan struct{})
+	srv := &ControlServer{GatewayUID: os.Getuid(), RequestLimit: testRequestLimit, ReplyLimit: 4096, IOTimeout: 5 * time.Second,
+		Handle: func(context.Context, []byte, Credential) Reply {
+			close(inHandle)
+			<-release
+			return Reply{Outcome: OutcomeDone}
+		},
+		AcceptStopped: func() { close(stopped) }}
+	done := make(chan error, 1)
+	go func() { done <- srv.Serve(context.Background(), ln) }()
+	go func() { _, _ = client(name, os.Getuid()).Exchange(context.Background(), []byte("{}"), testCred, soon()) }()
+	select {
+	case <-inHandle:
+	case <-time.After(5 * time.Second):
+		t.Fatal("the request never reached Handle")
+	}
+
+	require.NoError(t, ln.Close(), "the listener fails under an outstanding invocation")
+	select {
+	case <-stopped:
+	case <-time.After(5 * time.Second):
+		t.Fatal("AcceptStopped did not run while the invocation was outstanding")
+	}
+	select {
+	case err := <-done:
+		t.Fatalf("Serve returned before its connection finished: %v", err)
+	default:
+	}
+	close(release)
+	select {
+	case err := <-done:
+		require.Error(t, err, "a listener failure is reported")
+	case <-time.After(5 * time.Second):
+		t.Fatal("Serve did not return after its connection finished")
+	}
+}

spinifex/runtime/compute/lambda/testlink/daemon.go

@@ -90,10 +90,17 @@ type Daemon struct {
 	// InitializerSHA256 the digest it must have.
 	Initializer       string
 	InitializerSHA256 string
+	// Firecracker, when set, is the Firecracker binary each attempt boots
+	// under instead of QEMU.
+	Firecracker string
 
-	busy atomic.Bool
+	busy    atomic.Bool
+	serving atomic.Bool
 }
 
+// Serving reports whether the control socket is accepting invocations.
+func (d *Daemon) Serving() bool { return d.serving.Load() }
+
 // unresolved names an attempt whose marker is still on disk. A marker goes
 // only once the attempt's VM stop is proven and everything it held is
 // released, so it outlives a daemon restart; nothing resolves it on its own.
@@ -108,6 +115,12 @@ func (d *Daemon) unresolved() (string, error) {
 	return strings.TrimSuffix(filepath.Base(markers[0]), markerSuffix), nil
 }
 
+// validHandler is the handler a guest run accepts: 1..128 bytes, no
+// whitespace or NUL.
+func validHandler(h string) bool {
+	return h != "" && len(h) <= 128 && strings.IndexFunc(h, unicode.IsSpace) < 0 && !strings.ContainsRune(h, 0)
+}
+
 // controlIOTimeout bounds reading a request from, and writing a reply to, the
 // gateway on the control socket.
 const controlIOTimeout = 30 * time.Second
@@ -119,14 +132,24 @@ func (d *Daemon) ServeControl(ctx context.Context, name string, gatewayUID int)
 	if err != nil {
 		return err
 	}
-	srv := &ControlServer{GatewayUID: gatewayUID, RequestLimit: MaxFrameBytes, ReplyLimit: MaxFrameBytes,
-		IOTimeout: controlIOTimeout, Handle: d.Handle}
+	d.serve(ctx, ln, d.controlServer(gatewayUID))
+	return nil
+}
+
+// controlServer answers the gateway with d.Handle, and stops d reporting
+// serving as soon as it stops accepting, before its connections drain.
+func (d *Daemon) controlServer(gatewayUID int) *ControlServer {
+	return &ControlServer{GatewayUID: gatewayUID, RequestLimit: MaxFrameBytes, ReplyLimit: MaxFrameBytes,
+		IOTimeout: controlIOTimeout, Handle: d.Handle, AcceptStopped: func() { d.serving.Store(false) }}
+}
+
+func (d *Daemon) serve(ctx context.Context, ln *net.UnixListener, srv *ControlServer) {
+	d.serving.Store(true)
 	go func() {
 		if err := srv.Serve(ctx, ln); err != nil {
 			slog.Error("Lambda test link: control socket stopped; not serving new invocations", "err", err)
 		}
 	}()
-	return nil
 }
 
 // Handle runs one request under its grant, giving the guest cred in its first
@@ -293,8 +316,7 @@ func CheckRequest(req Request) error {
 	if req.RequestID == "" || req.RevisionID == "" {
 		return errors.New("request and revision IDs are required")
 	}
-	if req.Handler == "" || len(req.Handler) > 128 || strings.IndexFunc(req.Handler, unicode.IsSpace) >= 0 ||
-		strings.ContainsRune(req.Handler, 0) {
+	if !validHandler(req.Handler) {
 		return errors.New("handler must be 1..128 characters with no whitespace")
 	}
 	if req.TimeoutMs <= 0 || req.TimeoutMs > maxTimeoutMs {
@@ -404,24 +426,50 @@ func (d *Daemon) run(ctx context.Context, req Request, code []byte, lib *os.File
 	}
 	defer func() { _ = ln.Close() }()
 
-	cmd, err := cfg.Execute()
-	if err != nil {
-		return refused("VM command build failed"), d.release(a), marker
-	}
-	// A launch delayed past the fixed deadline is refused before QEMU starts.
+	// A launch delayed past the fixed deadline is refused before the VM starts.
 	if expired(hardDeadline) {
 		return refused(ErrDeadlineExpired.Error()), d.release(a), marker
 	}
-	hk, err := armLaunch(cmd, d.Image, lib, hardDeadline)
-	if err != nil {
-		return refused("hard-kill setup failed"), d.release(a), marker
+	var cmd *exec.Cmd
+	var hk *hardKill
+	var fcAck *net.UnixConn
+	fc := &vmHandle{fd: -1}
+	if d.Firecracker == "" {
+		if cmd, err = cfg.Execute(); err != nil {
+			return refused("VM command build failed"), d.release(a), marker
+		}
+		if hk, err = armLaunch(cmd, d.Image, lib, hardDeadline); err != nil {
+			return refused("hard-kill setup failed"), d.release(a), marker
+		}
+		defer hk.close()
+	} else {
+		var child *os.File
+		if fcAck, child, err = ackPair(); err != nil {
+			return refused("hard-kill setup failed"), d.release(a), marker
+		}
+		defer func() { _ = fcAck.Close() }()
+		cmd, err = d.supervisorCommand(filepath.Join(a.dir, fcConfigFile), hardDeadline, child)
+		defer func() { _ = child.Close() }()
+		if err != nil {
+			return refused("VM command build failed"), d.release(a), marker
+		}
 	}
-	defer hk.close()
 	var result Reply
 	unrecorded := false
 	out, runErr := attempt.Run(ctx, cmd, timeoutFor(req), func(sctx context.Context, pidfd int) error {
+		vmPidfd, vmPID := pidfd, cmd.Process.Pid
+		if fcAck != nil {
+			defer fc.callbackDone()
+			// Firecracker is the VM; its supervisor acks only once both kernel kills are armed.
+			fd, pid, err := awaitFirecracker(sctx, fcAck, cmd.Process.Pid, armWait)
+			if err != nil {
+				return err
+			}
+			fc.set(fd)
+			vmPidfd, vmPID = fd, pid
+		}
 		// The owner holds the launch identity before the guest is sent anything.
-		l, err := launchIdentity(pidfd, cmd.Process.Pid, a.tap)
+		l, err := launchIdentity(vmPidfd, vmPID, a.tap)
 		if err != nil {
 			return err
 		}
@@ -430,13 +478,20 @@ func (d *Daemon) run(ctx context.Context, req Request, code []byte, lib *os.File
 			return fmt.Errorf("%w: %w", ErrLaunchNotRecorded, err)
 		}
 		// No frame reaches the guest until QEMU proves its own kill timer is armed.
-		if err := hk.await(sctx, cmd.Process.Pid, armWait); err != nil {
-			return err
+		if hk != nil {
+			if err := hk.await(sctx, cmd.Process.Pid, armWait); err != nil {
+				return err
+			}
 		}
 		r, err := d.session(sctx, req, gen, ln, a.dir, hardDeadline, d.Grants.Accept)
 		result = r
 		return err
 	})
+	if fcAck != nil {
+		// Always settled, so a received pidfd is closed or handed to the callback.
+		fcGone := settleFirecracker(out.SessionDone, fc, cmd.Process != nil)
+		out.Stopped = out.Stopped && fcGone
+	}
 	if !out.Stopped {
 		slog.Error("Lambda test link: VM stop not proven; marker, tap, address and files held",
 			"attempt", req.AttemptID, "tap", tapName, "err", runErr)
@@ -527,6 +582,22 @@ func (d *Daemon) prepare(req Request, code []byte, dir, tap, mac, ip string) (*v
 	entries["test-package-sha256"] = []byte(req.CodeSHA256)
 	entries["test-grant"] = grantEntry(req.Grant)
 	entries["test-handler"] = []byte(req.Handler)
+	netcfgBytes := fmt.Appendf(nil, "NIC0_MAC=%s\nNIC0_CIDR=%s/24\nNIC0_DEFAULT=0\n", mac, ip)
+	if d.Firecracker != "" {
+		archive := make(map[string][]byte, len(entries)+1)
+		for name, value := range entries {
+			archive[lambdav1.FwCfgPrefix+name] = value
+		}
+		archive[fwCfgNetcfg] = netcfgBytes
+		cfgDisk := filepath.Join(dir, configArchive)
+		if err := writeConfigArchive(cfgDisk, archive); err != nil {
+			return fail(err)
+		}
+		if err := writeFirecrackerConfig(filepath.Join(dir, fcConfigFile), diskPath, cfgDisk, tap, mac, int(max(req.MemoryMiB, 128))); err != nil {
+			return fail(err)
+		}
+		return nil, gen, ln, nil
+	}
 	var fwCfg []vm.FwCfgEntry
 	for name, value := range entries {
 		path := filepath.Join(dir, "fwcfg-"+name)
@@ -536,7 +607,7 @@ func (d *Daemon) prepare(req Request, code []byte, dir, tap, mac, ip string) (*v
 		fwCfg = append(fwCfg, vm.FwCfgEntry{Name: lambdav1.FwCfgPrefix + name, File: path})
 	}
 	netcfg := filepath.Join(dir, "netcfg")
-	if err := os.WriteFile(netcfg, fmt.Appendf(nil, "NIC0_MAC=%s\nNIC0_CIDR=%s/24\nNIC0_DEFAULT=0\n", mac, ip), 0o600); err != nil {
+	if err := os.WriteFile(netcfg, netcfgBytes, 0o600); err != nil {
 		return fail(err)
 	}
 	fwCfg = append(fwCfg, vm.FwCfgEntry{Name: fwCfgNetcfg, File: netcfg})
@@ -588,6 +659,7 @@ func (d *Daemon) session(ctx context.Context, req Request, gen *guestconn.Genera
 	}
 	defer func() { _ = conn.Close() }()
 	_ = os.Remove(filepath.Join(dir, "fwcfg-"+lambdav1.EntryGuestKey))
+	_ = os.Remove(filepath.Join(dir, configArchive))
 
 	// The function timeout runs on the host from dispatch; the boot allowance
 	// covers startup only, so a guest that stops answering is cut off here.

spinifex/runtime/compute/lambda/testlink/firecracker.go

@@ -0,0 +1,396 @@
+//go:build lambdatestlink
+
+package testlink
+
+import (
+	"archive/tar"
+	"bytes"
+	"context"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"net"
+	"os"
+	"os/exec"
+	"path/filepath"
+	"runtime"
+	"sort"
+	"strconv"
+	"strings"
+	"sync"
+	"syscall"
+	"time"
+	"unsafe"
+
+	"golang.org/x/sys/unix"
+)
+
+// firecrackerArg0 is the argv[0] that turns the spx binary into the
+// Firecracker supervisor, so the supervisor is the already-pinned binary.
+const firecrackerArg0 = "spx-lambda-firecracker"
+
+// The supervisor inherits the sealed kernel and initramfs as fds 3 and 4,
+// which it passes on to Firecracker unchanged, and its ack socket as fd 5.
+const (
+	fcKernelFD = 3
+	fcInitrdFD = 4
+	fcAckFD    = 5
+	fcAck      = "armed"
+)
+
+// firecrackerCmdline is the guest kernel command line; reboot=k lets a guest
+// reboot end Firecracker, and Firecracker adds the virtio-mmio devices itself.
+const firecrackerCmdline = "console=ttyS0 quiet loglevel=3 reboot=k panic=1 pci=off"
+
+const (
+	configArchive = "config.tar"
+	fcConfigFile  = "firecracker.json"
+	// fcStopWait bounds proving Firecracker gone after its supervisor stopped.
+	fcStopWait = 5 * time.Second
+)
+
+func init() {
+	if filepath.Base(os.Args[0]) == firecrackerArg0 {
+		os.Exit(supervise(os.Args[1:]))
+	}
+}
+
+// supervise arms a CLOCK_BOOTTIME SIGKILL on itself at the absolute deadline,
+// then starts Firecracker with SIGKILL as its parent-death signal, so the
+// kernel ends both at the deadline even if this process and the daemon stop.
+func supervise(args []string) int {
+	if len(args) != 3 {
+		return 2
+	}
+	deadline, err := strconv.ParseInt(args[2], 10, 64)
+	if err != nil {
+		return 2
+	}
+	unix.CloseOnExec(fcAckFD)
+	// The parent-death signal follows the forking thread, which never exits.
+	runtime.LockOSThread()
+	if err := armBoottimeKill(time.Duration(deadline)); err != nil {
+		fmt.Fprintln(os.Stderr, "firecracker supervisor:", err)
+		return 3
+	}
+	pidfd := -1
+	cmd := exec.Command(args[0], "--no-api", "--config-file", args[1], "--level", "Warning")
+	cmd.Stdout, cmd.Stderr = os.Stdout, os.Stderr
+	cmd.ExtraFiles = []*os.File{os.NewFile(fcKernelFD, "kernel"), os.NewFile(fcInitrdFD, "initrd")}
+	cmd.SysProcAttr = &syscall.SysProcAttr{Pdeathsig: syscall.SIGKILL, PidFD: &pidfd}
+	if err := cmd.Start(); err != nil {
+		fmt.Fprintln(os.Stderr, "firecracker supervisor: start:", err)
+		return 4
+	}
+	if err := unix.Sendmsg(fcAckFD, []byte(fcAck), unix.UnixRights(pidfd), nil, 0); err != nil {
+		_ = cmd.Process.Kill()
+		_ = cmd.Wait()
+		return 5
+	}
+	if err := cmd.Wait(); err != nil {
+		var exit *exec.ExitError
+		if errors.As(err, &exit) {
+			return exit.ExitCode()
+		}
+		return 6
+	}
+	return 0
+}
+
+// sigevent is the kernel's struct sigevent on 64-bit Linux.
+type sigevent struct {
+	value  uintptr
+	signo  int32
+	notify int32
+	_      [48]byte
+}
+
+const (
+	sigevSignal  = 0
+	timerAbstime = 1
+)
+
+// armBoottimeKill arms a POSIX timer that delivers SIGKILL to this process at
+// the absolute CLOCK_BOOTTIME deadline; a passed deadline fires at once.
+func armBoottimeKill(deadline time.Duration) error {
+	ev := sigevent{signo: int32(unix.SIGKILL), notify: sigevSignal}
+	var id int32
+	if _, _, e := unix.Syscall(unix.SYS_TIMER_CREATE, unix.CLOCK_BOOTTIME,
+		uintptr(unsafe.Pointer(&ev)), uintptr(unsafe.Pointer(&id))); e != 0 {
+		return fmt.Errorf("timer_create: %w", e)
+	}
+	spec := unix.ItimerSpec{Value: unix.NsecToTimespec(int64(deadline))}
+	if _, _, e := unix.Syscall6(unix.SYS_TIMER_SETTIME, uintptr(id), timerAbstime,
+		uintptr(unsafe.Pointer(&spec)), 0, 0, 0); e != 0 {
+		return fmt.Errorf("timer_settime: %w", e)
+	}
+	return nil
+}
+
+// supervisorCommand builds the supervisor for one attempt: the spx binary
+// under firecrackerArg0, the sealed kernel and initramfs, and the ack socket.
+func (d *Daemon) supervisorCommand(configPath string, deadline time.Duration, ack *os.File) (*exec.Cmd, error) {
+	if d.Image == nil || d.Image.Kernel == nil || d.Image.Initrd == nil {
+		return nil, errors.New("runtime image not prepared")
+	}
+	self, err := os.Executable()
+	if err != nil {
+		return nil, err
+	}
+	cmd := exec.Command(self, d.Firecracker, configPath, strconv.FormatInt(int64(deadline), 10))
+	cmd.Args[0] = firecrackerArg0
+	cmd.Stdout, cmd.Stderr = os.Stdout, os.Stderr
+	cmd.ExtraFiles = []*os.File{d.Image.Kernel, d.Image.Initrd, ack}
+	return cmd, nil
+}
+
+// ackPair is the socket the supervisor acks on; the daemon keeps one end.
+func ackPair() (daemonEnd *net.UnixConn, child *os.File, err error) {
+	fds, err := unix.Socketpair(unix.AF_UNIX, unix.SOCK_SEQPACKET|unix.SOCK_CLOEXEC, 0)
+	if err != nil {
+		return nil, nil, err
+	}
+	child = os.NewFile(uintptr(fds[1]), "fc-ack")
+	f := os.NewFile(uintptr(fds[0]), "fc-ack-daemon")
+	c, err := net.FileConn(f)
+	_ = f.Close()
+	if err != nil {
+		_ = child.Close()
+		return nil, nil, err
+	}
+	return c.(*net.UnixConn), child, nil
+}
+
+// awaitFirecracker reads the supervisor's ack and the pidfd it carries, and
+// checks that pidfd names a live child of the supervisor.
+func awaitFirecracker(ctx context.Context, ack *net.UnixConn, supervisor int, wait time.Duration) (pidfd, pid int, err error) {
+	deadline := time.Now().Add(wait)
+	if d, ok := ctx.Deadline(); ok && d.Before(deadline) {
+		deadline = d
+	}
+	if err := ack.SetReadDeadline(deadline); err != nil {
+		return -1, 0, err
+	}
+	buf := make([]byte, 16)
+	oob := make([]byte, unix.CmsgSpace(4))
+	n, oobn, _, _, err := ack.ReadMsgUnix(buf, oob)
+	if err != nil {
+		return -1, 0, fmt.Errorf("%w: %w", ErrNotArmed, err)
+	}
+	msgs, err := unix.ParseSocketControlMessage(oob[:oobn])
+	if err != nil || len(msgs) != 1 {
+		return -1, 0, fmt.Errorf("%w: no pidfd in ack", ErrNotArmed)
+	}
+	fds, err := unix.ParseUnixRights(&msgs[0])
+	if err != nil || len(fds) != 1 {
+		return -1, 0, fmt.Errorf("%w: no pidfd in ack", ErrNotArmed)
+	}
+	pidfd = fds[0]
+	unix.CloseOnExec(pidfd)
+	if string(buf[:n]) != fcAck {
+		_ = unix.Close(pidfd)
+		return -1, 0, fmt.Errorf("%w: unexpected ack", ErrNotArmed)
+	}
+	if pid, err = pidfdPID(pidfd); err == nil {
+		err = checkParent(pid, supervisor)
+	}
+	if err != nil {
+		_ = unix.Close(pidfd)
+		return -1, 0, fmt.Errorf("%w: %w", ErrNotArmed, err)
+	}
+	return pidfd, pid, nil
+}
+
+// pidfdPID reads the PID a pidfd refers to from its fdinfo.
+func pidfdPID(pidfd int) (int, error) {
+	b, err := os.ReadFile(fmt.Sprintf("/proc/self/fdinfo/%d", pidfd))
+	if err != nil {
+		return 0, err
+	}
+	for line := range strings.SplitSeq(string(b), "\n") {
+		if v, ok := strings.CutPrefix(line, "Pid:"); ok {
+			pid, err := strconv.Atoi(strings.TrimSpace(v))
+			if err != nil || pid <= 0 {
+				return 0, fmt.Errorf("pidfd names no live process (%q)", strings.TrimSpace(v))
+			}
+			return pid, nil
+		}
+	}
+	return 0, errors.New("pidfd fdinfo has no Pid")
+}
+
+func checkParent(pid, parent int) error {
+	b, err := os.ReadFile(fmt.Sprintf("/proc/%d/status", pid))
+	if err != nil {
+		return err
+	}
+	for line := range strings.SplitSeq(string(b), "\n") {
+		if v, ok := strings.CutPrefix(line, "PPid:"); ok {
+			if strings.TrimSpace(v) != strconv.Itoa(parent) {
+				return fmt.Errorf("process %d is not the supervisor's child", pid)
+			}
+			return nil
+		}
+	}
+	return errors.New("process status has no PPid")
+}
+
+// stopFirecracker kills Firecracker through its pidfd and reports whether its
+// exit was proven; the supervisor's own exit proves nothing about it.
+func stopFirecracker(pidfd int) bool {
+	if pidfd < 0 {
+		return false
+	}
+	_ = unix.PidfdSendSignal(pidfd, unix.SIGKILL, nil, 0)
+	exited, err := pidfdExited(pidfd, fcStopWait)
+	return err == nil && exited
+}
+
+// settleFirecracker reports whether Firecracker is proven gone once Run has
+// returned. A supervisor that never started launched nothing; otherwise only
+// Firecracker's own pidfd proves its exit, since the supervisor's says nothing.
+func settleFirecracker(sessionDone <-chan struct{}, fc *vmHandle, supervisorStarted bool) bool {
+	select {
+	case <-sessionDone:
+	default:
+		// The callback may still receive the pidfd; it closes it, and the attempt is held.
+		fc.abandon()
+		return false
+	}
+	fd := fc.take()
+	if fd < 0 {
+		return !supervisorStarted
+	}
+	defer func() { _ = unix.Close(fd) }()
+	return stopFirecracker(fd)
+}
+
+// writeConfigArchive writes entries as a tar archive in the fw_cfg sysfs
+// layout, by_name/<entry>/raw, which the guest's init extracts.
+func writeConfigArchive(path string, entries map[string][]byte) error {
+	names := make([]string, 0, len(entries))
+	for name := range entries {
+		names = append(names, name)
+	}
+	sort.Strings(names)
+	var buf bytes.Buffer
+	tw := tar.NewWriter(&buf)
+	dirs := map[string]bool{}
+	for _, name := range names {
+		file := "by_name/" + name + "/raw"
+		var parents []string
+		for dir := filepath.Dir(file); dir != "." && !dirs[dir]; dir = filepath.Dir(dir) {
+			parents = append(parents, dir)
+			dirs[dir] = true
+		}
+		for i := len(parents) - 1; i >= 0; i-- {
+			if err := tw.WriteHeader(&tar.Header{Typeflag: tar.TypeDir, Name: parents[i] + "/", Mode: 0o700}); err != nil {
+				return err
+			}
+		}
+		b := entries[name]
+		if err := tw.WriteHeader(&tar.Header{Typeflag: tar.TypeReg, Name: file, Mode: 0o600, Size: int64(len(b))}); err != nil {
+			return err
+		}
+		if _, err := tw.Write(b); err != nil {
+			return err
+		}
+	}
+	if err := tw.Close(); err != nil {
+		return err
+	}
+	return os.WriteFile(path, buf.Bytes(), 0o600)
+}
+
+type fcConfig struct {
+	BootSource struct {
+		Kernel   string `json:"kernel_image_path"`
+		Initrd   string `json:"initrd_path"`
+		BootArgs string `json:"boot_args"`
+	} `json:"boot-source"`
+	Drives  []fcDrive `json:"drives"`
+	Machine struct {
+		VCPUs  int  `json:"vcpu_count"`
+		MemMiB int  `json:"mem_size_mib"`
+		SMT    bool `json:"smt"`
+	} `json:"machine-config"`
+	Net []fcNet `json:"network-interfaces"`
+}
+
+type fcDrive struct {
+	ID       string `json:"drive_id"`
+	Path     string `json:"path_on_host"`
+	Root     bool   `json:"is_root_device"`
+	ReadOnly bool   `json:"is_read_only"`
+}
+
+type fcNet struct {
+	ID  string `json:"iface_id"`
+	MAC string `json:"guest_mac"`
+	Tap string `json:"host_dev_name"`
+}
+
+// writeFirecrackerConfig writes the attempt's Firecracker config: the sealed
+// kernel and initramfs, the package disk then the configuration disk, and the tap.
+func writeFirecrackerConfig(path, pkg, cfgDisk, tap, mac string, memMiB int) error {
+	var c fcConfig
+	c.BootSource.Kernel = fmt.Sprintf("/proc/self/fd/%d", fcKernelFD)
+	c.BootSource.Initrd = fmt.Sprintf("/proc/self/fd/%d", fcInitrdFD)
+	c.BootSource.BootArgs = firecrackerCmdline
+	c.Drives = []fcDrive{{ID: "pkg", Path: pkg, ReadOnly: true}, {ID: "cfg", Path: cfgDisk}}
+	c.Machine.VCPUs, c.Machine.MemMiB = 1, memMiB
+	c.Net = []fcNet{{ID: "net0", MAC: mac, Tap: tap}}
+	b, err := json.Marshal(c)
+	if err != nil {
+		return err
+	}
+	return os.WriteFile(path, b, 0o600)
+}
+
+// vmHandle owns Firecracker's pidfd across the session callback, which can
+// still be running when attempt.Run returns; whichever side finishes last
+// closes an abandoned fd.
+type vmHandle struct {
+	mu        sync.Mutex
+	fd        int
+	finished  bool
+	abandoned bool
+}
+
+func (h *vmHandle) set(fd int) {
+	h.mu.Lock()
+	defer h.mu.Unlock()
+	h.fd = fd
+}
+
+// callbackDone runs as the callback returns.
+func (h *vmHandle) callbackDone() {
+	h.mu.Lock()
+	defer h.mu.Unlock()
+	h.finished = true
+	h.closeIfAbandoned()
+}
+
+// abandon gives up the fd while the callback may still run.
+func (h *vmHandle) abandon() {
+	h.mu.Lock()
+	defer h.mu.Unlock()
+	h.abandoned = true
+	h.closeIfAbandoned()
+}
+
+func (h *vmHandle) closeIfAbandoned() {
+	if h.abandoned && h.finished && h.fd >= 0 {
+		_ = unix.Close(h.fd)
+		h.fd = -1
+	}
+}
+
+// take hands the fd to teardown once the callback has returned.
+func (h *vmHandle) take() int {
+	h.mu.Lock()
+	defer h.mu.Unlock()
+	fd := h.fd
+	h.fd = -1
+	return fd
+}

spinifex/runtime/compute/lambda/testlink/firecracker_test.go

@@ -0,0 +1,190 @@
+//go:build lambdatestlink
+
+//test:in-package drives the unexported supervisor, ack and pidfd helpers against real processes
+
+package testlink
+
+import (
+	"archive/tar"
+	"bytes"
+	"context"
+	"errors"
+	"io"
+	"os"
+	"os/exec"
+	"path/filepath"
+	"strconv"
+	"strings"
+	"syscall"
+	"testing"
+	"time"
+
+	lambdav1 "github.com/mulgadc/spinifex/contracts/lambda/v1"
+	"github.com/mulgadc/spinifex/spinifex/agents/lambda/hostlink"
+	"github.com/mulgadc/spinifex/spinifex/runtime/compute/lambda/attempt"
+	"github.com/mulgadc/spinifex/spinifex/runtime/compute/lambda/guestconn"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"golang.org/x/sys/unix"
+)
+
+// supervised starts this test binary as the Firecracker supervisor over a
+// stand-in VMM that runs until killed, and returns the supervisor and the
+// VMM's pidfd from the ack.
+func supervised(t *testing.T, deadline time.Duration) (*exec.Cmd, int) {
+	t.Helper()
+	dir := t.TempDir()
+	vmm := filepath.Join(dir, "firecracker")
+	require.NoError(t, os.WriteFile(vmm, []byte("#!/bin/sh\nexec tail -f /dev/null\n"), 0o755))
+	kernel, err := os.Open(os.DevNull)
+	require.NoError(t, err)
+	initrd, err := os.Open(os.DevNull)
+	require.NoError(t, err)
+	t.Cleanup(func() { _ = kernel.Close(); _ = initrd.Close() })
+	d := &Daemon{Firecracker: vmm, Image: &Image{Kernel: kernel, Initrd: initrd}}
+	ack, child, err := ackPair()
+	require.NoError(t, err)
+	t.Cleanup(func() { _ = ack.Close() })
+	cmd, err := d.supervisorCommand(filepath.Join(dir, fcConfigFile), deadline, child)
+	require.NoError(t, err)
+	cmd.Path = os.Args[0]
+	cmd.Stdout, cmd.Stderr = io.Discard, io.Discard
+	require.NoError(t, cmd.Start())
+	_ = child.Close()
+	t.Cleanup(func() { _ = cmd.Process.Kill(); _ = cmd.Wait() })
+
+	pidfd, pid, err := awaitFirecracker(context.Background(), ack, cmd.Process.Pid, armWait)
+	require.NoError(t, err)
+	t.Cleanup(func() { _ = unix.PidfdSendSignal(pidfd, unix.SIGKILL, nil, 0); _ = unix.Close(pidfd) })
+	require.NotEqual(t, cmd.Process.Pid, pid)
+	return cmd, pidfd
+}
+
+func boottime(t *testing.T) time.Duration {
+	t.Helper()
+	now, err := bootNow()
+	require.NoError(t, err)
+	return now
+}
+
+func TestFirecrackerSupervisorKillsAtBoottimeDeadlineWhileStopped(t *testing.T) {
+	cmd, pidfd := supervised(t, boottime(t)+1500*time.Millisecond)
+
+	// The kernel holds a SIGKILL timer on CLOCK_BOOTTIME (7), so the deadline counts suspend.
+	timers, err := os.ReadFile("/proc/" + strconv.Itoa(cmd.Process.Pid) + "/timers")
+	require.NoError(t, err)
+	assert.Contains(t, string(timers), "signal: 9/")
+	assert.Contains(t, string(timers), "ClockID: 7")
+
+	// A stopped supervisor still loses the VMM at the deadline: both kills come from the kernel.
+	require.NoError(t, cmd.Process.Signal(syscall.SIGSTOP))
+	exited, err := pidfdExited(pidfd, 5*time.Second)
+	require.NoError(t, err)
+	assert.True(t, exited, "VMM outlived its deadline")
+	state, err := cmd.Process.Wait()
+	require.NoError(t, err)
+	assert.Equal(t, syscall.SIGKILL, state.Sys().(syscall.WaitStatus).Signal())
+}
+
+func TestFirecrackerDiesWithItsSupervisor(t *testing.T) {
+	cmd, pidfd := supervised(t, boottime(t)+time.Hour)
+	exited, err := pidfdExited(pidfd, 0)
+	require.NoError(t, err)
+	require.False(t, exited)
+
+	require.NoError(t, cmd.Process.Kill())
+	_, _ = cmd.Process.Wait()
+	// Only the kernel's parent-death signal can end the VMM here; nothing else signals it.
+	exited, err = pidfdExited(pidfd, 5*time.Second)
+	require.NoError(t, err)
+	assert.True(t, exited, "VMM outlived its supervisor")
+}
+
+// A supervisor that never started launched nothing, so the attempt is
+// released; one that started without handing over a pidfd may have started
+// Firecracker, so the attempt is held.
+func TestSettleFirecracker_OnlyANeverStartedSupervisorIsProvenStopped(t *testing.T) {
+	cancelled, cancel := context.WithCancel(context.Background())
+	cancel()
+	cases := map[string]struct {
+		ctx  context.Context
+		cmd  func() *exec.Cmd
+		want bool
+	}{
+		"cancelled before Start": {cancelled, func() *exec.Cmd { return exec.Command("/bin/true") }, true},
+		"failed Start":           {context.Background(), func() *exec.Cmd { return exec.Command(filepath.Join(t.TempDir(), "missing")) }, true},
+		"started, no pidfd":      {context.Background(), func() *exec.Cmd { return exec.Command("/bin/true") }, false},
+	}
+	for name, tc := range cases {
+		t.Run(name, func(t *testing.T) {
+			cmd := tc.cmd()
+			fc := &vmHandle{fd: -1}
+			out, _ := attempt.Run(tc.ctx, cmd, 10*time.Second, func(context.Context, int) error {
+				fc.callbackDone()
+				return errors.New("supervisor exited before its ack")
+			})
+			require.True(t, out.Stopped, "the supervisor itself is proven stopped")
+			<-out.SessionDone
+			assert.Equal(t, tc.want, settleFirecracker(out.SessionDone, fc, cmd.Process != nil))
+		})
+	}
+}
+
+func TestFirecrackerAckRequiresTheSupervisorsChild(t *testing.T) {
+	ack, child, err := ackPair()
+	require.NoError(t, err)
+	defer func() { _ = ack.Close() }()
+	self, err := unix.PidfdOpen(os.Getpid(), 0)
+	require.NoError(t, err)
+	require.NoError(t, unix.Sendmsg(int(child.Fd()), []byte(fcAck), unix.UnixRights(self), nil, 0))
+	_ = unix.Close(self)
+	_ = child.Close()
+
+	_, _, err = awaitFirecracker(context.Background(), ack, 1, time.Second)
+	require.ErrorIs(t, err, ErrNotArmed)
+	assert.Contains(t, err.Error(), "not the supervisor's child")
+}
+
+// The archive the daemon writes is read by the guest's own identity reader
+// once extracted, as the runtime image's init extracts it.
+func TestConfigArchiveIsReadByTheGuestReaders(t *testing.T) {
+	gen, err := guestconn.NewGeneration("ap-southeast-2/000000000001/fn/env/1", time.Hour)
+	require.NoError(t, err)
+	entries := map[string][]byte{fwCfgNetcfg: []byte("NIC0_MAC=02:00:00:00:00:01\n")}
+	for name, b := range gen.GuestEntries("192.0.2.1:4000") {
+		entries[lambdav1.FwCfgPrefix+name] = b
+	}
+	path := filepath.Join(t.TempDir(), configArchive)
+	require.NoError(t, writeConfigArchive(path, entries))
+
+	root := filepath.Join(t.TempDir(), "cfg")
+	extract(t, path, root)
+	_, err = hostlink.LoadIdentity(filepath.Join(root, "by_name", lambdav1.FwCfgPrefix))
+	require.NoError(t, err)
+	netcfg, err := os.ReadFile(filepath.Join(root, "by_name", fwCfgNetcfg, "raw"))
+	require.NoError(t, err)
+	assert.Equal(t, entries[fwCfgNetcfg], netcfg)
+}
+
+func extract(t *testing.T, archive, root string) {
+	t.Helper()
+	b, err := os.ReadFile(archive)
+	require.NoError(t, err)
+	tr := tar.NewReader(bytes.NewReader(b))
+	for {
+		h, err := tr.Next()
+		if err == io.EOF {
+			return
+		}
+		require.NoError(t, err)
+		require.False(t, strings.Contains(h.Name, ".."), h.Name)
+		p := filepath.Join(root, h.Name)
+		if h.Typeflag == tar.TypeDir {
+			require.NoError(t, os.MkdirAll(p, 0o700))
+			continue
+		}
+		data, err := io.ReadAll(tr)
+		require.NoError(t, err)
+		require.NoError(t, os.WriteFile(p, data, 0o600))
+	}
+}

spinifex/runtime/compute/lambda/testlink/readiness.go

@@ -0,0 +1,38 @@
+//go:build lambdatestlink
+
+package testlink
+
+import (
+	"context"
+
+	"github.com/mulgadc/spinifex/spinifex/domains/lambda/function"
+)
+
+// Readiness admits a function the test link can run, and only while d serves
+// the pinned image on the cluster's only node, so no gateway or daemon is
+// handed an Active function it cannot run.
+func (d *Daemon) Readiness(imagePin string, nodes int) function.Readiness {
+	return func(_ context.Context, rec function.Record) error {
+		if rec.EnvCipher != "" {
+			return notReady("The function cannot run here: test-link Lambda does not support environment variables.")
+		}
+		if !validHandler(rec.Handler) {
+			return notReady("The function cannot run here: test-link Lambda runs only a handler of 1 to 128 characters with no whitespace.")
+		}
+		if rec.Code.Size <= 0 || rec.Code.Size > MaxCodeBytes {
+			return notReady("The function cannot run here: its code is larger than test-link Lambda accepts.")
+		}
+		if nodes != 1 {
+			return notReady("The function cannot run here: test-link Lambda activates only on a single-node cluster.")
+		}
+		if !d.Serving() || d.Image == nil || d.Image.Digest != imagePin {
+			return notReady(function.ReasonCreating)
+		}
+		return nil
+	}
+}
+
+// notReady is an unmet activation condition, worded as the StateReason.
+type notReady string
+
+func (n notReady) Error() string { return string(n) }

spinifex/runtime/compute/lambda/testlink/readiness_test.go

@@ -0,0 +1,173 @@
+//go:build lambdatestlink
+
+//test:in-package — reuses the package fixtures writeImage, testImageDigest and controlName
+package testlink
+
+import (
+	"context"
+	"os"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/mulgadc/spinifex/internal/testkit"
+	"github.com/mulgadc/spinifex/spinifex/domains/lambda/function"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+const (
+	readyAccount = "000000000001"
+	readyRegion  = "ap-southeast-2"
+)
+
+// storedFunction creates a function through the real owner and records its
+// code as stored, leaving it Pending the way CreateFunction does.
+func storedFunction(t *testing.T, svc *function.MetaService, name string) {
+	t.Helper()
+	storedFunctionWith(t, svc, name, "", 9)
+}
+
+func storedFunctionWith(t *testing.T, svc *function.MetaService, name, envCipher string, size int64) {
+	t.Helper()
+	storedFunctionHandler(t, svc, name, "bootstrap", envCipher, size)
+}
+
+func storedFunctionHandler(t *testing.T, svc *function.MetaService, name, handler, envCipher string, size int64) {
+	t.Helper()
+	created, err := svc.Create(t.Context(), &function.CreateRequest{Region: readyRegion, Record: function.Record{
+		FunctionID: "id-" + name, FunctionName: name, Role: "arn:aws:iam::000000000001:role/r",
+		Runtime: "provided.al2023", Handler: handler, Timeout: 3, MemorySize: 128, EnvCipher: envCipher,
+		Code: function.Code{RevisionID: "rev-" + name},
+	}}, readyAccount)
+	require.NoError(t, err)
+	_, err = svc.CodeStored(t.Context(), &function.CodeStoredRequest{Region: readyRegion, FunctionName: name,
+		Revision: created.Revision, SHA256: testImageDigest, Size: size}, readyAccount)
+	require.NoError(t, err)
+}
+
+func functionState(t *testing.T, svc *function.MetaService, name string) function.Record {
+	t.Helper()
+	got, err := svc.Get(t.Context(), &function.GetRequest{Region: readyRegion, FunctionName: name}, readyAccount)
+	require.NoError(t, err)
+	require.True(t, got.Found)
+	return got.Record
+}
+
+func readyOwner(t *testing.T) *function.MetaService {
+	t.Helper()
+	_, nc, _ := testutil.StartTestJetStream(t)
+	return function.NewMetaService(testutil.NewJetStream(t, nc))
+}
+
+func TestReadiness_ActivatesOnlyOnceTheDaemonServesThePinnedImage(t *testing.T) {
+	svc := readyOwner(t)
+	im, err := PrepareImage(writeImage(t), testImageDigest)
+	require.NoError(t, err)
+	t.Cleanup(im.Close)
+	d := &Daemon{Node: "node1", Image: im}
+	svc.SetReadiness(d.Readiness(testImageDigest, 1))
+	storedFunction(t, svc, "hello")
+
+	require.NoError(t, svc.ActivatePending(t.Context()))
+	waiting := functionState(t, svc, "hello")
+	assert.Equal(t, function.StatePending, waiting.State, "a prepared image that is not served activates nothing")
+	assert.Equal(t, function.ReasonCreating, waiting.StateReason)
+
+	ctx, cancel := context.WithCancel(t.Context())
+	require.NoError(t, d.ServeControl(ctx, controlName(t), os.Getuid()))
+	require.NoError(t, svc.ActivatePending(t.Context()))
+	active := functionState(t, svc, "hello")
+	assert.Equal(t, function.StateActive, active.State)
+	assert.Empty(t, active.StateReason)
+
+	cancel()
+	require.Eventually(t, func() bool { return !d.Serving() }, 5*time.Second, 10*time.Millisecond,
+		"a stopped control socket is reported as not serving")
+	storedFunction(t, svc, "later")
+	require.NoError(t, svc.ActivatePending(t.Context()))
+	assert.Equal(t, function.StatePending, functionState(t, svc, "later").State)
+}
+
+func TestReadiness_RefusesAnotherPinOrAnotherNode(t *testing.T) {
+	im, err := PrepareImage(writeImage(t), testImageDigest)
+	require.NoError(t, err)
+	t.Cleanup(im.Close)
+	d := &Daemon{Node: "node1", Image: im}
+	require.NoError(t, d.ServeControl(t.Context(), controlName(t), os.Getuid()))
+
+	for _, tc := range []struct {
+		name   string
+		pin    string
+		nodes  int
+		reason string
+	}{
+		{"other pin", "0000000000000000000000000000000000000000000000000000000000000000", 1, function.ReasonCreating},
+		{"two nodes", testImageDigest, 2, "The function cannot run here: test-link Lambda activates only on a single-node cluster."},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			svc := readyOwner(t)
+			svc.SetReadiness(d.Readiness(tc.pin, tc.nodes))
+			storedFunction(t, svc, "hello")
+			require.NoError(t, svc.ActivatePending(t.Context()))
+			got := functionState(t, svc, "hello")
+			assert.Equal(t, function.StatePending, got.State)
+			assert.Equal(t, tc.reason, got.StateReason)
+		})
+	}
+}
+
+func TestReadiness_RefusesFunctionsTheTestLinkCannotRun(t *testing.T) {
+	im, err := PrepareImage(writeImage(t), testImageDigest)
+	require.NoError(t, err)
+	t.Cleanup(im.Close)
+	d := &Daemon{Node: "node1", Image: im}
+	require.NoError(t, d.ServeControl(t.Context(), controlName(t), os.Getuid()))
+	svc := readyOwner(t)
+	svc.SetReadiness(d.Readiness(testImageDigest, 1))
+	storedFunctionWith(t, svc, "env", "sealed", 9)
+	storedFunctionWith(t, svc, "huge", "", MaxCodeBytes+1)
+	storedFunctionWith(t, svc, "edge", "", MaxCodeBytes)
+	storedFunctionHandler(t, svc, "spaced", "boot\u00a0strap", "", 9)
+	storedFunctionHandler(t, svc, "longhandler", strings.Repeat("h", 129), "", 9)
+	storedFunctionHandler(t, svc, "edgehandler", strings.Repeat("h", 128), "", 9)
+
+	require.NoError(t, svc.ActivatePending(t.Context()))
+	assert.Equal(t, "The function cannot run here: test-link Lambda does not support environment variables.", functionState(t, svc, "env").StateReason)
+	assert.Equal(t, function.StatePending, functionState(t, svc, "env").State)
+	assert.Equal(t, "The function cannot run here: its code is larger than test-link Lambda accepts.", functionState(t, svc, "huge").StateReason)
+	assert.Equal(t, function.StatePending, functionState(t, svc, "huge").State)
+	assert.Equal(t, function.StateActive, functionState(t, svc, "edge").State, "the largest accepted code copy activates")
+	for _, name := range []string{"spaced", "longhandler"} {
+		assert.Equal(t, function.StatePending, functionState(t, svc, name).State, name)
+		assert.Equal(t, "The function cannot run here: test-link Lambda runs only a handler of 1 to 128 characters with no whitespace.", functionState(t, svc, name).StateReason, name)
+	}
+	assert.Equal(t, function.StateActive, functionState(t, svc, "edgehandler").State, "the longest accepted handler activates")
+}
+
+func TestReadiness_ServingEndsWhenAcceptingStops(t *testing.T) {
+	d := &Daemon{Node: "node1"}
+	name := controlName(t)
+	ln, err := ListenControl(name)
+	require.NoError(t, err)
+	inHandle, release := make(chan struct{}), make(chan struct{})
+	srv := d.controlServer(os.Getuid())
+	srv.Handle = func(context.Context, []byte, Credential) Reply {
+		close(inHandle)
+		<-release
+		return Reply{Outcome: OutcomeDone}
+	}
+	d.serve(t.Context(), ln, srv)
+	require.True(t, d.Serving())
+	go func() { _, _ = (&ControlClient{Name: name, DaemonUID: os.Getuid(), RequestLimit: MaxFrameBytes, ReplyLimit: MaxFrameBytes, IOTimeout: 5 * time.Second}).Exchange(context.Background(), []byte("{}"), testCred, time.Now().Add(5*time.Second)) }()
+	select {
+	case <-inHandle:
+	case <-time.After(5 * time.Second):
+		t.Fatal("the request never reached Handle")
+	}
+	defer close(release)
+
+	require.NoError(t, ln.Close(), "the listener fails under an outstanding invocation")
+	assert.Eventually(t, func() bool { return !d.Serving() }, 2*time.Second, 10*time.Millisecond,
+		"serving ends while the invocation is still held, not after the drain")
+}

spinifex/runtime/daemon/daemon.go

@@ -46,6 +46,7 @@ import (
 	acmdomain "github.com/mulgadc/spinifex/spinifex/domains/acm"
 	"github.com/mulgadc/spinifex/spinifex/domains/dns"
 	ec2account "github.com/mulgadc/spinifex/spinifex/domains/ec2/account"
+	ec2instanceapi "github.com/mulgadc/spinifex/spinifex/domains/ec2/awsapi/instance"
 	ec2eigw "github.com/mulgadc/spinifex/spinifex/domains/ec2/eigw"
 	ec2eip "github.com/mulgadc/spinifex/spinifex/domains/ec2/eip"
 	ec2igw "github.com/mulgadc/spinifex/spinifex/domains/ec2/igw"
@@ -1052,6 +1053,7 @@ func (d *Daemon) subscribeAll() error {
 		{"ec2.terminate", handleNATSRequest(d.node, d.instanceService.TerminateStoppedInstance), "spinifex-workers"},
 		{"ec2.DescribeStoppedInstances", handleNATSRequest(d.node, d.instanceService.DescribeStoppedInstances), "spinifex-workers"},
 		{"ec2.DescribeAccountInstanceRecords", handleNATSRequest(d.node, d.instanceService.DescribeAccountInstanceRecords), "spinifex-workers"},
+		{ec2instanceapi.AccountTeardownOutstandingSubject, handleNATSRequest(d.node, d.instanceService.AccountTeardownOutstandingRPC), "spinifex-workers"},
 		{"ec2.DescribeTerminatedInstances", handleNATSRequest(d.node, d.instanceService.DescribeTerminatedInstances), "spinifex-workers"},
 		// these fan out to all nodes and gateway aggregates the results. The
 		// handler only sees per-daemon local state (vmMgr/stoppedStore), so

spinifex/runtime/daemon/daemon_handlers.go

@@ -81,11 +81,18 @@ func respondWithError(nodeID string, msg *nats.Msg, errCode string) {
 // an opaque ServerInternal, leaving the reason visible only in the daemon log.
 // Mirrors natsmsg.ServeNATSRequestCtx, which has always preserved the message.
 func respondWithServiceError(nodeID string, msg *nats.Msg, err error) {
+	respondWithServiceErrorHeader(nodeID, msg, err, nil)
+}
+
+// respondWithServiceErrorHeader is respondWithServiceError with headers the
+// caller wants on the reply in addition to the ones the error implies.
+func respondWithServiceErrorHeader(nodeID string, msg *nats.Msg, err error, base nats.Header) {
 	_, message, _ := awserrors.ResolveErrorDetail(err)
 	payload := awserrors.GenerateErrorPayloadWithMessage(awserrors.ValidErrorCodeFromError(err), message)
 	// The AWS body stays as it is; the header is how ec2.start tells its
 	// forwarder this refusal must not be answered by starting elsewhere.
 	extra := nats.Header{}
+	maps.Copy(extra, base)
 	if _, ok := errors.AsType[*ec2instance.UnsealedWritesError](err); ok {
 		extra.Set(natsmsg.UnsealedWritesHeader, "true")
 	}
@@ -246,6 +253,10 @@ func (d *Daemon) dispatchEC2Command(msg *nats.Msg) (string, string) {
 		return name, d.handleSetInstanceTags(ctx, msg, command, instance)
 	case command.Attributes.SetInstanceMonitoring:
 		return name, d.handleSetInstanceMonitoring(ctx, msg, command, instance)
+	case command.Attributes.ModifyInstanceMetadataOptions:
+		return name, d.handleModifyInstanceMetadataOptions(ctx, msg, command, instance)
+	case command.Attributes.SetDisableApiTermination:
+		return name, d.handleSetDisableApiTermination(ctx, msg, command, instance)
 	case command.Attributes.StartInstance:
 		opCtx, opSpan := startOpSpan(ctx, "ec2.StartInstance", instance.ID)
 		err := d.instanceService.StartInstance(opCtx, instance, command)
@@ -314,6 +325,10 @@ func ec2CommandName(command ec2v1.EC2InstanceCommand) string {
 		return "RemoveInstanceTags"
 	case command.Attributes.SetInstanceMonitoring:
 		return "SetInstanceMonitoring"
+	case command.Attributes.ModifyInstanceMetadataOptions:
+		return "ModifyInstanceMetadataOptions"
+	case command.Attributes.SetDisableApiTermination:
+		return "SetDisableApiTermination"
 	case command.Attributes.StartInstance:
 		return "StartInstance"
 	case command.Attributes.RebootInstance:

spinifex/runtime/daemon/daemon_handlers_instance.go

@@ -360,3 +360,76 @@ func relayStartReply(msg, resp *nats.Msg) error {
 	maps.Copy(reply.Header, resp.Header)
 	return msg.RespondMsg(reply)
 }
+
+// handledHeader marks a reply as sent by the instance's owner, so the
+// forwarder can tell an owner's answer from a stale or missing responder.
+func handledHeader() nats.Header {
+	h := nats.Header{}
+	h.Set(ec2v1.HandledHeader, "1")
+	return h
+}
+
+// respondHandledError answers with err and the handled marker.
+func respondHandledError(nodeID string, msg *nats.Msg, err error) string {
+	respondWithServiceErrorHeader(nodeID, msg, err, handledHeader())
+	return outcomeForError(err)
+}
+
+// respondHandledJSON answers with a JSON body and the handled marker.
+func respondHandledJSON(nodeID string, msg *nats.Msg, out any) string {
+	body, err := json.Marshal(out)
+	if err != nil {
+		return respondHandledError(nodeID, msg, errors.New(awserrors.ErrorServerInternal))
+	}
+	respondNATSMsgWithHeader(nodeID, msg, body, handledHeader())
+	return outcomeSuccess
+}
+
+// handleModifyInstanceMetadataOptions applies a metadata-options change on the
+// node that runs the instance, through the same service method as the
+// gateway-local path.
+func (d *Daemon) handleModifyInstanceMetadataOptions(ctx context.Context, msg *nats.Msg, command ec2v1.EC2InstanceCommand, instance *vm.VM) string {
+	data := command.InstanceMetadataOptionsData
+	if data == nil {
+		return respondHandledError(d.node, msg, errors.New(awserrors.ErrorMissingParameter))
+	}
+	input := &ec2.ModifyInstanceMetadataOptionsInput{
+		InstanceId:              aws.String(instance.ID),
+		HttpPutResponseHopLimit: data.HttpPutResponseHopLimit,
+	}
+	if data.HttpTokens != "" {
+		input.HttpTokens = aws.String(data.HttpTokens)
+	}
+	if data.HttpEndpoint != "" {
+		input.HttpEndpoint = aws.String(data.HttpEndpoint)
+	}
+	if data.HttpProtocolIpv6 != "" {
+		input.HttpProtocolIpv6 = aws.String(data.HttpProtocolIpv6)
+	}
+	if data.InstanceMetadataTags != "" {
+		input.InstanceMetadataTags = aws.String(data.InstanceMetadataTags)
+	}
+	out, err := d.instanceService.ModifyInstanceMetadataOptions(ctx, input, natsmsg.AccountIDFromMsg(msg))
+	if err != nil {
+		return respondHandledError(d.node, msg, err)
+	}
+	return respondHandledJSON(d.node, msg, out)
+}
+
+// handleSetDisableApiTermination sets termination protection on the node that
+// runs the instance. Absent data is a validation error, never "false".
+func (d *Daemon) handleSetDisableApiTermination(ctx context.Context, msg *nats.Msg, command ec2v1.EC2InstanceCommand, instance *vm.VM) string {
+	data := command.DisableApiTerminationData
+	if data == nil {
+		return respondHandledError(d.node, msg, errors.New(awserrors.ErrorMissingParameter))
+	}
+	input := &ec2.ModifyInstanceAttributeInput{
+		InstanceId:            aws.String(instance.ID),
+		DisableApiTermination: &ec2.AttributeBooleanValue{Value: aws.Bool(data.Value)},
+	}
+	out, err := d.instanceService.ModifyInstanceAttribute(ctx, input, natsmsg.AccountIDFromMsg(msg))
+	if err != nil {
+		return respondHandledError(d.node, msg, err)
+	}
+	return respondHandledJSON(d.node, msg, out)
+}

spinifex/runtime/daemon/daemon_handlers_owner_routing_test.go

@@ -0,0 +1,323 @@
+package daemon
+
+import (
+	"bytes"
+	"encoding/json"
+	"log/slog"
+	"reflect"
+	"strings"
+	"testing"
+
+	"github.com/aws/aws-sdk-go/aws"
+	"github.com/aws/aws-sdk-go/service/ec2"
+	"github.com/mulgadc/spinifex/contracts/ec2/v1"
+	"github.com/mulgadc/spinifex/internal/testkit"
+	ec2instanceapi "github.com/mulgadc/spinifex/spinifex/domains/ec2/awsapi/instance"
+	ec2instance "github.com/mulgadc/spinifex/spinifex/domains/ec2/instance"
+	"github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
+	"github.com/mulgadc/spinifex/spinifex/providers/objectstore"
+	"github.com/mulgadc/spinifex/spinifex/runtime/compute/vm"
+	vmmock "github.com/mulgadc/spinifex/spinifex/runtime/compute/vm/mock"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func metadataCommand(id string, data *ec2v1.InstanceMetadataOptionsData) ec2v1.EC2InstanceCommand {
+	return ec2v1.EC2InstanceCommand{
+		ID:                          id,
+		Attributes:                  ec2v1.EC2CommandAttributes{ModifyInstanceMetadataOptions: true},
+		InstanceMetadataOptionsData: data,
+	}
+}
+
+func terminationCommand(id string, data *ec2v1.DisableApiTerminationData) ec2v1.EC2InstanceCommand {
+	return ec2v1.EC2InstanceCommand{
+		ID:                        id,
+		Attributes:                ec2v1.EC2CommandAttributes{SetDisableApiTermination: true},
+		DisableApiTerminationData: data,
+	}
+}
+
+func TestOwnerDispatch_ModifyInstanceMetadataOptions(t *testing.T) {
+	t.Run("success applies the change and sets the handled header", func(t *testing.T) {
+		const id = "i-or-meta-ok"
+		d, _ := monitoringTestDaemon(t, id, false)
+		cmd := metadataCommand(id, &ec2v1.InstanceMetadataOptionsData{
+			HttpTokens:              ec2.HttpTokensStateOptional,
+			HttpPutResponseHopLimit: aws.Int64(3),
+		})
+
+		reply := requestHandler(t, d.natsConn, "ec2.cmd."+id, d.handleEC2Events, testAccountID, mustMarshal(t, cmd))
+		assert.Equal(t, "1", reply.Header.Get(ec2v1.HandledHeader))
+
+		got, ok := d.vmMgr.Get(id)
+		require.True(t, ok)
+		require.NotNil(t, got.Instance.MetadataOptions)
+		assert.Equal(t, ec2.HttpTokensStateOptional, aws.StringValue(got.Instance.MetadataOptions.HttpTokens))
+		assert.Equal(t, int64(3), aws.Int64Value(got.Instance.MetadataOptions.HttpPutResponseHopLimit))
+	})
+
+	t.Run("service error still sets the handled header", func(t *testing.T) {
+		const id = "i-or-meta-bad"
+		d, _ := monitoringTestDaemon(t, id, false)
+		cmd := metadataCommand(id, &ec2v1.InstanceMetadataOptionsData{HttpTokens: "bogus"})
+
+		reply := requestHandler(t, d.natsConn, "ec2.cmd."+id, d.handleEC2Events, testAccountID, mustMarshal(t, cmd))
+		assert.Equal(t, "1", reply.Header.Get(ec2v1.HandledHeader))
+		assert.Equal(t, awserrors.ErrorUnsupportedOperation, decodeError(t, reply.Data)["Code"])
+	})
+
+	t.Run("missing data is a validation error with the handled header", func(t *testing.T) {
+		const id = "i-or-meta-nodata"
+		d, _ := monitoringTestDaemon(t, id, false)
+
+		reply := requestHandler(t, d.natsConn, "ec2.cmd."+id, d.handleEC2Events, testAccountID, mustMarshal(t, metadataCommand(id, nil)))
+		assert.Equal(t, "1", reply.Header.Get(ec2v1.HandledHeader))
+		assert.Equal(t, awserrors.ErrorMissingParameter, decodeError(t, reply.Data)["Code"])
+	})
+}
+
+func TestOwnerDispatch_SetDisableApiTermination(t *testing.T) {
+	protection := func(t *testing.T, d *Daemon, id string) *bool {
+		t.Helper()
+		got, ok := d.vmMgr.Get(id)
+		require.True(t, ok)
+		return got.RunInstancesInput.DisableApiTermination
+	}
+
+	t.Run("true then explicit false are both applied", func(t *testing.T) {
+		const id = "i-or-term-ok"
+		d, _ := monitoringTestDaemon(t, id, false)
+
+		reply := requestHandler(t, d.natsConn, "ec2.cmd."+id, d.handleEC2Events, testAccountID,
+			mustMarshal(t, terminationCommand(id, &ec2v1.DisableApiTerminationData{Value: true})))
+		assert.Equal(t, "1", reply.Header.Get(ec2v1.HandledHeader))
+		require.NotNil(t, protection(t, d, id))
+		assert.True(t, *protection(t, d, id))
+
+		reply = requestHandler(t, d.natsConn, "ec2.cmd."+id, d.handleEC2Events, testAccountID,
+			mustMarshal(t, terminationCommand(id, &ec2v1.DisableApiTerminationData{Value: false})))
+		assert.Equal(t, "1", reply.Header.Get(ec2v1.HandledHeader))
+		require.NotNil(t, protection(t, d, id))
+		assert.False(t, *protection(t, d, id))
+	})
+
+	t.Run("missing data is an error and never clears protection", func(t *testing.T) {
+		const id = "i-or-term-nodata"
+		d, _ := monitoringTestDaemon(t, id, false)
+		d.vmMgr.UpdateState(id, func(v *vm.VM) { v.RunInstancesInput.DisableApiTermination = aws.Bool(true) })
+
+		reply := requestHandler(t, d.natsConn, "ec2.cmd."+id, d.handleEC2Events, testAccountID, mustMarshal(t, terminationCommand(id, nil)))
+		assert.Equal(t, "1", reply.Header.Get(ec2v1.HandledHeader))
+		assert.Equal(t, awserrors.ErrorMissingParameter, decodeError(t, reply.Data)["Code"])
+		require.NotNil(t, protection(t, d, id))
+		assert.True(t, *protection(t, d, id))
+	})
+}
+
+// ownerRoutingDaemon is a daemon on url with its own vmMgr and the shared
+// stopped store.
+func ownerRoutingDaemon(t *testing.T, url string, stopped *vmmock.StateStore) *Daemon {
+	t.Helper()
+	d := createTestDaemon(t, url)
+	d.instanceService = ec2instance.NewInstanceServiceImpl(
+		d.config, d.resourceMgr.instanceTypes, d.natsConn,
+		objectstore.NewMemoryObjectStore(), d.vmMgr, d.resourceMgr, stopped)
+	return d
+}
+
+func TestModifyInstanceMetadataOptions_NonOwnerReceivesQueueOwnerAnswers(t *testing.T) {
+	const id = "i-or-multinode"
+	ns, gateway := testutil.StartTestNATS(t)
+	stopped := vmmock.New()
+
+	// B runs the instance; A does not and is the only queue-group member.
+	b := ownerRoutingDaemon(t, ns.ClientURL(), stopped)
+	b.vmMgr.Insert(&vm.VM{
+		ID:        id,
+		Status:    vm.StateRunning,
+		AccountID: testAccountID,
+		Instance:  &ec2.Instance{InstanceId: aws.String(id)},
+	})
+	subscribeOwner(t, b, id)
+
+	a := ownerRoutingDaemon(t, ns.ClientURL(), stopped)
+	sub, err := a.natsConn.QueueSubscribe("ec2.ModifyInstanceMetadataOptions", "spinifex-workers",
+		asMsgHandler(handleNATSRequest(a.node, a.instanceService.ModifyInstanceMetadataOptions)))
+	require.NoError(t, err)
+	t.Cleanup(func() { _ = sub.Unsubscribe() })
+	require.NoError(t, a.natsConn.Flush())
+
+	for range 9 {
+		out, err := ec2instanceapi.ModifyInstanceMetadataOptions(t.Context(), &ec2.ModifyInstanceMetadataOptionsInput{
+			InstanceId:              aws.String(id),
+			HttpPutResponseHopLimit: aws.Int64(3),
+		}, gateway, testAccountID)
+		require.NoError(t, err)
+		assert.Equal(t, id, aws.StringValue(out.InstanceId))
+		assert.Equal(t, int64(3), aws.Int64Value(out.InstanceMetadataOptions.HttpPutResponseHopLimit))
+	}
+
+	got, ok := b.vmMgr.Get(id)
+	require.True(t, ok)
+	assert.Equal(t, int64(3), aws.Int64Value(got.Instance.MetadataOptions.HttpPutResponseHopLimit))
+	_, onA := a.vmMgr.Get(id)
+	assert.False(t, onA, "the queue-group member never held the instance")
+}
+
+// n1CommandType is EC2InstanceCommand as the previous release defined it: the
+// current type minus the two attributes and two data blocks this change adds.
+func n1CommandType() reflect.Type {
+	newFields := map[string]bool{
+		"ModifyInstanceMetadataOptions": true, "SetDisableApiTermination": true,
+		"InstanceMetadataOptionsData": true, "DisableApiTerminationData": true,
+	}
+	strip := func(t reflect.Type, swap map[string]reflect.Type) reflect.Type {
+		var fields []reflect.StructField
+		for f := range t.Fields() {
+			if newFields[f.Name] {
+				continue
+			}
+			if st, ok := swap[f.Name]; ok {
+				f.Type = st
+			}
+			fields = append(fields, f)
+		}
+		return reflect.StructOf(fields)
+	}
+	attrs := strip(reflect.TypeFor[ec2v1.EC2CommandAttributes](), nil)
+	return strip(reflect.TypeFor[ec2v1.EC2InstanceCommand](), map[string]reflect.Type{"Attributes": attrs})
+}
+
+// An N-1 owner decodes a new-variant payload with the unknown attribute and
+// data dropped, so dispatch sees no attribute and takes the default branch.
+func TestS9_N1DispatchRejectsNewVariantsWithoutWriting(t *testing.T) {
+	var logs bytes.Buffer
+	prev := slog.Default()
+	slog.SetDefault(slog.New(slog.NewTextHandler(&logs, &slog.HandlerOptions{Level: slog.LevelWarn})))
+	t.Cleanup(func() { slog.SetDefault(prev) })
+
+	variants := map[string]func(id string) ec2v1.EC2InstanceCommand{
+		"metadata options": func(id string) ec2v1.EC2InstanceCommand {
+			return metadataCommand(id, &ec2v1.InstanceMetadataOptionsData{HttpPutResponseHopLimit: aws.Int64(3)})
+		},
+		"termination protection": func(id string) ec2v1.EC2InstanceCommand {
+			return terminationCommand(id, &ec2v1.DisableApiTerminationData{Value: true})
+		},
+	}
+	for name, build := range variants {
+		t.Run(name, func(t *testing.T) {
+			id := "i-or-n1-" + strings.ReplaceAll(name, " ", "-")
+			d, _ := monitoringTestDaemon(t, id, false)
+			before, _ := d.vmMgr.Get(id)
+			beforeInstance := *before.Instance
+			logs.Reset()
+
+			// Current encoder -> N-1 decoder -> what the N-1 owner re-dispatches.
+			n1 := reflect.New(n1CommandType())
+			require.NoError(t, json.Unmarshal(mustMarshal(t, build(id)), n1.Interface()))
+			require.Equal(t, id, n1.Elem().FieldByName("ID").String())
+			payload := mustMarshal(t, n1.Interface())
+			assert.NotContains(t, string(payload), "modify_instance_metadata_options")
+			assert.NotContains(t, string(payload), "set_disable_api_termination")
+
+			reply := requestHandler(t, d.natsConn, "ec2.cmd."+id, d.handleEC2Events, testAccountID, payload)
+			assert.Equal(t, awserrors.ErrorServerInternal, decodeError(t, reply.Data)["Code"])
+			assert.Empty(t, reply.Header.Get(ec2v1.HandledHeader), "an N-1 rejection carries no handled marker")
+			assert.Contains(t, logs.String(), "Unhandled EC2 instance command")
+
+			after, _ := d.vmMgr.Get(id)
+			assert.Equal(t, beforeInstance, *after.Instance)
+			assert.Nil(t, after.RunInstancesInput.DisableApiTermination)
+		})
+	}
+}
+
+func TestModifyInstanceMetadataOptions_NoOwnerAppliesToRealStoppedRecord(t *testing.T) {
+	const id = "i-or-stopped-real"
+	ns, gateway := testutil.StartTestNATS(t)
+	stopped := vmmock.New()
+	stopped.Stopped[id] = stoppedVMFixture(id, testAccountID)
+
+	// No daemon runs the instance; the only queue-group member is the real service.
+	a := ownerRoutingDaemon(t, ns.ClientURL(), stopped)
+	sub, err := a.natsConn.QueueSubscribe("ec2.ModifyInstanceMetadataOptions", "spinifex-workers",
+		asMsgHandler(handleNATSRequest(a.node, a.instanceService.ModifyInstanceMetadataOptions)))
+	require.NoError(t, err)
+	t.Cleanup(func() { _ = sub.Unsubscribe() })
+	require.NoError(t, a.natsConn.Flush())
+
+	out, err := ec2instanceapi.ModifyInstanceMetadataOptions(t.Context(), &ec2.ModifyInstanceMetadataOptionsInput{
+		InstanceId:              aws.String(id),
+		HttpPutResponseHopLimit: aws.Int64(2),
+	}, gateway, testAccountID)
+	require.NoError(t, err)
+	assert.Equal(t, int64(2), aws.Int64Value(out.InstanceMetadataOptions.HttpPutResponseHopLimit))
+
+	reloaded, err := stopped.LoadStoppedInstance(id)
+	require.NoError(t, err)
+	require.NotNil(t, reloaded.Instance.MetadataOptions)
+	assert.Equal(t, int64(2), aws.Int64Value(reloaded.Instance.MetadataOptions.HttpPutResponseHopLimit))
+	require.NotNil(t, stopped.WroteStopped[id], "the service persisted the stopped record")
+}
+
+func TestOwnerDispatch_RejectsOtherAccountWithoutWriting(t *testing.T) {
+	const id = "i-or-otheracct"
+	d, _ := monitoringTestDaemon(t, id, false)
+	before, _ := d.vmMgr.Get(id)
+	beforeInstance := *before.Instance
+
+	cmds := map[string]ec2v1.EC2InstanceCommand{
+		"metadata options": metadataCommand(id, &ec2v1.InstanceMetadataOptionsData{HttpPutResponseHopLimit: aws.Int64(3)}),
+		"termination":      terminationCommand(id, &ec2v1.DisableApiTerminationData{Value: true}),
+	}
+	for name, cmd := range cmds {
+		reply := requestHandler(t, d.natsConn, "ec2.cmd."+id, d.handleEC2Events, "999999999999", mustMarshal(t, cmd))
+		assert.Equal(t, awserrors.ErrorInvalidInstanceIDNotFound, decodeError(t, reply.Data)["Code"], name)
+	}
+
+	after, _ := d.vmMgr.Get(id)
+	assert.Equal(t, beforeInstance, *after.Instance)
+	assert.Nil(t, after.RunInstancesInput.DisableApiTermination)
+}
+
+func TestModifyInstanceAttribute_DisableApiTerminationRoutedToOwner(t *testing.T) {
+	const id = "i-or-termroute"
+	ns, gateway := testutil.StartTestNATS(t)
+	stopped := vmmock.New()
+
+	b := ownerRoutingDaemon(t, ns.ClientURL(), stopped)
+	b.vmMgr.Insert(&vm.VM{
+		ID:                id,
+		Status:            vm.StateRunning,
+		AccountID:         testAccountID,
+		Instance:          &ec2.Instance{InstanceId: aws.String(id)},
+		RunInstancesInput: &ec2.RunInstancesInput{},
+	})
+	subscribeOwner(t, b, id)
+
+	a := ownerRoutingDaemon(t, ns.ClientURL(), stopped)
+	sub, err := a.natsConn.QueueSubscribe("ec2.ModifyInstanceAttribute", "spinifex-workers",
+		asMsgHandler(handleNATSRequest(a.node, a.instanceService.ModifyInstanceAttribute)))
+	require.NoError(t, err)
+	t.Cleanup(func() { _ = sub.Unsubscribe() })
+	require.NoError(t, a.natsConn.Flush())
+
+	for range 9 {
+		_, err := ec2instanceapi.ModifyInstanceAttribute(t.Context(), &ec2.ModifyInstanceAttributeInput{
+			InstanceId:            aws.String(id),
+			DisableApiTermination: &ec2.AttributeBooleanValue{Value: aws.Bool(true)},
+		}, gateway, testAccountID)
+		require.NoError(t, err)
+	}
+
+	got, ok := b.vmMgr.Get(id)
+	require.True(t, ok)
+	require.NotNil(t, got.RunInstancesInput.DisableApiTermination)
+	assert.True(t, *got.RunInstancesInput.DisableApiTermination)
+
+	// The node that enforces the guard now refuses the terminate.
+	reply := requestHandler(t, b.natsConn, "ec2.cmd."+id+".enforce", b.handleEC2Events, testAccountID,
+		mustMarshal(t, ec2v1.EC2InstanceCommand{ID: id, Attributes: ec2v1.EC2CommandAttributes{TerminateInstance: true}}))
+	assert.Equal(t, awserrors.ErrorOperationNotPermitted, decodeError(t, reply.Data)["Code"])
+}

spinifex/runtime/daemon/daemon_lambda_testlink.go

@@ -3,14 +3,18 @@
 package daemon
 
 import (
+	"crypto/sha256"
 	"crypto/tls"
 	"crypto/x509"
+	"encoding/hex"
 	"errors"
 	"fmt"
+	"io"
 	"log/slog"
 	"net/http"
 	"os"
 	"path/filepath"
+	"time"
 
 	"github.com/mulgadc/spinifex/spinifex/domains/lambda/execution"
 	"github.com/mulgadc/spinifex/spinifex/domains/network/host"
@@ -27,6 +31,28 @@ var lambdaHardKillSHA256 string
 // image bytes and does not authenticate where they came from.
 var lambdaImageDigest string
 
+// lambdaFirecrackerSHA256 pins the Firecracker v1.17.0 x86_64 release binary.
+// A node whose test-link image directory holds a firecracker binary boots
+// every attempt under it, and does not serve unless the binary matches.
+const lambdaFirecrackerSHA256 = "99ad0f5cd0514a88aad0e9ae8cfdb3cc3b4ab9d190e1194602406c786b5de7a5"
+
+// checkFirecracker refuses a Firecracker binary other than the pinned release.
+func checkFirecracker(path string) error {
+	f, err := os.Open(path)
+	if err != nil {
+		return fmt.Errorf("firecracker binary: %w", err)
+	}
+	defer func() { _ = f.Close() }()
+	h := sha256.New()
+	if _, err := io.Copy(h, f); err != nil {
+		return fmt.Errorf("firecracker binary: %w", err)
+	}
+	if got := hex.EncodeToString(h.Sum(nil)); got != lambdaFirecrackerSHA256 {
+		return errors.New("firecracker binary does not match the pinned release")
+	}
+	return nil
+}
+
 // startLambdaTestLink serves controlled-test Lambda invocations for this node.
 // Recovery needs only the owner store, taps and addresses, so it starts even
 // when the bridge address, initializer, pin or download trust would not serve.
@@ -70,9 +96,20 @@ func (d *Daemon) startLambdaTestLink() {
 		InitializerSHA256: lambdaHardKillSHA256,
 		HTTP:              httpClient,
 	}
+	fc := filepath.Join(imageDir, "firecracker")
+	if _, err := os.Lstat(fc); err == nil {
+		tl.Firecracker = fc
+	}
 	err = runLambdaTestLink(
 		func() { go tl.RecoverLoop(d.ctx, testlink.RecoverFirst, testlink.RecoverMax) },
-		func() error { return lambdaTestLinkReady(tl, imageDir, lambdaImageDigest, trustErr) },
+		func() error {
+			if tl.Firecracker != "" {
+				if err := checkFirecracker(tl.Firecracker); err != nil {
+					return err
+				}
+			}
+			return lambdaTestLinkReady(tl, imageDir, lambdaImageDigest, trustErr)
+		},
 		func() error {
 			gatewayUID, err := testlink.LookupUID(testlink.GatewayUser)
 			if err != nil {
@@ -88,8 +125,17 @@ func (d *Daemon) startLambdaTestLink() {
 	}
 	slog.Warn("Lambda test link serving: test-only, not for production; the guest does not verify owner grant authority",
 		"node", d.node, "image_digest", tl.Image.Digest)
+	if d.lambdaMetaService != nil && d.clusterConfig != nil {
+		d.lambdaMetaService.SetReadiness(tl.Readiness(lambdaImageDigest, len(d.clusterConfig.Nodes)))
+		go d.lambdaMetaService.RunActivation(d.ctx, lambdaActivationInterval)
+	}
 }
 
+// lambdaActivationInterval bounds how long a function whose activation was
+// missed, by a crash or an earlier refusal, waits for the next pass.
+const lambdaActivationInterval = 30 * time.Second
+
+
 // lambdaTestLinkReady runs the serving checks and, only once they pass,
 // prepares the pinned image into tl; a refusal leaves tl.Image nil.
 func lambdaTestLinkReady(tl *testlink.Daemon, imageDir, imagePin string, trustErr error) error {

spinifex/runtime/daemon/nats_classification_test.go

@@ -137,6 +137,8 @@ func TestEC2CommandNameCoversEveryCommand(t *testing.T) {
 		{"SetInstanceTags", func(a *ec2v1.EC2CommandAttributes) { a.SetInstanceTags = true }},
 		{"RemoveInstanceTags", func(a *ec2v1.EC2CommandAttributes) { a.RemoveInstanceTags = true }},
 		{"SetInstanceMonitoring", func(a *ec2v1.EC2CommandAttributes) { a.SetInstanceMonitoring = true }},
+		{"ModifyInstanceMetadataOptions", func(a *ec2v1.EC2CommandAttributes) { a.ModifyInstanceMetadataOptions = true }},
+		{"SetDisableApiTermination", func(a *ec2v1.EC2CommandAttributes) { a.SetDisableApiTermination = true }},
 		{"StartInstance", func(a *ec2v1.EC2CommandAttributes) { a.StartInstance = true }},
 		{"RebootInstance", func(a *ec2v1.EC2CommandAttributes) { a.RebootInstance = true }},
 		{"StopInstance", func(a *ec2v1.EC2CommandAttributes) { a.StopInstance = true }},

spinifex/runtime/roles/awsgw/awsgw.go

@@ -736,6 +736,9 @@ func launchService(config *config.ClusterConfig) error {
 	operatorAdminDeps = withPlatformCleaner(operatorAdminDeps, func() (accountlifecycle.PlatformCleaner, error) {
 		return platformCleaner(janitorCtx, &nodeConfig, natsConn, js, operatorAdminDeps.NodeIDs)
 	})
+	operatorAdminDeps = withComputeCleaners(operatorAdminDeps, func() (accountlifecycle.ComputeStage, accountlifecycle.StorageStage, error) {
+		return computeCleaners(janitorCtx, &nodeConfig, natsConn, js, operatorAdminDeps.NodeIDs, objStore)
+	})
 	operatorControlPlane := operatoradmin.NewRegistration(operatorAdminDeps)
 	// Every gateway scans for account deletions whose holder died; the fenced
 	// takeover makes any number of them safe.

spinifex/runtime/roles/awsgw/compute_cleanup.go

@@ -0,0 +1,169 @@
+package awsgw
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"log/slog"
+	"slices"
+
+	"github.com/aws/aws-sdk-go/aws"
+	"github.com/aws/aws-sdk-go/service/ec2"
+	"github.com/aws/aws-sdk-go/service/eks"
+	"github.com/mulgadc/spinifex/spinifex/bootstrap/config"
+	"github.com/mulgadc/spinifex/spinifex/domains/ec2/accountcleanup"
+	"github.com/mulgadc/spinifex/spinifex/domains/ec2/accountinventory"
+	ec2instanceapi "github.com/mulgadc/spinifex/spinifex/domains/ec2/awsapi/instance"
+	ec2volumeapi "github.com/mulgadc/spinifex/spinifex/domains/ec2/awsapi/volume"
+	ec2metadata "github.com/mulgadc/spinifex/spinifex/domains/ec2/ebs/metadata"
+	ec2instance "github.com/mulgadc/spinifex/spinifex/domains/ec2/instance"
+	ec2key "github.com/mulgadc/spinifex/spinifex/domains/ec2/key"
+	ec2launchtemplate "github.com/mulgadc/spinifex/spinifex/domains/ec2/launchtemplate"
+	ec2placementgroup "github.com/mulgadc/spinifex/spinifex/domains/ec2/placementgroup"
+	ec2spotinstance "github.com/mulgadc/spinifex/spinifex/domains/ec2/spotinstance"
+	ec2tags "github.com/mulgadc/spinifex/spinifex/domains/ec2/tags"
+	ekscluster "github.com/mulgadc/spinifex/spinifex/domains/eks/cluster"
+	"github.com/mulgadc/spinifex/spinifex/operator/accountlifecycle"
+	operatoradmin "github.com/mulgadc/spinifex/spinifex/operator/admin"
+	"github.com/mulgadc/spinifex/spinifex/providers/objectstore"
+	"github.com/nats-io/nats.go"
+	"github.com/nats-io/nats.go/jetstream"
+)
+
+var (
+	errNoTerminateAck = errors.New("terminate was not acknowledged by the owning node")
+	errNoWorkerList   = errors.New("managed worker list is not available")
+)
+
+// strictInstances is the complete-or-error instance inventory.
+type strictInstances struct {
+	nc      *nats.Conn
+	nodeIDs []string
+}
+
+func (s strictInstances) AccountInstances(ctx context.Context, accountID string) (*ec2.DescribeInstancesOutput, error) {
+	return ec2instanceapi.DescribeAccountInstancesStrict(ctx, s.nc, s.nodeIDs, accountID)
+}
+
+// instanceTerminator sends the existing EC2 terminate and requires the owning
+// node to have acknowledged this instance, because that API skips one it could
+// not reach and still answers.
+type instanceTerminator struct{ nc *nats.Conn }
+
+func (t instanceTerminator) TerminateInstance(ctx context.Context, accountID, instanceID string) error {
+	out, err := ec2instanceapi.TerminateInstances(ctx, &ec2.TerminateInstancesInput{InstanceIds: []*string{aws.String(instanceID)}}, t.nc, accountID)
+	if err != nil {
+		return err
+	}
+	for _, change := range out.TerminatingInstances {
+		if aws.StringValue(change.InstanceId) == instanceID {
+			return nil
+		}
+	}
+	return errNoTerminateAck
+}
+
+// teardownReader reads unfinished instance teardowns through the owner.
+type teardownReader struct{ nc *nats.Conn }
+
+func (r teardownReader) AccountTeardownOutstanding(accountID string) ([]ec2instance.OutstandingInstance, error) {
+	return ec2instanceapi.AccountTeardownOutstanding(context.Background(), r.nc, accountID)
+}
+
+// guardedVolumeDeleter is the customer DeleteVolume path: it refuses unless
+// every node answered the live-holder listing, and never deletes from the store.
+type guardedVolumeDeleter struct {
+	nc      *nats.Conn
+	nodeIDs []string
+}
+
+func (d guardedVolumeDeleter) DeleteVolume(ctx context.Context, accountID, volumeID string) error {
+	_, err := ec2volumeapi.DeleteVolume(ctx, &ec2.DeleteVolumeInput{VolumeId: aws.String(volumeID)}, d.nc, d.nodeIDs, accountID)
+	return err
+}
+
+// eksManagedInstances is every worker a managed nodegroup of the account still
+// owns. A cluster that cannot be listed is an error, never an empty set.
+type eksManagedInstances struct{ svc ekscluster.EKSService }
+
+func (m eksManagedInstances) AccountManagedInstances(ctx context.Context, accountID string) (map[string]bool, error) {
+	if m.svc == nil {
+		return nil, errNoWorkerList
+	}
+	owned := map[string]bool{}
+	var token *string
+	for {
+		page, err := m.svc.ListClusters(ctx, &eks.ListClustersInput{NextToken: token}, accountID)
+		if err != nil {
+			return nil, fmt.Errorf("list clusters: %w", err)
+		}
+		for _, name := range page.Clusters {
+			workers, err := m.svc.ListManagedWorkers(ctx, &ekscluster.ListManagedWorkersInput{ClusterName: aws.StringValue(name)}, accountID)
+			if err != nil {
+				return nil, fmt.Errorf("list workers of cluster %s: %w", aws.StringValue(name), err)
+			}
+			for _, id := range workers.InstanceIDs {
+				owned[id] = true
+			}
+		}
+		if aws.StringValue(page.NextToken) == "" {
+			return owned, nil
+		}
+		token = page.NextToken
+	}
+}
+
+var (
+	_ accountcleanup.InstanceInventory     = strictInstances{}
+	_ accountcleanup.InstanceTerminator    = instanceTerminator{}
+	_ accountcleanup.TeardownReader        = teardownReader{}
+	_ accountcleanup.VolumeDeleter         = guardedVolumeDeleter{}
+	_ accountcleanup.ManagedInstanceReader = eksManagedInstances{}
+)
+
+// computeCleaners builds the EC2 compute and storage cleanup over the owners'
+// own services. Any service that cannot be built fails the whole bind, so
+// DeleteAccount refuses instead of running with part of the owners missing.
+func computeCleaners(ctx context.Context, node *config.Config, nc *nats.Conn, js jetstream.JetStream, nodeIDs []string, objects objectstore.ObjectStore) (accountlifecycle.ComputeStage, accountlifecycle.StorageStage, error) {
+	if nc == nil || js == nil || objects == nil || len(nodeIDs) == 0 {
+		return nil, nil, errors.New("compute cleanup needs NATS, JetStream, the object store and the node set")
+	}
+	tagsKV, err := ec2tags.GetOrCreateTagsBucket(ctx, js)
+	if err != nil {
+		return nil, nil, fmt.Errorf("tags bucket: %w", err)
+	}
+	tags := ec2tags.NewTagsServiceImpl(node, tagsKV)
+	spot, err := ec2spotinstance.NewSpotInstanceServiceImplWithNATS(ctx, node, nc)
+	if err != nil {
+		return nil, nil, fmt.Errorf("spot instance service: %w", err)
+	}
+	groups, err := ec2placementgroup.NewPlacementGroupServiceImplWithNATS(ctx, node, nc)
+	if err != nil {
+		return nil, nil, fmt.Errorf("placement group service: %w", err)
+	}
+	templates, err := ec2launchtemplate.NewLaunchTemplateServiceImplWithNATS(ctx, node, nc)
+	if err != nil {
+		return nil, nil, fmt.Errorf("launch template service: %w", err)
+	}
+	volumes := accountinventory.New(ec2key.NewKeyServiceImpl(node), groups, templates, tags).
+		WithCompute(spot, ec2metadata.NewStore(objects, node.Predastore.Bucket))
+
+	instances := strictInstances{nc: nc, nodeIDs: slices.Clone(nodeIDs)}
+	teardown := teardownReader{nc: nc}
+	compute := accountcleanup.NewCompute(spot, instances, instanceTerminator{nc: nc}, teardown,
+		eksManagedInstances{svc: ekscluster.NewNATSEKSService(nc)})
+	storage := accountcleanup.NewStorage(volumes, guardedVolumeDeleter{nc: nc, nodeIDs: slices.Clone(nodeIDs)}, instances, teardown, spot, tags)
+	return compute, storage, nil
+}
+
+// withComputeCleaners binds the compute and storage cleanup every DeleteAccount
+// runs. When it cannot be built both fields stay nil, which refuses the request.
+func withComputeCleaners(d operatoradmin.Deps, build func() (accountlifecycle.ComputeStage, accountlifecycle.StorageStage, error)) operatoradmin.Deps {
+	compute, storage, err := build()
+	if err != nil {
+		slog.Error("awsgw: EC2 compute and storage cleanup unavailable", "err", err)
+		return d
+	}
+	d.ComputeCleaner, d.StorageCleaner = compute, storage
+	return d
+}