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

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

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

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

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

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

build/microvm/lambda-runtime/init.sh

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

cmd/lambda-agent/connect.go

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

cmd/lambda-agent/connect_configdisk_test.go

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

contracts/ec2/v1/commands.go

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

contracts/ec2/v1/commands_test.go

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

contracts/lambda/v1/README.md

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

spinifex/agents/lambda/hostlink/configdisk.go

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

spinifex/agents/lambda/hostlink/configdisk_test.go

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

spinifex/domains/ec2/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/lambda/function/activate.go

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

spinifex/domains/lambda/function/activate_test.go

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

spinifex/domains/lambda/function/meta.go

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

spinifex/domains/rds/instance/delete.go

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

spinifex/domains/rds/instance/delete_test.go

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

spinifex/domains/rds/instance/lifecycle_test.go

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

spinifex/domains/rds/instance/reconciler.go

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

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

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

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

@@ -462,3 +462,45 @@ func TestControl_MixedProtocolNamesNeverConnect(t *testing.T) {
 		})
 	}
 }
+
+func TestControl_AcceptStoppedRunsBeforeTheDrain(t *testing.T) {
+	name := controlName(t)
+	ln, err := ListenControl(name)
+	require.NoError(t, err)
+	inHandle, release := make(chan struct{}), make(chan struct{})
+	stopped := make(chan struct{})
+	srv := &ControlServer{GatewayUID: os.Getuid(), RequestLimit: testRequestLimit, ReplyLimit: 4096, IOTimeout: 5 * time.Second,
+		Handle: func(context.Context, []byte, Credential) Reply {
+			close(inHandle)
+			<-release
+			return Reply{Outcome: OutcomeDone}
+		},
+		AcceptStopped: func() { close(stopped) }}
+	done := make(chan error, 1)
+	go func() { done <- srv.Serve(context.Background(), ln) }()
+	go func() { _, _ = client(name, os.Getuid()).Exchange(context.Background(), []byte("{}"), testCred, soon()) }()
+	select {
+	case <-inHandle:
+	case <-time.After(5 * time.Second):
+		t.Fatal("the request never reached Handle")
+	}
+
+	require.NoError(t, ln.Close(), "the listener fails under an outstanding invocation")
+	select {
+	case <-stopped:
+	case <-time.After(5 * time.Second):
+		t.Fatal("AcceptStopped did not run while the invocation was outstanding")
+	}
+	select {
+	case err := <-done:
+		t.Fatalf("Serve returned before its connection finished: %v", err)
+	default:
+	}
+	close(release)
+	select {
+	case err := <-done:
+		require.Error(t, err, "a listener failure is reported")
+	case <-time.After(5 * time.Second):
+		t.Fatal("Serve did not return after its connection finished")
+	}
+}

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

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

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

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

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

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

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

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

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

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

spinifex/runtime/daemon/daemon_handlers.go

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

spinifex/runtime/daemon/daemon_handlers_instance.go

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

spinifex/runtime/daemon/daemon_handlers_owner_routing_test.go

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

spinifex/runtime/daemon/daemon_lambda_testlink.go

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

spinifex/runtime/daemon/nats_classification_test.go

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