test(lambda): hold a daemon invocation while serving ends on listener failure

The wake test now waits for the start pass before storing code, so only
the code-stored wake can activate the second function.

contracts/ec2/v1/commands.go

@@ -12,11 +12,6 @@ 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.
@@ -28,9 +23,7 @@ func InstanceCommandSubject(instanceID string) string {
 }
 
 // EC2InstanceCommand is the NATS wire format for EC2 instance commands
-// (stop, terminate, start, reboot, attach/detach-volume, attach/detach-eni,
-// IAM profile, spot lineage, tags, monitoring, metadata options and
-// termination protection).
+// (stop, terminate, start, attach/detach-volume, attach/detach-eni).
 type EC2InstanceCommand struct {
 	ID                        string                     `json:"id"`
 	Attributes                EC2CommandAttributes       `json:"attributes"`
@@ -42,9 +35,6 @@ 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.
@@ -62,9 +52,6 @@ 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.
@@ -116,23 +103,6 @@ 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,9 +24,6 @@ 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},
@@ -36,15 +33,6 @@ 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)
@@ -64,9 +52,7 @@ func TestEC2InstanceCommandJSONContract(t *testing.T) {
 			"set_spot_lineage": true,
 			"set_instance_tags": true,
 			"remove_instance_tags": true,
-			"set_instance_monitoring": true,
-			"modify_instance_metadata_options": true,
-			"set_disable_api_termination": true
+			"set_instance_monitoring": true
 		},
 		"attach_volume_data": {"volume_id": "vol-1", "device": "/dev/sdf"},
 		"detach_volume_data": {"volume_id": "vol-2", "device": "/dev/sdg", "force": true},
@@ -75,62 +61,8 @@ 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_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)
+	}`, string(data))
 }
 
 func TestInstanceCommandSubjects(t *testing.T) {

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.
 

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

@@ -11,7 +11,6 @@ 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"
 )
@@ -66,20 +65,6 @@ 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,13 +3,10 @@ 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"
@@ -229,53 +226,3 @@ 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,16 +22,5 @@ func ModifyInstanceMetadataOptions(ctx context.Context, input *ec2.ModifyInstanc
 	if !strings.HasPrefix(*input.InstanceId, "i-") {
 		return nil, errors.New(awserrors.ErrorInvalidInstanceIDMalformed)
 	}
-	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)
+	return natsmsg.NATSRequest[ec2.ModifyInstanceMetadataOptionsOutput](ctx, natsConn, "ec2.ModifyInstanceMetadataOptions", input, 30*time.Second, accountID)
 }

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

@@ -2,16 +2,11 @@ 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"
 )
@@ -35,179 +30,3 @@ 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,46 +99,3 @@ 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,12 +66,6 @@ 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.
@@ -80,29 +74,29 @@ func (s *Service) DeleteDBInstance(ctx context.Context, input *rds.DeleteDBInsta
 		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)
+	// 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
 		}
-		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)
 	}
-	s.RecordEvent(ctx, accountID, EventSourceTypeDBInstance, id, "DB instance deleted.", EventCategoryDeletion)
 
-	// 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 {
+	if err := s.teardownDBInstance(ctx, 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
@@ -208,11 +202,6 @@ 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,11 +1,9 @@
 package instance
 
 import (
-	"context"
 	"errors"
 	"fmt"
 	"testing"
-	"time"
 
 	"github.com/aws/aws-sdk-go/aws"
 	"github.com/aws/aws-sdk-go/service/rds"
@@ -178,12 +176,13 @@ func TestDeleteDBInstance_ReservesTheFinalSnapshotBeforeCuttingIt(t *testing.T)
 }
 
 // A dead worker can leave the EC2 snapshot cut while its RDS record still says
-// 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) {
+// 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) {
 	t.Parallel()
 	h := newLifecycleHarness(t, false)
-	rec := abandonedDelete(availableRecord())
+	rec := availableRecord()
+	rec.Status = StatusDeleting
 	rec.FinalSnapshotIdentifier = "orders-db-final"
 	seedInstance(t, h.svc, rec)
 
@@ -202,9 +201,13 @@ func TestReconciler_AdoptsAnInterruptedFinalSnapshot(t *testing.T) {
 		},
 	}
 
-	require.NoError(t, onePass(t, NewReconciler(h.svc, "node-a")))
+	_, err = h.svc.DeleteDBInstance(t.Context(), &rds.DeleteDBInstanceInput{
+		DBInstanceIdentifier:      aws.String(testDBID),
+		FinalDBSnapshotIdentifier: aws.String(rec.FinalSnapshotIdentifier),
+	}, testAccountID)
+	require.NoError(t, err)
 
-	assert.Empty(t, h.snaps.created, "the resumed teardown must adopt the cut snapshot, not create another")
+	assert.Empty(t, h.snaps.created, "the retry 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)
@@ -322,18 +325,26 @@ 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 TestReconciler_IsIdempotentAcrossARetriedTeardown(t *testing.T) {
+func TestDeleteDBInstance_IsIdempotentAcrossARetriedTeardown(t *testing.T) {
 	t.Parallel()
 	h := newLifecycleHarness(t, false)
-	rec := abandonedDelete(availableRecord())
+	rec := availableRecord()
+	// The shape a delete leaves behind when its caller died mid-teardown.
+	rec.Status = StatusDeleting
 	rec.FinalSnapshotIdentifier = "orders-db-final"
 	seedInstance(t, h.svc, rec)
-	reconciler := NewReconciler(h.svc, "node-a")
-	require.NoError(t, onePass(t, reconciler))
+
+	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)
 
 	// The record left behind by the interrupted attempt, replayed.
 	seedInstance(t, h.svc, rec)
-	require.NoError(t, onePass(t, reconciler))
+	_, err = h.svc.DeleteDBInstance(t.Context(), input, testAccountID)
+	require.NoError(t, err)
 
 	assert.Len(t, h.snaps.created, 1, "a replayed teardown must not take a second final snapshot")
 	assert.False(t, h.recordExists(t, testDBID))
@@ -344,7 +355,9 @@ func TestReconciler_IsIdempotentAcrossARetriedTeardown(t *testing.T) {
 func TestReconciler_ResumesAnInterruptedDelete(t *testing.T) {
 	t.Parallel()
 	h := newLifecycleHarness(t, false)
-	seedInstance(t, h.svc, abandonedDelete(availableRecord()))
+	rec := availableRecord()
+	rec.Status = StatusDeleting
+	seedInstance(t, h.svc, rec)
 
 	reconciler := NewReconciler(h.svc, "node-a")
 	require.NoError(t, onePass(t, reconciler))
@@ -353,87 +366,6 @@ 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,9 +117,7 @@ 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
-	// The context deadline the last CreateSnapshot ran under.
-	deadline      time.Time
+	created       []*ec2.CreateSnapshotInput
 	holding       []string
 	deleted       []string
 	createErr     error
@@ -134,8 +132,7 @@ type fakeSnapshots struct {
 
 var _ snapshotProvider = (*fakeSnapshots)(nil)
 
-func (f *fakeSnapshots) CreateSnapshot(ctx context.Context, input *ec2.CreateSnapshotInput, _ string) (*ec2.Snapshot, error) {
-	f.deadline, _ = ctx.Deadline()
+func (f *fakeSnapshots) CreateSnapshot(_ context.Context, input *ec2.CreateSnapshotInput, _ string) (*ec2.Snapshot, error) {
 	if f.beforeCreate != nil {
 		f.beforeCreate()
 	}

spinifex/domains/rds/instance/reconciler.go

@@ -610,11 +610,6 @@ 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
@@ -800,12 +795,6 @@ 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

@@ -91,9 +91,13 @@ type Daemon struct {
 	Initializer       string
 	InitializerSHA256 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 +112,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 +129,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 +313,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 {

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,18 +81,11 @@ 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")
 	}
@@ -253,10 +246,6 @@ 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)
@@ -325,10 +314,6 @@ 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,76 +360,3 @@ 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

@@ -1,323 +0,0 @@
-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

@@ -11,6 +11,7 @@ import (
 	"net/http"
 	"os"
 	"path/filepath"
+	"time"
 
 	"github.com/mulgadc/spinifex/spinifex/domains/lambda/execution"
 	"github.com/mulgadc/spinifex/spinifex/domains/network/host"
@@ -88,8 +89,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,8 +137,6 @@ 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 }},