fix: route instance metadata options and termination protection to the owner (#1645)

* feat: add owner-routed commands for instance metadata options and termination protection to the ec2.cmd contract

* feat: dispatch metadata-options and termination-protection commands on the owner

* feat: route ModifyInstanceMetadataOptions to the instance owner first

* test: pin the N-1 rejection, account isolation and timeout-after-apply for owner routing

* feat: route DisableApiTermination changes to the instance owner first

* test: pin the N-1 rejection with the previous release's command type and exercise the real stopped fallback

Makefile

@@ -90,6 +90,7 @@ go_build:
 LAMBDA_TESTLINK_DIR := ./bin/lambda-testlink
 LAMBDA_TESTLINK_PKG := github.com/mulgadc/spinifex/spinifex/runtime/daemon
 LAMBDA_TESTLINK_PIN := $(LAMBDA_TESTLINK_PKG).lambdaHardKillSHA256
+LAMBDA_TESTLINK_GW := github.com/mulgadc/spinifex/spinifex/runtime/roles/awsgw
 LAMBDA_DIGEST_RE := ^[0-9a-f]{64}$$
 build-lambda-testlink: override export LAMBDA_IMAGE_DIGEST := $(value LAMBDA_IMAGE_DIGEST)
 build-lambda-testlink:
@@ -104,8 +105,8 @@ build-lambda-testlink:
 	@digest=$$(sha256sum $(LAMBDA_TESTLINK_DIR).tmp/hardkill.so | cut -d' ' -f1); \
 	test $${#digest} -eq 64 || { echo "initializer digest missing"; exit 1; }; \
 	echo "initializer sha256 $$digest"; \
-	pin="-X $(LAMBDA_TESTLINK_PIN)=$$digest -X $(LAMBDA_TESTLINK_PKG).lambdaImageDigest=$$LAMBDA_IMAGE_DIGEST"; \
-	LAMBDA_TESTLINK_INITIALIZER=$(abspath $(LAMBDA_TESTLINK_DIR).tmp/hardkill.so) go test -count=1 -tags lambdatestlink -ldflags "$$pin" -run '^TestLambda(HardKillPin|ImageDigestPin)$$' ./spinifex/runtime/daemon/ && \
+	pin="-X $(LAMBDA_TESTLINK_PIN)=$$digest -X $(LAMBDA_TESTLINK_PKG).lambdaImageDigest=$$LAMBDA_IMAGE_DIGEST -X $(LAMBDA_TESTLINK_GW).lambdaImageDigest=$$LAMBDA_IMAGE_DIGEST"; \
+	LAMBDA_TESTLINK_INITIALIZER=$(abspath $(LAMBDA_TESTLINK_DIR).tmp/hardkill.so) go test -count=1 -tags lambdatestlink -ldflags "$$pin" -run '^TestLambda(HardKillPin|ImageDigestPin)$$' ./spinifex/runtime/daemon/ ./spinifex/runtime/roles/awsgw/ && \
 	go build $(GO_BUILD_MOD) -tags lambdatestlink -ldflags "$(subst -s -w ,,$(LDFLAGS)) $$pin" -o $(LAMBDA_TESTLINK_DIR).tmp/$(GO_PROJECT_NAME) cmd/spinifex/main.go
 	@mkdir -p $(LAMBDA_TESTLINK_DIR)
 	mv $(LAMBDA_TESTLINK_DIR).tmp/hardkill.so $(LAMBDA_TESTLINK_DIR).tmp/$(GO_PROJECT_NAME) $(LAMBDA_TESTLINK_DIR)/

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"},

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

@@ -22,6 +22,11 @@ Tenant code may be able to reach the listener, but it cannot open the agent's co
 For each generation the executor creates ephemeral self-signed host and guest certificates whose single URI SAN is `spinifex-lambda://<Region>/<account>/<function ID>/<environment ID>/<generation>/<role>`, with role `host` or `guest`.
 Each end validates the peer certificate against the one pinned peer certificate and separately requires that URI SAN with the opposite role; hostname verification is not relied on.
 ALPN carries the L-GUEST major version, `spinifex-lambda-guest/1`, and an unsupported version fails the handshake. Builds with the `lambdatestlink` tag offer only `spinifex-lambda-testlink/2` on both host and guest, whose first frame carries the execution-role credential, so a test-link end and a release end fail the handshake; release builds are unchanged.
+In the same builds, the gateway reaches the host daemon on the local abstract socket `spinifex-lambda-testlink-control/3`, and each request carries the runtime-image digest the gateway was built with; the daemon refuses a missing or different digest before starting the attempt, and a gateway and daemon on different socket names never connect.
+On that socket a request's raw payload is at most 6 MiB, checked by the gateway before encoding and by the daemon after decoding.
+Its encoded frame is at most the base64 length of 6 MiB plus 64 KiB of metadata headroom and 64 bytes of field slack, and a receiver checks the frame length before allocating; metadata has no separate bound within the frame.
+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.
 
 **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.

spinifex/agents/lambda/hostlink/test_stream.go

@@ -321,6 +321,19 @@ func (s *TestStream) HostSend(req Request) error {
 	return s.write(frame{Type: typeInvoke, Request: &req})
 }
 
+// CheckInvokeFits encodes req as HostSend would and returns ErrFrameTooLarge
+// when the frame exceeds the bound, so a caller can refuse before launch.
+func CheckInvokeFits(req Request) error {
+	b, err := json.Marshal(frame{Type: typeInvoke, Request: &req})
+	if err != nil {
+		return err
+	}
+	if len(b) > maxFrame {
+		return ErrFrameTooLarge
+	}
+	return nil
+}
+
 // HostShutdown writes a shutdown frame, as the host side of the carrier.
 func (s *TestStream) HostShutdown() error {
 	return s.write(frame{Type: typeShutdown})

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/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/contract.go

@@ -49,6 +49,8 @@ type Request struct {
 	CodeURL     string          `json:"code_url"`
 	CodeSize    int64           `json:"code_size"`
 	CodeSHA256  string          `json:"code_sha256"`
+	// ImageDigest is the runtime image the gateway expects to run the code.
+	ImageDigest string `json:"image_digest"`
 	Payload     []byte          `json:"payload"`
 
 	// cred arrives beside the request on the control socket, never inside it.

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

@@ -23,7 +23,7 @@ import (
 // ControlName is the abstract Unix socket name the node's daemon serves. An
 // abstract name has no directory entry; the kernel frees it when the last
 // reference to the listening socket closes.
-const ControlName = "spinifex-lambda-testlink-control"
+const ControlName = "spinifex-lambda-testlink-control/3"
 
 // Service accounts at each end of the control socket.
 const (
@@ -31,10 +31,36 @@ const (
 	DaemonUser  = "spinifex-daemon"
 )
 
+// joinAfterFunc arms f on ctx's cancellation and returns a stop that, unlike
+// context.AfterFunc's, waits for f when it has already started.
+func joinAfterFunc(ctx context.Context, f func()) func() {
+	done := make(chan struct{})
+	stop := context.AfterFunc(ctx, func() { defer close(done); f() })
+	return func() {
+		if !stop() {
+			<-done
+		}
+	}
+}
+
 // MaxCredentialBytes bounds the encoded credential; a session's three values
 // fit well inside it.
 const MaxCredentialBytes = 8 << 10
 
+// Declared control-socket bounds, independent of any message bus. Request and
+// Reply payloads are capped raw; the frame cap leaves MetadataHeadroomBytes
+// beside a full base64 payload, and metadata has no bound of its own.
+const (
+	MaxPayloadBytes       = 6 << 20
+	MetadataHeadroomBytes = 64 << 10
+	// payloadFieldSlack covers the payload key, quotes and comma.
+	payloadFieldSlack = 64
+	MaxFrameBytes     = int64((MaxPayloadBytes+2)/3*4 + MetadataHeadroomBytes + payloadFieldSlack)
+	// MaxAdmitted bounds connections the daemon holds at once, from accept
+	// to close; the daemon runs one attempt, so one more is answered busy.
+	MaxAdmitted = 2
+)
+
 // Credential is one execution-role session for one attempt; it goes to the
 // guest in the first frame and never on NATS, to disk or into a log.
 type Credential = hostlink.Credential
@@ -118,8 +144,8 @@ func readFrame(r io.Reader, limit int64) ([]byte, error) {
 // ControlServer answers one request per connection from the gateway's UID.
 type ControlServer struct {
 	GatewayUID int
-	// RequestLimit bounds the encoded Request alone, as the NATS path did, and
-	// ReplyLimit the encoded Reply; a reply over it becomes unknown.
+	// RequestLimit bounds the encoded Request frame and ReplyLimit the
+	// encoded Reply; a reply over it, or over MaxPayloadBytes, becomes unknown.
 	RequestLimit int64
 	ReplyLimit   int64
 	// IOTimeout bounds reading the request and writing the reply.
@@ -129,14 +155,25 @@ type ControlServer struct {
 
 // Serve accepts until ctx ends or the listener fails, then waits for every
 // connection it started.
+// At most MaxAdmitted connections are held at once, each from accept to
+// close; others wait unaccepted in the kernel's listen backlog.
 func (s *ControlServer) Serve(ctx context.Context, ln *net.UnixListener) error {
-	stop := context.AfterFunc(ctx, func() { _ = ln.Close() })
-	defer stop()
+	// Serve may return before the callback runs, so it closes the listener
+	// too, and joins a callback already started before returning.
+	defer joinAfterFunc(ctx, func() { _ = ln.Close() })()
+	defer func() { _ = ln.Close() }()
 	var wg sync.WaitGroup
 	defer wg.Wait()
+	slots := make(chan struct{}, MaxAdmitted)
 	for {
+		select {
+		case slots <- struct{}{}:
+		case <-ctx.Done():
+			return nil
+		}
 		c, err := ln.AcceptUnix()
 		if err != nil {
+			<-slots
 			if ctx.Err() != nil {
 				return nil
 			}
@@ -145,6 +182,10 @@ func (s *ControlServer) Serve(ctx context.Context, ln *net.UnixListener) error {
 		wg.Add(1)
 		go func() {
 			defer wg.Done()
+			defer func() { <-slots }()
+			// Closing on cancellation unblocks a read or write in progress; the
+			// close is joined before the slot is released.
+			defer joinAfterFunc(ctx, func() { _ = c.Close() })()
 			s.serveConn(ctx, c)
 		}()
 	}
@@ -176,6 +217,10 @@ func (s *ControlServer) serveConn(ctx context.Context, c *net.UnixConn) {
 	_ = c.SetReadDeadline(time.Time{})
 
 	rep := s.Handle(ctx, req, cred)
+	// The raw payload is checked before the reply is encoded.
+	if len(rep.Payload) > MaxPayloadBytes {
+		rep = Reply{Outcome: OutcomeUnknown, Error: "result too large for the test path"}
+	}
 	body, err := json.Marshal(rep)
 	if err == nil && int64(len(body)) > s.ReplyLimit {
 		body, err = json.Marshal(Reply{Outcome: OutcomeUnknown, Error: "result too large for the test path"})
@@ -218,8 +263,11 @@ func (c *ControlClient) Exchange(ctx context.Context, req []byte, cred Credentia
 	if replyBy.IsZero() || !replyBy.After(time.Now()) {
 		return Reply{}, fmt.Errorf("%w: reply bound is zero or already past", ErrNotSent)
 	}
+	// The dial stops at replyBy even when ctx has no deadline.
+	dialCtx, cancelDial := context.WithDeadline(ctx, replyBy)
 	var d net.Dialer
-	conn, err := d.DialContext(ctx, "unix", "@"+c.Name)
+	conn, err := d.DialContext(dialCtx, "unix", "@"+c.Name)
+	cancelDial()
 	if err != nil {
 		return Reply{}, fmt.Errorf("%w: dial test-link control: %w", ErrNotSent, err)
 	}
@@ -269,6 +317,9 @@ func (c *ControlClient) Exchange(ctx context.Context, req []byte, cred Credentia
 	if err := json.Unmarshal(body, &rep); err != nil {
 		return Reply{}, errors.New("unreadable test-link reply")
 	}
+	if len(rep.Payload) > MaxPayloadBytes {
+		return Reply{}, fmt.Errorf("test-link reply payload of %d bytes exceeds %d", len(rep.Payload), MaxPayloadBytes)
+	}
 	return rep, nil
 }
 

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

@@ -0,0 +1,569 @@
+//go:build lambdatestlink
+
+//test:in-package — reads frames with the unexported readFrame and drives ControlServer limits directly
+package testlink
+
+import (
+	"bytes"
+	"context"
+	"encoding/binary"
+	"encoding/json"
+	"net"
+	"os"
+	"runtime"
+	"strings"
+	"sync"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/require"
+	"golang.org/x/sys/unix"
+
+	"github.com/mulgadc/spinifex/spinifex/agents/lambda/hostlink"
+)
+
+// fullServe runs a ControlServer at the declared bounds on ln and returns a
+// cancel that waits for Serve.
+func fullServe(t *testing.T, ln *net.UnixListener, handle func(context.Context, []byte, Credential) Reply) func() {
+	t.Helper()
+	srv := &ControlServer{GatewayUID: os.Getuid(), RequestLimit: MaxFrameBytes, ReplyLimit: MaxFrameBytes, IOTimeout: 60 * time.Second, Handle: handle}
+	ctx, cancel := context.WithCancel(context.Background())
+	done := make(chan error, 1)
+	go func() { done <- srv.Serve(ctx, ln) }()
+	var once sync.Once
+	stop := func() {
+		once.Do(func() {
+			cancel()
+			select {
+			case err := <-done:
+				require.NoError(t, err, "Serve returned an error on shutdown")
+			case <-time.After(20 * time.Second):
+				t.Fatal("Serve did not return")
+			}
+		})
+	}
+	t.Cleanup(stop)
+	return stop
+}
+
+func fullListen(t *testing.T, name string) *net.UnixListener {
+	t.Helper()
+	ln, err := ListenControl(name)
+	require.NoError(t, err)
+	return ln
+}
+
+func fullClient(name string) *ControlClient {
+	return &ControlClient{Name: name, DaemonUID: os.Getuid(), RequestLimit: MaxFrameBytes, ReplyLimit: MaxFrameBytes, IOTimeout: 60 * time.Second}
+}
+
+// requestOfSize returns an encoded Request with a payload of payloadLen and
+// a FunctionARN padded so the frame is exactly size bytes.
+func requestOfSize(t *testing.T, payloadLen, size int) []byte {
+	t.Helper()
+	req := Request{AttemptID: "0123456789ab", Payload: bytes.Repeat([]byte{7}, payloadLen)}
+	b, err := json.Marshal(req)
+	require.NoError(t, err)
+	req.FunctionARN = "a"
+	b, err = json.Marshal(req)
+	require.NoError(t, err)
+	require.LessOrEqual(t, len(b), size)
+	req.FunctionARN = strings.Repeat("a", 1+size-len(b))
+	b, err = json.Marshal(req)
+	require.NoError(t, err)
+	require.Len(t, b, size)
+	return b
+}
+
+// replyOfSize returns a Reply with a payload of payloadLen whose encoding is
+// exactly size bytes.
+func replyOfSize(t *testing.T, payloadLen, size int) Reply {
+	t.Helper()
+	rep := Reply{Outcome: OutcomeDone}
+	if payloadLen > 0 {
+		rep.Payload = bytes.Repeat([]byte{9}, payloadLen)
+	}
+	b, err := json.Marshal(rep)
+	require.NoError(t, err)
+	rep.Error = "e"
+	b, err = json.Marshal(rep)
+	require.NoError(t, err)
+	require.LessOrEqual(t, len(b), size)
+	rep.Error = strings.Repeat("e", 1+size-len(b))
+	b, err = json.Marshal(rep)
+	require.NoError(t, err)
+	require.Len(t, b, size)
+	return rep
+}
+
+func recvBounded[T any](t *testing.T, ch chan T) T {
+	t.Helper()
+	select {
+	case v := <-ch:
+		return v
+	case <-time.After(30 * time.Second):
+		t.Fatal("no value within 30 s")
+		var zero T
+		return zero
+	}
+}
+
+func allocatedDuring(f func()) uint64 {
+	var before, after runtime.MemStats
+	runtime.GC()
+	runtime.ReadMemStats(&before)
+	f()
+	runtime.ReadMemStats(&after)
+	return after.TotalAlloc - before.TotalAlloc
+}
+
+// Test 5: a frame over MaxFrameBytes is rejected on its header, before the
+// receiver allocates its body.
+func TestReadFrame_OverBoundRejectedBeforeAllocation(t *testing.T) {
+	var hdr [4]byte
+	binary.BigEndian.PutUint32(hdr[:], uint32(MaxFrameBytes+1))
+	var err error
+	alloc := allocatedDuring(func() { _, err = readFrame(bytes.NewReader(hdr[:]), MaxFrameBytes) })
+	require.ErrorContains(t, err, "outside 1..")
+	require.Less(t, alloc, uint64(1<<20), "no body buffer was allocated")
+}
+
+// Test 5: a direct peer's frame with a small payload and metadata far over
+// 64 KiB, within MaxFrameBytes, reaches Handle: metadata has no separate bound.
+func TestControl_LargeMetadataWithinTheFrameIsAccepted(t *testing.T) {
+	name := controlName(t)
+	got := make(chan []byte, 1)
+	fullServe(t, fullListen(t, name), func(_ context.Context, req []byte, _ Credential) Reply {
+		got <- req
+		return Reply{Outcome: OutcomeDone}
+	})
+	req := requestOfSize(t, 16, 1<<20)
+	_, err := fullClient(name).Exchange(context.Background(), req, testCred, time.Now().Add(30*time.Second))
+	require.NoError(t, err)
+	require.Equal(t, req, recvBounded(t, got))
+}
+
+// Test 6a: a Request and a Reply of exactly MaxFrameBytes with a full 6 MiB
+// payload are both carried byte-identical.
+func TestControl_AtBoundRequestAndReplyBothFit(t *testing.T) {
+	name := controlName(t)
+	want := replyOfSize(t, MaxPayloadBytes, int(MaxFrameBytes))
+	got := make(chan []byte, 1)
+	fullServe(t, fullListen(t, name), func(_ context.Context, req []byte, _ Credential) Reply {
+		got <- req
+		return want
+	})
+	req := requestOfSize(t, MaxPayloadBytes, int(MaxFrameBytes))
+	rep, err := fullClient(name).Exchange(context.Background(), req, testCred, time.Now().Add(60*time.Second))
+	require.NoError(t, err)
+	require.Equal(t, req, recvBounded(t, got))
+	require.Equal(t, want, rep)
+}
+
+// Test 6a: an at-bound reply with no payload, whose payload field the encoder
+// omits, still fits.
+func TestControl_AtBoundReplyWithoutPayloadFits(t *testing.T) {
+	name := controlName(t)
+	want := replyOfSize(t, 0, int(MaxFrameBytes))
+	fullServe(t, fullListen(t, name), func(context.Context, []byte, Credential) Reply { return want })
+	rep, err := fullClient(name).Exchange(context.Background(), []byte(`{"attempt_id":"a"}`), testCred, time.Now().Add(60*time.Second))
+	require.NoError(t, err)
+	require.Equal(t, want, rep)
+}
+
+// Test 6b: the real server answers a handler's oversized payload, and a reply
+// encoding over MaxFrameBytes, with the small unknown reply.
+func TestControl_ServerReplacesOversizedRepliesWithUnknown(t *testing.T) {
+	over := Reply{Outcome: OutcomeDone, Payload: make([]byte, MaxPayloadBytes+1)}
+	for name, rep := range map[string]Reply{
+		"payload over 6 MiB":        over,
+		"encoding over frame bound": replyOfSize(t, MaxPayloadBytes, int(MaxFrameBytes)+1),
+	} {
+		t.Run(name, func(t *testing.T) {
+			sock := controlName(t)
+			fullServe(t, fullListen(t, sock), func(context.Context, []byte, Credential) Reply { return rep })
+			got, err := fullClient(sock).Exchange(context.Background(), []byte(`{"attempt_id":"a"}`), testCred, time.Now().Add(60*time.Second))
+			require.NoError(t, err)
+			require.Equal(t, Reply{Outcome: OutcomeUnknown, Error: "result too large for the test path"}, got)
+		})
+	}
+	// The oversized payload is refused before the reply is encoded: one
+	// exchange allocates far less than an 8 MiB base64 encoding.
+	sock := controlName(t)
+	fullServe(t, fullListen(t, sock), func(context.Context, []byte, Credential) Reply { return over })
+	var err error
+	alloc := allocatedDuring(func() {
+		_, err = fullClient(sock).Exchange(context.Background(), []byte(`{"attempt_id":"a"}`), testCred, time.Now().Add(60*time.Second))
+	})
+	require.NoError(t, err)
+	require.Less(t, alloc, uint64(4<<20), "the oversized reply was encoded before the payload check")
+}
+
+// directPeer serves one connection by reading the request and credential,
+// then writing with body; the peer's own error is returned on the channel.
+func directPeer(t *testing.T, name string, body func(net.Conn) error) chan error {
+	t.Helper()
+	ln := fullListen(t, name)
+	t.Cleanup(func() { _ = ln.Close() })
+	peerErr := make(chan error, 1)
+	go func() {
+		peerErr <- func() error {
+			c, err := ln.AcceptUnix()
+			if err != nil {
+				return err
+			}
+			defer func() { _ = c.Close() }()
+			if _, err := readFrame(c, MaxFrameBytes); err != nil {
+				return err
+			}
+			if _, err := readFrame(c, MaxCredentialBytes); err != nil {
+				return err
+			}
+			return body(c)
+		}()
+	}()
+	return peerErr
+}
+
+// Test 6: replies from a direct peer daemon over the payload bound or the
+// frame bound are errors after sending, so the gateway reports unknown.
+func TestControl_ClientRejectsOversizedReplies(t *testing.T) {
+	overPayload, err := json.Marshal(Reply{Outcome: OutcomeDone, Payload: make([]byte, MaxPayloadBytes+1)})
+	require.NoError(t, err)
+	require.LessOrEqual(t, int64(len(overPayload)), MaxFrameBytes)
+	cases := map[string]struct {
+		send func(net.Conn) error
+		want string
+	}{
+		"payload over 6 MiB": {func(c net.Conn) error { return writeFrame(c, overPayload) }, "reply payload of 6291457 bytes exceeds 6291456"},
+		"frame over bound": {func(c net.Conn) error {
+			var hdr [4]byte
+			binary.BigEndian.PutUint32(hdr[:], uint32(MaxFrameBytes+1))
+			_, err := c.Write(hdr[:])
+			return err
+		}, "outside 1.."},
+	}
+	for name, tc := range cases {
+		t.Run(name, func(t *testing.T) {
+			sock := controlName(t)
+			peerErr := directPeer(t, sock, tc.send)
+			_, err := fullClient(sock).Exchange(context.Background(), []byte(`{"attempt_id":"a"}`), testCred, time.Now().Add(30*time.Second))
+			require.NoError(t, recvBounded(t, peerErr), "the peer read the request and wrote its reply")
+			require.ErrorContains(t, err, tc.want)
+			require.NotErrorIs(t, err, ErrNotSent)
+		})
+	}
+}
+
+// Test 12: the credential bound is unchanged beside a full-size request.
+func TestControl_CredentialOverBoundRefusedWithAFullRequest(t *testing.T) {
+	cred := testCred
+	b, err := json.Marshal(cred)
+	require.NoError(t, err)
+	cred.SessionToken += strings.Repeat("t", MaxCredentialBytes+1-len(b))
+	b, err = json.Marshal(cred)
+	require.NoError(t, err)
+	require.Len(t, b, MaxCredentialBytes+1)
+
+	name := controlName(t)
+	s, _ := serveControl(t, name, os.Getuid(), nil)
+	_, err = fullClient(name).Exchange(context.Background(), requestOfSize(t, MaxPayloadBytes, int(MaxFrameBytes)), cred, time.Now().Add(30*time.Second))
+	require.ErrorIs(t, err, ErrNotSent)
+	require.Zero(t, s.calls.Load())
+}
+
+// pausedHandlers counts concurrent Handle calls and blocks each until released.
+type pausedHandlers struct {
+	in, max atomic.Int32
+	calls   atomic.Int32
+	entered chan []byte
+	release chan struct{}
+	once    sync.Once
+}
+
+func newPaused(t *testing.T) *pausedHandlers {
+	p := &pausedHandlers{entered: make(chan []byte, 64), release: make(chan struct{})}
+	t.Cleanup(p.free)
+	return p
+}
+
+func (p *pausedHandlers) free() { p.once.Do(func() { close(p.release) }) }
+
+func (p *pausedHandlers) handle(ctx context.Context, req []byte, _ Credential) Reply {
+	n := p.in.Add(1)
+	for {
+		m := p.max.Load()
+		if n <= m || p.max.CompareAndSwap(m, n) {
+			break
+		}
+	}
+	p.calls.Add(1)
+	p.entered <- req
+	select {
+	case <-p.release:
+	case <-ctx.Done():
+	}
+	p.in.Add(-1)
+	return Reply{Outcome: OutcomeDone}
+}
+
+func (p *pausedHandlers) waitEntered(t *testing.T, n int) {
+	t.Helper()
+	for range n {
+		select {
+		case <-p.entered:
+		case <-time.After(30 * time.Second):
+			t.Fatal("handler not entered")
+		}
+	}
+}
+
+// serverConns reads the kernel's socket table for the abstract name: server
+// ends Accept returned are connected and held by this process; a connection
+// still in the listen backlog is connecting, with no inode until accepted.
+func serverConns(t *testing.T, name string) (accepted, backlog int) {
+	t.Helper()
+	held := map[string]bool{}
+	fds, err := os.ReadDir("/proc/self/fd")
+	require.NoError(t, err)
+	for _, fd := range fds {
+		if target, err := os.Readlink("/proc/self/fd/" + fd.Name()); err == nil && strings.HasPrefix(target, "socket:[") {
+			held[strings.TrimSuffix(strings.TrimPrefix(target, "socket:["), "]")] = true
+		}
+	}
+	table, err := os.ReadFile("/proc/net/unix")
+	require.NoError(t, err)
+	for _, line := range strings.Split(string(table), "\n")[1:] {
+		f := strings.Fields(line)
+		// Num RefCount Protocol Flags Type St Inode Path.
+		if len(f) != 8 || f[7] != "@"+name {
+			continue
+		}
+		switch {
+		case f[5] == "03" && held[f[6]]:
+			accepted++
+		case f[5] == "02" && f[6] == "0":
+			backlog++
+		}
+	}
+	return accepted, backlog
+}
+
+// Test 8: eight full-size requests at once; with two admitted connections
+// paused in Handle, the kernel shows two accepted and six still in the
+// backlog, and at most two Handle calls ever run at once.
+func TestControl_AdmissionHoldsAtMostTwoFullSizeRequests(t *testing.T) {
+	name := controlName(t)
+	p := newPaused(t)
+	fullServe(t, fullListen(t, name), p.handle)
+	req := requestOfSize(t, MaxPayloadBytes, int(MaxFrameBytes))
+
+	var wg sync.WaitGroup
+	errs := make(chan error, 8)
+	for range 8 {
+		wg.Go(func() {
+			_, err := fullClient(name).Exchange(context.Background(), req, testCred, time.Now().Add(90*time.Second))
+			errs <- err
+		})
+	}
+	p.waitEntered(t, MaxAdmitted)
+	deadline := time.Now().Add(30 * time.Second)
+	for {
+		accepted, backlog := serverConns(t, name)
+		require.LessOrEqual(t, accepted, MaxAdmitted, "a third connection was accepted")
+		if backlog == 8-MaxAdmitted {
+			break
+		}
+		require.True(t, time.Now().Before(deadline), "the other clients never reached the backlog: %d accepted, %d waiting", accepted, backlog)
+		runtime.Gosched()
+	}
+	for range 20 {
+		accepted, backlog := serverConns(t, name)
+		require.Equal(t, MaxAdmitted, accepted, "only the admitted connections are accepted")
+		require.Equal(t, 8-MaxAdmitted, backlog, "the other six stay unaccepted")
+		<-time.After(50 * time.Millisecond)
+	}
+	require.Equal(t, int32(MaxAdmitted), p.calls.Load())
+
+	p.free()
+	waitGroupBounded(t, &wg, 90*time.Second)
+	close(errs)
+	for err := range errs {
+		require.NoError(t, err)
+	}
+	require.Equal(t, int32(8), p.calls.Load())
+	require.LessOrEqual(t, p.max.Load(), int32(MaxAdmitted))
+}
+
+// Test 9: a gateway that wrote its whole request while unaccepted, then gave
+// up and closed, reports unknown; the daemon later admits it and Handle runs.
+func TestControl_GatewayGivesUpButBufferedRequestIsStillHandled(t *testing.T) {
+	name := controlName(t)
+	p := newPaused(t)
+	fullServe(t, fullListen(t, name), p.handle)
+	for range MaxAdmitted {
+		go func() {
+			_, _ = fullClient(name).Exchange(context.Background(), []byte(`{"attempt_id":"held"}`), testCred, time.Now().Add(60*time.Second))
+		}()
+	}
+	p.waitEntered(t, MaxAdmitted)
+
+	late := []byte(`{"attempt_id":"late"}`)
+	c := fullClient(name)
+	c.IOTimeout = 2 * time.Second
+	_, err := c.Exchange(context.Background(), late, testCred, time.Now().Add(2*time.Second))
+	require.Error(t, err)
+	require.NotErrorIs(t, err, ErrNotSent, "the request was written, so the outcome is unknown")
+
+	p.free()
+	deadline := time.After(30 * time.Second)
+	for {
+		select {
+		case req := <-p.entered:
+			if bytes.Equal(req, late) {
+				return // Handle ran for a request whose gateway reported unknown.
+			}
+		case <-deadline:
+			t.Fatal("the buffered request was never handled")
+		}
+	}
+}
+
+// smallBacklogListener binds name with a listen backlog of one.
+func smallBacklogListener(t *testing.T, name string) *net.UnixListener {
+	t.Helper()
+	fd, err := unix.Socket(unix.AF_UNIX, unix.SOCK_STREAM|unix.SOCK_CLOEXEC, 0)
+	require.NoError(t, err)
+	require.NoError(t, unix.Bind(fd, &unix.SockaddrUnix{Name: "@" + name}))
+	require.NoError(t, unix.Listen(fd, 1))
+	f := os.NewFile(uintptr(fd), name)
+	defer func() { _ = f.Close() }()
+	l, err := net.FileListener(f)
+	require.NoError(t, err)
+	return l.(*net.UnixListener)
+}
+
+// Test 10: with both slots held and the backlog full, a gateway connect made
+// with no context deadline fails or stops at replyBy as ErrNotSent, and no
+// request reaches Handle.
+func TestControl_FullBacklogConnectIsNotSent(t *testing.T) {
+	name := controlName(t)
+	p := newPaused(t)
+	fullServe(t, smallBacklogListener(t, name), p.handle)
+	for range MaxAdmitted {
+		go func() {
+			_, _ = fullClient(name).Exchange(context.Background(), []byte(`{"attempt_id":"held"}`), testCred, time.Now().Add(60*time.Second))
+		}()
+	}
+	p.waitEntered(t, MaxAdmitted)
+
+	// Fill the backlog with raw connections that never write.
+	var fillers []net.Conn
+	t.Cleanup(func() {
+		for _, c := range fillers {
+			_ = c.Close()
+		}
+	})
+	for range 64 {
+		d := net.Dialer{Timeout: 200 * time.Millisecond}
+		c, err := d.Dial("unix", "@"+name)
+		if err != nil {
+			break
+		}
+		fillers = append(fillers, c)
+	}
+	require.Less(t, len(fillers), 64, "the backlog never filled")
+
+	began := time.Now()
+	_, err := fullClient(name).Exchange(context.Background(), []byte(`{"attempt_id":"x"}`), testCred, time.Now().Add(2*time.Second))
+	require.ErrorIs(t, err, ErrNotSent)
+	require.Less(t, time.Since(began), 10*time.Second)
+	require.Equal(t, int32(MaxAdmitted), p.calls.Load(), "no further request reached Handle")
+}
+
+// Test 15 (measurement, not a bound): two admitted full-size lifetimes, each
+// decoding and fit-checking its request as the daemon does and answering with
+// a 6 MiB reply, measured from a post-GC baseline just before the first dial
+// until both replies are read.
+func TestControl_MeasureTwoAdmittedFullSizeLifetimes(t *testing.T) {
+	name := controlName(t)
+	var mu sync.Mutex
+	entered := 0
+	both := make(chan struct{})
+	fitErrs := make(chan error, MaxAdmitted)
+	fullServe(t, fullListen(t, name), func(ctx context.Context, b []byte, cred Credential) Reply {
+		var req Request
+		if err := json.Unmarshal(b, &req); err != nil {
+			return Reply{Outcome: OutcomeRefused, Error: err.Error()}
+		}
+		req.cred = cred
+		fitErrs <- hostlink.CheckInvokeFits(guestRequest(req, 0))
+		mu.Lock()
+		entered++
+		if entered == MaxAdmitted {
+			close(both)
+		}
+		mu.Unlock()
+		select {
+		case <-both:
+		case <-ctx.Done():
+			return Reply{Outcome: OutcomeUnknown}
+		}
+		return Reply{Outcome: OutcomeDone, Payload: bytes.Repeat([]byte{9}, MaxPayloadBytes)}
+	})
+	req := requestOfSize(t, MaxPayloadBytes, int(MaxFrameBytes))
+
+	var wg sync.WaitGroup
+	var reps [MaxAdmitted]Reply
+	var errs [MaxAdmitted]error
+	var before, after runtime.MemStats
+	runtime.GC()
+	runtime.ReadMemStats(&before)
+
+	// HeapAlloc is sampled every 2 ms from the baseline until both replies are
+	// read, so its maximum is a sampled process-wide figure, not a true peak.
+	var heapMax uint64
+	samples := 0
+	stopSampling, sampled := make(chan struct{}), make(chan struct{})
+	go func() {
+		defer close(sampled)
+		tick := time.NewTicker(2 * time.Millisecond)
+		defer tick.Stop()
+		for {
+			var m runtime.MemStats
+			runtime.ReadMemStats(&m)
+			heapMax, samples = max(heapMax, m.HeapAlloc), samples+1
+			select {
+			case <-tick.C:
+			case <-stopSampling:
+				return
+			}
+		}
+	}()
+	var stopOnce sync.Once
+	stop := func() {
+		stopOnce.Do(func() { close(stopSampling) })
+		<-sampled
+	}
+	t.Cleanup(stop)
+
+	for i := range MaxAdmitted {
+		wg.Go(func() {
+			reps[i], errs[i] = fullClient(name).Exchange(context.Background(), req, testCred, time.Now().Add(60*time.Second))
+		})
+	}
+	waitGroupBounded(t, &wg, 60*time.Second)
+	stop()
+	runtime.ReadMemStats(&after)
+
+	for i := range MaxAdmitted {
+		require.NoError(t, errs[i])
+		require.Equal(t, OutcomeDone, reps[i].Outcome)
+		require.Len(t, reps[i].Payload, MaxPayloadBytes, "each lifetime ends with its full-size reply read")
+		require.NoError(t, recvBounded(t, fitErrs), "the full-size request fits the guest link")
+	}
+	// Excludes the executor's marshaling, HostSend, the guest and QEMU.
+	t.Logf("two admitted full-size lifetimes, process-wide (client and server): %d bytes in %d allocations; sampled max HeapAlloc %d bytes over %d samples",
+		after.TotalAlloc-before.TotalAlloc, after.Mallocs-before.Mallocs, heapMax, samples)
+}

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

@@ -0,0 +1,96 @@
+//go:build lambdatestlink
+
+//test:in-package — reads the daemon's unexported slot and owner fixtures
+package testlink
+
+import (
+	"context"
+	"encoding/json"
+	"net/http"
+	"net/http/httptest"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/require"
+
+	"github.com/mulgadc/spinifex/spinifex/domains/lambda/execution"
+)
+
+// A direct peer's request whose raw payload is 6 MiB + 1 but whose frame is
+// within MaxFrameBytes passes the frame check, reaches the real daemon, and
+// is refused there before Start with no address or tap taken.
+func TestControlDaemon_RawPayloadOverBoundRefusedBeforeStart(t *testing.T) {
+	code := []byte("zip bytes")
+	srv := codeServer(t, code)
+	g := grant(t, code)
+	ips, taps := &fakeIPs{}, &fakeTaps{}
+	d := g.daemon(&Daemon{IPs: ips, Taps: taps, HTTP: srv.Client(), WorkDir: t.TempDir()})
+
+	req := baseRequest(srv.URL, code)
+	req.Grant, req.ImageDigest = g.att.Grant, g.image.Digest
+	req.Payload = make([]byte, MaxPayloadBytes+1)
+	b, err := json.Marshal(req)
+	require.NoError(t, err)
+	require.LessOrEqual(t, int64(len(b)), MaxFrameBytes, "the frame alone would admit it")
+
+	name := controlName(t)
+	fullServe(t, fullListen(t, name), d.Handle)
+	rep, err := fullClient(name).Exchange(context.Background(), b, testCred, time.Now().Add(30*time.Second))
+	require.NoError(t, err)
+	require.Equal(t, OutcomeRefused, rep.Outcome, rep.Error)
+	require.Contains(t, rep.Error, "payload")
+	require.Empty(t, ips.allocated)
+	require.Empty(t, taps.created)
+	require.Equal(t, execution.StateGranted, g.state(t))
+}
+
+// With one real attempt started and held in its code fetch, a second request
+// through the real control server is refused before Start as busy.
+func TestControlDaemon_SecondRequestRefusedWhileAnAttemptIsHeld(t *testing.T) {
+	code := []byte("zip bytes")
+	fetching, release := make(chan struct{}), make(chan struct{})
+	var fetchOnce, releaseOnce sync.Once
+	free := func() { releaseOnce.Do(func() { close(release) }) }
+	srv := httptest.NewTLSServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		fetchOnce.Do(func() { close(fetching) })
+		select {
+		case <-release:
+		case <-r.Context().Done():
+			return
+		}
+		_, _ = w.Write(code)
+	}))
+	// Cleanups run last-registered first: free the held fetch before Close.
+	t.Cleanup(srv.Close)
+	t.Cleanup(free)
+	g := grant(t, code)
+	ips, taps := &fakeIPs{}, &fakeTaps{}
+	d := g.daemon(&Daemon{IPs: ips, Taps: taps, HTTP: srv.Client(), WorkDir: t.TempDir()})
+	name := controlName(t)
+	fullServe(t, fullListen(t, name), d.Handle)
+
+	first := make(chan error, 1)
+	go func() {
+		_, err := fullClient(name).Exchange(context.Background(), g.request(srv.URL, code), testCred, time.Now().Add(60*time.Second))
+		first <- err
+	}()
+	// On every path the held attempt is released and its exchange joined.
+	t.Cleanup(func() {
+		free()
+		select {
+		case <-first:
+		case <-time.After(30 * time.Second):
+			t.Error("the held exchange did not return")
+		}
+	})
+	recvBounded(t, fetching)
+	require.Equal(t, execution.StateRunning, g.state(t), "the first attempt started")
+
+	rep, err := fullClient(name).Exchange(context.Background(), g.request(srv.URL, code), testCred, time.Now().Add(30*time.Second))
+	require.NoError(t, err)
+	require.Equal(t, OutcomeRefused, rep.Outcome)
+	require.Contains(t, rep.Error, "another")
+	require.Empty(t, ips.allocated, "nothing was allocated for either attempt yet")
+	require.Empty(t, taps.created)
+}

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

@@ -0,0 +1,169 @@
+//go:build lambdatestlink
+
+//test:in-package — exercises the unexported joinAfterFunc and Serve's admission lifecycle
+package testlink
+
+import (
+	"context"
+	"net"
+	"os"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/require"
+	"go.uber.org/goleak"
+)
+
+// waitClosed fails the test if ch is not closed within d.
+func waitClosed(t *testing.T, ch <-chan struct{}, d time.Duration, what string) {
+	t.Helper()
+	select {
+	case <-ch:
+	case <-time.After(d):
+		t.Fatal(what)
+	}
+}
+
+// waitGroupBounded fails the test if wg does not finish within d.
+func waitGroupBounded(t *testing.T, wg *sync.WaitGroup, d time.Duration) {
+	t.Helper()
+	done := make(chan struct{})
+	go func() { wg.Wait(); close(done) }()
+	waitClosed(t, done, d, "goroutines did not finish")
+}
+
+// A stop after the callback has started waits for it to finish, so no
+// close is still running when its caller releases a slot or returns.
+func TestJoinAfterFunc_StopWaitsForAStartedCallback(t *testing.T) {
+	ctx, cancel := context.WithCancel(t.Context())
+	started, release := make(chan struct{}), make(chan struct{})
+	var once sync.Once
+	unblock := func() { once.Do(func() { close(release) }) }
+	t.Cleanup(unblock)
+	stop := joinAfterFunc(ctx, func() { close(started); <-release })
+	cancel()
+	waitClosed(t, started, 5*time.Second, "callback never started")
+
+	returned := make(chan struct{})
+	go func() { stop(); close(returned) }()
+	select {
+	case <-returned:
+		t.Fatal("stop returned while the callback was still running")
+	case <-time.After(100 * time.Millisecond):
+	}
+	unblock()
+	waitClosed(t, returned, 5*time.Second, "stop never returned after the callback finished")
+}
+
+func TestJoinAfterFunc_StopBeforeCancelNeverRunsTheCallback(t *testing.T) {
+	ctx, cancel := context.WithCancel(t.Context())
+	ran := make(chan struct{}, 1)
+	stop := joinAfterFunc(ctx, func() { ran <- struct{}{} })
+	stop()
+	cancel()
+	select {
+	case <-ran:
+		t.Fatal("callback ran after stop")
+	case <-time.After(50 * time.Millisecond):
+	}
+}
+
+// startServe runs Serve with the given IOTimeout and handler and returns a
+// cancel that waits for Serve to return, reporting how long that took.
+func startServe(t *testing.T, name string, io time.Duration, handle func(context.Context, []byte, Credential) Reply) func() time.Duration {
+	t.Helper()
+	ln, err := ListenControl(name)
+	require.NoError(t, err)
+	srv := &ControlServer{GatewayUID: os.Getuid(), RequestLimit: testRequestLimit, ReplyLimit: 4096, IOTimeout: io, Handle: handle}
+	ctx, cancel := context.WithCancel(context.Background())
+	done := make(chan error, 1)
+	go func() { done <- srv.Serve(ctx, ln) }()
+	var once sync.Once
+	var took time.Duration
+	stop := func() time.Duration {
+		once.Do(func() {
+			began := time.Now()
+			cancel()
+			select {
+			case err := <-done:
+				require.NoError(t, err)
+			case <-time.After(20 * time.Second):
+				t.Fatal("Serve did not return after cancellation")
+			}
+			took = time.Since(began)
+		})
+		return took
+	}
+	t.Cleanup(func() { stop() })
+	return stop
+}
+
+// Shutdown closes an admitted peer stalled mid-request and a handler that
+// honours cancellation, well inside the 30 s read deadline, leaving no goroutine.
+func TestServe_ShutdownClosesAStalledPeerAndACancelledHandler(t *testing.T) {
+	defer goleak.VerifyNone(t, goleak.IgnoreCurrent())
+	name := controlName(t)
+	inHandle := make(chan struct{})
+	stop := startServe(t, name, 30*time.Second, func(ctx context.Context, _ []byte, _ Credential) Reply {
+		close(inHandle)
+		<-ctx.Done()
+		return Reply{Outcome: OutcomeUnknown}
+	})
+
+	stalled, err := net.DialUnix("unix", nil, &net.UnixAddr{Name: "@" + name, Net: "unix"})
+	require.NoError(t, err)
+	defer func() { _ = stalled.Close() }()
+	// Half a frame header: the server blocks reading the rest.
+	_, err = stalled.Write([]byte{0, 0})
+	require.NoError(t, err)
+
+	exchanged := make(chan error, 1)
+	go func() {
+		_, err := client(name, os.Getuid()).Exchange(context.Background(), []byte(`{"attempt_id":"a"}`), testCred, time.Now().Add(30*time.Second))
+		exchanged <- err
+	}()
+	select {
+	case <-inHandle:
+	case <-time.After(5 * time.Second):
+		t.Fatal("second admitted connection never reached Handle")
+	}
+
+	took := stop()
+	require.Less(t, took, 5*time.Second, "shutdown waited on the read deadline")
+	select {
+	case <-exchanged:
+	case <-time.After(5 * time.Second):
+		t.Fatal("gateway exchange never ended after shutdown")
+	}
+	require.NoError(t, stalled.SetReadDeadline(time.Now().Add(5*time.Second)))
+	_, err = stalled.Read(make([]byte, 1))
+	require.Error(t, err, "the stalled peer's connection was closed")
+}
+
+// Cancellation racing normal completion never leaves a close callback or
+// connection goroutine behind once Serve returns.
+func TestServe_CancellationRacingCompletionLeavesNothingRunning(t *testing.T) {
+	defer goleak.VerifyNone(t, goleak.IgnoreCurrent())
+	for i := range 50 {
+		name := controlName(t)
+		stop := startServe(t, name, 5*time.Second, func(context.Context, []byte, Credential) Reply {
+			return Reply{Outcome: OutcomeDone, Payload: []byte("ok")}
+		})
+		// Alternate cancelling before, during and after the exchanges.
+		if i%3 == 0 {
+			stop()
+		}
+		var wg sync.WaitGroup
+		for range MaxAdmitted {
+			wg.Go(func() {
+				_, _ = client(name, os.Getuid()).Exchange(context.Background(), []byte(`{"attempt_id":"a"}`), testCred, time.Now().Add(5*time.Second))
+			})
+		}
+		if i%3 == 1 {
+			stop()
+		}
+		waitGroupBounded(t, &wg, 15*time.Second)
+		stop()
+	}
+}

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

@@ -435,3 +435,30 @@ func TestControl_WriteErrorAfterDeliveryIsUnknown(t *testing.T) {
 	require.Equal(t, []byte(`{"attempt_id":"a"}`), s.req.Load())
 	require.Equal(t, testCred, s.cred.Load())
 }
+
+// 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 predecessors.
+func TestControl_MixedProtocolNamesNeverConnect(t *testing.T) {
+	prefix := controlName(t) + "/"
+	for name, tc := range map[string]struct{ daemon, gateway string }{
+		"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)
+			_, err := client(tc.gateway, os.Getuid()).Exchange(t.Context(), []byte(`{"attempt_id":"a"}`), testCred, soon())
+			require.ErrorIs(t, err, ErrNotSent)
+			assert.Equal(t, int32(0), s.calls.Load())
+		})
+	}
+}

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

@@ -11,6 +11,7 @@ import (
 	"fmt"
 	"io"
 	"log/slog"
+	"math"
 	"net"
 	"net/http"
 	"net/url"
@@ -113,12 +114,12 @@ const controlIOTimeout = 30 * time.Second
 
 // ServeControl binds the control socket name and answers the gateway's
 // requests on it until ctx ends. A bind error means this daemon does not serve.
-func (d *Daemon) ServeControl(ctx context.Context, name string, gatewayUID int, limit int64) error {
+func (d *Daemon) ServeControl(ctx context.Context, name string, gatewayUID int) error {
 	ln, err := ListenControl(name)
 	if err != nil {
 		return err
 	}
-	srv := &ControlServer{GatewayUID: gatewayUID, RequestLimit: limit, ReplyLimit: limit,
+	srv := &ControlServer{GatewayUID: gatewayUID, RequestLimit: MaxFrameBytes, ReplyLimit: MaxFrameBytes,
 		IOTimeout: controlIOTimeout, Handle: d.Handle}
 	go func() {
 		if err := srv.Serve(ctx, ln); err != nil {
@@ -149,6 +150,9 @@ func (d *Daemon) Handle(ctx context.Context, data []byte, cred Credential) Reply
 	if d.Image == nil {
 		return refused("runtime image not prepared")
 	}
+	if req.ImageDigest != d.Image.Digest {
+		return refused("runtime image digest differs from the prepared image")
+	}
 	if !d.busy.CompareAndSwap(false, true) {
 		return refused("another test-link attempt holds this node's slot")
 	}
@@ -162,6 +166,12 @@ func (d *Daemon) Handle(ctx context.Context, data []byte, cred Credential) Reply
 	if id, err := d.unresolved(); err != nil || id != "" {
 		return refused("an unresolved test-link attempt holds this node's slot")
 	}
+	// The guest frame is checked at its longest deadline encoding, MinInt64,
+	// before Start: control and guest frames bound different fields. It runs
+	// after the slot is taken so a busy refusal never encodes the payload.
+	if err := hostlink.CheckInvokeFits(guestRequest(req, math.MinInt64)); err != nil {
+		return refused("request does not fit the guest link")
+	}
 	// Without pidfs a restarted daemon could not identify the VM it left.
 	if _, err := selfInode(); err != nil {
 		return refused(err.Error())
@@ -193,7 +203,7 @@ func (d *Daemon) attempt(ctx context.Context, req Request, launch func(code []by
 	if err := d.Grants.Start(ctx, req.Grant, req.Identity()); err != nil {
 		if errors.Is(err, execution.ErrOutcomeUnknown) {
 			slog.Error("Lambda test link: grant start outcome unknown; node held", "attempt", req.AttemptID, "err", err)
-			return Reply{Outcome: OutcomeUnknown, Error: "grant start outcome unknown"}, true
+			return Reply{Outcome: OutcomeUnknown, Error: "grant start outcome unknown"}, false
 		}
 		return refused(err.Error()), true
 	}
@@ -251,8 +261,25 @@ func removeMarker(marker string) bool {
 	return true
 }
 
+// guestRequest is the invocation frame sent to the guest; the pre-start fit
+// check and the send share it so they cannot drift.
+func guestRequest(req Request, deadlineMs int64) hostlink.Request {
+	return hostlink.Request{
+		GrantID:     req.Grant.GrantID,
+		AttemptID:   req.AttemptID,
+		RequestID:   req.RequestID,
+		FunctionARN: req.FunctionARN,
+		DeadlineMs:  deadlineMs,
+		Payload:     req.Payload,
+		Credential:  &req.cred,
+	}
+}
+
 // CheckRequest is the daemon's validation of a test-link request.
 func CheckRequest(req Request) error {
+	if len(req.Payload) > MaxPayloadBytes {
+		return fmt.Errorf("payload of %d bytes exceeds %d", len(req.Payload), MaxPayloadBytes)
+	}
 	if !attemptIDPattern.MatchString(req.AttemptID) {
 		return errors.New("attempt ID must be 12 lowercase hex digits")
 	}
@@ -577,15 +604,7 @@ func (d *Daemon) session(ctx context.Context, req Request, gen *guestconn.Genera
 	if expired(hardDeadline) {
 		return Reply{}, ErrDeadlineExpired
 	}
-	if err := stream.HostSend(hostlink.Request{
-		GrantID:     req.Grant.GrantID,
-		AttemptID:   req.AttemptID,
-		RequestID:   req.RequestID,
-		FunctionARN: req.FunctionARN,
-		DeadlineMs:  deadline.UnixMilli(),
-		Payload:     req.Payload,
-		Credential:  &req.cred,
-	}); err != nil {
+	if err := stream.HostSend(guestRequest(req, deadline.UnixMilli())); err != nil {
 		return Reply{}, pastDeadline(deadline, fmt.Errorf("send invocation: %w", err))
 	}
 	resp, err := stream.HostResult(rctx)

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

@@ -0,0 +1,146 @@
+//go:build lambdatestlink
+
+//test:in-package — drives Daemon.Handle with the package's grant fixtures and guestRequest
+package testlink
+
+import (
+	"context"
+	"encoding/json"
+	"math"
+	"net"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/require"
+
+	"github.com/mulgadc/spinifex/spinifex/agents/lambda/hostlink"
+	"github.com/mulgadc/spinifex/spinifex/domains/lambda/execution"
+)
+
+// maxCredential encodes to exactly MaxCredentialBytes.
+func maxCredential(t *testing.T) Credential {
+	t.Helper()
+	c := Credential{AccessKeyID: "ASIA" + strings.Repeat("K", 16), SecretAccessKey: strings.Repeat("s", 40)}
+	b, err := json.Marshal(c)
+	require.NoError(t, err)
+	c.SessionToken = strings.Repeat("t", MaxCredentialBytes-len(b))
+	b, err = json.Marshal(c)
+	require.NoError(t, err)
+	require.Len(t, b, MaxCredentialBytes)
+	return c
+}
+
+// A request the control socket accepts but whose guest frame would not fit
+// is refused before Start, so no grant starts and nothing is allocated.
+func TestDaemon_GuestFrameOverflowRefusedBeforeStart(t *testing.T) {
+	code := []byte("zip bytes")
+	good := codeServer(t, code)
+	g := grant(t, code)
+	ips, taps := &fakeIPs{}, &fakeTaps{}
+	d := g.daemon(&Daemon{IPs: ips, Taps: taps, HTTP: good.Client(), WorkDir: t.TempDir()})
+
+	req := baseRequest(good.URL, code)
+	req.Grant, req.ImageDigest = g.att.Grant, g.image.Digest
+	req.Payload = make([]byte, MaxPayloadBytes)
+	req.FunctionARN = strings.Repeat("a", 60000)
+	b, err := json.Marshal(req)
+	require.NoError(t, err)
+	require.LessOrEqual(t, int64(len(b)), MaxFrameBytes, "the control frame fits")
+	require.NoError(t, CheckRequest(req))
+
+	rep := d.Handle(context.Background(), b, maxCredential(t))
+	require.Equal(t, OutcomeRefused, rep.Outcome, rep.Error)
+	require.Contains(t, rep.Error, "does not fit the guest link")
+	require.Equal(t, execution.StateGranted, g.state(t), "the grant was never started")
+	require.False(t, d.busy.Load(), "a fit refusal releases the slot")
+	require.Empty(t, ips.allocated)
+	require.Empty(t, taps.created)
+
+	// A later valid request for the same grant is not refused as busy.
+	rep = d.Handle(context.Background(), g.request(good.URL, code), testCred)
+	require.NotContains(t, rep.Error, "holds this node's slot")
+	require.NotEqual(t, execution.StateGranted, g.state(t), "the valid request started the grant")
+}
+
+// The largest Invoke ARN, a full payload, a maximal credential and full IDs
+// reach a guest reader byte-identical through HostSend.
+func TestGuestRequest_LargestInvokeFitsTheGuestLink(t *testing.T) {
+	cred := maxCredential(t)
+	req := Request{
+		Grant:     execution.Grant{GrantID: strings.Repeat("f", 32)},
+		AttemptID: strings.Repeat("a", 12),
+		RequestID: strings.Repeat("b", 12),
+		// Region of representative length; name and qualifier at Invoke's maxima.
+		FunctionARN: "arn:aws:lambda:ap-southeast-2:123456789012:function:" + strings.Repeat("n", 64) + ":" + strings.Repeat("q", 128),
+		Payload:     make([]byte, MaxPayloadBytes),
+		cred:        cred,
+	}
+	for i := range req.Payload {
+		req.Payload[i] = byte(i)
+	}
+	sent := guestRequest(req, math.MinInt64)
+	require.NoError(t, hostlink.CheckInvokeFits(sent))
+
+	hostEnd, guestEnd := net.Pipe()
+	t.Cleanup(func() { _ = hostEnd.Close(); _ = guestEnd.Close() })
+	host, guest := hostlink.NewTestStream(hostEnd), hostlink.NewTestStream(guestEnd)
+	sendErr := make(chan error, 1)
+	go func() { sendErr <- host.HostSend(sent) }()
+
+	ctx, cancel := context.WithTimeout(t.Context(), 30*time.Second)
+	defer cancel()
+	got, err := guest.Next(ctx)
+	require.NoError(t, err)
+	select {
+	case err := <-sendErr:
+		require.NoError(t, err)
+	case <-time.After(30 * time.Second):
+		t.Fatal("HostSend never returned")
+	}
+	require.Equal(t, sent, got)
+}
+
+// At a request size where only the minus sign of MinInt64 overflows the guest
+// frame, the daemon still refuses before Start: its check uses the longest
+// int64 encoding, not MaxInt64.
+func TestDaemon_GuestFrameCheckUsesTheLongestDeadlineEncoding(t *testing.T) {
+	code := []byte("zip bytes")
+	good := codeServer(t, code)
+	g := grant(t, code)
+	ips, taps := &fakeIPs{}, &fakeTaps{}
+	d := g.daemon(&Daemon{IPs: ips, Taps: taps, HTTP: good.Client(), WorkDir: t.TempDir()})
+	cred := maxCredential(t)
+
+	req := baseRequest(good.URL, code)
+	req.Grant, req.ImageDigest = g.att.Grant, g.image.Digest
+	req.Payload = make([]byte, MaxPayloadBytes)
+	req.cred = cred
+	fits := func(n int, deadline int64) bool {
+		r := req
+		r.FunctionARN = strings.Repeat("a", n)
+		return hostlink.CheckInvokeFits(guestRequest(r, deadline)) == nil
+	}
+	// The longest ARN that fits at MaxInt64.
+	lo, hi := 0, 70000
+	require.True(t, fits(lo, math.MaxInt64))
+	require.False(t, fits(hi, math.MaxInt64))
+	for hi-lo > 1 {
+		if mid := (lo + hi) / 2; fits(mid, math.MaxInt64) {
+			lo = mid
+		} else {
+			hi = mid
+		}
+	}
+	require.False(t, fits(lo, math.MinInt64), "MinInt64 is one byte longer")
+
+	req.FunctionARN = strings.Repeat("a", lo)
+	b, err := json.Marshal(req)
+	require.NoError(t, err)
+	require.LessOrEqual(t, int64(len(b)), MaxFrameBytes)
+	rep := d.Handle(context.Background(), b, cred)
+	require.Equal(t, OutcomeRefused, rep.Outcome, rep.Error)
+	require.Contains(t, rep.Error, "does not fit the guest link")
+	require.Equal(t, execution.StateGranted, g.state(t))
+	require.False(t, d.busy.Load(), "a fit refusal releases the slot")
+}

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

@@ -121,13 +121,18 @@ func baseRequest(url string, code []byte) Request {
 	return Request{
 		AttemptID: "0123456789ab", RequestID: "r", Region: "ap-southeast-2", AccountID: "123456789012",
 		FunctionID: "fn1", RevisionID: "rev1", Handler: "bootstrap.handler", TimeoutMs: 3000, CodeURL: url, CodeSize: int64(len(code)),
-		CodeSHA256: hex.EncodeToString(sum[:]),
+		CodeSHA256: hex.EncodeToString(sum[:]), ImageDigest: testImageDigest,
 	}
 }
 
 func (g *granted) request(url string, code []byte) []byte {
+	return g.requestFor(url, code, g.image.Digest)
+}
+
+func (g *granted) requestFor(url string, code []byte, imageDigest string) []byte {
 	req := baseRequest(url, code)
 	req.Grant = g.att.Grant
+	req.ImageDigest = imageDigest
 	b, _ := json.Marshal(req)
 	return b
 }
@@ -148,6 +153,10 @@ func TestDaemon_RefusesBeforeAllocating(t *testing.T) {
 			b, _ := json.Marshal(baseRequest(good.URL, code))
 			return b
 		}, execution.StateGranted},
+		// A gateway built for another runtime image, or an old one that sends
+		// no digest, is refused before Start.
+		"image digest differs": {func(g *granted) []byte { return g.requestFor(good.URL, code, strings.Repeat("0", 64)) }, execution.StateGranted},
+		"image digest missing": {func(g *granted) []byte { return g.requestFor(good.URL, code, "") }, execution.StateGranted},
 		// Refused after Start with nothing launched: the daemon records the end.
 		"digest mismatch": {func(g *granted) []byte { return g.request(wrong.URL, code) }, execution.StateEnded},
 		"size mismatch":   {func(g *granted) []byte { return g.request(short.URL, code) }, execution.StateEnded},
@@ -164,6 +173,7 @@ func TestDaemon_RefusesBeforeAllocating(t *testing.T) {
 			rep := g.daemon(&Daemon{IPs: ips, Taps: taps, HTTP: client, WorkDir: t.TempDir()}).Handle(context.Background(), tc.data(g), testCred)
 			require.Equal(t, OutcomeRefused, rep.Outcome, rep.Error)
 			require.Empty(t, ips.allocated)
+			require.Empty(t, taps.created)
 			require.NotContains(t, rep.Error, good.URL)
 			require.Equal(t, tc.state, g.state(t))
 		})
@@ -198,6 +208,7 @@ func TestDaemon_OwnerRefusalComesBeforeAnyAllocation(t *testing.T) {
 			ips := &fakeIPs{}
 			rep := g.daemon(&Daemon{IPs: ips, Taps: &fakeTaps{}, HTTP: srv.Client(), WorkDir: t.TempDir()}).Handle(context.Background(), data, testCred)
 			require.Equal(t, OutcomeRefused, rep.Outcome)
+			require.NotContains(t, rep.Error, "image digest", "the owner, not the digest check, refuses")
 			require.False(t, fetched)
 			require.Empty(t, ips.allocated)
 			require.Equal(t, execution.StateGranted, g.state(t))
@@ -447,3 +458,65 @@ func TestCheckRequest_RefusesUnusableHandler(t *testing.T) {
 		})
 	}
 }
+
+// unconfirmedStart lets the owner write the start, then reports the outcome
+// as unknown, as a lost acknowledgement would.
+type unconfirmedStart struct{ *execution.Store }
+
+func (u unconfirmedStart) Start(ctx context.Context, g execution.Grant, want execution.Identity) error {
+	if err := u.Store.Start(ctx, g, want); err != nil {
+		return err
+	}
+	return execution.ErrOutcomeUnknown
+}
+
+func TestDaemon_UnknownStartOutcomeHoldsTheSlot(t *testing.T) {
+	code := []byte("x")
+	g := grant(t, code)
+	ips, taps := &fakeIPs{}, &fakeTaps{}
+	d := g.daemon(&Daemon{IPs: ips, Taps: taps, WorkDir: t.TempDir()})
+	d.Grants = unconfirmedStart{g.store}
+
+	rep := d.Handle(context.Background(), g.request("https://h/x", code), testCred)
+	require.Equal(t, OutcomeUnknown, rep.Outcome)
+	require.True(t, d.busy.Load(), "an unconfirmed start may have taken the node")
+	require.Empty(t, ips.allocated)
+	require.Equal(t, execution.StateRunning, g.state(t))
+
+	rep = d.Handle(context.Background(), g.request("https://h/x", code), testCred)
+	require.Equal(t, OutcomeRefused, rep.Outcome)
+	require.Contains(t, rep.Error, "holds this node's slot")
+}
+
+// A daemon whose image is replaced between two Invokes refuses the second
+// before Start, though the first passed the digest check.
+func TestDaemon_ImageReplacedBetweenInvokesRefusesBeforeStart(t *testing.T) {
+	code := []byte("zip bytes")
+	wrong := codeServer(t, []byte("other one"))
+	g := grant(t, code)
+	ips, taps := &fakeIPs{}, &fakeTaps{}
+	d := g.daemon(&Daemon{IPs: ips, Taps: taps, HTTP: wrong.Client(), WorkDir: t.TempDir()})
+
+	rep := d.Handle(context.Background(), g.request(wrong.URL, code), testCred)
+	require.Equal(t, OutcomeRefused, rep.Outcome)
+	require.Equal(t, execution.StateEnded, g.state(t), "the first request passed the digest check and started")
+
+	second, err := g.store.Issue(context.Background(), execution.Attempt{
+		Grant:    execution.Grant{AttemptID: "0123456789ac", EnvironmentID: "env-0123456789ac", Generation: 1, Node: testNode},
+		Identity: baseRequest("", code).Identity(),
+		LaunchBy: time.Now().Add(time.Minute),
+	})
+	require.NoError(t, err)
+	d.Image = &Image{Digest: strings.Repeat("1", 64)}
+	req := baseRequest(wrong.URL, code)
+	req.AttemptID, req.Grant, req.ImageDigest = second.Grant.AttemptID, second.Grant, g.image.Digest
+	b, err := json.Marshal(req)
+	require.NoError(t, err)
+
+	rep = d.Handle(context.Background(), b, testCred)
+	require.Equal(t, OutcomeRefused, rep.Outcome)
+	require.Contains(t, rep.Error, "image digest differs")
+	require.Equal(t, execution.StateGranted, g.state(t), "the second grant was never started")
+	require.Empty(t, ips.allocated)
+	require.Empty(t, taps.created)
+}

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

@@ -78,7 +78,7 @@ func (d *Daemon) startLambdaTestLink() {
 			if err != nil {
 				return err
 			}
-			return tl.ServeControl(d.ctx, testlink.ControlName, gatewayUID, d.natsConn.MaxPayload())
+			return tl.ServeControl(d.ctx, testlink.ControlName, gatewayUID)
 		},
 		func() { releaseLambdaTestLinkImage(tl) },
 	)

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
+}

spinifex/runtime/roles/awsgw/lambda_testlink.go

@@ -24,17 +24,15 @@ func lambdaTestExecutor(nc *nats.Conn, node string, codes *codestore.Manager, se
 		slog.Error("Lambda test link: control peer account unavailable; Invoke is not served", "err", err)
 		return nil
 	}
+	if !testlink.ValidDigest(lambdaImageDigest) {
+		slog.Error("Lambda test link: runtime image pin missing or malformed; every Invoke is refused")
+	}
 	return &testLinkExecutor{
-		Control: &testlink.ControlClient{
-			Name:         testlink.ControlName,
-			DaemonUID:    daemonUID,
-			RequestLimit: nc.MaxPayload(),
-			ReplyLimit:   nc.MaxPayload(),
-			IOTimeout:    controlWriteTimeout,
-		},
-		RequestLimit: nc.MaxPayload(),
+		Control:      testLinkControl(daemonUID),
+		RequestLimit: testlink.MaxFrameBytes,
 		Sessions:     sessions,
 		Node:         node,
+		ImageDigest:  lambdaImageDigest,
 		Codes: func(ctx context.Context) (codePresigner, error) {
 			store, err := codes.Store(ctx)
 			if err != nil {
@@ -52,6 +50,21 @@ func lambdaTestExecutor(nc *nats.Conn, node string, codes *codestore.Manager, se
 	}
 }
 
+// testLinkControl is the gateway's client for this node's daemon socket.
+func testLinkControl(daemonUID int) *testlink.ControlClient {
+	return &testlink.ControlClient{
+		Name:         testlink.ControlName,
+		DaemonUID:    daemonUID,
+		RequestLimit: testlink.MaxFrameBytes,
+		ReplyLimit:   testlink.MaxFrameBytes,
+		IOTimeout:    controlWriteTimeout,
+	}
+}
+
+// lambdaImageDigest pins the runtime image this gateway expects the daemon
+// to run; make build-lambda-testlink sets it to the daemon's pin.
+var lambdaImageDigest string
+
 // controlWriteTimeout bounds writing one request to the local daemon.
 const controlWriteTimeout = 30 * time.Second
 

spinifex/runtime/roles/awsgw/lambda_testlink_executor.go

@@ -55,11 +55,12 @@ var _ grantIssuer = (*execution.Store)(nil)
 // testLinkExecutor sends each invocation to the test-link daemon on Node.
 type testLinkExecutor struct {
 	Control controlExchanger
-	// RequestLimit bounds the encoded request without its credential: the
-	// NATS message limit the path used before the socket.
+	// RequestLimit bounds the encoded request frame, without its credential.
 	RequestLimit int64
 	Sessions     lambdaEnvironmentRoleIssuer
 	Node         string
+	// ImageDigest is the runtime image every request requires of the daemon.
+	ImageDigest string
 	Codes  func(context.Context) (codePresigner, error)
 	Grants func(context.Context) (grantIssuer, error)
 	now    func() time.Time
@@ -73,6 +74,10 @@ var _ awsapi.Executor = (*testLinkExecutor)(nil)
 // session is revoked on every path after it is issued.
 func (e *testLinkExecutor) InvokeSync(ctx context.Context, in awsapi.SyncInvocation) (awsapi.SyncResult, error) {
 	rec := in.Function
+	if !testlink.ValidDigest(e.ImageDigest) {
+		return awsapi.SyncResult{}, &awsapi.RefusalError{Err: awserrors.Errorf(awserrors.ErrorServiceException,
+			"the test execution path has no valid runtime image pin")}
+	}
 	// The test link cannot deliver environment variables, so it refuses a
 	// function that has any before presigning, granting or sending.
 	if rec.EnvCipher != "" {
@@ -159,6 +164,11 @@ func (e *testLinkExecutor) InvokeSync(ctx context.Context, in awsapi.SyncInvocat
 		return err
 	}
 
+	// The raw payload is checked before encoding allocates its base64 copy.
+	if len(in.Payload) > testlink.MaxPayloadBytes {
+		_ = revoke()
+		return refuse(fmt.Errorf("payload of %d bytes exceeds %d", len(in.Payload), testlink.MaxPayloadBytes))
+	}
 	body, err := json.Marshal(testlink.Request{
 		Grant:       att.Grant,
 		AttemptID:   attemptID,
@@ -174,13 +184,14 @@ func (e *testLinkExecutor) InvokeSync(ctx context.Context, in awsapi.SyncInvocat
 		CodeURL:     url,
 		CodeSize:    rec.Code.Size,
 		CodeSHA256:  rec.Code.SHA256,
+		ImageDigest: e.ImageDigest,
 		Payload:     in.Payload,
 	})
 	if err != nil {
 		_ = revoke()
 		return refuse(errors.New("encode test-link request"))
 	}
-	// The encoded request, base64 payload included, keeps the NATS-era bound.
+	// The encoded frame, base64 payload included, is checked after encoding.
 	if int64(len(body)) > e.RequestLimit {
 		_ = revoke()
 		return refuse(fmt.Errorf("encoded request of %d bytes exceeds the %d-byte message limit", len(body), e.RequestLimit))

spinifex/runtime/roles/awsgw/lambda_testlink_executor_test.go

@@ -10,6 +10,7 @@ import (
 	"io"
 	"net"
 	"os"
+	"runtime"
 	"strings"
 	"testing"
 	"time"
@@ -114,8 +115,11 @@ func executor(n *fakeControl, presignErr error, store *execution.Store) *testLin
 	return executorWith(n, &fakeSessions{}, presignErr, store)
 }
 
+// testImageDigest stands in for the build's pinned runtime image digest.
+const testImageDigest = "ab12ab12ab12ab12ab12ab12ab12ab12ab12ab12ab12ab12ab12ab12ab12ab12"
+
 func executorWith(n *fakeControl, sessions *fakeSessions, presignErr error, store *execution.Store) *testLinkExecutor {
-	return &testLinkExecutor{Control: n, RequestLimit: n.max, Sessions: sessions, Node: "node1",
+	return &testLinkExecutor{Control: n, RequestLimit: n.max, Sessions: sessions, Node: "node1", ImageDigest: testImageDigest,
 		Codes:  func(context.Context) (codePresigner, error) { return fakePresigner{err: presignErr}, nil },
 		Grants: func(context.Context) (grantIssuer, error) { return store, nil },
 	}
@@ -161,6 +165,25 @@ func TestLambdaTestLinkExecutor_RefusesWithoutSending(t *testing.T) {
 	}
 }
 
+// A gateway built without a valid image pin refuses before presigning,
+// granting, issuing or connecting.
+func TestLambdaTestLinkExecutor_InvalidImagePinRefusesBeforeConnecting(t *testing.T) {
+	for name, pin := range map[string]string{"missing": "", "malformed": "not-a-digest"} {
+		t.Run(name, func(t *testing.T) {
+			n, store := &fakeControl{max: 1 << 20, reply: &testlink.Reply{Outcome: testlink.OutcomeDone}}, grantStore(t)
+			sessions := &fakeSessions{}
+			e := executorWith(n, sessions, nil, store)
+			e.ImageDigest = pin
+			_, err := e.InvokeSync(context.Background(), invocation(nil))
+			var refusal *awsapi.RefusalError
+			require.ErrorAs(t, err, &refusal)
+			require.Empty(t, n.sent)
+			require.Empty(t, sessions.issued)
+			require.Empty(t, grantState(t, store), "no grant taken")
+		})
+	}
+}
+
 func TestLambdaTestLinkExecutor_BusyNodeSendsNothing(t *testing.T) {
 	store := grantStore(t)
 	held := &fakeControl{max: 1 << 20, err: errors.New("no reply")}
@@ -277,6 +300,7 @@ func TestLambdaTestLinkExecutor_IssuesForTheAttemptAndSendsOnTheSocket(t *testin
 	var sent testlink.Request
 	require.NoError(t, json.Unmarshal(n.sent[0], &sent))
 	require.Equal(t, testScope.FunctionARN+":$LATEST", sent.FunctionARN, "the runtime still sees the invoked ARN")
+	require.Equal(t, testImageDigest, sent.ImageDigest, "the daemon checks the gateway's pinned image")
 	require.NotContains(t, string(n.sent[0]), testSecret, "the credential travels beside the request, not in it")
 	require.Equal(t, testSecret, n.creds[0].SecretAccessKey)
 
@@ -399,3 +423,161 @@ func TestLambdaTestLinkExecutor_SilentDaemonIsUnknownAndRevoked(t *testing.T) {
 	require.Less(t, time.Since(start), 10*time.Second)
 	require.Len(t, sessions.revoked, 1)
 }
+
+// withdrawFault fails Withdraw before or after the owner writes it.
+type withdrawFault struct {
+	*execution.Store
+	afterWrite bool
+}
+
+func (w withdrawFault) Withdraw(ctx context.Context, g execution.Grant) error {
+	if !w.afterWrite {
+		return errors.New("owner unavailable")
+	}
+	if err := w.Store.Withdraw(ctx, g); err != nil {
+		return err
+	}
+	return execution.ErrOutcomeUnknown
+}
+
+// A daemon refusal with a failed Withdraw leaves the grant startable: a
+// replayed Start succeeds. A lost Withdraw acknowledgement leaves it fenced.
+func TestLambdaTestLinkExecutor_RefusalWithdrawFaults(t *testing.T) {
+	for name, tc := range map[string]struct {
+		afterWrite bool
+		state      string
+		replay     bool
+	}{
+		"fails before write": {state: execution.StateGranted, replay: true},
+		"ack lost":           {afterWrite: true, state: execution.StateWithdrawn},
+	} {
+		t.Run(name, func(t *testing.T) {
+			store := grantStore(t)
+			n := &fakeControl{max: 1 << 20, reply: &testlink.Reply{Outcome: testlink.OutcomeRefused, Error: "image differs"}}
+			e := executor(n, nil, store)
+			e.Grants = func(context.Context) (grantIssuer, error) { return withdrawFault{store, tc.afterWrite}, nil }
+			_, err := e.InvokeSync(context.Background(), invocation(nil))
+			require.Error(t, err)
+			require.Len(t, n.sent, 1)
+			require.Equal(t, tc.state, grantState(t, store))
+
+			var sent testlink.Request
+			require.NoError(t, json.Unmarshal(n.sent[0], &sent))
+			err = store.Start(context.Background(), sent.Grant, sent.Identity())
+			if tc.replay {
+				require.NoError(t, err, "an unwithdrawn grant can still start")
+			} else {
+				require.Error(t, err, "a withdrawn grant cannot start")
+			}
+		})
+	}
+}
+
+// The digest field counts toward the request bound: the exact encoded size
+// is sent and one byte less is refused unsent.
+func TestLambdaTestLinkExecutor_RequestBoundIncludesImageDigest(t *testing.T) {
+	probe := &fakeControl{max: 1 << 20, reply: &testlink.Reply{Outcome: testlink.OutcomeDone}}
+	_, err := executor(probe, nil, grantStore(t)).InvokeSync(context.Background(), invocation(make([]byte, 1000)))
+	require.NoError(t, err)
+	require.Len(t, probe.sent, 1)
+	require.Contains(t, string(probe.sent[0]), `"image_digest":"`+testImageDigest+`"`)
+	size := int64(len(probe.sent[0]))
+
+	at := &fakeControl{max: size, reply: &testlink.Reply{Outcome: testlink.OutcomeDone}}
+	_, err = executor(at, nil, grantStore(t)).InvokeSync(context.Background(), invocation(make([]byte, 1000)))
+	require.NoError(t, err)
+	require.Len(t, at.sent, 1)
+	require.Equal(t, size, int64(len(at.sent[0])), "the generated IDs are fixed-length, so the at-limit request is the probed size")
+
+	under := &fakeControl{max: size - 1}
+	_, err = executor(under, nil, grantStore(t)).InvokeSync(context.Background(), invocation(make([]byte, 1000)))
+	var refusal *awsapi.RefusalError
+	require.ErrorAs(t, err, &refusal)
+	require.Empty(t, under.sent)
+}
+
+// realCodeStore signs download links with the production code store, using a
+// session of the shape STS issues and a representative IPv6 off-host endpoint.
+func realCodeStore(t *testing.T) *codestore.Store {
+	t.Helper()
+	store, err := codestore.Open(codestore.OpenConfig{
+		Region: "ap-southeast-2", Endpoint: "http://127.0.0.1:1", SigningRegion: "ap-southeast-2",
+		PresignEndpoint: "https://[2001:db8:ffff:ffff:ffff:ffff:ffff:ffff]:65535",
+		Assume: func() (*sts.AssumeRoleOutput, error) {
+			return &sts.AssumeRoleOutput{Credentials: &sts.Credentials{
+				AccessKeyId: aws.String("ASIA" + strings.Repeat("K", 16)), SecretAccessKey: aws.String(strings.Repeat("s", 40)),
+				SessionToken: aws.String(strings.Repeat("t", 44)), Expiration: aws.Time(time.Now().Add(time.Hour)),
+			}}, nil
+		},
+	})
+	require.NoError(t, err)
+	return store
+}
+
+// longestInvocation carries the longest function name and qualifier Invoke
+// accepts, full-length generated IDs, and the given payload and handler.
+func longestInvocation(payload []byte, handler string) awsapi.SyncInvocation {
+	in := invocation(payload)
+	in.Qualifier = strings.Repeat("q", 128)
+	in.Function.FunctionName = strings.Repeat("n", 64)
+	in.Function.FunctionID = strings.Repeat("f", 32)
+	in.Function.Code.RevisionID = strings.Repeat("r", 32)
+	in.Function.Handler = handler
+	return in
+}
+
+func TestLambdaTestLinkExecutor_SenderBoundsAtFrameLimit(t *testing.T) {
+	codes := realCodeStore(t)
+	run := func(in awsapi.SyncInvocation) (*fakeControl, *fakeSessions, *execution.Store, error) {
+		n := &fakeControl{max: testlink.MaxFrameBytes, reply: &testlink.Reply{Outcome: testlink.OutcomeDone}}
+		sessions, store := &fakeSessions{}, grantStore(t)
+		e := executorWith(n, sessions, nil, store)
+		e.Codes = func(context.Context) (codePresigner, error) { return codes, nil }
+		_, err := e.InvokeSync(context.Background(), in)
+		return n, sessions, store, err
+	}
+	refusedUnsent := func(n *fakeControl, sessions *fakeSessions, store *execution.Store, err error) {
+		t.Helper()
+		var refusal *awsapi.RefusalError
+		require.ErrorAs(t, err, &refusal)
+		require.Empty(t, n.sent)
+		require.Len(t, sessions.revoked, 1, "the issued session is revoked")
+		require.Equal(t, execution.StateWithdrawn, grantState(t, store))
+	}
+
+	full, _, _, err := run(longestInvocation(make([]byte, testlink.MaxPayloadBytes), "bootstrap.h"))
+	require.NoError(t, err)
+	require.Len(t, full.sent, 1)
+	sent := string(full.sent[0])
+	// Asserted without printing, since the frame holds a signed URL.
+	require.True(t, strings.Contains(sent, `"code_url":"https://[2001:db8:ffff:ffff:ffff:ffff:ffff:ffff]:65535/`), "the off-host endpoint is signed")
+	require.True(t, strings.Contains(sent, "X-Amz-Signature="), "the code URL is a real presigned link")
+	headroom := testlink.MaxFrameBytes - int64(len(full.sent[0]))
+	t.Logf("full-size request: %d of %d frame bytes, %d bytes of headroom", len(full.sent[0]), testlink.MaxFrameBytes, headroom)
+
+	// The generated IDs and signed URL are fixed-length, so a longer handler
+	// moves the frame by exactly its extra bytes.
+	atLimit := "bootstrap.h" + strings.Repeat("h", int(headroom))
+	at, _, _, err := run(longestInvocation(make([]byte, testlink.MaxPayloadBytes), atLimit))
+	require.NoError(t, err)
+	require.Len(t, at.sent, 1)
+	require.Equal(t, testlink.MaxFrameBytes, int64(len(at.sent[0])))
+
+	refusedUnsent(run(longestInvocation(make([]byte, testlink.MaxPayloadBytes), atLimit+"h")))
+
+	// Refused before encoding: base64 alone would allocate 8 MiB, so a run
+	// that allocates under the payload size never encoded it.
+	over := longestInvocation(make([]byte, testlink.MaxPayloadBytes+1), "bootstrap.h")
+	n := &fakeControl{max: testlink.MaxFrameBytes}
+	sessions, store := &fakeSessions{}, grantStore(t)
+	e := executorWith(n, sessions, nil, store)
+	e.Codes = func(context.Context) (codePresigner, error) { return codes, nil }
+	var before, after runtime.MemStats
+	runtime.ReadMemStats(&before)
+	_, err = e.InvokeSync(context.Background(), over)
+	runtime.ReadMemStats(&after)
+	allocated := after.TotalAlloc - before.TotalAlloc
+	t.Logf("6 MiB + 1 refusal: %d bytes in %d allocations, process-wide", allocated, after.Mallocs-before.Mallocs)
+	require.Less(t, allocated, uint64(testlink.MaxPayloadBytes))
+	refusedUnsent(n, sessions, store, err)
+}

spinifex/runtime/roles/awsgw/lambda_testlink_pin_test.go

@@ -0,0 +1,35 @@
+//go:build lambdatestlink
+
+//test:in-package — reads the unexported ldflags image pin
+package awsgw
+
+import (
+	"os"
+	"testing"
+
+	"github.com/mulgadc/spinifex/spinifex/runtime/compute/lambda/testlink"
+)
+
+// TestLambdaImageDigestPin runs from make build-lambda-testlink with the
+// daemon's ldflags: the gateway carries the same image digest as the daemon.
+func TestLambdaImageDigestPin(t *testing.T) {
+	want := os.Getenv("LAMBDA_IMAGE_DIGEST")
+	if want == "" {
+		t.Skip("run by make build-lambda-testlink with a pinned image digest")
+	}
+	if lambdaImageDigest != want || !testlink.ValidDigest(lambdaImageDigest) {
+		t.Fatalf("pinned image digest %q, want %q", lambdaImageDigest, want)
+	}
+}
+
+// The gateway dials the production control name, so a mixed-version daemon on
+// the previous name is never reached.
+func TestLambdaTestLinkControlUsesTheProductionName(t *testing.T) {
+	c := testLinkControl(4242)
+	if c.Name != testlink.ControlName || c.DaemonUID != 4242 {
+		t.Fatalf("control client %q uid %d, want %q uid 4242", c.Name, c.DaemonUID, testlink.ControlName)
+	}
+	if c.RequestLimit != testlink.MaxFrameBytes || c.ReplyLimit != testlink.MaxFrameBytes {
+		t.Fatalf("control limits %d/%d, want the declared frame bound %d", c.RequestLimit, c.ReplyLimit, testlink.MaxFrameBytes)
+	}
+}

spinifex/runtime/roles/awsgw/lambda_testlink_seam_env_test.go

@@ -161,7 +161,7 @@ func startSeamDaemon(t *testing.T, d *testlink.Daemon, guest guestFunc) *seamDae
 	ln, err := testlink.ListenControl(name)
 	require.NoError(t, err)
 	s := &seamDaemon{entered: make(chan testlink.Request, 4), done: make(chan testlink.Reply, 4)}
-	srv := &testlink.ControlServer{GatewayUID: os.Getuid(), RequestLimit: 1 << 20, ReplyLimit: 1 << 20, IOTimeout: 5 * time.Second,
+	srv := &testlink.ControlServer{GatewayUID: os.Getuid(), RequestLimit: testlink.MaxFrameBytes, ReplyLimit: testlink.MaxFrameBytes, IOTimeout: 5 * time.Second,
 		Handle: func(ctx context.Context, req []byte, cred testlink.Credential) testlink.Reply {
 			var parsed testlink.Request
 			_ = json.Unmarshal(req, &parsed)
@@ -189,7 +189,7 @@ func startSeamDaemon(t *testing.T, d *testlink.Daemon, guest guestFunc) *seamDae
 			t.Errorf("control server did not stop within %s", seamWait)
 		}
 	})
-	s.client = &testlink.ControlClient{Name: name, DaemonUID: os.Getuid(), RequestLimit: 1 << 20, ReplyLimit: 1 << 20, IOTimeout: 5 * time.Second}
+	s.client = &testlink.ControlClient{Name: name, DaemonUID: os.Getuid(), RequestLimit: testlink.MaxFrameBytes, ReplyLimit: testlink.MaxFrameBytes, IOTimeout: 5 * time.Second}
 	return s
 }
 

spinifex/runtime/roles/awsgw/lambda_testlink_seam_payload_test.go

@@ -0,0 +1,113 @@
+//go:build lambdatestlink && lambdatestlinkseam
+
+//test:in-package — drives the unexported test-link executor and seam fixtures
+package awsgw
+
+import (
+	"bytes"
+	"fmt"
+	"io"
+	"net/http"
+	"net/http/httptest"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/require"
+
+	"github.com/mulgadc/spinifex/spinifex/agents/lambda/runtimeapi"
+	lambdaawsapi "github.com/mulgadc/spinifex/spinifex/domains/lambda/awsapi"
+	"github.com/mulgadc/spinifex/spinifex/runtime/compute/lambda/testlink"
+	"github.com/mulgadc/spinifex/internal/testkit/gwfix"
+)
+
+// jsonString returns a JSON string literal of exactly n bytes.
+func jsonString(n int) []byte {
+	return []byte(`"` + strings.Repeat("a", n-2) + `"`)
+}
+
+// invokeComposedBody is invokeComposedIn with a signed Invoke carrying body.
+func invokeComposedBody(t *testing.T, body []byte, codeURL string, client *http.Client, guest guestFunc) seamRun {
+	t.Helper()
+	env := newSeamEnv(t)
+	exec, daemon, grants, presigns := composedExecutor(t, env, codeURL, client, guest)
+	reg := lambdaawsapi.NewRegistration(lambdaawsapi.Deps{Meta: seamFunction{timeout: 30, role: env.roleARN}, Exec: exec})
+	req := httptest.NewRequest(http.MethodPost, "/2015-03-31/functions/f/invocations", bytes.NewReader(body))
+	req.Host = "localhost:9999"
+	gwfix.SignRequestAs(t, req, body, gwfix.AccessKey, gwfix.SecretKey, lambdaawsapi.ServiceName, time.Now().UTC())
+	w := httptest.NewRecorder()
+	defaultGateway(t, lambdaTestRoutes(reg)...).ServeHTTP(w, req)
+	resp := w.Result()
+	out, err := io.ReadAll(resp.Body)
+	require.NoError(t, err)
+	reqs, replies := daemon.recorded()
+	return seamRun{env: env, resp: resp, body: out, requests: reqs, replies: replies, presigns: presigns.Load(), grants: grants.issued.Load()}
+}
+
+// echoGuest is a function runtime that returns its event unchanged and
+// reports its own result on done.
+func echoGuest(done chan<- error) func(base string) {
+	client := &http.Client{Timeout: 30 * time.Second}
+	return func(base string) {
+		done <- func() error {
+			resp, err := client.Get(base + "invocation/next")
+			if err != nil {
+				return err
+			}
+			event, err := io.ReadAll(resp.Body)
+			_ = resp.Body.Close()
+			if err != nil {
+				return err
+			}
+			if resp.StatusCode != http.StatusOK {
+				return fmt.Errorf("next: status %d", resp.StatusCode)
+			}
+			id := resp.Header.Get("Lambda-Runtime-Aws-Request-Id")
+			r, err := client.Post(base+"invocation/"+id+"/response", "application/json", bytes.NewReader(event))
+			if err != nil {
+				return err
+			}
+			_, _ = io.Copy(io.Discard, r.Body)
+			_ = r.Body.Close()
+			if r.StatusCode != http.StatusAccepted {
+				return fmt.Errorf("response: status %d", r.StatusCode)
+			}
+			return nil
+		}()
+	}
+}
+
+// A 6 MiB event crosses the real gateway, control socket and guest link,
+// and the 6 MiB response returns byte-identical. The seam daemon uses
+// HandleWithGuest, so the busy, image and fit checks are tested elsewhere.
+func TestLambdaTestLinkSeam_SixMiBRequestAndResponseRoundTrip(t *testing.T) {
+	require.Equal(t, testlink.MaxPayloadBytes, runtimeapi.MaxResponseBytes)
+	event := jsonString(testlink.MaxPayloadBytes)
+	srv := codeServer(t)
+	guestDone := make(chan error, 1)
+	run := invokeComposedBody(t, event, srv.URL+"/code", srv.Client(), agentGuest(0, echoGuest(guestDone)))
+	require.Equal(t, http.StatusOK, run.resp.StatusCode, string(run.body[:min(len(run.body), 300)]))
+	require.Empty(t, run.resp.Header.Get("X-Amz-Function-Error"))
+	require.Len(t, run.body, testlink.MaxPayloadBytes)
+	require.True(t, bytes.Equal(event, run.body), "response differs from the event")
+	require.Len(t, run.replies, 1)
+	select {
+	case err := <-guestDone:
+		require.NoError(t, err, "the echo runtime completed")
+	case <-time.After(30 * time.Second):
+		t.Fatal("the echo runtime did not complete")
+	}
+}
+
+// A 6 MiB + 1 event is refused with 413 before anything is presigned,
+// granted or sent to the daemon.
+func TestLambdaTestLinkSeam_OverSixMiBRequestRefusedBeforeDispatch(t *testing.T) {
+	srv := codeServer(t)
+	run := invokeComposedBody(t, jsonString(testlink.MaxPayloadBytes+1), srv.URL+"/code", srv.Client(), agentGuest(0, echoGuest(make(chan error, 1))))
+	require.Equal(t, http.StatusRequestEntityTooLarge, run.resp.StatusCode)
+	require.Contains(t, run.resp.Header.Get("X-Amzn-Errortype"), "RequestEntityTooLargeException")
+	require.Zero(t, run.presigns)
+	require.Zero(t, run.grants)
+	require.Empty(t, run.requests)
+	require.Zero(t, run.env.sessionCount(t))
+}

spinifex/runtime/roles/awsgw/lambda_testlink_seam_test.go

@@ -212,7 +212,7 @@ func composedExecutor(t *testing.T, env *seamEnv, codeURL string, client *http.C
 	grants := &countedGrants{grantIssuer: env.grants}
 	presigns := &atomic.Int32{}
 	daemon := startSeamDaemon(t, &testlink.Daemon{Node: "node1", Grants: env.grants, HTTP: client}, guest)
-	exec := &testLinkExecutor{Control: daemon.client, RequestLimit: 1 << 20, Sessions: env.sts, Node: "node1",
+	exec := &testLinkExecutor{Control: daemon.client, RequestLimit: testlink.MaxFrameBytes, Sessions: env.sts, Node: "node1", ImageDigest: testImageDigest,
 		Codes:  func(context.Context) (codePresigner, error) { return urlPresigner{codeURL, presigns}, nil },
 		Grants: func(context.Context) (grantIssuer, error) { return grants, nil },
 	}