fix(rds): run one teardown of a DB instance delete at a time (#1647)

* fix(rds): run one teardown of a DB instance delete at a time

The handler that accepted a delete now owns its teardown for its request
budget, a repeat of an accepted delete answers deleting without a second
teardown, and the reconciler resumes only a delete older than that budget.
Two teardowns racing let the handler see the VM mid-shutdown as gone and
answer 500 on the still-attached data volume.

* fix(rds): end the delete handler's teardown exactly when its claim does

The handler's deadline now derives from the persisted transition stamp the
reconciler measures from, and a teardown past that deadline starts no
terminate, since the graceful engine stop swallows the cancellation.

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)

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