test: run the admin package's timeout-bound tests in parallel (#1564)

Six account-delete tests each wait one 5s teardown drain poll, and the
network-test and storage-status tests wait out fixed 5s and 2s timeouts.
Each uses its own embedded NATS, so they run concurrently: the package
drops from about 37s to 8s.

.github/workflows/codex-review.yml

@@ -0,0 +1,21 @@
+# Codex review gate: calls bluebottle's reusable workflows from main, so a PR cannot change its own review.
+name: codex-review
+
+on:
+  pull_request_target:
+    branches: [dev]
+    types: [opened, synchronize, reopened, ready_for_review, edited]
+  issue_comment:
+    types: [created]
+
+permissions: {}
+
+jobs:
+  review:
+    if: github.event_name == 'pull_request_target'
+    uses: mulgadc/bluebottle/.github/workflows/codex-pr-review.yml@main
+    secrets: inherit
+  waive:
+    if: github.event_name == 'issue_comment' && github.event.issue.pull_request
+    uses: mulgadc/bluebottle/.github/workflows/codex-waive.yml@main
+    secrets: inherit

.github/workflows/e2e-baremetal.yml

@@ -71,9 +71,16 @@ jobs:
       ARTIFACT_DIR: /tmp/spinifex-e2e-nightly-cell-${{ inputs.cell || '20' }}-${{ github.run_id }}
       SPINIFEX_OTEL_ATTRS: ${{ inputs.otel-attrs || format('ci.run_id={0},ci.workflow={1},ci.cell={2},ci.matrix=baremetal/pxe/real-hw', github.run_id, github.workflow, inputs.cell || '20') }}
     steps:
+      # A cancelled job's root steps outlive it, since the runner cannot reach
+      # them once it kills sudo, and would keep writing here and to the cache.
+      # A rootfs killed before bin/bash exists is one the build refuses to reuse.
       - name: Clean Workspace
         run: |
+          sudo systemctl stop spinifex-bm-pxe.scope spinifex-bm-e2e.scope 2>/dev/null || true
           sudo umount -R -l "${BM_ISO_CACHE}/rootfs" 2>/dev/null || true
+          if [ -d "${BM_ISO_CACHE}/rootfs" ] && [ ! -x "${BM_ISO_CACHE}/rootfs/bin/bash" ]; then
+            sudo rm -rf --one-file-system "${BM_ISO_CACHE}/rootfs"
+          fi
           if [ ! -c /dev/null ]; then sudo rm -f /dev/null; sudo mknod -m 666 /dev/null c 1 3; fi
           sudo rm -rf --one-file-system "$GITHUB_WORKSPACE"/*
 
@@ -113,10 +120,15 @@ jobs:
             "$GITHUB_WORKSPACE" \
             "${RUNNER_TEMP}/spinifex-tests.tar" \
             "${RUNNER_TEMP}/bm-bins"
+          # Built here, as the runner user, so the PXE build finds it current.
+          make -C "$GITHUB_WORKSPACE" build-microvm-image
 
       - name: Build PXE Artifacts
         run: |
-          sudo make -C mulga/scripts/iso-builder \
+          # -o build-spinifex: the binaries come from Build Artifacts, and a
+          # rebuild here would write root-owned files into the workspace.
+          sudo systemd-run --scope --collect --quiet --unit=spinifex-bm-pxe -- \
+            make -C mulga/scripts/iso-builder -o build-spinifex \
             build-rootfs build-squashfs build-initramfs \
             CACHE="${BM_ISO_CACHE}" \
             SPINIFEX_BIN="${RUNNER_TEMP}/bm-bins/spx" \
@@ -124,6 +136,10 @@ jobs:
             SPINIFEX_SRC="${GITHUB_WORKSPACE}" \
             VIPERBLOCK_SRC="${RUNNER_TEMP}/bm-bins/viperblock" \
             VERSION="$(cd "$GITHUB_WORKSPACE" && git describe --tags --always --exclude '*-dev' 2>/dev/null || echo dev)"
+          if root_owned="$(find "$GITHUB_WORKSPACE" -user root -print -quit)" && [ -n "$root_owned" ]; then
+            echo "::error::the PXE build wrote root-owned files into the workspace, e.g. $root_owned"
+            exit 1
+          fi
 
       - name: Write ePDU credentials
         run: |
@@ -195,13 +211,18 @@ jobs:
           done
           export BM_TEST_TIMEOUT
           echo "suites=${BM_SUITES} test-timeout=${BM_TEST_TIMEOUT}"
-          sudo -E mulga/scripts/baremetal/run-baremetal-e2e.sh \
+          sudo -E systemd-run --scope --collect --quiet --unit=spinifex-bm-e2e -- \
+            mulga/scripts/baremetal/run-baremetal-e2e.sh \
             --vmlinuz   "${BM_ISO_CACHE}/vmlinuz" \
             --initrd    "${BM_ISO_CACHE}/initrd.img" \
             --squashfs  "${BM_ISO_CACHE}/spinifex.squashfs" \
             --tests-tar "${RUNNER_TEMP}/spinifex-tests.tar" \
             2>&1 | tee "${ARTIFACT_DIR}/test-baremetal.log"
 
+      - name: Stop this job's root steps
+        if: always()
+        run: sudo systemctl stop spinifex-bm-pxe.scope spinifex-bm-e2e.scope 2>/dev/null || true
+
       - name: Generate JUnit report
         if: always()
         run: |

Makefile

@@ -42,11 +42,15 @@ ifdef QUIET
   _COVQ  = 2>&1 | { grep -Ev '^\s*(ok|PASS|\?|=== RUN|--- PASS:)\s' | grep -v 'coverage: 0\.0%' || true; }; exit $${PIPESTATUS[0]}
   _RACEQ = 2>&1 | { grep -Ev '^\s*(ok|PASS|\?|=== RUN|--- PASS:)\s' || true; }; exit $${PIPESTATUS[0]}
   _SECQ  = >
+  _HDR   = :
+  _RUNSH = sh "$$t" >"$$log" 2>&1 || { echo "-- $$t"; cat "$$log"; exit 1; }
 else
   _Q     =
   _COVQ  =
   _RACEQ =
   _SECQ  = 2>&1 | tee
+  _HDR   = echo -e
+  _RUNSH = echo "-- $$t"; sh "$$t" || exit 1
 endif
 
 build: go_build build-installer build-lb-agent generate-aws-model-coverage
@@ -240,27 +244,24 @@ preflight:
 NODE_SCRIPTS := scripts/setup.sh scripts/setup-ovn.sh scripts/install-node.sh scripts/smoke-test.sh scripts/node-reset.sh scripts/uninstall-spx.sh scripts/lib/distro.sh scripts/lib/stage-sqs-store-microvm.sh
 
 test-build-scripts:
-	@echo -e "\n....Running build/scripts/**/*_test.sh...."
-	@for t in $$(find build/scripts -name '*_test.sh' | sort); do \
-		echo "-- $$t"; \
-		sh "$$t" || exit 1; \
+	@$(_HDR) "\n....Running build/scripts/**/*_test.sh...."
+	@log=$$(mktemp); trap 'rm -f "$$log"' EXIT; for t in $$(find build/scripts -name '*_test.sh' | sort); do \
+		$(_RUNSH); \
 	done
-	@echo -e "\n....Running scripts/*_test.sh, scripts/lib/*_test.sh and scripts/appliance/*_test.sh...."
-	@for t in $$( (find scripts -maxdepth 1 -name '*_test.sh'; find scripts/lib scripts/appliance -name '*_test.sh') | sort); do \
-		echo "-- $$t"; \
-		sh "$$t" || exit 1; \
+	@$(_HDR) "\n....Running scripts/*_test.sh, scripts/lib/*_test.sh and scripts/appliance/*_test.sh...."
+	@log=$$(mktemp); trap 'rm -f "$$log"' EXIT; for t in $$( (find scripts -maxdepth 1 -name '*_test.sh'; find scripts/lib scripts/appliance -name '*_test.sh') | sort); do \
+		$(_RUNSH); \
 	done
-	@echo -e "\n....Running .github/scripts/*_test.sh...."
-	@for t in $$(find .github/scripts -maxdepth 1 -name '*_test.sh' | sort); do \
-		echo "-- $$t"; \
-		sh "$$t" || exit 1; \
+	@$(_HDR) "\n....Running .github/scripts/*_test.sh...."
+	@log=$$(mktemp); trap 'rm -f "$$log"' EXIT; for t in $$(find .github/scripts -maxdepth 1 -name '*_test.sh' | sort); do \
+		$(_RUNSH); \
 	done
-	shellcheck -S warning .github/scripts/hypervisor-disk-preflight.sh .github/scripts/hypervisor-disk-preflight_test.sh \
+	$(_Q)shellcheck -S warning .github/scripts/hypervisor-disk-preflight.sh .github/scripts/hypervisor-disk-preflight_test.sh \
 		.github/scripts/wait-for-memory.sh .github/scripts/wait-for-memory_test.sh
-	@echo -e "\n....Running shellcheck over build/scripts/**/*.sh + the node scripts...."
-	shellcheck -S warning $$(find build/scripts -name '*.sh' | sort) $(NODE_SCRIPTS)
-	shellcheck -S warning -x scripts/build-oracle-appliance.sh $$(find scripts/appliance -name '*.sh' | sort)
-	@echo "  test-build-scripts ok"
+	@$(_HDR) "\n....Running shellcheck over build/scripts/**/*.sh + the node scripts...."
+	$(_Q)shellcheck -S warning $$(find build/scripts -name '*.sh' | sort) $(NODE_SCRIPTS)
+	$(_Q)shellcheck -S warning -x scripts/build-oracle-appliance.sh $$(find scripts/appliance -name '*.sh' | sort)
+	@$(_HDR) "  test-build-scripts ok"
 
 # E2E harness unit tests. Build-tagged `e2e` so they're skipped by the
 # default `go test ./spinifex/...`. Runs with mocked AWS clients — no

architecture/layering/allowlist_test.go

@@ -15,9 +15,6 @@ var allowlist = []allowance{
 	{From: "operator/accountlifecycle", To: "domains/ec2/awsapi/igw", Debt: "accountteardown (dependency)", Reason: "EC2 reapers call the awsapi action adapters; needs domain-owned cleanup capabilities"},
 	{From: "operator/accountlifecycle", To: "domains/ec2/awsapi/image", Debt: "accountteardown (dependency)", Reason: "EC2 reapers call the awsapi action adapters; needs domain-owned cleanup capabilities"},
 	{From: "operator/accountlifecycle", To: "domains/ec2/awsapi/instance", Debt: "accountteardown (dependency)", Reason: "EC2 reapers call the awsapi action adapters; needs domain-owned cleanup capabilities"},
-	{From: "operator/accountlifecycle", To: "domains/ec2/awsapi/key", Debt: "accountteardown (dependency)", Reason: "EC2 reapers call the awsapi action adapters; needs domain-owned cleanup capabilities"},
-	{From: "operator/accountlifecycle", To: "domains/ec2/awsapi/launchtemplate", Debt: "accountteardown (dependency)", Reason: "EC2 reapers call the awsapi action adapters; needs domain-owned cleanup capabilities"},
-	{From: "operator/accountlifecycle", To: "domains/ec2/awsapi/placementgroup", Debt: "accountteardown (dependency)", Reason: "EC2 reapers call the awsapi action adapters; needs domain-owned cleanup capabilities"},
 	{From: "operator/accountlifecycle", To: "domains/ec2/awsapi/routetable", Debt: "accountteardown (dependency)", Reason: "EC2 reapers call the awsapi action adapters; needs domain-owned cleanup capabilities"},
 	{From: "operator/accountlifecycle", To: "domains/ec2/awsapi/snapshot", Debt: "accountteardown (dependency)", Reason: "EC2 reapers call the awsapi action adapters; needs domain-owned cleanup capabilities"},
 	{From: "operator/accountlifecycle", To: "domains/ec2/awsapi/volume", Debt: "accountteardown (dependency)", Reason: "EC2 reapers call the awsapi action adapters; needs domain-owned cleanup capabilities"},

contracts/ec2/v1/README.md

@@ -23,7 +23,7 @@ subject shape are wire compatibility. Callers and receivers must use this
 package rather than rebuild those details. `commands_test.go` preserves the
 version-1 JSON shape and subject form.
 
-This package owns shared command payloads and the drain-volume response. It
+This package owns shared command payloads. It
 does not define every action's reply body, authorize the request, store EC2
 state or implement NATS transport. Those remain responsibilities of the EC2
 domain and its runtime composition.
@@ -37,6 +37,13 @@ contract or an approved compatibility plan; it is not a package refactor.
 Only the subject is owned here today.
 The request (`account_id`) and reply (`vpc_id`, `error`) bodies are still declared as anonymous structs on each side; moving them into this package is a contract change that needs its own characterization first.
 
+## Volume held elsewhere
+
+A start refused because another node holds one of the instance's volumes is an `IncorrectState` error payload on the `ec2.start`, `ec2.start.<node>` and `ec2.cmd.<instance-id>` replies, and the AWS body cannot say why.
+The owning node adds the `X-Volume-Held-Elsewhere: true` header to that reply, the `ec2.start` forwarder relays it, and `StartInstances` wraps `ErrVolumeHeldElsewhere` around the error it rebuilds, so a caller matches it with `errors.Is` rather than reading the message.
+The owning node sets the header only from the EBS provider's `volume_held_elsewhere` code, which every provider returns for this refusal; no reader matches the message text.
+`VolumeHeldElsewhereText` is the wording viperblockd puts in that refusal's message, not a signal.
+
 ## Platform inventory
 
 `PlatformInventorySubject` (`ec2.v1.PlatformInventory`) is a request/reply route on which a daemon in the `spinifex-workers` queue group reports one account's key pairs, placement groups, launch templates and their `ec2-tags` entries from the EC2 owner stores.

contracts/ec2/v1/commands.go

@@ -29,7 +29,6 @@ type EC2InstanceCommand struct {
 	Attributes                EC2CommandAttributes       `json:"attributes"`
 	AttachVolumeData          *AttachVolumeData          `json:"attach_volume_data,omitempty"`
 	DetachVolumeData          *DetachVolumeData          `json:"detach_volume_data,omitempty"`
-	DrainVolumeData           *DrainVolumeData           `json:"drain_volume_data,omitempty"`
 	AttachENIData             *AttachENIData             `json:"attach_eni_data,omitempty"`
 	DetachENIData             *DetachENIData             `json:"detach_eni_data,omitempty"`
 	IamProfileAssociationData *IamProfileAssociationData `json:"iam_profile_association_data,omitempty"`
@@ -45,7 +44,6 @@ type EC2CommandAttributes struct {
 	StartInstance               bool `json:"start_instance"`
 	AttachVolume                bool `json:"attach_volume"`
 	DetachVolume                bool `json:"detach_volume"`
-	DrainVolume                 bool `json:"drain_volume,omitempty"`
 	RebootInstance              bool `json:"reboot_instance"`
 	AttachENI                   bool `json:"attach_eni"`
 	DetachENI                   bool `json:"detach_eni"`
@@ -69,33 +67,6 @@ type DetachVolumeData struct {
 	Force    bool   `json:"force,omitempty"`
 }
 
-// DrainVolumeData carries parameters for a drain-volume command, which flushes
-// the volume's in-flight writes to S3 before a snapshot reads them. The command
-// is addressed to the instance the volume is attached to because the drain
-// socket only exists on the node hosting it.
-type DrainVolumeData struct {
-	VolumeID string `json:"volume_id"`
-}
-
-const (
-	// DrainVolumeStatusDrained is the ack a node returns once the volume's
-	// in-flight writes have reached S3.
-	DrainVolumeStatusDrained = "drained"
-
-	// DrainVolumeStatusNotRunning is the ack a node returns when it still holds
-	// the instance but the VM is not running, so no process is writing to the
-	// volume and its sealed checkpoint is already current. An attachment record
-	// outlives the writer — stop deliberately leaves boot volumes attached — so
-	// the caller cannot tell this from the record alone.
-	DrainVolumeStatusNotRunning = "not-running"
-)
-
-// DrainVolumeResponse is the reply to a drain-volume command.
-type DrainVolumeResponse struct {
-	VolumeID string `json:"volume_id"`
-	Status   string `json:"status"`
-}
-
 // AttachENIData carries parameters for an attach-network-interface command.
 type AttachENIData struct {
 	NetworkInterfaceID string `json:"network_interface_id"`

contracts/ec2/v1/commands_test.go

@@ -16,7 +16,6 @@ func TestEC2InstanceCommandJSONContract(t *testing.T) {
 			StartInstance:               true,
 			AttachVolume:                true,
 			DetachVolume:                true,
-			DrainVolume:                 true,
 			RebootInstance:              true,
 			AttachENI:                   true,
 			DetachENI:                   true,
@@ -28,7 +27,6 @@ func TestEC2InstanceCommandJSONContract(t *testing.T) {
 		},
 		AttachVolumeData:          &AttachVolumeData{VolumeID: "vol-1", Device: "/dev/sdf"},
 		DetachVolumeData:          &DetachVolumeData{VolumeID: "vol-2", Device: "/dev/sdg", Force: true},
-		DrainVolumeData:           &DrainVolumeData{VolumeID: "vol-3"},
 		AttachENIData:             &AttachENIData{NetworkInterfaceID: "eni-1", DeviceIndex: 2},
 		DetachENIData:             &DetachENIData{AttachmentID: "eni-attach-1", Force: true},
 		IamProfileAssociationData: &IamProfileAssociationData{InstanceProfileArn: "arn:aws:iam::123456789012:instance-profile/example"},
@@ -47,7 +45,6 @@ func TestEC2InstanceCommandJSONContract(t *testing.T) {
 			"start_instance": true,
 			"attach_volume": true,
 			"detach_volume": true,
-			"drain_volume": true,
 			"reboot_instance": true,
 			"attach_eni": true,
 			"detach_eni": true,
@@ -59,7 +56,6 @@ func TestEC2InstanceCommandJSONContract(t *testing.T) {
 		},
 		"attach_volume_data": {"volume_id": "vol-1", "device": "/dev/sdf"},
 		"detach_volume_data": {"volume_id": "vol-2", "device": "/dev/sdg", "force": true},
-		"drain_volume_data": {"volume_id": "vol-3"},
 		"attach_eni_data": {"network_interface_id": "eni-1", "device_index": 2},
 		"detach_eni_data": {"attachment_id": "eni-attach-1", "force": true},
 		"iam_profile_association_data": {"instance_profile_arn": "arn:aws:iam::123456789012:instance-profile/example"},
@@ -69,16 +65,6 @@ func TestEC2InstanceCommandJSONContract(t *testing.T) {
 	}`, string(data))
 }
 
-func TestDrainVolumeResponseJSONContract(t *testing.T) {
-	data, err := json.Marshal(DrainVolumeResponse{
-		VolumeID: "vol-1",
-		Status:   DrainVolumeStatusDrained,
-	})
-	require.NoError(t, err)
-	require.JSONEq(t, `{"volume_id":"vol-1","status":"drained"}`, string(data))
-	require.Equal(t, "not-running", DrainVolumeStatusNotRunning)
-}
-
 func TestInstanceCommandSubjects(t *testing.T) {
 	require.Equal(t, "ec2.cmd.i-0123456789abcdef0", InstanceCommandSubject("i-0123456789abcdef0"))
 	require.Equal(t, "ec2.cmd.*", InstanceCommandSubjectWildcard)

contracts/ec2/v1/volume_held.go

@@ -1,7 +1,13 @@
 package ec2v1
 
-// VolumeHeldElsewhereText is the text of a start refused because another node
-// holds one of the instance's volumes. It crosses NATS as message text, so
-// readers match on it; the provider that refuses and every reader share this
-// string so they cannot drift.
+import "errors"
+
+// VolumeHeldElsewhereText is the message viperblockd gives a lease refusal. It
+// is wording for the operator; callers classify the refusal by its code.
 const VolumeHeldElsewhereText = "volume is leased by another owner"
+
+// ErrVolumeHeldElsewhere marks a start refused because another node still
+// holds one of the instance's volumes. The owning node sets it from the EBS
+// provider's held-elsewhere code, and a start reply carries it across NATS as
+// the X-Volume-Held-Elsewhere header rather than as message text.
+var ErrVolumeHeldElsewhere = errors.New("a volume is leased by another node")

contracts/operator/v1/admin.go

@@ -84,8 +84,8 @@ type ListAccountsResponse struct {
 }
 
 // IAMQuiescence is the coordinator's recorded view of IAM quiescence for one
-// fence. Recording only: it admits nothing. QuiescedScope is filled only for a
-// non-finalizing run, and holds as of EvaluatedAt: a write after that is not caught.
+// fence. Recording only: it admits nothing. QuiescedScope is filled only by a run
+// that guarantees a final inventory re-read, and holds as of EvaluatedAt.
 type IAMQuiescence struct {
 	OperationID   string            `json:"operationId"`
 	Fence         uint64            `json:"fence"`

contracts/viperblockd/legacy/v1/README.md

@@ -51,19 +51,6 @@ request/reply replies.
 removed its local state and export. These routes and JSON shapes are deployed
 legacy compatibility behaviour, not a generic EBS-provider interface.
 
-## Fence route
-
-`VolumeFencedSubject(node)` is `ebs.<node>.fenced`, or `ebs.fenced` with no
-node for a single-node daemon. The Viperblock adapter publishes
-`VolumeFencedEvent` on it after tearing down an export whose volume lease
-moved to another node; the local daemon subscribes to stop the guest that
-was using it. Node-addressed, like mount and unmount: the guest is local, and
-no other node's daemon has anything to stop.
-
-The event is sent as a request and repeated until the daemon replies with `VolumeFencedAck`, because the fence often runs while the node's NATS server is restarting and a single publish can reach no subscriber.
-The daemon replies only when the event carries a reply subject, so a plain publish is still accepted.
-Repeats are harmless, since fencing a guest that is already fenced changes nothing.
-
 ## Abandon route
 
 `VolumeAbandonSubject(node)` is `ebs.<node>.abandon`. The daemon uses

contracts/viperblockd/legacy/v1/abandon_test.go

@@ -9,9 +9,8 @@ import (
 	"github.com/stretchr/testify/require"
 )
 
-// TestVolumeAbandonSubject_BothForms pins the abandon route the same way the
-// fenced route is pinned: node-addressed, with an empty-node fallback for a
-// single-node daemon.
+// TestVolumeAbandonSubject_BothForms pins the abandon route: node-addressed,
+// with an empty-node fallback for a single-node daemon.
 func TestVolumeAbandonSubject_BothForms(t *testing.T) {
 	assert.Equal(t, "ebs.node-a.abandon", viperblocklegacyv1.VolumeAbandonSubject("node-a"))
 	assert.Equal(t, "ebs.abandon", viperblocklegacyv1.VolumeAbandonSubject(""))

contracts/viperblockd/legacy/v1/fence.go

@@ -1,29 +0,0 @@
-package viperblocklegacyv1
-
-// VolumeFencedSubject is where a fenced volume is announced so the daemon on
-// this node can stop the guest that was using it. Node-addressed: the guest is
-// local, and every other node's daemon has nothing to do about it.
-func VolumeFencedSubject(node string) string {
-	if node == "" {
-		return "ebs.fenced"
-	}
-	return "ebs." + node + ".fenced"
-}
-
-// VolumeFencedEvent announces that a node tore down a volume's export because
-// the lease moved. Winner is who holds it now, for the operator reading the
-// guest's stop reason.
-type VolumeFencedEvent struct {
-	Volume string `json:"volume"`
-	Node   string `json:"node"`
-	Winner string `json:"winner"`
-	Reason string `json:"reason"`
-}
-
-// VolumeFencedAck is the daemon's answer to a VolumeFencedEvent. Any answer
-// ends the sender's retries; Instance names the guest stopped, if one was, and
-// Error says why the daemon could not act on the event.
-type VolumeFencedAck struct {
-	Instance string `json:"instance,omitempty"`
-	Error    string `json:"error,omitempty"`
-}

contracts/viperblockd/legacy/v1/fence_test.go

@@ -1,54 +0,0 @@
-package viperblocklegacyv1_test
-
-import (
-	"encoding/json"
-	"testing"
-
-	viperblocklegacyv1 "github.com/mulgadc/spinifex/contracts/viperblockd/legacy/v1"
-	"github.com/stretchr/testify/assert"
-	"github.com/stretchr/testify/require"
-)
-
-// TestVolumeFencedSubject_IsNodeAddressed pins the routing. A fenced guest is on
-// the node that lost the volume, so a subject without the node in it would reach
-// daemons with nothing to stop and, on a queue group, miss the one that has.
-func TestVolumeFencedSubject_IsNodeAddressed(t *testing.T) {
-	assert.Equal(t, "ebs.node-a.fenced", viperblocklegacyv1.VolumeFencedSubject("node-a"))
-	assert.Equal(t, "ebs.fenced", viperblocklegacyv1.VolumeFencedSubject(""),
-		"a single-node daemon has no node name, and still has to hear its own fences")
-}
-
-// TestVolumeFencedEvent_JSON pins the exact field spelling of the fenced
-// event, published by viperblockd and consumed by the daemon.
-func TestVolumeFencedEvent_JSON(t *testing.T) {
-	golden := `{"volume":"vol-1","node":"node-a","winner":"node-b","reason":"volume lease moved to node-b while this node had the volume open"}`
-	event := viperblocklegacyv1.VolumeFencedEvent{
-		Volume: "vol-1",
-		Node:   "node-a",
-		Winner: "node-b",
-		Reason: "volume lease moved to node-b while this node had the volume open",
-	}
-
-	marshaled, err := json.Marshal(event)
-	require.NoError(t, err)
-	assert.JSONEq(t, golden, string(marshaled))
-
-	var decoded viperblocklegacyv1.VolumeFencedEvent
-	require.NoError(t, json.Unmarshal([]byte(golden), &decoded))
-	assert.Equal(t, event, decoded)
-}
-
-// TestVolumeFencedAck_JSON pins the daemon's answer, which is what ends
-// viperblockd's retries of a fence notice.
-func TestVolumeFencedAck_JSON(t *testing.T) {
-	golden := `{"instance":"i-1","error":"bad event"}`
-	ack := viperblocklegacyv1.VolumeFencedAck{Instance: "i-1", Error: "bad event"}
-
-	marshaled, err := json.Marshal(ack)
-	require.NoError(t, err)
-	assert.JSONEq(t, golden, string(marshaled))
-
-	empty, err := json.Marshal(viperblocklegacyv1.VolumeFencedAck{})
-	require.NoError(t, err)
-	assert.JSONEq(t, `{}`, string(empty), "an ack for a volume with no local guest carries nothing")
-}

docs/COMMANDS.md

@@ -95,7 +95,7 @@ Operational commands for inspecting cluster state. These fan out NATS requests t
 
 | Command | Flags | Description |
 |---------|-------|-------------|
-| `spx admin init` | `--nodes`, `--node`, `--bind`, `--advertise` (external IP off-host clients dial; auto-detected from the default route), `--port`, `--region`, `--az`, `--cluster-name`, `--cluster-bind`, `--cluster-routes`, `--predastore-nodes`, `--predastore-compaction-interval` (seconds; test clusters set a short one), `--services`, `--formation-timeout`, `--token-ttl`, `--force`, `--discard-jetstream` (with `--nodes` >= 2, remove this node's existing JetStream streams once every node has joined; without it a multi-node init over a store that holds streams is refused, and it never proceeds while NATS is running), `--skip-host-dns`, `--email` (operator address for update and security notices), `--no-telemetry`, `--gpu-passthrough`, `--ipsec` (default true; encrypt intra-AZ Geneve via OVN native IPsec cluster-wide — disable only for a trusted single-rack lab; on a host whose IPsec stack is libreswan it defaults to false and `--ipsec=true` is refused), `--host` (leader to join; falls back to multicast discovery), `--external-mode` (`pool` \| `nat` routed NAT for non-bridgeable uplinks, single-node; pair with `setup-ovn.sh --nat-uplink`; note: system instances (ECS/EKS/load-balancer agents) are unsupported in nat v1 — they require `pool` mode), `--external-iface` (WAN NIC for br-external, auto-detected), `--external-source` (`dhcp` \| `static`), `--external-pool` (`start-end`), `--external-prefix-len` (default 24), `--external-gateway` (auto-detected), `--external-dns` (comma-separated IPv4 addresses, at most 3; the DNS servers written to the external pool's `dns_servers` and handed to guests, verbatim; when omitted they are detected from the host resolver at init time, so set it to give every node of a cluster the same value), `--external-bind-bridge` (default `br-wan` when `--external-source=dhcp`), `--gateway-ip` (OVN gateway router's external IP for SNAT; defaults to the pool's range start, required for `nat` mode without DHCP) | Generates root IAM credentials (AKIA-prefixed access key + secret) → creates master.key (AES-256, 32 bytes, 0600) → writes bootstrap.json (consumed on first start) → generates CA + server TLS certificates → generates join token (written to `join-token` file, displayed in join command) → creates NATS config with auth token → writes spinifex.toml, awsgw.toml, predastore.toml → configures AWS CLI `spx` profile → creates directory structure under `~/spinifex/` → points the host resolver at this node's local northstar so the Spinifex zones (ELB/EKS names) resolve from the node itself (opt out with `--skip-host-dns`). `--force` on an already-initialized node is idempotent for crypto: it preserves the existing keys (master.key, predastore/viperblock encryption keys), system + admin credentials, and CA, and refreshes only the config files and the CA-signed server certificate (so a changed bind IP / SANs is picked up without breaking already-joined nodes or CA-baked AMIs). A genuine clean slate requires removing the data dirs first. |
+| `spx admin init` | `--nodes`, `--node`, `--bind`, `--advertise` (external IP off-host clients dial; auto-detected from the default route), `--port`, `--region`, `--az`, `--cluster-name`, `--cluster-bind`, `--cluster-routes`, `--predastore-nodes`, `--predastore-compaction-interval` (seconds; test clusters set a short one), `--services`, `--formation-timeout`, `--token-ttl`, `--force`, `--discard-jetstream` (with `--nodes` >= 2, remove this node's existing JetStream streams once every node has joined; without it a multi-node init over a store that holds streams is refused, and it never proceeds while NATS is running), `--skip-host-dns`, `--email` (operator address for update and security notices), `--no-telemetry`, `--gpu-passthrough`, `--ipsec` (default true; encrypt intra-AZ Geneve via OVN native IPsec cluster-wide — disable only for a trusted single-rack lab; on a host whose IPsec stack is libreswan it defaults to false and `--ipsec=true` is refused), `--external-mode` (`pool` \| `nat` routed NAT for non-bridgeable uplinks, single-node; pair with `setup-ovn.sh --nat-uplink`; note: system instances (ECS/EKS/load-balancer agents) are unsupported in nat v1 — they require `pool` mode), `--external-iface` (WAN NIC for br-external, auto-detected), `--external-source` (`dhcp` \| `static`), `--external-pool` (`start-end`), `--external-prefix-len` (default 24), `--external-gateway` (auto-detected), `--external-dns` (comma-separated IPv4 addresses, at most 3; the DNS servers written to the external pool's `dns_servers` and handed to guests, verbatim; when omitted they are detected from the host resolver at init time, so set it to give every node of a cluster the same value), `--external-bind-bridge` (default `br-wan` when `--external-source=dhcp`), `--gateway-ip` (OVN gateway router's external IP for SNAT; defaults to the pool's range start, required for `nat` mode without DHCP) | Generates root IAM credentials (AKIA-prefixed access key + secret) → creates master.key (AES-256, 32 bytes, 0600) → writes bootstrap.json (consumed on first start) → generates CA + server TLS certificates → generates join token (written to `join-token` file, displayed in join command) → creates NATS config with auth token → writes spinifex.toml, awsgw.toml, predastore.toml → configures AWS CLI `spx` profile → creates directory structure under `~/spinifex/` → points the host resolver at this node's local northstar so the Spinifex zones (ELB/EKS names) resolve from the node itself (opt out with `--skip-host-dns`). `--force` on an already-initialized node is idempotent for crypto: it preserves the existing keys (master.key, predastore/viperblock encryption keys), system + admin credentials, and CA, and refreshes only the config files and the CA-signed server certificate (so a changed bind IP / SANs is picked up without breaking already-joined nodes or CA-baked AMIs). A genuine clean slate requires removing the data dirs first. |
 | `spx admin join` | `--host` (required), `--node` (required), `--token` (required), `--bind`, `--advertise` (defaults to `--bind`, or the auto-detected WAN IP when `--bind` is 0.0.0.0), `--port`, `--region`, `--az`, `--cluster-bind`, `--cluster-routes`, `--data-dir`, `--services`, `--skip-host-dns`, `--email`, `--no-telemetry`, `--predastore-compaction-interval`, `--join-timeout` (how long to keep retrying while the formation server is unreachable, default 20m), `--force` (join even though this node is already initialized or holds JetStream streams, **discarding its own CA, master key and JetStream store**; the store is removed only after formation succeeds, and the join is refused while NATS is running even with `--force`) | Connects to leader node with join token (Authorization: Bearer header) → retrieves cluster configuration → configures local node to join cluster and participate in distributed operations → points the host resolver at this node's local northstar so the Spinifex zones resolve from the node itself (opt out with `--skip-host-dns`) |
 
 ### Version
@@ -135,7 +135,7 @@ Operational commands for inspecting cluster state. These fan out NATS requests t
 
 | Command | Flags | Description |
 |---------|-------|-------------|
-| `spx admin volumes orphans` | — | Lists volumes the storage provider holds that the control plane has no record of. Their blocks consume space but they have no API handle, so they can be neither described nor deleted through the EC2 API. **Reports only, never deletes**: the evidence that an orphan is still wanted is the record that went missing, so removal is an operator decision taken per volume. |
+| `spx admin volumes orphans` | — | Lists volumes the storage provider holds that the control plane has no record of. Their blocks consume space but they have no API handle, so they can be neither described nor deleted through the EC2 API. **Reports only, never deletes**: the evidence that an orphan is still wanted is the record that went missing, so removal is an operator decision taken per volume. It also lists UEFI variable stores nothing owns: a `uefi/i-*/` prefix with no instance record and no store record, a `uefi/ami-*/` prefix with no AMI document, and a store record that is not `deleting` and has no instance record. These are reported and never deleted, because an absent instance record is not proof of a terminate. |
 | `spx admin volumes unsealed` | — | Lists volumes a node has open, or had open, without a confirmed seal to the backend — the node named holds writes predastore may not have. Starting such a volume elsewhere is refused while that node is alive (its daemon heartbeat is under 60 seconds old, or cannot be read) and is otherwise allowed, in which case it opens from the last checkpoint that did reach the backend. This is the list of volumes where that would cost something. |
 | `spx admin storage status` | `--endpoint` (default: this node's AWS gateway), `--region` (default: this node's region), `--ca-bundle` (default: this node's CA) | Sends the signed `GetStorageStatus` action to the AWS gateway with credentials from the standard AWS chain (e.g. `AWS_PROFILE=spinifex`) and prints what the console's Service Metrics page shows: predastore encoding, meta and blob node counts, cluster health, a meta node table (health, raft state, leader, term, commit and applied index) and the blob nodes. Cluster health is `Degraded` when any meta node is unreachable or the nodes reporting raft state do not agree on one leader. A meta node read only through its host's `/readyz` shows healthy with blank raft fields. |
 

docs/DESIGN.md

@@ -112,14 +112,14 @@ When a daemon handles `RunInstances`:
 
 1. **Resource Check**: Validates CPU/memory availability via `ResourceManager`
 2. **Volume Generation**: Creates boot and EFI volumes via Viperblock
-3. **Volume Mount**: Sends a mount request to the colocated Viperblock worker on `ebs.{node}.mount`, receives NBD URI
+3. **Volume Mount**: Publishes each volume through the EBS provider contract (`PublishVolume` on `ebs.provider.v1.{node}.mount`), receives its export
 4. **QEMU Launch**: Builds and executes QEMU command with NBD-backed disks
 5. **QMP Monitoring**: Establishes QEMU Machine Protocol connection for VM management
 6. **Command subscription**: Subscribes to `ec2.cmd.{instanceID}` for subsequent commands on this VM
 
 ```go
-// Volume mounting via NATS (runtime/daemon/vm_adapters.go)
-reply, err := ebsRequestWithTrace(ctx, a.nc, instance.AccountID, viperblocklegacyv1.MountSubject(a.node), ebsMountRequest, 30*time.Second)
+// Volume mounting through the EBS provider contract (runtime/daemon/vm_adapters.go)
+published, err := a.provider.PublishVolume(ebsprovider.WithAccountID(ctx, accountID), ebsprovider.PublishVolumeRequest{Versioned: ebsprovider.NewLatestVersioned(), VolumeID: req.Name, NodeID: a.node})
 
 // QEMU launch with NBD-backed drives in args (runtime/compute/vm/vm.go)
 cmd := exec.Command(qemuArchitecture, args...) // qemu-system-x86_64 or qemu-system-aarch64

docs/compute/launching-instances/README.md

@@ -23,6 +23,7 @@ resources:
 - [Overview](#overview)
 - [Launch](#launch)
 - [Manage](#manage)
+- [Stopping and Data Volumes](#stopping-and-data-volumes)
 - [Modify Instance Attributes](#modify-instance-attributes)
 - [Instance Metadata Options](#instance-metadata-options)
 - [Spot Instances](#spot-instances)
@@ -73,6 +74,31 @@ aws ec2 terminate-instances --instance-ids $INSTANCE_ID
 aws ec2 reboot-instances --instance-ids $INSTANCE_ID
 ```
 
+## Stopping and Data Volumes
+
+A guest that powers itself off (`poweroff`, `shutdown -h`) goes `stopping` then `stopped`, the same as `stop-instances`.
+It is not restarted automatically.
+`InstanceInitiatedShutdownBehavior=terminate` is not honoured yet: a guest shutdown always stops the instance.
+
+Data volumes stay `in-use` and attached to a stopped instance, and `start-instances` mounts them again.
+While the instance is stopped, `attach-volume` of one of its data volumes to another instance returns `VolumeInUse`, and `delete-volume` is refused.
+
+`detach-volume` works on a stopped instance, with or without `--instance-id`.
+The volume becomes `available` and the next start does not attach it.
+The root volume of a stopped instance can be detached too, as on AWS.
+The internal EFI variable store is not a customer volume: it never appears in `describe-volumes` or `BlockDeviceMappings`, and detaching it is refused with `OperationNotPermitted`.
+The instance then cannot start: `start-instances` returns `InvalidParameterValue` with "Instance does not have a volume attached at root (/dev/sda1)".
+
+`attach-volume` works on a stopped instance too.
+The volume must be `available` and in the instance's availability zone, otherwise the call returns `VolumeInUse` or `InvalidVolume.ZoneMismatch`.
+The attachment is recorded on the stopped instance and the volume becomes `in-use` straight away.
+Attaching at the root device name (for example `/dev/sda1`) makes the volume the root again, and the next `start-instances` boots from it.
+
+An instance stopped by an earlier build had its data volumes released to `available` while its record still listed them.
+On start, a listed data volume that was deleted or is now attached to another instance is dropped with a warning in the daemon log, and the start continues.
+A listed volume that is still `available` and unattached is attached again.
+A missing root volume fails the start with the same `InvalidParameterValue` answer.
+
 ## Modify Instance Attributes
 
 Change instance type, user data, or termination protection. Instance type and user data require the instance to be **stopped** first.
@@ -135,6 +161,30 @@ aws ec2 get-console-output --instance-id $INSTANCE_ID \
   --query 'Output' --output text | base64 -d
 ```
 
+Like AWS, the call returns the most recent 64 KiB of the console.
+While an instance is shutting down after a terminate request, the call reads the renamed file described below.
+Once the instance has left the node's running set, which includes a stopped or terminated instance, the node no longer answers and the call returns `InvalidInstanceID.NotFound`.
+AWS still lists a terminated instance for a while after termination, so a caller may expect this call to work then; it does not here, a known gap.
+Whether the API should answer for a stopped instance is not settled here; only the file on the host is kept.
+
+### Where console logs live and how big they get
+
+The serial console of each instance is written on its host to `/run/spinifex/console-<instance-id>.log`.
+`/run` is memory-backed, so these files are cleared when the host reboots, including any left by older releases.
+The daemon keeps them bounded.
+A file over 4 MiB is cut back in place to its newest 1 MiB, and the guest keeps writing while it happens.
+The console is opened in append mode, so a reboot or auto-restart of the guest adds to the same file instead of starting over.
+A live `console-<instance-id>.log` is deleted only when it has not changed for an hour and no QEMU process on the host has it open, so a stopped instance keeps its console for an hour and a quiet running guest keeps it indefinitely.
+If the daemon cannot read the open files of every QEMU process on the host, it deletes nothing on that pass and logs a warning.
+This also clears the file of an instance that was stopped and then terminated, or that was recovered on another node, within about an hour of its last write.
+When an instance is terminated, including a failed first launch and the system instances behind load balancers, queues and EKS control planes, its file is renamed to `console-<instance-id>.log.terminated`.
+The daemon removes a `.terminated` file an hour after its last write.
+
+Known limits:
+
+- The log shipper follows only `console-*.log`, so it stops following a file when it is renamed on terminate, and the last few seconds of output that it had not yet read can be lost.
+- The instance password retrieval reads only the live file name, so it does not find a console renamed after terminate.
+
 ## Instance Types
 
 List instance types available on the current host. The catalog is generated from the host CPU (Intel, AMD, or ARM) and includes burstable (t-family), general purpose (m-family), compute optimised (c-family), and memory optimised (r-family) types.

internal/awsmodel/model.go

@@ -92,6 +92,7 @@ type Shape struct {
 	Min             *float64
 	Max             *float64
 	Pattern         string
+	Documentation   string
 	LocationName    string
 	TimestampFormat string
 	XMLNamespace    XMLNamespace

internal/awsmodel/smithy.go

@@ -443,6 +443,9 @@ func (t *smithyTranslator) applyShapeTraits(shape *Shape, traits smithyTraits) e
 	if shape.Pattern, err = traits.string("smithy.api#pattern"); err != nil {
 		return err
 	}
+	if shape.Documentation, err = traits.string("smithy.api#documentation"); err != nil {
+		return err
+	}
 	if shape.LocationName, err = traits.string("smithy.api#xmlName"); err != nil {
 		return err
 	}

internal/ecsgw/client.go

@@ -7,9 +7,10 @@ package ecsgw
 
 import (
 	"bytes"
-	"context"
+	"crypto/sha256"
 	"crypto/tls"
 	"crypto/x509"
+	"encoding/hex"
 	"fmt"
 	"io"
 	"net/http"
@@ -17,10 +18,8 @@ import (
 	"strings"
 	"time"
 
-	"github.com/aws/aws-sdk-go/aws/credentials"
-	v4 "github.com/aws/aws-sdk-go/aws/signer/v4"
-
 	"github.com/mulgadc/bluebottle/pkg/tlsconfig"
+	"github.com/mulgadc/spinifex/internal/gwsign"
 )
 
 // targetPrefix is the X-Amz-Target service prefix the gateway's ECS dispatch
@@ -28,31 +27,26 @@ import (
 // suffix); this one mirrors the AWS JSON 1.1 convention.
 const targetPrefix = "AmazonEC2ContainerServiceV20141113."
 
-// CredentialsFunc yields the current SigV4 credentials. It is called once per
-// request so a provider backed by IMDS can rotate instance-role creds (and their
-// session token) transparently. Must be safe for concurrent use.
-type CredentialsFunc func(ctx context.Context) (accessKey, secretKey, sessionToken string, err error)
-
 // Client posts SigV4-signed AWS JSON 1.1 requests to the gateway for service
 // "ecs". One client is reused for register/heartbeat/state/poll.
 type Client struct {
 	baseURL    string
-	creds      CredentialsFunc
+	signer     *gwsign.Signer
 	region     string
 	httpClient *http.Client
 }
 
 // New builds a client. caPath optionally pins the gateway TLS CA; empty relies on
-// the system trust store. creds supplies the per-call instance-role credentials.
+// the system trust store. signer supplies the per-call instance-role credentials.
 // region defaults to us-east-1 when empty (SigV4 requires a non-empty region).
 // timeout bounds a single call — callers needing a long-poll pass a larger value
 // than the default register/heartbeat timeout.
-func New(baseURL, caPath string, creds CredentialsFunc, region string, timeout time.Duration) (*Client, error) {
+func New(baseURL, caPath string, signer *gwsign.Signer, region string, timeout time.Duration) (*Client, error) {
 	if baseURL == "" {
 		return nil, fmt.Errorf("ecsgw: baseURL is required")
 	}
-	if creds == nil {
-		return nil, fmt.Errorf("ecsgw: credentials func is required")
+	if signer == nil {
+		return nil, fmt.Errorf("ecsgw: signer is required")
 	}
 	if region == "" {
 		region = "us-east-1"
@@ -79,7 +73,7 @@ func New(baseURL, caPath string, creds CredentialsFunc, region string, timeout t
 
 	return &Client{
 		baseURL: strings.TrimRight(baseURL, "/"),
-		creds:   creds,
+		signer:  signer,
 		region:  region,
 		httpClient: &http.Client{
 			Timeout:   timeout,
@@ -90,7 +84,7 @@ func New(baseURL, caPath string, creds CredentialsFunc, region string, timeout t
 
 // Call SigV4-signs and POSTs body under X-Amz-Target "<prefix><action>" to the
 // gateway root, returning the 2xx response body. Credentials are fetched per call
-// and the v4 signer emits X-Amz-Security-Token for instance-role (temporary)
+// and the signer emits X-Amz-Security-Token for instance-role (temporary)
 // creds, so the gateway's ASIA path verifies them. No retry; callers wrap.
 func (c *Client) Call(action string, body []byte) ([]byte, error) {
 	req, err := http.NewRequest(http.MethodPost, c.baseURL+"/", bytes.NewReader(body))
@@ -100,12 +94,8 @@ func (c *Client) Call(action string, body []byte) ([]byte, error) {
 	req.Header.Set("Content-Type", "application/x-amz-json-1.1")
 	req.Header.Set("X-Amz-Target", targetPrefix+action)
 
-	akid, secret, token, err := c.creds(req.Context())
-	if err != nil {
-		return nil, fmt.Errorf("retrieve credentials: %w", err)
-	}
-	signer := v4.NewSigner(credentials.NewStaticCredentials(akid, secret, token))
-	if _, err := signer.Sign(req, bytes.NewReader(body), "ecs", c.region, time.Now()); err != nil {
+	sum := sha256.Sum256(body)
+	if err := c.signer.Sign(req, hex.EncodeToString(sum[:]), "ecs", c.region); err != nil {
 		return nil, fmt.Errorf("sign request: %w", err)
 	}
 

internal/ecsgw/client_test.go

@@ -1,24 +1,29 @@
 package ecsgw
 
 import (
-	"context"
 	"io"
 	"net/http"
 	"net/http/httptest"
 	"strings"
 	"testing"
 	"time"
+
+	"github.com/aws/aws-sdk-go-v2/credentials"
+	"github.com/mulgadc/bluebottle/pkg/sigv4"
+	"github.com/mulgadc/spinifex/internal/gwsign"
 )
 
-// testCreds returns a fixed temporary credential set (with a session token, as
-// IMDS instance-role creds carry).
-func testCreds(context.Context) (string, string, string, error) {
-	return "ASIATEST", "secret", "session-token-xyz", nil
+const testSecret = "secret"
+
+// testSigner signs with a fixed temporary credential set (with a session token,
+// as IMDS instance-role creds carry).
+func testSigner() *gwsign.Signer {
+	return gwsign.New(credentials.NewStaticCredentialsProvider("ASIATEST", testSecret, "session-token-xyz"))
 }
 
 func newTestClient(t *testing.T, srv *httptest.Server) *Client {
 	t.Helper()
-	c, err := New(srv.URL, "", testCreds, "ap-southeast-2", time.Second)
+	c, err := New(srv.URL, "", testSigner(), "ap-southeast-2", time.Second)
 	if err != nil {
 		t.Fatalf("New: %v", err)
 	}
@@ -30,7 +35,14 @@ func newTestClient(t *testing.T, srv *httptest.Server) *Client {
 // 2xx body.
 func TestClient_CallSignsAndTargets(t *testing.T) {
 	var gotTarget, gotAuth, gotBody, gotToken string
+	var verifyErr error
 	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		// sigv4.Parse reads and hashes the body itself, then restores it.
+		sr, err := sigv4.Parse(r)
+		if err == nil {
+			_, err = sr.Verify(testSecret, "ap-southeast-2", "ecs")
+		}
+		verifyErr = err
 		gotTarget = r.Header.Get("X-Amz-Target")
 		gotAuth = r.Header.Get("Authorization")
 		gotToken = r.Header.Get("X-Amz-Security-Token")
@@ -45,6 +57,9 @@ func TestClient_CallSignsAndTargets(t *testing.T) {
 	if err != nil {
 		t.Fatalf("Call: %v", err)
 	}
+	if verifyErr != nil {
+		t.Fatalf("gateway-side SigV4 verify failed: %v", verifyErr)
+	}
 	if string(out) != `{"ok":true}` {
 		t.Errorf("body = %q", out)
 	}
@@ -78,10 +93,10 @@ func TestClient_CallErrorStatus(t *testing.T) {
 }
 
 func TestNew_Validation(t *testing.T) {
-	if _, err := New("", "", testCreds, "r", 0); err == nil {
+	if _, err := New("", "", testSigner(), "r", 0); err == nil {
 		t.Error("want error for empty baseURL")
 	}
 	if _, err := New("https://gw:9999", "", nil, "r", 0); err == nil {
-		t.Error("want error for nil credentials func")
+		t.Error("want error for nil signer")
 	}
 }

internal/gwsign/signer.go

@@ -15,8 +15,7 @@ import (
 	v4 "github.com/aws/aws-sdk-go-v2/aws/signer/v4"
 	awsconfig "github.com/aws/aws-sdk-go-v2/config"
 	"github.com/aws/aws-sdk-go-v2/credentials"
-	"github.com/aws/aws-sdk-go-v2/credentials/ec2rolecreds"
-	"github.com/aws/aws-sdk-go-v2/feature/ec2/imds"
+	"github.com/mulgadc/spinifex/internal/imdscreds"
 )
 
 // Signer holds a credential provider and signs requests in place.
@@ -24,6 +23,12 @@ type Signer struct {
 	provider aws.CredentialsProvider
 }
 
+// New returns a Signer over provider. Wrap a provider that fetches remotely in
+// aws.NewCredentialsCache, since Sign retrieves on every call.
+func New(provider aws.CredentialsProvider) *Signer {
+	return &Signer{provider: provider}
+}
+
 // NewStatic returns a Signer using fixed long-lived keys (no session token).
 func NewStatic(accessKey, secretKey string) *Signer {
 	return &Signer{provider: credentials.NewStaticCredentialsProvider(accessKey, secretKey, "")}
@@ -46,12 +51,14 @@ func NewIMDS(ctx context.Context, region string) (*Signer, error) {
 
 // NewInstanceRole returns a Signer whose credentials come only from the
 // instance metadata service: no environment, profile or static-key fallback.
+// A failed refresh of expired credentials is an error, not the stale key.
 func NewInstanceRole() *Signer {
-	return newInstanceRole(imds.New(imds.Options{}))
+	return newInstanceRole("")
 }
 
-func newInstanceRole(client *imds.Client) *Signer {
-	return &Signer{provider: aws.NewCredentialsCache(ec2rolecreds.New(func(o *ec2rolecreds.Options) { o.Client = client }))}
+// newInstanceRole reads from the IMDS root base; "" means the default endpoint.
+func newInstanceRole(base string) *Signer {
+	return &Signer{provider: aws.NewCredentialsCache(imdscreds.NewProvider(nil, base))}
 }
 
 // EnsureCredentials retrieves once from the provider, surfacing the error so a

internal/gwsign/signer_test.go

@@ -12,7 +12,6 @@ import (
 
 	"github.com/aws/aws-sdk-go-v2/aws"
 	"github.com/aws/aws-sdk-go-v2/credentials"
-	"github.com/aws/aws-sdk-go-v2/feature/ec2/imds"
 	"github.com/mulgadc/bluebottle/pkg/sigv4"
 )
 
@@ -145,7 +144,7 @@ func TestNewInstanceRole_UsesOnlyIMDS(t *testing.T) {
 	}))
 	t.Cleanup(srv.Close)
 
-	s := newInstanceRole(imds.New(imds.Options{Endpoint: srv.URL}))
+	s := newInstanceRole(srv.URL)
 	req := httptest.NewRequest(http.MethodPost, "https://gw.example/", nil)
 	if err := s.Sign(req, emptySHA, testService, testRegion); err != nil {
 		t.Fatal(err)
@@ -155,10 +154,42 @@ func TestNewInstanceRole_UsesOnlyIMDS(t *testing.T) {
 	}
 
 	srv.Close()
-	cold := newInstanceRole(imds.New(imds.Options{Endpoint: srv.URL}))
+	cold := newInstanceRole(srv.URL)
 	if err := cold.EnsureCredentials(context.Background()); err == nil {
 		t.Fatal("environment keys were used when IMDS was unreachable")
 	}
 }
 
+// Once the cached instance-role key has expired, a refresh that cannot reach
+// IMDS must fail the sign rather than reuse the expired key with a new expiry.
+func TestNewInstanceRole_FailedRefreshErrorsInsteadOfStaleKey(t *testing.T) {
+	const staleKey = "ASIASTALEEXAMPLE0001"
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		switch r.URL.Path {
+		case "/latest/api/token":
+			_, _ = w.Write([]byte("token"))
+		case "/latest/meta-data/iam/security-credentials/":
+			_, _ = w.Write([]byte("sqsWorkerRole"))
+		case "/latest/meta-data/iam/security-credentials/sqsWorkerRole":
+			_, _ = w.Write([]byte(`{"Code":"Success","AccessKeyId":"` + staleKey + `","SecretAccessKey":"s","Token":"t","Expiration":"2000-01-01T00:00:00Z"}`))
+		default:
+			http.NotFound(w, r)
+		}
+	}))
+	s := newInstanceRole(srv.URL)
+	if err := s.EnsureCredentials(t.Context()); err != nil {
+		t.Fatalf("first retrieve: %v", err)
+	}
+
+	srv.Close()
+	req := httptest.NewRequest(http.MethodPost, "https://gw.example/", nil)
+	err := s.Sign(req, emptySHA, testService, testRegion)
+	if err == nil {
+		t.Fatalf("Sign after failed refresh = nil, signed with %q", req.Header.Get("Authorization"))
+	}
+	if auth := req.Header.Get("Authorization"); auth != "" {
+		t.Fatalf("request carries Authorization %q after a failed refresh", auth)
+	}
+}
+
 const emptySHA = "e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855"

internal/imdscreds/imdscreds.go

@@ -21,31 +21,18 @@ import (
 	"github.com/aws/aws-sdk-go-v2/feature/ec2/imds"
 )
 
-// Credentials is a resolved instance-role credential set with parsed expiry.
-type Credentials struct {
-	AccessKeyID     string
-	SecretAccessKey string
-	SessionToken    string
-	Expiration      time.Time
-}
-
 // Fetch resolves the instance role then its credentials via the SDK's IMDSv2
 // client and the ec2rolecreds provider. base is the IMDS root, e.g.
 // http://169.254.169.254/latest.
-func Fetch(client *http.Client, base string) (Credentials, error) {
+func Fetch(client *http.Client, base string) (aws.Credentials, error) {
 	creds, err := NewProvider(client, base).Retrieve(context.Background())
 	if err != nil {
-		return Credentials{}, fmt.Errorf("fetch IMDS credentials: %w", err)
+		return aws.Credentials{}, fmt.Errorf("fetch IMDS credentials: %w", err)
 	}
 	if creds.AccessKeyID == "" || creds.SecretAccessKey == "" {
-		return Credentials{}, fmt.Errorf("IMDS credentials missing access/secret key")
+		return aws.Credentials{}, fmt.Errorf("IMDS credentials missing access/secret key")
 	}
-	return Credentials{
-		AccessKeyID:     creds.AccessKeyID,
-		SecretAccessKey: creds.SecretAccessKey,
-		SessionToken:    creds.SessionToken,
-		Expiration:      creds.Expires,
-	}, nil
+	return creds, nil
 }
 
 // NewProvider builds an ec2rolecreds provider against base, an IMDS root URL

internal/imdscreds/imdscreds_test.go

@@ -1,9 +1,11 @@
 package imdscreds
 
 import (
+	"context"
 	"encoding/json"
 	"net/http"
 	"net/http/httptest"
+	"sync/atomic"
 	"testing"
 	"time"
 
@@ -60,8 +62,8 @@ func TestFetch_OK(t *testing.T) {
 	if got.AccessKeyID != "AKIATEST" || got.SecretAccessKey != "secrettest" || got.SessionToken != "sessiontoken" {
 		t.Errorf("creds mismatch: %+v", got)
 	}
-	if !got.Expiration.Equal(exp) {
-		t.Errorf("Expiration = %v, want %v", got.Expiration, exp)
+	if !got.Expires.Equal(exp) {
+		t.Errorf("Expires = %v, want %v", got.Expires, exp)
 	}
 }
 
@@ -132,3 +134,46 @@ func TestNewProvider_WithholdsExtendOnFailure(t *testing.T) {
 		t.Error("provider dropped AdjustExpiresBy, which a caller using an ExpiryWindow relies on")
 	}
 }
+
+// Behind aws.CredentialsCache, as the ecs-agent uses it, a refresh that fails
+// against an expired credential must surface as an error rather than hand back
+// the dead key with a pushed-out expiry.
+func TestNewProvider_CachedFailedRefreshErrors(t *testing.T) {
+	var hits atomic.Int32
+	mux := http.NewServeMux()
+	mux.HandleFunc("/latest/api/token", func(w http.ResponseWriter, r *http.Request) {
+		w.Header().Set("X-Aws-Ec2-Metadata-Token-Ttl-Seconds", "21600")
+		_, _ = w.Write([]byte("v2-token"))
+	})
+	mux.HandleFunc("/latest/meta-data/iam/security-credentials/", func(w http.ResponseWriter, r *http.Request) {
+		if r.URL.Path == "/latest/meta-data/iam/security-credentials/" {
+			_, _ = w.Write([]byte("node-role"))
+			return
+		}
+		if hits.Add(1) > 1 {
+			w.WriteHeader(http.StatusInternalServerError)
+			return
+		}
+		_ = json.NewEncoder(w).Encode(map[string]string{
+			"Code":            "Success",
+			"AccessKeyId":     "AKIA",
+			"SecretAccessKey": "secret",
+			"Expiration":      time.Now().Add(-time.Minute).UTC().Format(time.RFC3339),
+		})
+	})
+	srv := httptest.NewServer(mux)
+	t.Cleanup(srv.Close)
+
+	cache := aws.NewCredentialsCache(NewProvider(srv.Client(), srv.URL+"/latest"))
+	first, err := cache.Retrieve(context.Background())
+	if err != nil {
+		t.Fatalf("first Retrieve: %v", err)
+	}
+	got, err := cache.Retrieve(context.Background())
+	if err == nil {
+		t.Fatalf("refresh failure was reported as success: %+v", got)
+	}
+	if got.Expires.After(first.Expires) {
+		t.Errorf("expiry extended on refresh failure: %v -> %v", first.Expires, got.Expires)
+	}
+}

internal/stsauth/stsauth.go

@@ -8,8 +8,8 @@ package stsauth
 import (
 	"fmt"
 	"net/http"
-	"time"
 
+	awsv2 "github.com/aws/aws-sdk-go-v2/aws"
 	"github.com/aws/aws-sdk-go/aws"
 	awscreds "github.com/aws/aws-sdk-go/aws/credentials"
 	"github.com/aws/aws-sdk-go/aws/session"
@@ -20,24 +20,15 @@ import (
 // scope and the gateway does not enforce a specific value.
 const defaultRegion = "us-east-1"
 
-// Credentials is the decoded result of AssumeRole: a SigV4 credential set bound
-// to the assumed role, with its expiry.
-type Credentials struct {
-	AccessKeyID     string
-	SecretAccessKey string
-	SessionToken    string
-	Expiration      time.Time
-}
-
 // AssumeRole builds an aws-sdk STS client pointed at the gateway and calls
 // AssumeRole with the supplied static (instance-role) credentials, returning the
 // assumed-role credential set. httpClient must trust the gateway CA.
-func AssumeRole(region, gatewayURL string, httpClient *http.Client, akid, secret, sessionToken, roleARN, sessionName string) (Credentials, error) {
+func AssumeRole(region, gatewayURL string, httpClient *http.Client, akid, secret, sessionToken, roleARN, sessionName string) (awsv2.Credentials, error) {
 	if region == "" {
 		region = defaultRegion
 	}
 	if roleARN == "" || sessionName == "" {
-		return Credentials{}, fmt.Errorf("stsauth: roleARN and sessionName are required")
+		return awsv2.Credentials{}, fmt.Errorf("stsauth: roleARN and sessionName are required")
 	}
 	sess, err := session.NewSession(&aws.Config{
 		Region:      aws.String(region),
@@ -47,7 +38,7 @@ func AssumeRole(region, gatewayURL string, httpClient *http.Client, akid, secret
 		DisableSSL:  aws.Bool(false),
 	})
 	if err != nil {
-		return Credentials{}, fmt.Errorf("new session: %w", err)
+		return awsv2.Credentials{}, fmt.Errorf("new session: %w", err)
 	}
 
 	out, err := sts.New(sess).AssumeRole(&sts.AssumeRoleInput{
@@ -55,16 +46,17 @@ func AssumeRole(region, gatewayURL string, httpClient *http.Client, akid, secret
 		RoleSessionName: aws.String(sessionName),
 	})
 	if err != nil {
-		return Credentials{}, fmt.Errorf("AssumeRole: %w", err)
+		return awsv2.Credentials{}, fmt.Errorf("AssumeRole: %w", err)
 	}
 	if out.Credentials == nil {
-		return Credentials{}, fmt.Errorf("AssumeRole returned no credentials")
+		return awsv2.Credentials{}, fmt.Errorf("AssumeRole returned no credentials")
 	}
 	c := out.Credentials
-	return Credentials{
+	return awsv2.Credentials{
 		AccessKeyID:     aws.StringValue(c.AccessKeyId),
 		SecretAccessKey: aws.StringValue(c.SecretAccessKey),
 		SessionToken:    aws.StringValue(c.SessionToken),
-		Expiration:      aws.TimeValue(c.Expiration),
+		CanExpire:       c.Expiration != nil,
+		Expires:         aws.TimeValue(c.Expiration),
 	}, nil
 }

internal/stsauth/stsauth_test.go

@@ -41,8 +41,8 @@ func TestAssumeRole_DecodesCredentials(t *testing.T) {
 		t.Errorf("bad creds: %+v", creds)
 	}
 	want, _ := time.Parse(time.RFC3339, "2031-01-02T03:04:05Z")
-	if !creds.Expiration.Equal(want) {
-		t.Errorf("Expiration = %v, want %v", creds.Expiration, want)
+	if !creds.Expires.Equal(want) {
+		t.Errorf("Expires = %v, want %v", creds.Expires, want)
 	}
 	// The request must carry the AssumeRole action and the role ARN we asked for.
 	if !strings.Contains(gotBody, "Action=AssumeRole") {

internal/testkit/awserror.go

@@ -0,0 +1,29 @@
+package testutil
+
+import (
+	"testing"
+
+	awserrors "github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+// RequireAWSError asserts err carries the AWS code and exact client-facing
+// message; an empty message asserts the bare code with no message attached.
+func RequireAWSError(t *testing.T, err error, code, message string) {
+	t.Helper()
+	require.Error(t, err)
+	got, msg, ok := awserrors.ResolveErrorDetail(err)
+	require.True(t, ok, "error %v carries no AWS code", err)
+	assert.Equal(t, code, got)
+	assert.Equal(t, message, msg)
+}
+
+// RequireAWSCode asserts err carries the AWS code, whatever its message.
+func RequireAWSCode(t *testing.T, err error, code string) {
+	t.Helper()
+	require.Error(t, err)
+	got, ok := awserrors.ResolveErrorCode(err)
+	require.True(t, ok, "error %v carries no AWS code", err)
+	assert.Equal(t, code, got)
+}

internal/testkit/drainsocket.go

@@ -5,6 +5,7 @@ import (
 	"os"
 	"path/filepath"
 	"sync"
+	"sync/atomic"
 	"testing"
 
 	"github.com/stretchr/testify/require"
@@ -28,10 +29,11 @@ func SocketTempDir(t *testing.T) string {
 // StartDrainSocket stands in for the NBD plugin's drain socket, at the path the
 // plugin serves it on the node hosting volumeID ({dataDir}/viperblock/{volumeID}
 // /snapshot.sock). Every connection is answered with ack ("OK\n" for a drain
-// that reached S3, "ERR\n" for one that did not) and closed.
-func StartDrainSocket(t *testing.T, dataDir, volumeID, ack string) {
+// that reached S3, "ERR\n" for one that did not) and closed. connections
+// reports how many drains have been asked for so far.
+func StartDrainSocket(t *testing.T, dataDir, volumeID, ack string) (connections func() int) {
 	t.Helper()
-	serveDrainSocket(t, dataDir, volumeID, ack, nil, nil)
+	return serveDrainSocket(t, dataDir, volumeID, ack, nil, nil)
 }
 
 // StartBlockingDrainSocket stands in for a plugin whose flush is still running:
@@ -43,7 +45,7 @@ func StartBlockingDrainSocket(t *testing.T, dataDir, volumeID, ack string) (acce
 
 	acceptedCh := make(chan struct{})
 	releaseCh := make(chan struct{})
-	serveDrainSocket(t, dataDir, volumeID, ack, acceptedCh, releaseCh)
+	_ = serveDrainSocket(t, dataDir, volumeID, ack, acceptedCh, releaseCh)
 
 	var once sync.Once
 	t.Cleanup(func() { once.Do(func() { close(releaseCh) }) })
@@ -52,8 +54,9 @@ func StartBlockingDrainSocket(t *testing.T, dataDir, volumeID, ack string) (acce
 }
 
 // serveDrainSocket answers every connection with ack, optionally closing
-// accepted on the first one and waiting for release before it replies.
-func serveDrainSocket(t *testing.T, dataDir, volumeID, ack string, accepted chan struct{}, release <-chan struct{}) {
+// accepted on the first one and waiting for release before it replies. The
+// returned func counts the connections accepted so far.
+func serveDrainSocket(t *testing.T, dataDir, volumeID, ack string, accepted chan struct{}, release <-chan struct{}) func() int {
 	t.Helper()
 
 	dir := filepath.Join(dataDir, "viperblock", volumeID)
@@ -63,6 +66,7 @@ func serveDrainSocket(t *testing.T, dataDir, volumeID, ack string, accepted chan
 	require.NoError(t, err)
 	t.Cleanup(func() { _ = ln.Close() })
 
+	var count atomic.Int64
 	go func() {
 		var once sync.Once
 		for {
@@ -70,6 +74,7 @@ func serveDrainSocket(t *testing.T, dataDir, volumeID, ack string, accepted chan
 			if err != nil {
 				return // listener closed by the cleanup above
 			}
+			count.Add(1)
 			if accepted != nil {
 				once.Do(func() { close(accepted) })
 			}
@@ -80,4 +85,5 @@ func serveDrainSocket(t *testing.T, dataDir, volumeID, ack string, accepted chan
 			_ = conn.Close()
 		}
 	}()
+	return func() int { return int(count.Load()) }
 }

internal/testkit/nats.go

@@ -167,11 +167,11 @@ func registerVpcdStub(t *testing.T, nc *nats.Conn, defaultPayload []byte) {
 	})
 }
 
-// SeedKV creates a KV bucket and populates it with the given entries.
+// SeedKV creates or updates a KV bucket and populates it with the given entries.
 // Returns the KV handle for further use.
 func SeedKV(t *testing.T, js jetstream.JetStream, bucket string, entries map[string][]byte) jetstream.KeyValue {
 	t.Helper()
-	kv, err := js.CreateKeyValue(t.Context(), jetstream.KeyValueConfig{Bucket: bucket, History: 1})
+	kv, err := js.CreateOrUpdateKeyValue(t.Context(), jetstream.KeyValueConfig{Bucket: bucket, History: 1})
 	require.NoError(t, err)
 	for k, v := range entries {
 		_, err := kv.Put(t.Context(), k, v)

internal/testkit/nats_test.go

@@ -0,0 +1,19 @@
+package testutil
+
+import (
+	"testing"
+
+	"github.com/nats-io/nats.go/jetstream"
+	"github.com/stretchr/testify/require"
+)
+
+func TestSeedKV_ExistingBucket(t *testing.T) {
+	_, _, js := StartTestJetStream(t)
+	_, err := js.CreateKeyValue(t.Context(), jetstream.KeyValueConfig{Bucket: "seeded", History: 5})
+	require.NoError(t, err)
+
+	kv := SeedKV(t, js, "seeded", map[string][]byte{"k": []byte("v")})
+	entry, err := kv.Get(t.Context(), "k")
+	require.NoError(t, err)
+	require.Equal(t, []byte("v"), entry.Value())
+}

scripts/images/ecs-agent-gpu/ecs-agent.service

@@ -9,8 +9,7 @@ Wants=network-online.target
 
 [Service]
 Type=simple
-# Static config (gateway URL, CA, region, cluster, seeded IAM creds,
-# containerd socket) comes from /etc/spinifex-ecs/agent.env, dropped by
+# Static config (gateway URL, CA, region, cluster, containerd socket) comes from /etc/spinifex-ecs/agent.env, dropped by
 # cloud-init at launch. Absent file is tolerated — the agent falls back to
 # its built-in defaults.
 EnvironmentFile=-/etc/spinifex-ecs/agent.env

scripts/images/ecs-agent/ecs-agent.initd

@@ -3,8 +3,7 @@
 # IMDSv2, registers + heartbeats + polls task assignments through the AWS gateway
 # over TLS+SigV4 (no NATS), and pulls task images through containerd.
 #
-# Static config (gateway URL, CA, region, cluster, seeded IAM creds, containerd
-# socket) comes from /etc/spinifex-ecs/agent.env, dropped by cloud-init at launch.
+# Static config (gateway URL, CA, region, cluster, containerd socket) comes from /etc/spinifex-ecs/agent.env, dropped by cloud-init at launch.
 
 description="Spinifex ECS container-instance agent"
 command="/usr/local/bin/ecs-agent"
@@ -28,7 +27,7 @@ start_pre() {
         # shellcheck disable=SC1091
         . /etc/spinifex-ecs/agent.env
         export ECS_GATEWAY_URL ECS_GATEWAY_CA ECS_REGION ECS_CLUSTER \
-            ECS_ACCESS_KEY ECS_SECRET_KEY ECS_IMDS_BASE ECS_CONTAINERD_SOCKET \
+            ECS_IMDS_BASE ECS_CONTAINERD_SOCKET \
             ECS_HEARTBEAT_INTERVAL ECS_POLL_INTERVAL
     fi
 }

scripts/images/eks-node/eks-node-role.sh

@@ -159,6 +159,7 @@ case "${ROLE}" in
         svc_start mulga-eks-state-report
         svc_start mulga-eks-addon-sync
         svc_start mulga-eks-node-gc
+        svc_optional mulga-eks-log-forward
         svc_optional eks-pod-identity-webhook
         ;;
     server-join)
@@ -178,12 +179,14 @@ case "${ROLE}" in
         svc_start konnectivity-server
         svc_start mulga-eks-state-report
         svc_start mulga-eks-node-gc
+        svc_optional mulga-eks-log-forward
         svc_optional eks-pod-identity-webhook
         ;;
     agent)
         log "configuring agent role"
         svc_enable k3s-agent
         svc_start k3s-agent
+        svc_optional mulga-eks-log-forward
         ;;
     *)
         die "SPINIFEX_K3S_ROLE missing or invalid: '${ROLE}'"

scripts/images/eks-node/eks-node-role_test.sh

@@ -146,6 +146,15 @@ for INIT in openrc systemd; do
     )"
     check "server(${INIT}): enables node-gc" "yes" "$(present "${CALLS}" "$(enable_cmd "${INIT}" mulga-eks-node-gc)")"
     check "server(${INIT}): starts node-gc" "yes" "$(present "${CALLS}" "$(start_cmd "${INIT}" mulga-eks-node-gc)")"
+    check "server(${INIT}): enables log-forward" "yes" "$(present "${CALLS}" "$(enable_cmd "${INIT}" mulga-eks-log-forward)")"
+    check "server(${INIT}): starts log-forward" "yes" "$(present "${CALLS}" "$(start_cmd "${INIT}" mulga-eks-log-forward)")"
+    check "server(${INIT}): starts log-forward before the webhook" "yes" "$(
+        awk -v f="$(start_cmd "${INIT}" mulga-eks-log-forward)" -v w="$(start_cmd "${INIT}" eks-pod-identity-webhook)" '
+            index($0, f) == 1 { fn = NR }
+            index($0, w) == 1 { wn = NR }
+            END { print (fn > 0 && wn > fn) ? "yes" : "no" }
+        ' "${CALLS}"
+    )"
     check "server(${INIT}): no k3s-agent" "no" "$(present "${CALLS}" "$(enable_cmd "${INIT}" k3s-agent)")"
     check "server(${INIT}): self-disables" "yes" "$(present "${CALLS}" "$(disable_cmd "${INIT}" eks-node-role)")"
     check "server(${INIT}): pod-identity-webhook is the last call" "$(start_cmd "${INIT}" eks-pod-identity-webhook)" "$(grep -v "$(disable_cmd "${INIT}" eks-node-role)" "${CALLS}" | tail -n 1)"
@@ -196,6 +205,15 @@ for INIT in openrc systemd; do
     # published the cluster-identical artifacts, and a join re-publish would
     # only race the bootstrap bus. This is deliberate, not an oversight.
     check "server-join(${INIT}): pod-identity-webhook is the last call" "$(start_cmd "${INIT}" eks-pod-identity-webhook)" "$(grep -v "$(disable_cmd "${INIT}" eks-node-role)" "${CALLS}" | tail -n 1)"
+    check "server-join(${INIT}): enables log-forward" "yes" "$(present "${CALLS}" "$(enable_cmd "${INIT}" mulga-eks-log-forward)")"
+    check "server-join(${INIT}): starts log-forward" "yes" "$(present "${CALLS}" "$(start_cmd "${INIT}" mulga-eks-log-forward)")"
+    check "server-join(${INIT}): starts log-forward before the webhook" "yes" "$(
+        awk -v f="$(start_cmd "${INIT}" mulga-eks-log-forward)" -v w="$(start_cmd "${INIT}" eks-pod-identity-webhook)" '
+            index($0, f) == 1 { fn = NR }
+            index($0, w) == 1 { wn = NR }
+            END { print (fn > 0 && wn > fn) ? "yes" : "no" }
+        ' "${CALLS}"
+    )"
     check "server-join(${INIT}): no k3s-first-boot" "yes" "$(absent_all "${CALLS}" "k3s-first-boot")"
     check "server-join(${INIT}): no k3s-agent" "yes" "$(absent_all "${CALLS}" "$(enable_cmd "${INIT}" k3s-agent)")"
     check "server-join(${INIT}): self-disables" "yes" "$(present "${CALLS}" "$(disable_cmd "${INIT}" eks-node-role)")"
@@ -207,8 +225,39 @@ for INIT in openrc systemd; do
     check "agent(${INIT}): exit 0" 0 "${rc}"
     check "agent(${INIT}): role file" "agent" "$(cat "${ROLE_FILE}")"
     check "agent(${INIT}): enables k3s-agent" "yes" "$(present "${CALLS}" "$(enable_cmd "${INIT}" k3s-agent)")"
+    check "agent(${INIT}): enables log-forward" "yes" "$(present "${CALLS}" "$(enable_cmd "${INIT}" mulga-eks-log-forward)")"
+    check "agent(${INIT}): starts log-forward" "yes" "$(present "${CALLS}" "$(start_cmd "${INIT}" mulga-eks-log-forward)")"
+    check "agent(${INIT}): log-forward is enabled after k3s-agent" "yes" "$(
+        awk -v k="$(enable_cmd "${INIT}" k3s-agent)" -v f="$(enable_cmd "${INIT}" mulga-eks-log-forward)" '
+            index($0, k) == 1 { kn = NR }
+            index($0, f) == 1 { fn = NR }
+            END { print (kn > 0 && fn > kn) ? "yes" : "no" }
+        ' "${CALLS}"
+    )"
+
     check "agent(${INIT}): no server services" "yes" "$(absent_all "${CALLS}" "eks-token-webhook" "eks-pod-identity-webhook" "k3s-first-boot" "mulga-eks-node-gc" "$(enable_cmd "${INIT}" k3s)")"
 
+    # A broken log forwarder must never stop a node joining.
+    for FAILING in enable start; do
+        run_case "${INIT}" agent_logfwd_${FAILING}_fail
+        printf 'SPINIFEX_K3S_ROLE=agent\n' > "${EKS_NODE_ENVFILE}"
+        if [ "${FAILING}" = enable ]; then
+            STUB_FAIL="$(enable_cmd "${INIT}" mulga-eks-log-forward)"
+            export STUB_FAIL
+            STUB_FAIL="${STUB_FAIL#rc-update add }"; STUB_FAIL="${STUB_FAIL#systemctl enable }"
+        else
+            STUB_FAIL="$(start_cmd "${INIT}" mulga-eks-log-forward)"
+            export STUB_FAIL
+            STUB_FAIL="${STUB_FAIL#rc-service }"; STUB_FAIL="${STUB_FAIL#systemctl start --no-block }"
+        fi
+        invoke
+        unset STUB_FAIL
+        check "agent(${INIT}): log-forward ${FAILING} failure still exits 0" 0 "${rc}"
+        check "agent(${INIT}): log-forward ${FAILING} failure still enables k3s-agent" "yes" "$(present "${CALLS}" "$(enable_cmd "${INIT}" k3s-agent)")"
+        check "agent(${INIT}): log-forward ${FAILING} failure still starts k3s-agent" "yes" "$(present "${CALLS}" "$(start_cmd "${INIT}" k3s-agent)")"
+        check "agent(${INIT}): log-forward ${FAILING} failure still self-disables" "yes" "$(present "${CALLS}" "$(disable_cmd "${INIT}" eks-node-role)")"
+    done
+
     # --- agent (inferred from agent.env, no explicit role) ---
     run_case "${INIT}" agent_inferred
     printf 'K3S_URL=https://nlb:443\nK3S_TOKEN=abc\n' > "${EKS_NODE_AGENT_ENVFILE}"

scripts/images/eks-node/manifest.conf

@@ -45,6 +45,8 @@ scripts/images/eks-node/mulga-eks-addon-sync.initd:/etc/init.d/mulga-eks-addon-s
 scripts/images/eks-node/mulga-eks-addon-sync.sh:/usr/local/sbin/mulga-eks-addon-sync \
 scripts/images/eks-node/mulga-eks-node-gc.initd:/etc/init.d/mulga-eks-node-gc \
 scripts/images/eks-node/mulga-eks-node-gc.sh:/usr/local/sbin/mulga-eks-node-gc \
+scripts/images/eks-node/mulga-eks-log-forward.initd:/etc/init.d/mulga-eks-log-forward \
+scripts/images/eks-node/mulga-eks-log-forward.sh:/usr/local/sbin/mulga-eks-log-forward \
 scripts/images/eks-node/konnectivity-server.initd:/etc/init.d/konnectivity-server \
 scripts/images/eks-node/konnectivity-server-prestart.sh:/usr/local/sbin/konnectivity-server-prestart \
 scripts/images/eks-node/konnectivity/konnectivity-agent.yaml:/usr/share/spinifex-eks/konnectivity/konnectivity-agent.yaml \

scripts/images/eks-node/mulga-eks-log-forward.initd

@@ -0,0 +1,30 @@
+#!/sbin/openrc-run
+# mulga-eks-log-forward — copies k3s warnings and errors to the serial console
+# so they reach the Elasticsearch sink. Best effort: nothing depends on it and
+# a failure here never affects the node.
+
+description="Forward k3s warnings and errors to the serial console"
+command="/bin/sh"
+command_background="yes"
+supervisor="supervise-daemon"
+respawn_delay=5
+pidfile="/run/mulga-eks-log-forward.pid"
+output_log="/dev/null"
+error_log="/dev/null"
+
+depend() {
+    after k3s k3s-agent
+}
+
+start_pre() {
+    role=$(cat /etc/spinifex-eks/role 2>/dev/null || true)
+    if [ "${role}" = "agent" ]; then
+        log=/var/log/k3s-agent.log
+        FORWARD_PREFIX="[k3s-agent]"
+    else
+        log=/var/log/k3s.log
+        FORWARD_PREFIX="[k3s]"
+    fi
+    export FORWARD_PREFIX
+    command_args="-c 'tail -n +1 -F ${log} 2>/dev/null | /usr/local/sbin/mulga-eks-log-forward'"
+}

scripts/images/eks-node/mulga-eks-log-forward.sh

@@ -0,0 +1,90 @@
+#!/bin/sh
+# mulga-eks-log-forward — copy the warnings and errors of a k3s log to the
+# serial console, where the host picks them up for the Elasticsearch sink.
+#
+# Reads the tailed log on stdin. Lines that matter are redacted first, then
+# cut to FORWARD_MAX_BYTES, then rate limited: at most FORWARD_RATE_PER_MIN a
+# minute reach the console and the rest are reported as one "N lines dropped"
+# line. Every console write has a timeout, so a stuck console loses lines
+# instead of stalling the tail or k3s.
+#
+# Env: FORWARD_CONSOLE (default /dev/console), FORWARD_PREFIX, FORWARD_MAX_BYTES
+# (512), FORWARD_RATE_PER_MIN (60), FORWARD_WRITE_TIMEOUT seconds (2) and
+# FORWARD_CLOCK_CMD (default: seconds of uptime, read without forking).
+
+CONSOLE=${FORWARD_CONSOLE:-/dev/console}
+PREFIX=${FORWARD_PREFIX:-[k3s]}
+MAX_BYTES=${FORWARD_MAX_BYTES:-512}
+RATE=${FORWARD_RATE_PER_MIN:-60}
+WRITE_TIMEOUT=${FORWARD_WRITE_TIMEOUT:-2}
+CLOCK_CMD=${FORWARD_CLOCK_CMD:-}
+
+now() {
+    if [ -n "${CLOCK_CMD}" ]; then
+        sh -c "${CLOCK_CMD}"
+    else
+        read -r up _ < /proc/uptime && echo "${up%%.*}"
+    fi
+}
+
+# emit TEXT: one console line, abandoned after WRITE_TIMEOUT. Returns non-zero
+# when the console did not take it.
+emit() {
+    timeout "${WRITE_TIMEOUT}" sh -c 'printf "%s\n" "$1" >> "$2"' sh "$1" "${CONSOLE}" 2>/dev/null
+}
+
+# Filter, redact and truncate in one awk pass so lines that do not matter cost
+# no fork. Redaction runs before the cut so a secret straddling the limit
+# cannot leave a readable prefix behind.
+select_lines() {
+    LC_ALL=C awk -v max="${MAX_BYTES}" '
+    {
+        line = $0
+        gsub(/\r/, "", line)
+        keep = 0
+        if (line ~ /level=(warning|error)/) keep = 1
+        else if (line ~ /^[WE][0-9][0-9][0-9][0-9] /) keep = 1
+        else if (line ~ /NodeNotReady|PLEG|[Ll]ease|[Nn]ode status/) keep = 1
+        if (!keep) next
+
+        gsub(/[Bb]earer +[A-Za-z0-9._~+\/=-]+/, "Bearer [REDACTED]", line)
+        gsub(/K10[0-9a-f]+(::[A-Za-z0-9-]+:[^ "\047]+)?/, "[REDACTED]", line)
+        gsub(/eyJ[A-Za-z0-9_.-]+/, "[REDACTED]", line)
+        gsub(/(AKIA|ASIA)[A-Z0-9]+/, "[REDACTED]", line)
+        gsub(/X-Amz-Signature=[^& "\047]*/, "X-Amz-Signature=[REDACTED]", line)
+        gsub(/X-Amz-Credential=[^& "\047]*/, "X-Amz-Credential=[REDACTED]", line)
+        gsub(/X-Amz-Security-Token=[^& "\047]*/, "X-Amz-Security-Token=[REDACTED]", line)
+        gsub(/[^[:print:]\t]/, " ", line)
+        print substr(line, 1, max)
+    }'
+}
+
+win_start=0
+sent=0
+dropped=0
+
+flush_dropped() {
+    [ "${dropped}" -gt 0 ] || return 0
+    emit "${PREFIX} ${dropped} lines dropped"
+    dropped=0
+}
+
+select_lines | {
+    while IFS= read -r line; do
+        t=$(now) || t=0
+        if [ "${win_start}" -eq 0 ] || [ $((t - win_start)) -ge 60 ]; then
+            flush_dropped
+            win_start=${t}
+            sent=0
+            [ "${win_start}" -eq 0 ] && win_start=1
+        fi
+        if [ "${sent}" -ge "${RATE}" ]; then
+            dropped=$((dropped + 1))
+            continue
+        fi
+        sent=$((sent + 1))
+        emit "${PREFIX} ${line}" || dropped=$((dropped + 1))
+    done
+    flush_dropped
+}
+exit 0

scripts/images/eks-node/mulga-eks-log-forward_test.sh

@@ -0,0 +1,190 @@
+#!/bin/sh
+# Self-contained POSIX test for mulga-eks-log-forward.sh: the line filter,
+# redaction, truncation, rate cap and blocked-console behaviour. The script
+# reads the tailed log on stdin and writes to FORWARD_CONSOLE, which the test
+# points at a temp file (or a reader-less FIFO), so no root or tty is needed.
+#
+# Run: sh scripts/images/eks-node/mulga-eks-log-forward_test.sh
+set -eu
+
+SCRIPT_DIR=$(cd "$(dirname "$0")" && pwd)
+SCRIPT="${SCRIPT_DIR}/mulga-eks-log-forward.sh"
+WORK=$(mktemp -d)
+trap 'rm -rf "${WORK}"' EXIT
+
+PASS=0
+FAIL=0
+check() { # check <desc> <expected> <actual>
+    if [ "$2" = "$3" ]; then
+        PASS=$((PASS + 1))
+    else
+        FAIL=$((FAIL + 1))
+        echo "FAIL: $1"
+        echo "  expected: [$2]"
+        echo "  actual:   [$3]"
+    fi
+}
+
+# forward <name>: feeds stdin to the script with a per-case console file and a
+# fixed clock, leaving the forwarded text in ${OUT}.
+forward() {
+    CASE="${WORK}/$1"
+    mkdir -p "${CASE}"
+    OUT="${CASE}/console"
+    : > "${OUT}"
+    FORWARD_CONSOLE="${OUT}" FORWARD_PREFIX="[k3s-agent]" \
+        FORWARD_CLOCK_CMD="echo 1000" sh "${SCRIPT}" 2>/dev/null || true
+}
+forwarded() { grep -c . "${OUT}" || true; }
+
+# --- filter: lines that must pass ---
+forward filter_pass <<'EOF'
+time="2026-11-03T03:00:00Z" level=warning msg="slow response from apiserver"
+time="2026-11-03T03:00:01Z" level=error msg="failed to renew"
+W1103 03:00:02.123456    1234 reflector.go:561] watch of *v1.Pod ended
+E1103 03:00:03.123456    1234 kubelet.go:2400] node not found
+I1103 03:00:04.123456    1234 kubelet_node_status.go:73] Node status update failed
+I1103 03:00:05.123456    1234 controller.go:193] Failed to update lease
+I1103 03:00:06.123456    1234 kubelet.go:2200] PLEG is not healthy
+time="2026-11-03T03:00:07Z" level=info msg="Event: NodeNotReady kubelet"
+EOF
+check "filter: all eight interesting lines pass" "8" "$(forwarded)"
+check "filter: lines carry the prefix" "8" "$(grep -c '^\[k3s-agent\] ' "${OUT}" || true)"
+
+# --- filter: lines that must be dropped ---
+forward filter_drop <<'EOF'
+time="2026-11-03T03:00:00Z" level=info msg="Starting k3s agent"
+time="2026-11-03T03:00:01Z" level=debug msg="tick"
+I1103 03:00:02.123456    1234 server.go:100] Version: v1.32.5
+I1103 03:00:03.123456    1234 kubelet.go:1000] SyncLoop (UPDATE, "api")
+EOF
+check "filter: info and debug lines are dropped" "0" "$(forwarded)"
+
+# --- redaction: one case per secret shape ---
+redact() { # redact <name> <line> <secret-that-must-not-appear>
+    forward "redact_$1" <<EOF
+time="2026-11-03T03:00:00Z" level=error msg="$2"
+EOF
+    check "redact(${1}): line still forwarded" "1" "$(forwarded)"
+    if grep -qF "$3" "${OUT}"; then
+        check "redact(${1}): secret removed" "absent" "present"
+    else
+        check "redact(${1}): secret removed" "absent" "absent"
+    fi
+    check "redact(${1}): marker present" "1" "$(grep -c 'REDACTED' "${OUT}" || true)"
+}
+redact bearer "request failed Authorization: Bearer abc123SECRETvalue456" "abc123SECRETvalue456"
+redact k3s_token "join failed token K1009f3c2d4e5a6b7c8d9e0f1a2b3c4d5e6f7a8b9c0d1e2f3a4b5c6d7e8f9a0b1c2::server:s3cr3tPassw0rd" "s3cr3tPassw0rd"
+redact jwt "bad token eyJhbGciOiJIUzI1NiJ9.eyJzdWIiOiIxMjM0NTY3ODkwIn0.c2lnbmF0dXJlLXNlY3JldA" "eyJhbGciOiJIUzI1NiJ9"
+redact akia "credentials AKIAIOSFODNN7EXAMPLE rejected" "AKIAIOSFODNN7EXAMPLE"
+redact asia "credentials ASIAIOSFODNN7EXAMPLE rejected" "ASIAIOSFODNN7EXAMPLE"
+redact amz_signature "GET /obj?X-Amz-Signature=0123456789abcdef0123456789abcdef&X-Amz-Expires=60 failed" "0123456789abcdef0123456789abcdef"
+redact amz_credential "GET /obj?X-Amz-Credential=ZZZsecretscope%2F20261103%2Fap%2Fs3%2Faws4_request failed" "ZZZsecretscope"
+
+# --- truncation: text over 512 bytes is cut ---
+LONG=$(awk 'BEGIN { for (i = 0; i < 2000; i++) printf "a"; }')
+forward truncate <<EOF
+time="2026-11-03T03:00:00Z" level=error msg="${LONG}"
+EOF
+check "truncate: still forwarded" "1" "$(forwarded)"
+BODY=$(sed 's/^\[k3s-agent\] //' "${OUT}" | head -n 1)
+check "truncate: body is exactly 512 bytes" "512" "$(printf '%s' "${BODY}" | wc -c | tr -d ' ')"
+check "truncate: keeps the start of the line" "time=\"2026-11-03T03:00:00Z\"" "$(printf '%s' "${BODY}" | cut -c1-27)"
+
+# --- redaction runs before the cut: a secret over the limit leaves no prefix ---
+# The pad puts the 512-byte cut right after the bare marker (AKIA, eyJ, K10).
+# A cut fragment that long matches no redaction pattern, so redacting after the
+# cut would leave it in the output.
+HEAD='time="2026-11-03T03:00:00Z" level=error msg="'
+straddle() { # straddle <name> <secret> <bytes of the secret kept by the cut>
+    PAD=$(awk -v n=$((512 - ${#HEAD} - 1 - $3)) 'BEGIN { for (i = 0; i < n; i++) printf "x"; }')
+    forward "straddle_$1" <<EOF
+${HEAD}${PAD} $2 tail"
+EOF
+    check "straddle(${1}): line forwarded" "1" "$(forwarded)"
+    check "straddle(${1}): cut keeps the marker length" "$3" "$(sed 's/^\[k3s-agent\] //' "${OUT}" | head -n 1 | awk -v p="${PAD} " '{ print length($0) - index($0, p) - length(p) + 1 }')"
+    if grep -qF "$(printf '%s' "$2" | cut -c1-"$3")" "${OUT}"; then
+        check "straddle(${1}): no fragment survives the cut" "absent" "present"
+    else
+        check "straddle(${1}): no fragment survives the cut" "absent" "absent"
+    fi
+}
+straddle jwt "eyJhbGciOiJIUzI1NiJ9.eyJzdWIiOiIxMjM0In0.c2ln" 3
+straddle akia "AKIAIOSFODNN7EXAMPLE" 4
+straddle k3s_token "K1009f3c2d4e5a6b7c8d9e0f1a2b3c4d5e6f7a8b9c0d1e2f3a4b5c6d7e8f9a0b1c2::server:s3cr3tPassw0rd" 3
+
+# --- rate cap: 60 a minute, then one summary of what was dropped ---
+gen() { # gen <count>
+    awk -v n="$1" 'BEGIN { for (i = 0; i < n; i++) printf "time=\"t\" level=error msg=\"loop %d\"\n", i }'
+}
+gen 100 > "${WORK}/burst"
+forward rate_cap < "${WORK}/burst"
+check "rate cap: 60 lines forwarded plus one summary" "61" "$(forwarded)"
+check "rate cap: one dropped summary" "1" "$(grep -c '40 lines dropped' "${OUT}" || true)"
+check "rate cap: first line kept" "1" "$(grep -c 'loop 0"' "${OUT}" || true)"
+check "rate cap: line past the cap is not forwarded" "0" "$(grep -c 'loop 60"' "${OUT}" || true)"
+
+# --- rate cap: a new minute starts a new window and reports the old drops ---
+# The clock command advances 60 s every 70 calls, so lines 0-69 share a window
+# (10 dropped) and lines 70-74 land in the next one.
+CASE="${WORK}/rate_window"
+mkdir -p "${CASE}"
+OUT="${CASE}/console"
+: > "${OUT}"
+echo 0 > "${CASE}/calls"
+cat > "${CASE}/clock" <<EOF
+#!/bin/sh
+n=\$(cat "${CASE}/calls")
+echo \$((n + 1)) > "${CASE}/calls"
+echo \$((1000 + n / 70 * 60))
+EOF
+chmod +x "${CASE}/clock"
+gen 75 | FORWARD_CONSOLE="${OUT}" FORWARD_PREFIX="[k3s-agent]" FORWARD_CLOCK_CMD="${CASE}/clock" sh "${SCRIPT}" 2>/dev/null || true
+check "window: 60 + 5 forwarded plus one summary" "66" "$(forwarded)"
+check "window: summary names the 10 dropped" "1" "$(grep -c '10 lines dropped' "${OUT}" || true)"
+check "window: new window forwards its lines" "1" "$(grep -c 'loop 74"' "${OUT}" || true)"
+
+# --- blocked console: lines are dropped, the forwarder neither stalls nor fails ---
+CASE="${WORK}/blocked"
+mkdir -p "${CASE}"
+mkfifo "${CASE}/console"
+gen 3 > "${CASE}/in"
+(
+    FORWARD_CONSOLE="${CASE}/console" FORWARD_PREFIX="[k3s-agent]" FORWARD_CLOCK_CMD="echo 1000" \
+        FORWARD_WRITE_TIMEOUT=1 sh "${SCRIPT}" < "${CASE}/in"
+) &
+PID=$!
+i=0
+while kill -0 "${PID}" 2>/dev/null && [ "${i}" -lt 20 ]; do
+    sleep 1
+    i=$((i + 1))
+done
+if kill -0 "${PID}" 2>/dev/null; then
+    kill "${PID}" 2>/dev/null || true
+    check "blocked console: forwarder finished" "finished" "stalled"
+else
+    check "blocked console: forwarder finished" "finished" "finished"
+fi
+rc=0
+wait "${PID}" || rc=$?
+check "blocked console: exit 0" "0" "${rc}"
+
+# --- the service is wired in, and nothing depends on it ---
+INITD="${SCRIPT_DIR}/mulga-eks-log-forward.initd"
+check "initd ships" "yes" "$([ -f "${INITD}" ] && echo yes || echo no)"
+check "manifest installs the initd" "1" "$(grep -c 'mulga-eks-log-forward.initd:/etc/init.d/mulga-eks-log-forward' "${SCRIPT_DIR}/manifest.conf" || true)"
+check "manifest installs the script" "1" "$(grep -c 'mulga-eks-log-forward.sh:/usr/local/sbin/mulga-eks-log-forward' "${SCRIPT_DIR}/manifest.conf" || true)"
+DEPS=0
+for f in "${SCRIPT_DIR}"/*.initd; do
+    [ "${f}" = "${INITD}" ] && continue
+    if grep -E '^[[:space:]]*(need|use|want)[[:space:]].*mulga-eks-log-forward' "${f}" >/dev/null 2>&1; then
+        DEPS=$((DEPS + 1))
+    fi
+done
+check "nothing depends on the forwarder" "0" "${DEPS}"
+if [ -f "${INITD}" ]; then
+    check "initd does not need anything it could block boot on" "0" "$(grep -cE '^[[:space:]]*need[[:space:]]' "${INITD}" || true)"
+fi
+
+echo "mulga-eks-log-forward_test: ${PASS} passed, ${FAIL} failed"
+[ "${FAIL}" -eq 0 ]

scripts/images/eks-node/setup.sh

@@ -45,10 +45,10 @@ ln -sf /usr/local/bin/k3s /usr/local/bin/ctr
 chmod 0755 /etc/init.d/eks-node-role /etc/init.d/k3s /etc/init.d/k3s-agent \
     /etc/init.d/eks-token-webhook /etc/init.d/k3s-first-boot /etc/init.d/mulga-eks-state-report \
     /etc/init.d/mulga-eks-addon-sync /etc/init.d/mulga-eks-node-gc /etc/init.d/konnectivity-server \
-    /etc/init.d/eks-pod-identity-webhook
+    /etc/init.d/eks-pod-identity-webhook /etc/init.d/mulga-eks-log-forward
 chmod 0755 /usr/local/sbin/eks-node-role /usr/local/sbin/k3s-first-boot \
     /usr/local/sbin/mulga-eks-state-report /usr/local/sbin/mulga-eks-addon-sync \
-    /usr/local/sbin/mulga-eks-node-gc
+    /usr/local/sbin/mulga-eks-node-gc /usr/local/sbin/mulga-eks-log-forward
 # start_pre helpers shared with the systemd wrappers that will eventually
 # replace these runscripts.
 chmod 0755 /usr/local/sbin/k3s-prestart /usr/local/sbin/k3s-agent-prestart \

spinifex/agents/ecs/agent/agent.go

@@ -12,8 +12,9 @@ import (
 	"strings"
 	"time"
 
+	"github.com/aws/aws-sdk-go-v2/aws"
 	ecsv1 "github.com/mulgadc/spinifex/contracts/ecs/v1"
-	"github.com/mulgadc/spinifex/spinifex/agents/ecs/credentials"
+	"github.com/mulgadc/spinifex/internal/imdscreds"
 	ctrruntime "github.com/mulgadc/spinifex/spinifex/agents/ecs/runtime"
 )
 
@@ -102,7 +103,9 @@ func New(cfg Config) (*Agent, error) {
 		AgentVersion: cfg.AgentVersion,
 	}
 
-	creds := credentials.NewIMDSProvider(imdsClient, cfg.IMDSBase)
+	// The cache refetches only once the real expiry passes, and a failed
+	// refresh is an error rather than an extended expiry.
+	creds := aws.NewCredentialsCache(imdscreds.NewProvider(imdsClient, cfg.IMDSBase))
 	resolver := newLazyECRResolver(creds, cfg.Region, cfg.GatewayURL, cfg.GatewayCA)
 
 	// A dial that fails here is not fatal and not final: the node starts the

spinifex/agents/ecs/agent/config.go

@@ -21,17 +21,14 @@ const (
 
 // Config holds the static settings the agent reads at boot. Cluster identity
 // (account ID, instance ID) is discovered at runtime from IMDS, not configured;
-// only the cluster *name* the instance was launched into is static. AccessKey /
-// SecretKey are legacy seeded creds, optional now that the agent SigV4-signs
-// gateway calls with instance-role credentials from IMDS; the agent never holds
-// a NATS token.
+// only the cluster *name* the instance was launched into is static. The agent
+// SigV4-signs gateway calls with instance-role credentials from IMDS and never
+// holds a NATS token.
 type Config struct {
 	GatewayURL       string
 	GatewayCA        string
 	Region           string
 	ClusterName      string
-	AccessKey        string
-	SecretKey        string
 	IMDSBase         string
 	ContainerdSocket string
 	Heartbeat        time.Duration
@@ -53,8 +50,6 @@ func LoadConfig(envFile string) Config {
 		GatewayCA:        get("ECS_GATEWAY_CA"),
 		Region:           get("ECS_REGION"),
 		ClusterName:      get("ECS_CLUSTER"),
-		AccessKey:        get("ECS_ACCESS_KEY"),
-		SecretKey:        get("ECS_SECRET_KEY"),
 		IMDSBase:         get("ECS_IMDS_BASE"),
 		ContainerdSocket: get("ECS_CONTAINERD_SOCKET"),
 	}

spinifex/agents/ecs/agent/config_test.go

@@ -10,7 +10,7 @@ import (
 func TestLoadConfig_Defaults(t *testing.T) {
 	// Ensure env vars don't leak in from the host.
 	for _, k := range []string{"ECS_GATEWAY_URL", "ECS_GATEWAY_CA", "ECS_REGION",
-		"ECS_CLUSTER", "ECS_ACCESS_KEY", "ECS_SECRET_KEY", "ECS_IMDS_BASE",
+		"ECS_CLUSTER", "ECS_IMDS_BASE",
 		"ECS_CONTAINERD_SOCKET", "ECS_HEARTBEAT_INTERVAL", "ECS_POLL_INTERVAL"} {
 		t.Setenv(k, "")
 	}
@@ -43,7 +43,6 @@ func TestLoadConfig_FileThenEnvOverride(t *testing.T) {
 		t.Fatal(err)
 	}
 	t.Setenv("ECS_GATEWAY_URL", "https://gw.env")
-	t.Setenv("ECS_ACCESS_KEY", "AKIAENV")
 
 	cfg := LoadConfig(envFile)
 	if cfg.GatewayURL != "https://gw.env" {
@@ -55,9 +54,6 @@ func TestLoadConfig_FileThenEnvOverride(t *testing.T) {
 	if cfg.ClusterName != "prod" {
 		t.Errorf("ClusterName = %q, want from file", cfg.ClusterName)
 	}
-	if cfg.AccessKey != "AKIAENV" {
-		t.Errorf("AccessKey = %q, want env", cfg.AccessKey)
-	}
 }
 
 func TestLoadConfig_HeartbeatOverride(t *testing.T) {

spinifex/agents/ecs/agent/controlplane.go

@@ -1,17 +1,17 @@
 package agent
 
 import (
-	"context"
 	"encoding/json"
 	"fmt"
 	"time"
 
+	awsv2 "github.com/aws/aws-sdk-go-v2/aws"
 	"github.com/aws/aws-sdk-go/aws"
 	"github.com/aws/aws-sdk-go/service/ecs"
 
 	ecsv1 "github.com/mulgadc/spinifex/contracts/ecs/v1"
 	"github.com/mulgadc/spinifex/internal/ecsgw"
-	"github.com/mulgadc/spinifex/spinifex/agents/ecs/credentials"
+	"github.com/mulgadc/spinifex/internal/gwsign"
 	ecsdomain "github.com/mulgadc/spinifex/spinifex/domains/ecs"
 )
 
@@ -47,26 +47,14 @@ const pollTimeout = 30 * time.Second
 
 // newGatewayControlPlane builds the SigV4 client signing with the instance-role
 // credentials from IMDS (fetched per call so rotation is transparent) + pinned CA.
-func newGatewayControlPlane(cfg Config, creds credentials.CredentialsProvider) (*gatewayControlPlane, error) {
-	client, err := ecsgw.New(cfg.GatewayURL, cfg.GatewayCA, credentialsFunc(creds), cfg.Region, pollTimeout)
+func newGatewayControlPlane(cfg Config, creds awsv2.CredentialsProvider) (*gatewayControlPlane, error) {
+	client, err := ecsgw.New(cfg.GatewayURL, cfg.GatewayCA, gwsign.New(creds), cfg.Region, pollTimeout)
 	if err != nil {
 		return nil, err
 	}
 	return &gatewayControlPlane{client: client}, nil
 }
 
-// credentialsFunc adapts the agent's IMDS CredentialsProvider to the ecsgw
-// per-call credentials hook.
-func credentialsFunc(p credentials.CredentialsProvider) ecsgw.CredentialsFunc {
-	return func(ctx context.Context) (string, string, string, error) {
-		c, err := p.Retrieve(ctx)
-		if err != nil {
-			return "", "", "", err
-		}
-		return c.AccessKeyID, c.SecretAccessKey, c.SessionToken, nil
-	}
-}
-
 func (g *gatewayControlPlane) Register(id identity) error {
 	in := &ecs.RegisterContainerInstanceInput{
 		Cluster:                  aws.String(id.ClusterName),

spinifex/agents/ecs/agent/controlplane_test.go

@@ -8,17 +8,17 @@ import (
 	"net/http/httptest"
 	"testing"
 
+	"github.com/aws/aws-sdk-go-v2/aws"
 	"github.com/aws/aws-sdk-go/service/ecs"
-	"github.com/mulgadc/spinifex/spinifex/agents/ecs/credentials"
 	ecsdomain "github.com/mulgadc/spinifex/spinifex/domains/ecs"
 )
 
-// fixedCredsProvider is a static credentials.CredentialsProvider for tests
+// fixedCredsProvider is a static aws.CredentialsProvider for tests
 // that only need a valid SigV4 signature, not real IMDS rotation.
 type fixedCredsProvider struct{}
 
-func (fixedCredsProvider) Retrieve(context.Context) (credentials.Credentials, error) {
-	return credentials.Credentials{AccessKeyID: "AKIATEST", SecretAccessKey: "secret"}, nil
+func (fixedCredsProvider) Retrieve(context.Context) (aws.Credentials, error) {
+	return aws.Credentials{AccessKeyID: "AKIATEST", SecretAccessKey: "secret"}, nil
 }
 
 func newTestGatewayControlPlane(t *testing.T, srv *httptest.Server) *gatewayControlPlane {

spinifex/agents/ecs/agent/credendpoint.go

@@ -13,9 +13,9 @@ import (
 	"sync"
 	"time"
 
+	"github.com/aws/aws-sdk-go-v2/aws"
 	"github.com/mulgadc/spinifex/internal/ecrauth"
 	"github.com/mulgadc/spinifex/internal/stsauth"
-	"github.com/mulgadc/spinifex/spinifex/agents/ecs/credentials"
 )
 
 const (
@@ -56,7 +56,7 @@ type credRole struct {
 
 type credEntry struct {
 	roleARN string
-	creds   stsauth.Credentials
+	creds   aws.Credentials
 }
 
 // credEndpoint is the host-netns HTTP server that hands a task its IAM role
@@ -64,7 +64,7 @@ type credEntry struct {
 // mints/caches credentials on demand by assuming the role over the gateway with
 // the agent's instance credentials.
 type credEndpoint struct {
-	provider   credentials.CredentialsProvider
+	provider   aws.CredentialsProvider
 	region     string
 	gatewayURL string
 	caPath     string
@@ -75,7 +75,7 @@ type credEndpoint struct {
 
 	// assume mints fresh credentials for a role; defaults to AssumeRole over the
 	// gateway and is overridable in tests.
-	assume func(ctx context.Context, roleARN, sessionName string) (stsauth.Credentials, error)
+	assume func(ctx context.Context, roleARN, sessionName string) (aws.Credentials, error)
 
 	mu         sync.Mutex
 	roles      map[string]credRole  // credID -> task role
@@ -89,7 +89,7 @@ type credEndpoint struct {
 // newCredEndpoint builds the endpoint. An empty ip/port falls back to the
 // AWS-reserved 169.254.170.2:80; tests pass loopback + an ephemeral port. run
 // adds the link-local address to the dummy interface (real or fake).
-func newCredEndpoint(provider credentials.CredentialsProvider, region, gatewayURL, caPath, ip string, port int, run netCmdRunner) *credEndpoint {
+func newCredEndpoint(provider aws.CredentialsProvider, region, gatewayURL, caPath, ip string, port int, run netCmdRunner) *credEndpoint {
 	if ip == "" {
 		ip = defaultCredEndpointIP
 	}
@@ -165,7 +165,7 @@ func (c *credEndpoint) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 		AccessKeyID:     creds.AccessKeyID,
 		SecretAccessKey: creds.SecretAccessKey,
 		Token:           creds.SessionToken,
-		Expiration:      creds.Expiration.UTC().Format(time.RFC3339),
+		Expiration:      creds.Expires.UTC().Format(time.RFC3339),
 	}); err != nil {
 		slog.Warn("ecs-agent: encode credential response", "task", role.taskID, "err", err)
 	}
@@ -176,12 +176,12 @@ var errUnknownCredID = fmt.Errorf("unknown credId")
 
 // fetch returns valid credentials for credID, reusing the cache until within the
 // refresh margin and otherwise assuming the task role over the gateway.
-func (c *credEndpoint) fetch(ctx context.Context, credID string) (stsauth.Credentials, credRole, error) {
+func (c *credEndpoint) fetch(ctx context.Context, credID string) (aws.Credentials, credRole, error) {
 	c.mu.Lock()
 	role, known := c.roles[credID]
 	if !known {
 		c.mu.Unlock()
-		return stsauth.Credentials{}, credRole{}, errUnknownCredID
+		return aws.Credentials{}, credRole{}, errUnknownCredID
 	}
 	if e, ok := c.cache[credID]; ok && e.roleARN == role.roleARN && credValid(e.creds, credRefreshMargin) {
 		c.mu.Unlock()
@@ -191,7 +191,7 @@ func (c *credEndpoint) fetch(ctx context.Context, credID string) (stsauth.Creden
 
 	creds, err := c.assume(ctx, role.roleARN, sessionName(role.taskID))
 	if err != nil {
-		return stsauth.Credentials{}, role, err
+		return aws.Credentials{}, role, err
 	}
 
 	c.mu.Lock()
@@ -204,7 +204,7 @@ func (c *credEndpoint) fetch(ctx context.Context, credID string) (stsauth.Creden
 // roleARN over the gateway — the execution-role path for ECR image pulls. It
 // caches the assumed set and re-assumes within the refresh margin. session scopes
 // the STS session name.
-func (c *credEndpoint) AssumeProvider(roleARN, session string) credentials.CredentialsProvider {
+func (c *credEndpoint) AssumeProvider(roleARN, session string) aws.CredentialsProvider {
 	return &assumeProvider{ep: c, roleARN: roleARN, session: session}
 }
 
@@ -216,39 +216,34 @@ type assumeProvider struct {
 	session string
 
 	mu     sync.Mutex
-	cached stsauth.Credentials
+	cached aws.Credentials
 }
 
-var _ credentials.CredentialsProvider = (*assumeProvider)(nil)
+var _ aws.CredentialsProvider = (*assumeProvider)(nil)
 
-func (p *assumeProvider) Retrieve(ctx context.Context) (credentials.Credentials, error) {
+func (p *assumeProvider) Retrieve(ctx context.Context) (aws.Credentials, error) {
 	p.mu.Lock()
 	defer p.mu.Unlock()
 	if !credValid(p.cached, credRefreshMargin) {
 		creds, err := p.ep.assume(ctx, p.roleARN, p.session)
 		if err != nil {
-			return credentials.Credentials{}, err
+			return aws.Credentials{}, err
 		}
 		p.cached = creds
 	}
-	return credentials.Credentials{
-		AccessKeyID:     p.cached.AccessKeyID,
-		SecretAccessKey: p.cached.SecretAccessKey,
-		SessionToken:    p.cached.SessionToken,
-		Expiration:      p.cached.Expiration,
-	}, nil
+	return p.cached, nil
 }
 
 // assumeOverGateway mints credentials for roleARN by SigV4-signing AssumeRole
 // against the gateway with the agent's instance credentials.
-func (c *credEndpoint) assumeOverGateway(ctx context.Context, roleARN, sessionName string) (stsauth.Credentials, error) {
+func (c *credEndpoint) assumeOverGateway(ctx context.Context, roleARN, sessionName string) (aws.Credentials, error) {
 	instCreds, err := c.provider.Retrieve(ctx)
 	if err != nil {
-		return stsauth.Credentials{}, fmt.Errorf("instance credentials: %w", err)
+		return aws.Credentials{}, fmt.Errorf("instance credentials: %w", err)
 	}
 	client, err := c.client()
 	if err != nil {
-		return stsauth.Credentials{}, err
+		return aws.Credentials{}, err
 	}
 	return stsauth.AssumeRole(c.region, c.gatewayURL, client,
 		instCreds.AccessKeyID, instCreds.SecretAccessKey, instCreds.SessionToken,
@@ -387,14 +382,14 @@ func (c *credEndpoint) delEndpointAddr() {
 }
 
 // credValid reports whether creds are usable now with margin to spare.
-func credValid(c stsauth.Credentials, margin time.Duration) bool {
+func credValid(c aws.Credentials, margin time.Duration) bool {
 	if c.AccessKeyID == "" || c.SecretAccessKey == "" {
 		return false
 	}
-	if c.Expiration.IsZero() {
+	if c.Expires.IsZero() {
 		return true
 	}
-	return time.Now().Add(margin).Before(c.Expiration)
+	return time.Now().Add(margin).Before(c.Expires)
 }
 
 // sessionName derives an AssumeRole RoleSessionName from the task ID, satisfying the

spinifex/agents/ecs/agent/credendpoint_test.go

@@ -9,15 +9,15 @@ import (
 	"testing"
 	"time"
 
-	"github.com/mulgadc/spinifex/internal/stsauth"
+	"github.com/aws/aws-sdk-go-v2/aws"
 )
 
 // newTestCredEndpoint builds an endpoint whose assume seam returns canned creds,
 // counting calls so cache behaviour is observable. run is nil (no iface plumbing).
-func newTestCredEndpoint(creds stsauth.Credentials) (*credEndpoint, *int32) {
+func newTestCredEndpoint(creds aws.Credentials) (*credEndpoint, *int32) {
 	c := newCredEndpoint(nil, "us-east-1", "https://gw", "", "127.0.0.1", 0, nil)
 	var calls int32
-	c.assume = func(ctx context.Context, roleARN, sessionName string) (stsauth.Credentials, error) {
+	c.assume = func(ctx context.Context, roleARN, sessionName string) (aws.Credentials, error) {
 		atomic.AddInt32(&calls, 1)
 		return creds, nil
 	}
@@ -26,8 +26,8 @@ func newTestCredEndpoint(creds stsauth.Credentials) (*credEndpoint, *int32) {
 
 func TestCredEndpoint_ServesRegisteredCredID(t *testing.T) {
 	exp := time.Now().Add(time.Hour).UTC().Truncate(time.Second)
-	c, _ := newTestCredEndpoint(stsauth.Credentials{
-		AccessKeyID: "AKIA", SecretAccessKey: "secret", SessionToken: "tok", Expiration: exp,
+	c, _ := newTestCredEndpoint(aws.Credentials{
+		AccessKeyID: "AKIA", SecretAccessKey: "secret", SessionToken: "tok", Expires: exp,
 	})
 	c.Register("cred-1", "task-1", "arn:aws:iam::111122223333:role/task")
 
@@ -54,11 +54,11 @@ func TestCredEndpoint_ServesRegisteredCredID(t *testing.T) {
 // The STS session name lands in the assumed-role ARN and in STS logs, so it is
 // built from the public task ID and never from the secret credID.
 func TestCredEndpoint_SessionNameIsTaskIDNotCredID(t *testing.T) {
-	c, _ := newTestCredEndpoint(stsauth.Credentials{})
+	c, _ := newTestCredEndpoint(aws.Credentials{})
 	var gotSession string
-	c.assume = func(ctx context.Context, roleARN, session string) (stsauth.Credentials, error) {
+	c.assume = func(ctx context.Context, roleARN, session string) (aws.Credentials, error) {
 		gotSession = session
-		return stsauth.Credentials{AccessKeyID: "AKIA", SecretAccessKey: "secret"}, nil
+		return aws.Credentials{AccessKeyID: "AKIA", SecretAccessKey: "secret"}, nil
 	}
 	c.Register("secret-cred", "task-1", "arn:role")
 	rr := httptest.NewRecorder()
@@ -72,7 +72,7 @@ func TestCredEndpoint_SessionNameIsTaskIDNotCredID(t *testing.T) {
 }
 
 func TestCredEndpoint_UnknownCredIDIs404(t *testing.T) {
-	c, calls := newTestCredEndpoint(stsauth.Credentials{AccessKeyID: "x", SecretAccessKey: "y"})
+	c, calls := newTestCredEndpoint(aws.Credentials{AccessKeyID: "x", SecretAccessKey: "y"})
 	rr := httptest.NewRecorder()
 	c.ServeHTTP(rr, httptest.NewRequest(http.MethodGet, "/v2/credentials/nope", nil))
 	if rr.Code != http.StatusNotFound {
@@ -84,7 +84,7 @@ func TestCredEndpoint_UnknownCredIDIs404(t *testing.T) {
 }
 
 func TestCredEndpoint_WrongMethodIs405(t *testing.T) {
-	c, _ := newTestCredEndpoint(stsauth.Credentials{AccessKeyID: "x", SecretAccessKey: "y"})
+	c, _ := newTestCredEndpoint(aws.Credentials{AccessKeyID: "x", SecretAccessKey: "y"})
 	c.Register("cred-1", "task-1", "arn:role")
 	rr := httptest.NewRecorder()
 	c.ServeHTTP(rr, httptest.NewRequest(http.MethodPost, "/v2/credentials/cred-1", nil))
@@ -94,8 +94,8 @@ func TestCredEndpoint_WrongMethodIs405(t *testing.T) {
 }
 
 func TestCredEndpoint_CachesUntilExpiry(t *testing.T) {
-	c, calls := newTestCredEndpoint(stsauth.Credentials{
-		AccessKeyID: "AKIA", SecretAccessKey: "secret", Expiration: time.Now().Add(time.Hour),
+	c, calls := newTestCredEndpoint(aws.Credentials{
+		AccessKeyID: "AKIA", SecretAccessKey: "secret", Expires: time.Now().Add(time.Hour),
 	})
 	c.Register("cred-1", "task-1", "arn:role")
 	for hit := range 3 {
@@ -112,8 +112,8 @@ func TestCredEndpoint_CachesUntilExpiry(t *testing.T) {
 
 func TestCredEndpoint_RefreshesNearExpiry(t *testing.T) {
 	// Expiry inside the refresh margin -> every hit re-assumes.
-	c, calls := newTestCredEndpoint(stsauth.Credentials{
-		AccessKeyID: "AKIA", SecretAccessKey: "secret", Expiration: time.Now().Add(time.Minute),
+	c, calls := newTestCredEndpoint(aws.Credentials{
+		AccessKeyID: "AKIA", SecretAccessKey: "secret", Expires: time.Now().Add(time.Minute),
 	})
 	c.Register("cred-1", "task-1", "arn:role")
 	for range 2 {
@@ -126,7 +126,7 @@ func TestCredEndpoint_RefreshesNearExpiry(t *testing.T) {
 }
 
 func TestCredEndpoint_DeregisterDropsCredID(t *testing.T) {
-	c, _ := newTestCredEndpoint(stsauth.Credentials{AccessKeyID: "x", SecretAccessKey: "y"})
+	c, _ := newTestCredEndpoint(aws.Credentials{AccessKeyID: "x", SecretAccessKey: "y"})
 	c.Register("cred-1", "task-1", "arn:role")
 	c.Deregister("cred-1")
 	rr := httptest.NewRecorder()
@@ -137,9 +137,9 @@ func TestCredEndpoint_DeregisterDropsCredID(t *testing.T) {
 }
 
 func TestCredEndpoint_AssumeFailureIs502(t *testing.T) {
-	c, _ := newTestCredEndpoint(stsauth.Credentials{})
-	c.assume = func(ctx context.Context, roleARN, sessionName string) (stsauth.Credentials, error) {
-		return stsauth.Credentials{}, context.DeadlineExceeded
+	c, _ := newTestCredEndpoint(aws.Credentials{})
+	c.assume = func(ctx context.Context, roleARN, sessionName string) (aws.Credentials, error) {
+		return aws.Credentials{}, context.DeadlineExceeded
 	}
 	c.Register("cred-1", "task-1", "arn:role")
 	rr := httptest.NewRecorder()
@@ -150,8 +150,8 @@ func TestCredEndpoint_AssumeFailureIs502(t *testing.T) {
 }
 
 func TestCredEndpoint_StartServesOverLoopback(t *testing.T) {
-	c, _ := newTestCredEndpoint(stsauth.Credentials{
-		AccessKeyID: "AKIA", SecretAccessKey: "secret", Expiration: time.Now().Add(time.Hour),
+	c, _ := newTestCredEndpoint(aws.Credentials{
+		AccessKeyID: "AKIA", SecretAccessKey: "secret", Expires: time.Now().Add(time.Hour),
 	})
 	c.Register("cred-1", "task-1", "arn:role")
 	c.port = 0 // bind an ephemeral port instead of privileged 80

spinifex/agents/ecs/agent/credinject_test.go

@@ -5,7 +5,7 @@ import (
 	"net/http/httptest"
 	"testing"
 
-	"github.com/mulgadc/spinifex/internal/stsauth"
+	"github.com/aws/aws-sdk-go-v2/aws"
 
 	ecsv1 "github.com/mulgadc/spinifex/contracts/ecs/v1"
 )
@@ -30,7 +30,7 @@ func TestTaskCredID(t *testing.T) {
 // The task ID is public (it is the task ARN suffix), so naming another task's ID
 // at the credential endpoint must not yield that task's role.
 func TestRegisterTaskCreds_TaskIDDoesNotServeCredentials(t *testing.T) {
-	c, _ := newTestCredEndpoint(stsauth.Credentials{AccessKeyID: "AKIA", SecretAccessKey: "secret"})
+	c, _ := newTestCredEndpoint(aws.Credentials{AccessKeyID: "AKIA", SecretAccessKey: "secret"})
 	a := Agent{cred: c}
 	get := func(id string) int {
 		rr := httptest.NewRecorder()

spinifex/agents/ecs/agent/doc.go

@@ -5,8 +5,7 @@
 // task assignments, runs them through containerd, and reports state back over
 // the gateway.
 //
-// Static config (gateway URL, CA, region, cluster, seeded IAM creds, containerd
-// socket) is read from the cloud-init env file /etc/spinifex-ecs/agent.env
+// Static config (gateway URL, CA, region, cluster, containerd socket) is read from the cloud-init env file /etc/spinifex-ecs/agent.env
 // (KEY=value); real env vars override it. Host identity (account, instance, AZ)
 // comes from IMDS at boot.
 package agent

spinifex/agents/ecs/agent/ecrresolver.go

@@ -7,8 +7,8 @@ import (
 	"strings"
 	"sync"
 
+	"github.com/aws/aws-sdk-go-v2/aws"
 	"github.com/mulgadc/spinifex/internal/ecrauth"
-	"github.com/mulgadc/spinifex/spinifex/agents/ecs/credentials"
 	"github.com/mulgadc/spinifex/spinifex/agents/ecs/runtime"
 )
 
@@ -22,7 +22,7 @@ import (
 // pulled, so the agent can still register + heartbeat on a host with absent or
 // wrong ECR config.
 type ecrResolver struct {
-	creds      credentials.CredentialsProvider
+	creds      aws.CredentialsProvider
 	region     string
 	gatewayURL string
 	caPath     string
@@ -35,13 +35,13 @@ var _ runtime.Resolver = (*ecrResolver)(nil)
 
 // newECRResolver builds a resolver with an injected HTTP client (tests). The
 // client is used as-is and never rebuilt from caPath.
-func newECRResolver(creds credentials.CredentialsProvider, region, gatewayURL string, httpClient *http.Client) *ecrResolver {
+func newECRResolver(creds aws.CredentialsProvider, region, gatewayURL string, httpClient *http.Client) *ecrResolver {
 	return &ecrResolver{creds: creds, region: region, gatewayURL: gatewayURL, httpClient: httpClient}
 }
 
 // newLazyECRResolver builds a resolver that constructs its gateway HTTP client
 // from caPath on first use (production path).
-func newLazyECRResolver(creds credentials.CredentialsProvider, region, gatewayURL, caPath string) *ecrResolver {
+func newLazyECRResolver(creds aws.CredentialsProvider, region, gatewayURL, caPath string) *ecrResolver {
 	return &ecrResolver{creds: creds, region: region, gatewayURL: gatewayURL, caPath: caPath}
 }
 

spinifex/agents/ecs/agent/ecrresolver_test.go

@@ -14,16 +14,16 @@ import (
 	"testing"
 	"time"
 
-	"github.com/mulgadc/spinifex/spinifex/agents/ecs/credentials"
+	"github.com/aws/aws-sdk-go-v2/aws"
 )
 
-// stubCreds is a credentials.CredentialsProvider returning fixed creds or error.
+// stubCreds is a aws.CredentialsProvider returning fixed creds or error.
 type stubCreds struct {
-	c   credentials.Credentials
+	c   aws.Credentials
 	err error
 }
 
-func (s stubCreds) Retrieve(context.Context) (credentials.Credentials, error) {
+func (s stubCreds) Retrieve(context.Context) (aws.Credentials, error) {
 	return s.c, s.err
 }
 
@@ -58,9 +58,9 @@ func trusting(srv *httptest.Server) *http.Client {
 
 func TestECRResolver_Authorize(t *testing.T) {
 	srv := gatewayStub(t, "AWS:jwt", "https://123456789012.dkr.ecr.us-east-1.spinifex.internal", http.StatusOK)
-	creds := stubCreds{c: credentials.Credentials{
+	creds := stubCreds{c: aws.Credentials{
 		AccessKeyID: "AKIA", SecretAccessKey: "secret", SessionToken: "sess",
-		Expiration: time.Now().Add(time.Hour),
+		Expires: time.Now().Add(time.Hour),
 	}}
 	r := newECRResolver(creds, "us-east-1", srv.URL, trusting(srv))
 
@@ -85,7 +85,7 @@ func TestECRResolver_CredsError(t *testing.T) {
 
 func TestECRResolver_GatewayError(t *testing.T) {
 	srv := gatewayStub(t, "", "", http.StatusInternalServerError)
-	creds := stubCreds{c: credentials.Credentials{AccessKeyID: "A", SecretAccessKey: "B"}}
+	creds := stubCreds{c: aws.Credentials{AccessKeyID: "A", SecretAccessKey: "B"}}
 	r := newECRResolver(creds, "us-east-1", srv.URL, trusting(srv))
 	if _, _, _, err := r.Authorize(context.Background(), "123456789012.dkr.ecr.us-east-1.spinifex.internal/app:1"); err == nil {
 		t.Fatal("expected gateway error")
@@ -93,7 +93,7 @@ func TestECRResolver_GatewayError(t *testing.T) {
 }
 
 func TestECRResolver_LazyClient_BadCADeferredToAuthorize(t *testing.T) {
-	creds := stubCreds{c: credentials.Credentials{AccessKeyID: "A", SecretAccessKey: "B"}}
+	creds := stubCreds{c: aws.Credentials{AccessKeyID: "A", SecretAccessKey: "B"}}
 	// Construction with a bogus CA path must not fail — only Authorize does.
 	r := newLazyECRResolver(creds, "us-east-1", "https://gw", filepath.Join(t.TempDir(), "absent.pem"))
 	if _, _, _, err := r.Authorize(context.Background(), "123456789012.dkr.ecr.us-east-1.spinifex.internal/app:1"); err == nil {
@@ -141,9 +141,9 @@ func TestIsECRRef(t *testing.T) {
 // registry, even though it lacks a ".dkr.ecr." name.
 func TestECRResolver_Authorize_BareHostGateway(t *testing.T) {
 	srv := gatewayStub(t, "AWS:jwt", "https://192.168.1.13:9999", http.StatusOK)
-	creds := stubCreds{c: credentials.Credentials{
+	creds := stubCreds{c: aws.Credentials{
 		AccessKeyID: "AKIA", SecretAccessKey: "secret", SessionToken: "sess",
-		Expiration: time.Now().Add(time.Hour),
+		Expires: time.Now().Add(time.Hour),
 	}}
 	r := newECRResolver(creds, "us-east-1", srv.URL, trusting(srv))
 

spinifex/agents/ecs/credentials/imds.go

@@ -1,41 +0,0 @@
-package credentials
-
-import (
-	"context"
-	"net/http"
-
-	"github.com/aws/aws-sdk-go-v2/aws"
-	"github.com/mulgadc/spinifex/internal/imdscreds"
-)
-
-// IMDSProvider fetches instance-role credentials from IMDS via the AWS SDK's
-// EC2 IMDS client and ec2rolecreds provider, cached with aws.CredentialsCache,
-// which refetches once the real expiry passes. A failed refresh is an error
-// here, not an extended expiry. Safe for concurrent use.
-type IMDSProvider struct {
-	cache *aws.CredentialsCache
-}
-
-var _ CredentialsProvider = (*IMDSProvider)(nil)
-
-// NewIMDSProvider builds a provider against the given IMDS base URL (e.g.
-// "http://169.254.169.254/latest"). A nil client uses the SDK IMDS client's
-// default HTTP client.
-func NewIMDSProvider(client *http.Client, base string) *IMDSProvider {
-	provider := imdscreds.NewProvider(client, base)
-	return &IMDSProvider{cache: aws.NewCredentialsCache(provider)}
-}
-
-// Retrieve returns cached credentials when still valid, else fetches fresh ones.
-func (p *IMDSProvider) Retrieve(ctx context.Context) (Credentials, error) {
-	creds, err := p.cache.Retrieve(ctx)
-	if err != nil {
-		return Credentials{}, err
-	}
-	return Credentials{
-		AccessKeyID:     creds.AccessKeyID,
-		SecretAccessKey: creds.SecretAccessKey,
-		SessionToken:    creds.SessionToken,
-		Expiration:      creds.Expires,
-	}, nil
-}

spinifex/agents/ecs/credentials/imds_test.go

@@ -1,164 +0,0 @@
-package credentials
-
-import (
-	"context"
-	"encoding/json"
-	"net/http"
-	"net/http/httptest"
-	"sync/atomic"
-	"testing"
-	"time"
-)
-
-// imdsStub serves the IMDSv2 token + role + credentials endpoints under the
-// real /latest prefix the SDK IMDS client always requests.
-func imdsStub(t *testing.T, hits *int32, creds map[string]string) *httptest.Server {
-	t.Helper()
-	mux := http.NewServeMux()
-	mux.HandleFunc("/latest/api/token", func(w http.ResponseWriter, r *http.Request) {
-		// Real IMDS echoes the requested TTL back as a response header; the SDK
-		// client parses it from there, not the body.
-		w.Header().Set("X-Aws-Ec2-Metadata-Token-Ttl-Seconds", "21600")
-		_, _ = w.Write([]byte("v2-token"))
-	})
-	mux.HandleFunc("/latest/meta-data/iam/security-credentials/", func(w http.ResponseWriter, r *http.Request) {
-		if r.URL.Path == "/latest/meta-data/iam/security-credentials/" {
-			_, _ = w.Write([]byte("node-role"))
-			return
-		}
-		if hits != nil {
-			atomic.AddInt32(hits, 1)
-		}
-		_ = json.NewEncoder(w).Encode(creds)
-	})
-	srv := httptest.NewServer(mux)
-	t.Cleanup(srv.Close)
-	return srv
-}
-
-func TestRetrieve_FetchesAndCaches(t *testing.T) {
-	var hits int32
-	// Kept under an hour: ec2rolecreds caps Expires to now+1h.
-	exp := time.Now().Add(30 * time.Minute).UTC().Round(time.Second)
-	srv := imdsStub(t, &hits, map[string]string{
-		"Code":            "Success",
-		"AccessKeyId":     "AKIA",
-		"SecretAccessKey": "secret",
-		"Token":           "session",
-		"Expiration":      exp.Format(time.RFC3339),
-	})
-
-	p := NewIMDSProvider(srv.Client(), srv.URL+"/latest")
-	got, err := p.Retrieve(context.Background())
-	if err != nil {
-		t.Fatalf("Retrieve: %v", err)
-	}
-	if got.AccessKeyID != "AKIA" || !got.Expiration.Equal(exp) {
-		t.Errorf("creds mismatch: %+v", got)
-	}
-
-	// Second call well within validity must not re-hit IMDS.
-	if _, err := p.Retrieve(context.Background()); err != nil {
-		t.Fatalf("Retrieve 2: %v", err)
-	}
-	if hits != 1 {
-		t.Errorf("IMDS credential hits = %d, want 1 (cached)", hits)
-	}
-}
-
-func TestRetrieve_RefetchesWhenExpired(t *testing.T) {
-	var hits int32
-	// Already expired: aws.CredentialsCache refetches once the stored
-	// credentials' real expiry has passed, regardless of ExpiryWindow.
-	srv := imdsStub(t, &hits, map[string]string{
-		"Code":            "Success",
-		"AccessKeyId":     "AKIA",
-		"SecretAccessKey": "secret",
-		"Expiration":      time.Now().Add(-time.Minute).UTC().Format(time.RFC3339),
-	})
-	p := NewIMDSProvider(srv.Client(), srv.URL+"/latest")
-	if _, err := p.Retrieve(context.Background()); err != nil {
-		t.Fatal(err)
-	}
-	if _, err := p.Retrieve(context.Background()); err != nil {
-		t.Fatal(err)
-	}
-	if hits != 2 {
-		t.Errorf("hits = %d, want 2 (expired refetch)", hits)
-	}
-}
-
-// imdsStubFailAfterFirst serves creds on the first credential request and 500s
-// on every one after it, so a test can drive a refresh that fails against a
-// credential the cache has already stored.
-func imdsStubFailAfterFirst(t *testing.T, creds map[string]string) *httptest.Server {
-	t.Helper()
-	var hits atomic.Int32
-	mux := http.NewServeMux()
-	mux.HandleFunc("/latest/api/token", func(w http.ResponseWriter, r *http.Request) {
-		w.Header().Set("X-Aws-Ec2-Metadata-Token-Ttl-Seconds", "21600")
-		_, _ = w.Write([]byte("v2-token"))
-	})
-	mux.HandleFunc("/latest/meta-data/iam/security-credentials/", func(w http.ResponseWriter, r *http.Request) {
-		if r.URL.Path == "/latest/meta-data/iam/security-credentials/" {
-			_, _ = w.Write([]byte("node-role"))
-			return
-		}
-		if hits.Add(1) > 1 {
-			w.WriteHeader(http.StatusInternalServerError)
-			return
-		}
-		_ = json.NewEncoder(w).Encode(creds)
-	})
-	srv := httptest.NewServer(mux)
-	t.Cleanup(srv.Close)
-	return srv
-}
-
-// A refresh that fails must surface as an error. Left to itself the SDK's
-// ec2rolecreds implements HandleFailToRefresh, which hands back the same dead
-// key with Expires pushed 5-15 minutes out, so the agent would sign requests
-// with a credential the control plane has already rejected — indefinitely,
-// since every subsequent refresh fails the same way.
-func TestRetrieve_FailedRefreshDoesNotExtendExpiry(t *testing.T) {
-	srv := imdsStubFailAfterFirst(t, map[string]string{
-		"Code":            "Success",
-		"AccessKeyId":     "AKIA",
-		"SecretAccessKey": "secret",
-		"Token":           "session",
-		// Already expired, so the next Retrieve must go back to IMDS.
-		"Expiration": time.Now().Add(-time.Minute).UTC().Format(time.RFC3339),
-	})
-
-	p := NewIMDSProvider(srv.Client(), srv.URL+"/latest")
-	first, err := p.Retrieve(context.Background())
-	if err != nil {
-		t.Fatalf("first Retrieve: %v", err)
-	}
-
-	got, err := p.Retrieve(context.Background())
-	if err == nil {
-		t.Fatalf("refresh failure was reported as success: %+v", got)
-	}
-	if got.Expiration.After(first.Expiration) {
-		t.Errorf("expiry extended on refresh failure: %v -> %v", first.Expiration, got.Expiration)
-	}
-}
-
-func TestRetrieve_CancelledContext(t *testing.T) {
-	srv := imdsStub(t, nil, map[string]string{"Code": "Success", "AccessKeyId": "A", "SecretAccessKey": "B"})
-	p := NewIMDSProvider(srv.Client(), srv.URL+"/latest")
-	ctx, cancel := context.WithCancel(context.Background())
-	cancel()
-	if _, err := p.Retrieve(ctx); err == nil {
-		t.Fatal("expected error from cancelled context")
-	}
-}
-
-func TestRetrieve_CodeNotSuccessErrors(t *testing.T) {
-	srv := imdsStub(t, nil, map[string]string{"Code": "Failure", "AccessKeyId": "A", "SecretAccessKey": "B"})
-	p := NewIMDSProvider(srv.Client(), srv.URL+"/latest")
-	if _, err := p.Retrieve(context.Background()); err == nil {
-		t.Fatal("expected error when Code != Success")
-	}
-}

spinifex/agents/ecs/credentials/provider.go

@@ -1,24 +0,0 @@
-// Package credentials supplies AWS SigV4 credentials to the ecs-agent. The agent
-// signs gateway calls (ECR GetAuthorizationToken, instance register/heartbeat)
-// with the host's instance-role credentials, fetched from IMDS and cached until
-// near expiry.
-package credentials
-
-import (
-	"context"
-	"time"
-)
-
-// Credentials is a SigV4 credential set with an expiry.
-type Credentials struct {
-	AccessKeyID     string
-	SecretAccessKey string
-	SessionToken    string
-	Expiration      time.Time
-}
-
-// CredentialsProvider yields current credentials, refreshing as needed. Retrieve
-// must be safe for concurrent use.
-type CredentialsProvider interface {
-	Retrieve(ctx context.Context) (Credentials, error)
-}

spinifex/agents/eks/konnectivitycert/cert.go

@@ -80,7 +80,17 @@ func ensureCert(dir, caCertPath, caKeyPath, cn string, dns []string, ips []net.I
 	if err != nil {
 		return "", "", err
 	}
-	certPEM, keyPEM, err := issueLeaf(caCert, caKey, cn, dns, ips)
+	now := time.Now().UTC()
+	certPEM, keyPEM, err := Mint(&x509.Certificate{
+		Subject:               pkix.Name{CommonName: cn},
+		NotBefore:             now.Add(-time.Hour),
+		NotAfter:              now.AddDate(10, 0, 0),
+		KeyUsage:              x509.KeyUsageDigitalSignature | x509.KeyUsageKeyEncipherment,
+		ExtKeyUsage:           []x509.ExtKeyUsage{x509.ExtKeyUsageServerAuth},
+		BasicConstraintsValid: true,
+		DNSNames:              dns,
+		IPAddresses:           ips,
+	}, caCert, caKey)
 	if err != nil {
 		return "", "", err
 	}
@@ -181,8 +191,9 @@ func parseECKey(der []byte) (*ecdsa.PrivateKey, error) {
 	return ec, nil
 }
 
-// issueLeaf mints an ECDSA-P256 ServerAuth leaf for cn + SANs, signed by caCert/caKey.
-func issueLeaf(caCert *x509.Certificate, caKey *ecdsa.PrivateKey, cn string, dns []string, ips []net.IP) (certPEM, keyPEM []byte, err error) {
+// Mint generates an ECDSA-P256 key, issues tmpl for it with a random serial and
+// signs it with parentKey, or self-signs when parent is nil. tmpl is not modified.
+func Mint(tmpl, parent *x509.Certificate, parentKey *ecdsa.PrivateKey) (certPEM, keyPEM []byte, err error) {
 	priv, err := ecdsa.GenerateKey(elliptic.P256(), rand.Reader)
 	if err != nil {
 		return nil, nil, fmt.Errorf("generate key: %w", err)
@@ -191,19 +202,12 @@ func issueLeaf(caCert *x509.Certificate, caKey *ecdsa.PrivateKey, cn string, dns
 	if err != nil {
 		return nil, nil, fmt.Errorf("serial: %w", err)
 	}
-	now := time.Now().UTC()
-	tmpl := x509.Certificate{
-		SerialNumber:          serial,
-		Subject:               pkix.Name{CommonName: cn},
-		NotBefore:             now.Add(-time.Hour),
-		NotAfter:              now.AddDate(10, 0, 0),
-		KeyUsage:              x509.KeyUsageDigitalSignature | x509.KeyUsageKeyEncipherment,
-		ExtKeyUsage:           []x509.ExtKeyUsage{x509.ExtKeyUsageServerAuth},
-		BasicConstraintsValid: true,
-		DNSNames:              dns,
-		IPAddresses:           ips,
+	t := *tmpl
+	t.SerialNumber = serial
+	if parent == nil {
+		parent, parentKey = &t, priv
 	}
-	der, err := x509.CreateCertificate(rand.Reader, &tmpl, caCert, &priv.PublicKey, caKey)
+	der, err := x509.CreateCertificate(rand.Reader, &t, parent, &priv.PublicKey, parentKey)
 	if err != nil {
 		return nil, nil, fmt.Errorf("create cert: %w", err)
 	}

spinifex/agents/eks/podidentity/certs.go

@@ -1,15 +1,11 @@
 package podidentity
 
 import (
-	"crypto/ecdsa"
-	"crypto/elliptic"
-	"crypto/rand"
 	"crypto/tls"
 	"crypto/x509"
 	"crypto/x509/pkix"
 	"encoding/pem"
 	"fmt"
-	"math/big"
 	"net"
 	"os"
 	"path/filepath"
@@ -83,17 +79,8 @@ func ensureLeaf(certPath, keyPath, caCertPath, caKeyPath string, spec leafSpec)
 	if leafCurrent(certPath, keyPath, ca, spec) {
 		return nil
 	}
-	priv, err := ecdsa.GenerateKey(elliptic.P256(), rand.Reader)
-	if err != nil {
-		return fmt.Errorf("generate key: %w", err)
-	}
-	serial, err := rand.Int(rand.Reader, new(big.Int).Lsh(big.NewInt(1), 128))
-	if err != nil {
-		return fmt.Errorf("serial: %w", err)
-	}
 	now := time.Now().UTC()
-	tmpl := x509.Certificate{
-		SerialNumber:          serial,
+	certPEM, keyPEM, err := konnectivitycert.Mint(&x509.Certificate{
 		Subject:               pkix.Name{CommonName: spec.cn},
 		NotBefore:             now.Add(-time.Hour),
 		NotAfter:              now.Add(certValidity),
@@ -101,20 +88,15 @@ func ensureLeaf(certPath, keyPath, caCertPath, caKeyPath string, spec leafSpec)
 		ExtKeyUsage:           []x509.ExtKeyUsage{spec.usage},
 		BasicConstraintsValid: true,
 		IPAddresses:           spec.ips,
-	}
-	der, err := x509.CreateCertificate(rand.Reader, &tmpl, ca, &priv.PublicKey, caKey)
+	}, ca, caKey)
 	if err != nil {
-		return fmt.Errorf("create certificate: %w", err)
-	}
-	keyDER, err := x509.MarshalECPrivateKey(priv)
-	if err != nil {
-		return fmt.Errorf("marshal key: %w", err)
-	}
-	// Key first: a cert without its key is never published.
-	if err := writeAtomic(keyPath, pem.EncodeToMemory(&pem.Block{Type: "EC PRIVATE KEY", Bytes: keyDER}), 0o600); err != nil {
 		return err
 	}
-	return writeAtomic(certPath, pem.EncodeToMemory(&pem.Block{Type: "CERTIFICATE", Bytes: der}), 0o644)
+	// Key first: a cert without its key is never published.
+	if err := writeAtomic(keyPath, keyPEM, 0o600); err != nil {
+		return err
+	}
+	return writeAtomic(certPath, certPEM, 0o644)
 }
 
 // leafCurrent reports whether the cached pair loads, matches, was signed by ca,

spinifex/agents/eks/tokenwebhook/cert.go

@@ -1,21 +1,17 @@
 package tokenwebhook
 
 import (
-	"crypto/ecdsa"
-	"crypto/elliptic"
-	"crypto/rand"
 	"crypto/tls"
 	"crypto/x509"
 	"crypto/x509/pkix"
 	"encoding/base64"
-	"encoding/pem"
 	"fmt"
-	"math/big"
 	"net"
 	"os"
 	"time"
 
 	"github.com/mulgadc/bluebottle/pkg/tlsconfig"
+	"github.com/mulgadc/spinifex/spinifex/agents/eks/konnectivitycert"
 )
 
 // serverTLSConfig is the webhook's loopback serving TLS config.
@@ -41,7 +37,18 @@ func ensureServingCert(certPath, keyPath string) (tls.Certificate, []byte, error
 		}
 	}
 
-	certPEM, keyPEM, err := generateSelfSigned()
+	now := time.Now().UTC()
+	certPEM, keyPEM, err := konnectivitycert.Mint(&x509.Certificate{
+		Subject:               pkix.Name{CommonName: "eks-token-webhook"},
+		NotBefore:             now.Add(-time.Hour),
+		NotAfter:              now.AddDate(10, 0, 0),
+		KeyUsage:              x509.KeyUsageDigitalSignature | x509.KeyUsageKeyEncipherment | x509.KeyUsageCertSign,
+		ExtKeyUsage:           []x509.ExtKeyUsage{x509.ExtKeyUsageServerAuth},
+		BasicConstraintsValid: true,
+		IsCA:                  true,
+		IPAddresses:           []net.IP{net.IPv4(127, 0, 0, 1), net.IPv6loopback},
+		DNSNames:              []string{"localhost"},
+	}, nil, nil)
 	if err != nil {
 		return tls.Certificate{}, nil, err
 	}
@@ -58,41 +65,6 @@ func ensureServingCert(certPath, keyPath string) (tls.Certificate, []byte, error
 	return cert, certPEM, nil
 }
 
-func generateSelfSigned() (certPEM, keyPEM []byte, err error) {
-	priv, err := ecdsa.GenerateKey(elliptic.P256(), rand.Reader)
-	if err != nil {
-		return nil, nil, fmt.Errorf("generate key: %w", err)
-	}
-	serial, err := rand.Int(rand.Reader, new(big.Int).Lsh(big.NewInt(1), 128))
-	if err != nil {
-		return nil, nil, fmt.Errorf("serial: %w", err)
-	}
-	now := time.Now().UTC()
-	tmpl := x509.Certificate{
-		SerialNumber:          serial,
-		Subject:               pkix.Name{CommonName: "eks-token-webhook"},
-		NotBefore:             now.Add(-time.Hour),
-		NotAfter:              now.AddDate(10, 0, 0),
-		KeyUsage:              x509.KeyUsageDigitalSignature | x509.KeyUsageKeyEncipherment | x509.KeyUsageCertSign,
-		ExtKeyUsage:           []x509.ExtKeyUsage{x509.ExtKeyUsageServerAuth},
-		BasicConstraintsValid: true,
-		IsCA:                  true,
-		IPAddresses:           []net.IP{net.IPv4(127, 0, 0, 1), net.IPv6loopback},
-		DNSNames:              []string{"localhost"},
-	}
-	der, err := x509.CreateCertificate(rand.Reader, &tmpl, &tmpl, &priv.PublicKey, priv)
-	if err != nil {
-		return nil, nil, fmt.Errorf("create cert: %w", err)
-	}
-	keyDER, err := x509.MarshalECPrivateKey(priv)
-	if err != nil {
-		return nil, nil, fmt.Errorf("marshal key: %w", err)
-	}
-	certPEM = pem.EncodeToMemory(&pem.Block{Type: "CERTIFICATE", Bytes: der})
-	keyPEM = pem.EncodeToMemory(&pem.Block{Type: "EC PRIVATE KEY", Bytes: keyDER})
-	return certPEM, keyPEM, nil
-}
-
 // writeAPIServerKubeconfig writes the kubeconfig kube-apiserver loads via
 // --authentication-token-webhook-config-file. It points the apiserver at the
 // loopback webhook and trusts its self-signed serving cert as the CA bundle.

spinifex/agents/eks/webhookcert/cert.go

@@ -14,9 +14,6 @@ package webhookcert
 
 import (
 	"context"
-	"crypto/ecdsa"
-	"crypto/elliptic"
-	"crypto/rand"
 	"crypto/x509"
 	"crypto/x509/pkix"
 	"encoding/base64"
@@ -24,11 +21,12 @@ import (
 	"errors"
 	"fmt"
 	"io"
-	"math/big"
 	"os"
 	"path/filepath"
 	"strings"
 	"time"
+
+	"github.com/mulgadc/spinifex/spinifex/agents/eks/konnectivitycert"
 )
 
 // Config names the webhook certificate and where it is cached.
@@ -69,13 +67,25 @@ func ensureCert(dir, cn string, dnsNames []string) (certPEM, keyPEM []byte, err
 
 	if c, cerr := os.ReadFile(certPath); cerr == nil {
 		if k, kerr := os.ReadFile(keyPath); kerr == nil {
-			if _, perr := x509.ParseCertificate(decodeFirst(c)); perr == nil && len(k) > 0 {
-				return c, k, nil
+			if blk, _ := pem.Decode(c); blk != nil && len(k) > 0 {
+				if _, perr := x509.ParseCertificate(blk.Bytes); perr == nil {
+					return c, k, nil
+				}
 			}
 		}
 	}
 
-	certPEM, keyPEM, err = generateSelfSigned(cn, dnsNames)
+	now := time.Now().UTC()
+	certPEM, keyPEM, err = konnectivitycert.Mint(&x509.Certificate{
+		Subject:               pkix.Name{CommonName: cn},
+		NotBefore:             now.Add(-time.Hour),
+		NotAfter:              now.AddDate(10, 0, 0),
+		KeyUsage:              x509.KeyUsageDigitalSignature | x509.KeyUsageKeyEncipherment | x509.KeyUsageCertSign,
+		ExtKeyUsage:           []x509.ExtKeyUsage{x509.ExtKeyUsageServerAuth},
+		BasicConstraintsValid: true,
+		IsCA:                  true,
+		DNSNames:              dnsNames,
+	}, nil, nil)
 	if err != nil {
 		return nil, nil, err
 	}
@@ -91,51 +101,6 @@ func ensureCert(dir, cn string, dnsNames []string) (certPEM, keyPEM []byte, err
 	return certPEM, keyPEM, nil
 }
 
-// generateSelfSigned mints an ECDSA-P256 self-signed cert valid as both a CA and
-// a server cert, with the given common name and DNS SANs.
-func generateSelfSigned(cn string, dnsNames []string) (certPEM, keyPEM []byte, err error) {
-	priv, err := ecdsa.GenerateKey(elliptic.P256(), rand.Reader)
-	if err != nil {
-		return nil, nil, fmt.Errorf("generate key: %w", err)
-	}
-	serial, err := rand.Int(rand.Reader, new(big.Int).Lsh(big.NewInt(1), 128))
-	if err != nil {
-		return nil, nil, fmt.Errorf("serial: %w", err)
-	}
-	now := time.Now().UTC()
-	tmpl := x509.Certificate{
-		SerialNumber:          serial,
-		Subject:               pkix.Name{CommonName: cn},
-		NotBefore:             now.Add(-time.Hour),
-		NotAfter:              now.AddDate(10, 0, 0),
-		KeyUsage:              x509.KeyUsageDigitalSignature | x509.KeyUsageKeyEncipherment | x509.KeyUsageCertSign,
-		ExtKeyUsage:           []x509.ExtKeyUsage{x509.ExtKeyUsageServerAuth},
-		BasicConstraintsValid: true,
-		IsCA:                  true,
-		DNSNames:              dnsNames,
-	}
-	der, err := x509.CreateCertificate(rand.Reader, &tmpl, &tmpl, &priv.PublicKey, priv)
-	if err != nil {
-		return nil, nil, fmt.Errorf("create cert: %w", err)
-	}
-	keyDER, err := x509.MarshalECPrivateKey(priv)
-	if err != nil {
-		return nil, nil, fmt.Errorf("marshal key: %w", err)
-	}
-	certPEM = pem.EncodeToMemory(&pem.Block{Type: "CERTIFICATE", Bytes: der})
-	keyPEM = pem.EncodeToMemory(&pem.Block{Type: "EC PRIVATE KEY", Bytes: keyDER})
-	return certPEM, keyPEM, nil
-}
-
-// decodeFirst returns the DER bytes of the first PEM block, or nil.
-func decodeFirst(pemBytes []byte) []byte {
-	blk, _ := pem.Decode(pemBytes)
-	if blk == nil {
-		return nil
-	}
-	return blk.Bytes
-}
-
 func splitCSV(s string) []string {
 	var out []string
 	for p := range strings.SplitSeq(s, ",") {

spinifex/agents/eks/webhookcert/cert_test.go

@@ -2,6 +2,7 @@ package webhookcert
 
 import (
 	"crypto/x509"
+	"encoding/pem"
 	"os"
 	"path/filepath"
 	"testing"
@@ -14,7 +15,11 @@ const (
 
 func parseLeaf(t *testing.T, certPEM []byte) *x509.Certificate {
 	t.Helper()
-	leaf, err := x509.ParseCertificate(decodeFirst(certPEM))
+	blk, _ := pem.Decode(certPEM)
+	if blk == nil {
+		t.Fatal("no PEM block in cert")
+	}
+	leaf, err := x509.ParseCertificate(blk.Bytes)
 	if err != nil {
 		t.Fatalf("parse cert: %v", err)
 	}

spinifex/domains/acm/awsapi/certificates.go

@@ -8,6 +8,7 @@ import (
 
 	"github.com/aws/aws-sdk-go/service/acm"
 	"github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
+	"github.com/mulgadc/spinifex/spinifex/ingress/aws/envelope"
 	"github.com/nats-io/nats.go"
 )
 
@@ -17,7 +18,7 @@ const natsTimeout = 30 * time.Second
 // RequestCertificate — CertificateManager.RequestCertificate.
 func RequestCertificate(ctx context.Context, natsConn *nats.Conn, accountID string, body []byte) (*acm.RequestCertificateOutput, error) {
 	input := new(acm.RequestCertificateInput)
-	if err := unmarshalIfBody(body, input); err != nil {
+	if err := envelope.UnmarshalJSONBody(body, input); err != nil {
 		return nil, errors.New(awserrors.ErrorInvalidParameterValue)
 	}
 	return natsmsg.NATSRequest[acm.RequestCertificateOutput](ctx, natsConn, "acm.RequestCertificate", input, natsTimeout, accountID)
@@ -26,7 +27,7 @@ func RequestCertificate(ctx context.Context, natsConn *nats.Conn, accountID stri
 // ImportCertificate — CertificateManager.ImportCertificate.
 func ImportCertificate(ctx context.Context, natsConn *nats.Conn, accountID string, body []byte) (*acm.ImportCertificateOutput, error) {
 	input := new(acm.ImportCertificateInput)
-	if err := unmarshalIfBody(body, input); err != nil {
+	if err := envelope.UnmarshalJSONBody(body, input); err != nil {
 		return nil, errors.New(awserrors.ErrorInvalidParameter)
 	}
 	return natsmsg.NATSRequest[acm.ImportCertificateOutput](ctx, natsConn, "acm.ImportCertificate", input, natsTimeout, accountID)
@@ -35,7 +36,7 @@ func ImportCertificate(ctx context.Context, natsConn *nats.Conn, accountID strin
 // DescribeCertificate — CertificateManager.DescribeCertificate.
 func DescribeCertificate(ctx context.Context, natsConn *nats.Conn, accountID string, body []byte) (*acm.DescribeCertificateOutput, error) {
 	input := new(acm.DescribeCertificateInput)
-	if err := unmarshalIfBody(body, input); err != nil {
+	if err := envelope.UnmarshalJSONBody(body, input); err != nil {
 		return nil, errors.New(awserrors.ErrorInvalidParameter)
 	}
 	return natsmsg.NATSRequest[acm.DescribeCertificateOutput](ctx, natsConn, "acm.DescribeCertificate", input, natsTimeout, accountID)
@@ -44,7 +45,7 @@ func DescribeCertificate(ctx context.Context, natsConn *nats.Conn, accountID str
 // GetCertificate — CertificateManager.GetCertificate.
 func GetCertificate(ctx context.Context, natsConn *nats.Conn, accountID string, body []byte) (*acm.GetCertificateOutput, error) {
 	input := new(acm.GetCertificateInput)
-	if err := unmarshalIfBody(body, input); err != nil {
+	if err := envelope.UnmarshalJSONBody(body, input); err != nil {
 		return nil, errors.New(awserrors.ErrorInvalidParameter)
 	}
 	return natsmsg.NATSRequest[acm.GetCertificateOutput](ctx, natsConn, "acm.GetCertificate", input, natsTimeout, accountID)
@@ -53,7 +54,7 @@ func GetCertificate(ctx context.Context, natsConn *nats.Conn, accountID string,
 // ListCertificates — CertificateManager.ListCertificates.
 func ListCertificates(ctx context.Context, natsConn *nats.Conn, accountID string, body []byte) (*acm.ListCertificatesOutput, error) {
 	input := new(acm.ListCertificatesInput)
-	if err := unmarshalIfBody(body, input); err != nil {
+	if err := envelope.UnmarshalJSONBody(body, input); err != nil {
 		return nil, errors.New(awserrors.ErrorInvalidParameter)
 	}
 	return natsmsg.NATSRequest[acm.ListCertificatesOutput](ctx, natsConn, "acm.ListCertificates", input, natsTimeout, accountID)
@@ -62,7 +63,7 @@ func ListCertificates(ctx context.Context, natsConn *nats.Conn, accountID string
 // DeleteCertificate — CertificateManager.DeleteCertificate.
 func DeleteCertificate(ctx context.Context, natsConn *nats.Conn, accountID string, body []byte) (*acm.DeleteCertificateOutput, error) {
 	input := new(acm.DeleteCertificateInput)
-	if err := unmarshalIfBody(body, input); err != nil {
+	if err := envelope.UnmarshalJSONBody(body, input); err != nil {
 		return nil, errors.New(awserrors.ErrorInvalidParameter)
 	}
 	return natsmsg.NATSRequest[acm.DeleteCertificateOutput](ctx, natsConn, "acm.DeleteCertificate", input, natsTimeout, accountID)
@@ -71,7 +72,7 @@ func DeleteCertificate(ctx context.Context, natsConn *nats.Conn, accountID strin
 // ListTagsForCertificate — CertificateManager.ListTagsForCertificate.
 func ListTagsForCertificate(ctx context.Context, natsConn *nats.Conn, accountID string, body []byte) (*acm.ListTagsForCertificateOutput, error) {
 	input := new(acm.ListTagsForCertificateInput)
-	if err := unmarshalIfBody(body, input); err != nil {
+	if err := envelope.UnmarshalJSONBody(body, input); err != nil {
 		return nil, errors.New(awserrors.ErrorInvalidParameter)
 	}
 	return natsmsg.NATSRequest[acm.ListTagsForCertificateOutput](ctx, natsConn, "acm.ListTagsForCertificate", input, natsTimeout, accountID)
@@ -80,7 +81,7 @@ func ListTagsForCertificate(ctx context.Context, natsConn *nats.Conn, accountID
 // AddTagsToCertificate — CertificateManager.AddTagsToCertificate.
 func AddTagsToCertificate(ctx context.Context, natsConn *nats.Conn, accountID string, body []byte) (*acm.AddTagsToCertificateOutput, error) {
 	input := new(acm.AddTagsToCertificateInput)
-	if err := unmarshalIfBody(body, input); err != nil {
+	if err := envelope.UnmarshalJSONBody(body, input); err != nil {
 		return nil, errors.New(awserrors.ErrorInvalidParameter)
 	}
 	return natsmsg.NATSRequest[acm.AddTagsToCertificateOutput](ctx, natsConn, "acm.AddTagsToCertificate", input, natsTimeout, accountID)
@@ -89,7 +90,7 @@ func AddTagsToCertificate(ctx context.Context, natsConn *nats.Conn, accountID st
 // RemoveTagsFromCertificate — CertificateManager.RemoveTagsFromCertificate.
 func RemoveTagsFromCertificate(ctx context.Context, natsConn *nats.Conn, accountID string, body []byte) (*acm.RemoveTagsFromCertificateOutput, error) {
 	input := new(acm.RemoveTagsFromCertificateInput)
-	if err := unmarshalIfBody(body, input); err != nil {
+	if err := envelope.UnmarshalJSONBody(body, input); err != nil {
 		return nil, errors.New(awserrors.ErrorInvalidParameter)
 	}
 	return natsmsg.NATSRequest[acm.RemoveTagsFromCertificateOutput](ctx, natsConn, "acm.RemoveTagsFromCertificate", input, natsTimeout, accountID)

spinifex/domains/acm/awsapi/certificates_test.go

@@ -2,7 +2,6 @@ package awsapi
 
 import (
 	"context"
-	"net/http/httptest"
 	"testing"
 
 	"github.com/mulgadc/spinifex/internal/testkit"
@@ -11,23 +10,6 @@ import (
 	"github.com/stretchr/testify/require"
 )
 
-func TestUnmarshalIfBody(t *testing.T) {
-	var out struct {
-		A string `json:"a"`
-	}
-	require.NoError(t, unmarshalIfBody(nil, &out))
-	require.NoError(t, unmarshalIfBody([]byte(`{"a":"x"}`), &out))
-	assert.Equal(t, "x", out.A)
-	require.Error(t, unmarshalIfBody([]byte("{bad"), &out))
-}
-
-func TestWriteJSONResponse(t *testing.T) {
-	w := httptest.NewRecorder()
-	WriteJSONResponse(w, map[string]string{"CertificateArn": "arn:aws:acm:x"})
-	assert.Equal(t, JSONContentType, w.Header().Get("Content-Type"))
-	assert.Contains(t, w.Body.String(), "CertificateArn")
-}
-
 // Invalid JSON bodies are rejected before any NATS round-trip.
 func TestOps_InvalidBodyRejected(t *testing.T) {
 	bad := []byte("{not json")

spinifex/domains/acm/awsapi/handler.go

@@ -2,40 +2,3 @@
 // NATS-backed service contract. Generic gateway routing and authorization stay
 // at the ingress boundary.
 package awsapi
-
-import (
-	"encoding/json"
-	"log/slog"
-	"net/http"
-
-	"github.com/aws/aws-sdk-go/private/protocol/json/jsonutil"
-	"github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
-)
-
-// JSONContentType is the AWS JSON 1.1 content type ACM clients expect.
-const JSONContentType = "application/x-amz-json-1.1"
-
-// WriteJSONResponse serialises obj as a 200 AWS JSON 1.1 response using the
-// SDK's own jsonutil marshaler, which encodes time.Time as epoch seconds
-// (encoding/json emits RFC3339 strings the SDK rejects).
-func WriteJSONResponse(w http.ResponseWriter, obj any) {
-	body, err := jsonutil.BuildJSON(obj)
-	if err != nil {
-		slog.Error("ACM: failed to marshal response JSON", "err", err)
-		http.Error(w, awserrors.ErrorInternalError, http.StatusInternalServerError)
-		return
-	}
-	w.Header().Set("Content-Type", JSONContentType)
-	w.WriteHeader(http.StatusOK)
-	if _, err := w.Write(body); err != nil {
-		slog.Error("ACM: failed to write response", "err", err)
-	}
-}
-
-// unmarshalIfBody decodes body into out only when non-empty.
-func unmarshalIfBody(body []byte, out any) error {
-	if len(body) == 0 {
-		return nil
-	}
-	return json.Unmarshal(body, out)
-}

spinifex/domains/acm/awsapi/registration.go

@@ -3,16 +3,14 @@ package awsapi
 import (
 	"context"
 	"errors"
-	"io"
 	"log/slog"
 	"maps"
 	"net/http"
 	"slices"
-	"strings"
 
-	"github.com/mulgadc/bluebottle/pkg/sigv4"
 	"github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
 	"github.com/mulgadc/spinifex/spinifex/ingress/aws/dispatch"
+	"github.com/mulgadc/spinifex/spinifex/ingress/aws/envelope"
 	"github.com/nats-io/nats.go"
 )
 
@@ -66,7 +64,7 @@ var Actions = map[string]Handler{
 func NewRegistration(deps Deps) dispatch.Registration {
 	return dispatch.Registration{
 		Service:       ServiceName,
-		ResolveAction: resolveAction,
+		ResolveAction: dispatch.ResolveTargetAction,
 		Dispatch:      deps.dispatch,
 		Errors:        dispatch.ErrorEnvelopeJSON,
 		Inventory:     OperationInventory(),
@@ -81,25 +79,11 @@ func OperationInventory() dispatch.Inventory {
 	}
 }
 
-// resolveAction is the one X-Amz-Target parser telemetry and dispatch share.
-func resolveAction(r *http.Request) string {
-	return actionFromTarget(r.Header.Get("X-Amz-Target"))
-}
-
-// actionFromTarget extracts the action suffix from an X-Amz-Target header.
-// Any "<Prefix>.<Action>" or bare "<Action>" form is accepted.
-func actionFromTarget(target string) string {
-	if _, after, ok := strings.CutLast(target, "."); ok {
-		return after
-	}
-	return target
-}
-
 // dispatch resolves, scopes and authorizes the request from one read of its
 // body, then runs the action. Errors are awserrors codes the gateway renders.
 func (d Deps) dispatch(w http.ResponseWriter, inv dispatch.Invocation) error {
 	r := inv.Request
-	action := resolveAction(r)
+	action := dispatch.ResolveTargetAction(r)
 	if action == "" {
 		return errors.New(awserrors.ErrorMissingAction)
 	}
@@ -118,7 +102,7 @@ func (d Deps) dispatch(w http.ResponseWriter, inv dispatch.Invocation) error {
 		return errors.New(awserrors.ErrorInternalError)
 	}
 
-	body, err := readBoundedBody(r)
+	body, err := dispatch.ReadBoundedBody(r)
 	if err != nil {
 		slog.ErrorContext(r.Context(), "ACM dispatch: failed to read body", "err", err)
 		return err
@@ -141,19 +125,6 @@ func (d Deps) dispatch(w http.ResponseWriter, inv dispatch.Invocation) error {
 		return err
 	}
 
-	WriteJSONResponse(w, output)
+	envelope.WriteJSONResult(w, output)
 	return nil
 }
-
-// readBoundedBody reads the body under the SigV4 payload cap, so a body read
-// ahead of the policy gate can neither bypass it nor exhaust memory.
-func readBoundedBody(r *http.Request) ([]byte, error) {
-	body, err := io.ReadAll(io.LimitReader(r.Body, sigv4.MaxPayloadLen+1))
-	if err != nil {
-		return nil, errors.New(awserrors.ErrorInvalidParameterValue)
-	}
-	if int64(len(body)) > sigv4.MaxPayloadLen {
-		return nil, errors.New(awserrors.ErrorRequestEntityTooLarge)
-	}
-	return body, nil
-}

spinifex/domains/acm/awsapi/registration_test.go

@@ -58,18 +58,6 @@ func dispatchWith(t *testing.T, deps Deps, inv dispatch.Invocation) (*httptest.R
 	return w, NewRegistration(deps).Dispatch(w, inv)
 }
 
-func TestActionFromTarget(t *testing.T) {
-	for target, want := range map[string]string{
-		"CertificateManager.ListCertificates": "ListCertificates",
-		"a.b.ListCertificates":                "ListCertificates",
-		"ListCertificates":                    "ListCertificates",
-		"CertificateManager.":                 "",
-		"":                                    "",
-	} {
-		assert.Equal(t, want, actionFromTarget(target), "target %q", target)
-	}
-}
-
 // TestActionsScopeTableIsExhaustive is what stops the next ACM action being
 // added with a silent account-wide grant. It asserts both directions, so a
 // scope left behind by a deleted or renamed action fails too.

spinifex/domains/admission/quota/override.go

@@ -230,7 +230,7 @@ func (s *Service) clearAccountQuota(ctx context.Context, accountID string) error
 			return nil
 		case errors.Is(err, jetstream.ErrKeyNotFound):
 			return nil
-		case isRevisionConflict(err):
+		case kvutil.IsRevisionConflict(err):
 			// A concurrent write landed between the read and the guarded
 			// delete; retry against whatever revision now holds.
 			continue

spinifex/domains/admission/quota/seal.go

@@ -62,20 +62,6 @@ func (s *Service) checkTerminating(ctx context.Context, accountID string) error
 	return nil
 }
 
-// isRevisionConflict reports a lost CAS race to retry, including the bare
-// wrong-last-sequence APIError from Create's retry against a delete marker.
-// A private copy of domains/ec2/account's helper, which this package may not import.
-func isRevisionConflict(err error) bool {
-	if errors.Is(err, jetstream.ErrKeyExists) || errors.Is(err, jetstream.ErrKeyRevisionMismatch) {
-		return true
-	}
-	if apiErr, ok := errors.AsType[*jetstream.APIError](err); ok {
-		return apiErr.ErrorCode == jetstream.JSErrCodeStreamWrongLastSequence ||
-			apiErr.ErrorCode == jetstream.JSErrCodeStreamWrongLastSequenceConstant
-	}
-	return false
-}
-
 // sealBucket returns this gateway's override bucket, or get-or-creates it
 // when quotas are disabled here. Called only after checkTerminating accepts,
 // so a refused seal never creates the bucket.
@@ -150,7 +136,7 @@ func (s *Service) SealAccountOverrides(ctx context.Context, accountID string) (*
 		switch {
 		case writeErr == nil:
 			return &OverrideSeal{Revision: newRev}, nil
-		case isRevisionConflict(writeErr):
+		case kvutil.IsRevisionConflict(writeErr):
 			continue
 		default:
 			slog.ErrorContext(ctx, "Failed to write sealed quota override record", "accountID", accountID, "err", writeErr)

spinifex/domains/admission/quota/seal_test.go

@@ -4,7 +4,6 @@ import (
 	"context"
 	"encoding/json"
 	"errors"
-	"fmt"
 	"sync"
 	"testing"
 
@@ -433,16 +432,6 @@ func TestSealStoreErrorWritesNothing(t *testing.T) {
 	assert.Equal(t, before, lastOverrideSeq(t, js))
 }
 
-func TestIsRevisionConflictRecognisesTombstoneRetry(t *testing.T) {
-	assert.True(t, isRevisionConflict(jetstream.ErrKeyExists))
-	assert.True(t, isRevisionConflict(jetstream.ErrKeyRevisionMismatch))
-	assert.True(t, isRevisionConflict(&jetstream.APIError{ErrorCode: jetstream.JSErrCodeStreamWrongLastSequence}))
-	assert.True(t, isRevisionConflict(&jetstream.APIError{ErrorCode: jetstream.JSErrCodeStreamWrongLastSequenceConstant}))
-	assert.True(t, isRevisionConflict(fmt.Errorf("wrapped: %w", &jetstream.APIError{ErrorCode: jetstream.JSErrCodeStreamWrongLastSequence})))
-	assert.False(t, isRevisionConflict(errors.New("unrelated failure")))
-	assert.False(t, isRevisionConflict(nil))
-}
-
 // Clearing never writes when there is nothing active to clear: an absent key,
 // a tombstoned key, and a sealed key hidden behind a stale not-found Get all
 // leave the stream exactly where they found it.

spinifex/domains/ec2/account/seal.go

@@ -8,6 +8,7 @@ import (
 
 	awserrors "github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
 	awsidentifiers "github.com/mulgadc/spinifex/spinifex/foundation/aws/identifiers"
+	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
 	"github.com/nats-io/nats.go/jetstream"
 )
 
@@ -58,20 +59,6 @@ func (s *AccountSettingsServiceImpl) checkTerminating(ctx context.Context, accou
 	return nil
 }
 
-// isRevisionConflict reports a lost CAS race SealAccount should retry. Besides
-// the two sentinels, Create's retry against a delete marker returns a bare
-// *jetstream.APIError carrying a wrong-last-sequence code.
-func isRevisionConflict(err error) bool {
-	if errors.Is(err, jetstream.ErrKeyExists) || errors.Is(err, jetstream.ErrKeyRevisionMismatch) {
-		return true
-	}
-	if apiErr, ok := errors.AsType[*jetstream.APIError](err); ok {
-		return apiErr.ErrorCode == jetstream.JSErrCodeStreamWrongLastSequence ||
-			apiErr.ErrorCode == jetstream.JSErrCodeStreamWrongLastSequenceConstant
-	}
-	return false
-}
-
 // SealAccount seals accountID's settings record so no later mutation can
 // recreate it. It never relies on a read for its claim: the only success is a
 // leader-accepted write, evidenced by the returned revision.
@@ -122,7 +109,7 @@ func (s *AccountSettingsServiceImpl) SealAccount(ctx context.Context, accountID
 		switch {
 		case writeErr == nil:
 			return &AccountSettingsSeal{Revision: newRev}, nil
-		case isRevisionConflict(writeErr):
+		case kvutil.IsRevisionConflict(writeErr):
 			continue
 		default:
 			slog.ErrorContext(ctx, "Failed to write sealed account settings record", "accountID", accountID, "err", writeErr)

spinifex/domains/ec2/account/seal_test.go

@@ -4,7 +4,6 @@ import (
 	"context"
 	"encoding/json"
 	"errors"
-	"fmt"
 	"sync"
 	"testing"
 
@@ -397,16 +396,6 @@ func TestSealStoreErrorWritesNothing(t *testing.T) {
 	assert.Equal(t, before, lastSettingsSeq(t, svc))
 }
 
-func TestIsRevisionConflictRecognisesTombstoneRetry(t *testing.T) {
-	assert.True(t, isRevisionConflict(jetstream.ErrKeyExists))
-	assert.True(t, isRevisionConflict(jetstream.ErrKeyRevisionMismatch))
-	assert.True(t, isRevisionConflict(&jetstream.APIError{ErrorCode: jetstream.JSErrCodeStreamWrongLastSequence}))
-	assert.True(t, isRevisionConflict(&jetstream.APIError{ErrorCode: jetstream.JSErrCodeStreamWrongLastSequenceConstant}))
-	assert.True(t, isRevisionConflict(fmt.Errorf("wrapped: %w", &jetstream.APIError{ErrorCode: jetstream.JSErrCodeStreamWrongLastSequence})))
-	assert.False(t, isRevisionConflict(errors.New("unrelated failure")))
-	assert.False(t, isRevisionConflict(nil))
-}
-
 // An account IAM does not hold as TERMINATING refuses the seal before any
 // settings-KV write, whether the key is present or absent.
 func TestSealRefusedForActiveAccount(t *testing.T) {

spinifex/domains/ec2/accountcleanup/cleanup.go

@@ -49,11 +49,13 @@ type PlacementGroupOwner interface {
 	CleanupAccount(ctx context.Context, accountID string, referenced func(groupID, groupName string) string) (*ec2placementgroup.AccountCleanup, error)
 }
 
-// LaunchTemplateOwner is the launch-template owner's account scan and residue
-// cleanup; live templates are never deleted by account cleanup.
+// LaunchTemplateOwner is the launch-template owner's account scan, residue
+// cleanup and live-template delete; a run deletes live templates only when its
+// request asks and the consumer check completed.
 type LaunchTemplateOwner interface {
 	ScanAccount(ctx context.Context, accountID string) (*ec2launchtemplate.AccountScan, error)
 	CleanupAccountResidue(ctx context.Context, accountID string, settledBefore time.Time) (*ec2launchtemplate.AccountCleanup, error)
+	DeleteLiveTemplates(ctx context.Context, accountID string, referenced func(id, name string) string) (*ec2launchtemplate.AccountTemplateDeletes, error)
 }
 
 // TagIndexOwner lists and clears the account's tag-index entries.
@@ -153,6 +155,11 @@ func (c *Cleanup) CleanupPlatform(ctx context.Context, req *Request) (*Result, e
 	var keyRun *keyPairRun
 	var groupRun *placementGroupRun
 	var templateRun *launchTemplateRun
+	// Live templates go first when asked: a version that names a key pair or
+	// group blocks it, so the reference read must see what remains.
+	if req.DeleteLiveTemplates && slices.Contains(kinds, ec2v1.PlatformKindLaunchTemplate) {
+		templateRun = c.cleanLaunchTemplates(ctx, acct, settledBefore, true)
+	}
 	if slices.Contains(kinds, ec2v1.PlatformKindKeyPair) || slices.Contains(kinds, ec2v1.PlatformKindPlacementGroup) {
 		refs := c.readReferences(ctx, acct)
 		if slices.Contains(kinds, ec2v1.PlatformKindKeyPair) {
@@ -162,8 +169,8 @@ func (c *Cleanup) CleanupPlatform(ctx context.Context, req *Request) (*Result, e
 			groupRun = c.cleanPlacementGroups(ctx, acct, refs)
 		}
 	}
-	if slices.Contains(kinds, ec2v1.PlatformKindLaunchTemplate) {
-		templateRun = c.cleanLaunchTemplates(ctx, acct, settledBefore)
+	if templateRun == nil && slices.Contains(kinds, ec2v1.PlatformKindLaunchTemplate) {
+		templateRun = c.cleanLaunchTemplates(ctx, acct, settledBefore, false)
 	}
 
 	// Tags are listed before the owner re-scans, so a resource created in

spinifex/domains/ec2/accountcleanup/cleanup_live_templates_test.go

@@ -0,0 +1,159 @@
+package accountcleanup
+
+//test:in-package — shares the owner fixtures and the injected clock.
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	"github.com/aws/aws-sdk-go/aws"
+	"github.com/aws/aws-sdk-go/service/ec2"
+	ec2v1 "github.com/mulgadc/spinifex/contracts/ec2/v1"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+// fakeConsumers answers what nodegroups name; a non-nil err is returned with
+// whatever partial answer it also holds.
+type fakeConsumers struct {
+	ids, names []string
+	err        error
+}
+
+func (f *fakeConsumers) AccountTemplateConsumers(context.Context, string) (*TemplateConsumers, error) {
+	out := &TemplateConsumers{TemplateIDs: map[string]bool{}, TemplateNames: map[string]bool{}}
+	markAll(out.TemplateIDs, f.ids)
+	markAll(out.TemplateNames, f.names)
+	return out, f.err
+}
+
+var _ TemplateConsumerReader = (*fakeConsumers)(nil)
+
+// liveDelete runs a cleanup that deletes live templates, with consumers as the
+// nodegroup reader (nil means none supplied).
+func (o *owners) liveDelete(t *testing.T, consumers TemplateConsumerReader) *Result {
+	t.Helper()
+	readers := o.readers()
+	readers.Consumers = consumers
+	c := New(o.keys, o.groups, o.templates, o.tags, readers)
+	c.now = func() time.Time { return time.Now().Add(time.Hour) }
+	res, err := c.CleanupPlatform(t.Context(), &Request{AccountID: testAccount, CorrelationRef: "ref-1", DeleteLiveTemplates: true})
+	require.NoError(t, err)
+	return res
+}
+
+func (o *owners) templateNamingKeyAndGroup(t *testing.T, name, key, group string) string {
+	t.Helper()
+	ltID := o.createTemplate(t, testAccount, name)
+	_, err := o.templates.CreateLaunchTemplateVersion(t.Context(), &ec2.CreateLaunchTemplateVersionInput{
+		LaunchTemplateId: aws.String(ltID),
+		LaunchTemplateData: &ec2.RequestLaunchTemplateData{
+			KeyName:   aws.String(key),
+			Placement: &ec2.LaunchTemplatePlacementRequest{GroupName: aws.String(group)},
+		},
+	}, testAccount)
+	require.NoError(t, err)
+	return ltID
+}
+
+func (o *owners) liveTemplateIDs(t *testing.T) []string {
+	t.Helper()
+	scan, err := o.templates.ScanAccount(t.Context(), testAccount)
+	require.NoError(t, err)
+	return scan.LiveTemplateIDs
+}
+
+// A template no nodegroup names is deleted in the same run as the key pair and
+// group its version names, which the reference check no longer sees.
+func TestCleanupPlatformDeletesAnUnnamedTemplateAndWhatItNames(t *testing.T) {
+	o := newOwners(t)
+	key := o.importKey(t, testAccount, "tkey")
+	group := o.createGroup(t, testAccount, "tgroup")
+	lt := o.templateNamingKeyAndGroup(t, "web", "tkey", "tgroup")
+
+	res := o.liveDelete(t, &fakeConsumers{})
+	assert.ElementsMatch(t, []Item{
+		item(lt, ec2v1.PlatformKindLaunchTemplate, DispositionDeleted, ""),
+		item(key, ec2v1.PlatformKindKeyPair, DispositionDeleted, ""),
+		item(group, ec2v1.PlatformKindPlacementGroup, DispositionDeleted, ""),
+	}, res.Items)
+	assert.Empty(t, o.liveTemplateIDs(t))
+}
+
+// A template a nodegroup names, by ID or by name, is kept and blocked, and so
+// are the key pair and group its versions name.
+func TestCleanupPlatformKeepsATemplateANodegroupNames(t *testing.T) {
+	for name, consumers := range map[string]*fakeConsumers{
+		"by id":   {},
+		"by name": {names: []string{"web"}},
+	} {
+		t.Run(name, func(t *testing.T) {
+			o := newOwners(t)
+			key := o.importKey(t, testAccount, "tkey")
+			group := o.createGroup(t, testAccount, "tgroup")
+			lt := o.templateNamingKeyAndGroup(t, "web", "tkey", "tgroup")
+			other := o.createTemplate(t, testAccount, "free")
+			if name == "by id" {
+				consumers.ids = []string{lt}
+			}
+
+			res := o.liveDelete(t, consumers)
+			assert.ElementsMatch(t, []Item{
+				item(lt, ec2v1.PlatformKindLaunchTemplate, DispositionBlocked, ReasonLaunchTemplateReference),
+				item(other, ec2v1.PlatformKindLaunchTemplate, DispositionDeleted, ""),
+				item(key, ec2v1.PlatformKindKeyPair, DispositionBlocked, ReasonLaunchTemplateReference),
+				item(group, ec2v1.PlatformKindPlacementGroup, DispositionBlocked, ReasonLaunchTemplateReference),
+			}, res.Items)
+			assert.Equal(t, []string{lt}, o.liveTemplateIDs(t))
+		})
+	}
+}
+
+// A consumer read that fails, returns a partial answer with an error, or is not
+// supplied deletes no template, key pair or group, and the kind is incomplete.
+func TestCleanupPlatformUnreadableConsumersDeleteNoTemplate(t *testing.T) {
+	cases := map[string]TemplateConsumerReader{
+		"error":        &fakeConsumers{err: errInjected},
+		"partial":      &fakeConsumers{ids: []string{"lt-0other"}, err: errInjected},
+		"not supplied": nil,
+	}
+	for name, consumers := range cases {
+		t.Run(name, func(t *testing.T) {
+			o := newOwners(t)
+			key := o.importKey(t, testAccount, "tkey")
+			lt := o.templateNamingKeyAndGroup(t, "web", "tkey", "none")
+
+			res := o.liveDelete(t, consumers)
+			assert.Contains(t, res.Items, item(lt, ec2v1.PlatformKindLaunchTemplate, DispositionBlocked, ReasonReferenceCheckFailed))
+			assert.Contains(t, res.Items, item(key, ec2v1.PlatformKindKeyPair, DispositionBlocked, ReasonLaunchTemplateReference))
+			assert.Equal(t, ec2v1.PlatformStatusScanIncomplete, statuses(res)[ec2v1.PlatformKindLaunchTemplate])
+			assert.Equal(t, []string{lt}, o.liveTemplateIDs(t))
+		})
+	}
+}
+
+// A template only a DELETING nodegroup still names is as blocked as any other:
+// the reader reports every status, so the cleanup never sees a status to trust.
+func TestCleanupPlatformKeepsATemplateADeletingNodegroupStillNames(t *testing.T) {
+	o := newOwners(t)
+	lt := o.createTemplate(t, testAccount, "draining")
+
+	res := o.liveDelete(t, &fakeConsumers{ids: []string{lt}})
+	assert.Equal(t, []Item{item(lt, ec2v1.PlatformKindLaunchTemplate, DispositionBlocked, ReasonLaunchTemplateReference)}, res.Items)
+	assert.Equal(t, []string{lt}, o.liveTemplateIDs(t))
+}
+
+// Without the request flag a supplied consumer reader changes nothing: live
+// templates are never deleted, as the exercise relies on.
+func TestCleanupPlatformLiveTemplatesSurviveWithoutTheRequestFlag(t *testing.T) {
+	o := newOwners(t)
+	lt := o.createTemplate(t, testAccount, "web")
+	readers := o.readers()
+	readers.Consumers = &fakeConsumers{}
+	c := New(o.keys, o.groups, o.templates, o.tags, readers)
+
+	res := run(t, c)
+	assert.Equal(t, []Item{item(lt, ec2v1.PlatformKindLaunchTemplate, DispositionBlocked, ReasonReferenceUnknown)}, res.Items)
+	assert.Equal(t, []string{lt}, o.liveTemplateIDs(t))
+}

spinifex/domains/ec2/accountcleanup/kinds.go

@@ -204,13 +204,18 @@ type launchTemplateRun struct {
 	before  *ec2launchtemplate.AccountScan
 	cleaned *ec2launchtemplate.AccountCleanup
 	did     attempts
+	// deleteLive is true when this run tried to delete live templates.
+	deleteLive bool
 }
 
-func (c *Cleanup) cleanLaunchTemplates(ctx context.Context, acct string, settledBefore time.Time) *launchTemplateRun {
-	run := &launchTemplateRun{kind: c.startKind(ec2v1.PlatformKindLaunchTemplate), did: newAttempts()}
+func (c *Cleanup) cleanLaunchTemplates(ctx context.Context, acct string, settledBefore time.Time, deleteLive bool) *launchTemplateRun {
+	run := &launchTemplateRun{kind: c.startKind(ec2v1.PlatformKindLaunchTemplate), did: newAttempts(), deleteLive: deleteLive}
 	if before, err := c.templates.ScanAccount(ctx, acct); err == nil {
 		run.before = before
 	}
+	if deleteLive {
+		c.deleteLiveTemplates(ctx, acct, run)
+	}
 	cleaned, err := c.templates.CleanupAccountResidue(ctx, acct, settledBefore)
 	if err != nil {
 		slog.WarnContext(ctx, "Platform cleanup: launch-template residue cleanup failed", "err", err)
@@ -218,10 +223,50 @@ func (c *Cleanup) cleanLaunchTemplates(ctx context.Context, acct string, settled
 		return run
 	}
 	run.cleaned = cleaned
-	keepAll(run.did.kept, cleaned.KeptTemplateIDs, ReasonReferenceUnknown)
+	for _, id := range cleaned.KeptTemplateIDs {
+		if run.did.kept[id] == "" {
+			run.did.kept[id] = ReasonReferenceUnknown
+		}
+	}
 	return run
 }
 
+// deleteLiveTemplates deletes the live templates no consumer names. Without a
+// complete consumer answer nothing is deleted and every live template is kept
+// as unchecked, so an unreadable consumer never reads as "no references".
+func (c *Cleanup) deleteLiveTemplates(ctx context.Context, acct string, run *launchTemplateRun) {
+	var consumers *TemplateConsumers
+	if c.refs.Consumers != nil {
+		var err error
+		consumers, err = c.refs.Consumers.AccountTemplateConsumers(ctx, acct)
+		if err != nil {
+			slog.WarnContext(ctx, "Platform cleanup: template consumer check failed", "err", err)
+			consumers = nil
+		}
+	}
+	if consumers == nil {
+		if run.before != nil {
+			keepAll(run.did.kept, run.before.LiveTemplateIDs, ReasonReferenceCheckFailed)
+		}
+		run.kind.incomplete(ReasonReferenceCheckFailed)
+		return
+	}
+	deletes, err := c.templates.DeleteLiveTemplates(ctx, acct, func(id, name string) string {
+		if consumers.names(id, name) {
+			return ReasonLaunchTemplateReference
+		}
+		return ""
+	})
+	if err != nil {
+		slog.WarnContext(ctx, "Platform cleanup: live template delete failed", "err", err)
+		run.kind.incomplete(ec2v1.PlatformReasonStoreListFailed)
+		return
+	}
+	markAll(run.did.deleted, deletes.DeletedTemplateIDs)
+	markAll(run.did.failed, deletes.FailedTemplateIDs)
+	maps.Copy(run.did.kept, deletes.Referenced)
+}
+
 func (c *Cleanup) finishLaunchTemplates(ctx context.Context, acct string, run *launchTemplateRun, res *Result) ownedIDs {
 	defer c.finishKind(run.kind)
 	after, err := c.templates.ScanAccount(ctx, acct)
@@ -250,8 +295,10 @@ func (c *Cleanup) finishLaunchTemplates(ctx context.Context, acct string, run *l
 	}
 	// Live templates are never deleted here, so every one still present is
 	// blocked on the unknown EKS nodegroup reference, whenever it appeared.
-	for _, id := range afterLive {
-		run.did.kept[id] = ReasonReferenceUnknown
+	if !run.deleteLive {
+		for _, id := range afterLive {
+			run.did.kept[id] = ReasonReferenceUnknown
+		}
 	}
 	for _, id := range sortedUnion(beforeLive, kept, afterLive) {
 		disposition, reason := itemDisposition(id, run.did, present[id], after != nil)

spinifex/domains/ec2/accountcleanup/references.go

@@ -35,17 +35,38 @@ type TemplateReferenceReader interface {
 	AccountTemplateReferences(ctx context.Context, accountID string) (*ec2launchtemplate.TemplateReferences, error)
 }
 
+// TemplateConsumers are the IDs and names of launch templates that records
+// outside EC2, such as EKS nodegroups, still name. They select what to keep and
+// are never reported.
+type TemplateConsumers struct {
+	TemplateIDs   map[string]bool
+	TemplateNames map[string]bool
+}
+
+// TemplateConsumerReader reads which launch templates records outside EC2 name,
+// every status included, complete or an error. The consumer's owner answers it;
+// EC2 never reads those records.
+type TemplateConsumerReader interface {
+	AccountTemplateConsumers(ctx context.Context, accountID string) (*TemplateConsumers, error)
+}
+
+func (c *TemplateConsumers) names(id, name string) bool {
+	return c.TemplateIDs[id] || c.TemplateNames[name]
+}
+
 var (
 	_ SpotReferenceReader     = (*ec2spotinstance.SpotInstanceServiceImpl)(nil)
 	_ TemplateReferenceReader = (*ec2launchtemplate.LaunchTemplateServiceImpl)(nil)
 )
 
 // ReferenceReaders are the checks run before key pairs and placement groups
-// are deleted. A nil reader counts as a failed check.
+// are deleted. A nil reader counts as a failed check. Consumers is read only
+// by a request that deletes live templates, where nil is also a failed check.
 type ReferenceReaders struct {
 	Instances InstanceReferenceReader
 	Spot      SpotReferenceReader
 	Templates TemplateReferenceReader
+	Consumers TemplateConsumerReader
 }
 
 func newReferences() *References {

spinifex/domains/ec2/accountcleanup/types.go

@@ -10,6 +10,11 @@ type Request struct {
 
 	// Kinds limits the run to these platform kinds. Empty means all three.
 	Kinds []string `json:"kinds,omitempty"`
+
+	// DeleteLiveTemplates lets the run delete live launch templates that no
+	// nodegroup names, using the TemplateConsumerReader. Without it live
+	// templates are never deleted.
+	DeleteLiveTemplates bool `json:"delete_live_templates,omitempty"`
 }
 
 // Result is what the owner stores hold after one run. Kinds reuse the ec2.v1
@@ -88,7 +93,8 @@ const (
 	ReasonInstanceReference = "instance-reference"
 	// ReasonSpotRequestReference: an open or active spot request names it.
 	ReasonSpotRequestReference = "spot-request-reference"
-	// ReasonLaunchTemplateReference: a version of a live launch template names it.
+	// ReasonLaunchTemplateReference: a version of a live launch template names a
+	// key pair or group, or a nodegroup names a template.
 	ReasonLaunchTemplateReference = "launch-template-reference"
 	// ReasonReferenceUnknown: a record EC2 cannot read may name it, so it is
 	// never deleted by account cleanup.

spinifex/domains/ec2/accountinventory/compute.go

@@ -0,0 +1,88 @@
+package accountinventory
+
+import (
+	"context"
+	"fmt"
+
+	ec2metadata "github.com/mulgadc/spinifex/spinifex/domains/ec2/ebs/metadata"
+	ec2spot "github.com/mulgadc/spinifex/spinifex/domains/ec2/spotinstance"
+)
+
+// SpotRequestScanner reads one account's spot requests, failing on any record
+// it cannot read.
+type SpotRequestScanner interface {
+	ScanAccount(ctx context.Context, accountID string) (*ec2spot.AccountScan, error)
+}
+
+// VolumeScanner reads one account's volume documents, failing on any document
+// it cannot read.
+type VolumeScanner interface {
+	ListVolumesStrict(ctx context.Context, accountID string) ([]ec2metadata.Volume, error)
+}
+
+var (
+	_ SpotRequestScanner = (*ec2spot.SpotInstanceServiceImpl)(nil)
+	_ VolumeScanner      = (*ec2metadata.Store)(nil)
+)
+
+// SpotRequestItem is one request the account holds.
+type SpotRequestItem struct {
+	RequestID string
+	State     string
+	Active    bool
+}
+
+// VolumeItem is one volume document the account holds.
+type VolumeItem struct {
+	VolumeID            string
+	State               string
+	AttachedInstance    string
+	DeleteOnTermination bool
+}
+
+// WithCompute adds the spot-request and volume owner stores.
+func (i *Inventory) WithCompute(spot SpotRequestScanner, volumes VolumeScanner) *Inventory {
+	i.spot = spot
+	i.volumes = volumes
+	return i
+}
+
+// SpotRequestInventory returns every spot request the account holds. A scan
+// that cannot be completed is an error, never a shorter list.
+func (i *Inventory) SpotRequestInventory(ctx context.Context, accountID string) ([]SpotRequestItem, error) {
+	if i.spot == nil {
+		return nil, fmt.Errorf("spot request inventory is not configured")
+	}
+	scan, err := i.spot.ScanAccount(ctx, accountID)
+	if err != nil {
+		return nil, fmt.Errorf("spot request inventory incomplete: %w", err)
+	}
+	items := make([]SpotRequestItem, 0, len(scan.Active)+len(scan.Terminal))
+	for _, r := range scan.Active {
+		items = append(items, SpotRequestItem{RequestID: r.RequestID, State: r.State, Active: true})
+	}
+	for _, r := range scan.Terminal {
+		items = append(items, SpotRequestItem{RequestID: r.RequestID, State: r.State})
+	}
+	return items, nil
+}
+
+// VolumeInventory returns every volume document the account holds, including
+// those already marked deleting. An unreadable document is an error.
+func (i *Inventory) VolumeInventory(ctx context.Context, accountID string) ([]VolumeItem, error) {
+	if i.volumes == nil {
+		return nil, fmt.Errorf("volume inventory is not configured")
+	}
+	docs, err := i.volumes.ListVolumesStrict(ctx, accountID)
+	if err != nil {
+		return nil, fmt.Errorf("volume inventory incomplete: %w", err)
+	}
+	items := make([]VolumeItem, 0, len(docs))
+	for _, v := range docs {
+		items = append(items, VolumeItem{
+			VolumeID: v.VolumeID, State: v.State, AttachedInstance: v.AttachedInstance,
+			DeleteOnTermination: v.DeleteOnTermination,
+		})
+	}
+	return items, nil
+}

spinifex/domains/ec2/accountinventory/compute_test.go

@@ -0,0 +1,82 @@
+package accountinventory
+
+import (
+	"context"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+
+	ec2metadata "github.com/mulgadc/spinifex/spinifex/domains/ec2/ebs/metadata"
+	ec2spot "github.com/mulgadc/spinifex/spinifex/domains/ec2/spotinstance"
+)
+
+type fakeSpot struct {
+	scan *ec2spot.AccountScan
+	err  error
+}
+
+func (f fakeSpot) ScanAccount(context.Context, string) (*ec2spot.AccountScan, error) {
+	return f.scan, f.err
+}
+
+type fakeVolumes struct {
+	docs []ec2metadata.Volume
+	err  error
+}
+
+func (f fakeVolumes) ListVolumesStrict(context.Context, string) ([]ec2metadata.Volume, error) {
+	return f.docs, f.err
+}
+
+func TestSpotRequestInventoryMarksActiveAndTerminal(t *testing.T) {
+	inv := New(nil, nil, nil, nil).WithCompute(fakeSpot{scan: &ec2spot.AccountScan{
+		Active:   []ec2spot.AccountSpotRequest{{RequestID: "sir-1", State: "active"}},
+		Terminal: []ec2spot.AccountSpotRequest{{RequestID: "sir-2", State: "cancelled"}},
+	}}, nil)
+
+	items, err := inv.SpotRequestInventory(t.Context(), testAccount)
+	require.NoError(t, err)
+	assert.Equal(t, []SpotRequestItem{
+		{RequestID: "sir-1", State: "active", Active: true},
+		{RequestID: "sir-2", State: "cancelled"},
+	}, items)
+}
+
+func TestSpotRequestInventoryScanFailureIsAnErrorNotAShorterList(t *testing.T) {
+	inv := New(nil, nil, nil, nil).WithCompute(fakeSpot{err: errList}, nil)
+
+	items, err := inv.SpotRequestInventory(t.Context(), testAccount)
+	require.ErrorIs(t, err, errList)
+	assert.Nil(t, items)
+}
+
+func TestVolumeInventoryReportsAttachmentAndDeletingDocuments(t *testing.T) {
+	inv := New(nil, nil, nil, nil).WithCompute(nil, fakeVolumes{docs: []ec2metadata.Volume{
+		{VolumeID: "vol-1", State: "in-use", AttachedInstance: "i-1", DeleteOnTermination: true},
+		{VolumeID: "vol-2", State: "deleting"},
+	}})
+
+	items, err := inv.VolumeInventory(t.Context(), testAccount)
+	require.NoError(t, err)
+	assert.Equal(t, []VolumeItem{
+		{VolumeID: "vol-1", State: "in-use", AttachedInstance: "i-1", DeleteOnTermination: true},
+		{VolumeID: "vol-2", State: "deleting"},
+	}, items)
+}
+
+func TestVolumeInventoryListFailureIsAnErrorNotAShorterList(t *testing.T) {
+	inv := New(nil, nil, nil, nil).WithCompute(nil, fakeVolumes{err: errList})
+
+	items, err := inv.VolumeInventory(t.Context(), testAccount)
+	require.ErrorIs(t, err, errList)
+	assert.Nil(t, items)
+}
+
+func TestComputeInventoriesRefuseWhenNotConfigured(t *testing.T) {
+	inv := New(nil, nil, nil, nil)
+	_, err := inv.SpotRequestInventory(t.Context(), testAccount)
+	require.Error(t, err)
+	_, err = inv.VolumeInventory(t.Context(), testAccount)
+	require.Error(t, err)
+}

spinifex/domains/ec2/accountinventory/inventory.go

@@ -61,6 +61,8 @@ type Inventory struct {
 	groups    PlacementGroupScanner
 	templates LaunchTemplateScanner
 	tags      TagIndexScanner
+	spot      SpotRequestScanner
+	volumes   VolumeScanner
 	now       func() time.Time
 }
 

spinifex/domains/ec2/awsapi/account/DescribeAccountAttributes.go

@@ -4,6 +4,7 @@
 package account
 
 import (
+	"github.com/mulgadc/spinifex/spinifex/foundation/aws/identifiers"
 	"log/slog"
 
 	"github.com/aws/aws-sdk-go/aws"
@@ -14,12 +15,7 @@ import (
 // but default-vpc is static; defaultVPCID looks that one up, is called only
 // when it is asked for, and returns "" for an account with no default VPC.
 func DescribeAccountAttributes(input *ec2.DescribeAccountAttributesInput, defaultVPCID func() (string, error)) (*ec2.DescribeAccountAttributesOutput, error) {
-	requestedAttrs := make(map[string]bool)
-	for _, name := range input.AttributeNames {
-		if name != nil {
-			requestedAttrs[*name] = true
-		}
-	}
+	requestedAttrs := identifiers.Set(input.AttributeNames)
 
 	returnAll := len(requestedAttrs) == 0
 

spinifex/domains/ec2/awsapi/actions.go

@@ -151,7 +151,7 @@ var ec2Actions = map[string]ec2Action{
 		launchQuotaCheck := func() error {
 			return c.Quota.EnforceLaunch(ctx, accountID, aws.StringValue(input.InstanceType), int(aws.Int64Value(input.MaxCount)))
 		}
-		reservation, err := ec2instanceapi.RunInstances(ctx, input, c.NATS, c.IAM, accountID, passRoleCheck, launchQuotaCheck, c.ExpectedNodes)
+		reservation, err := ec2instanceapi.RunInstances(ctx, input, c.NATS, c.clientTokenBucket, c.IAM, accountID, passRoleCheck, launchQuotaCheck, c.ExpectedNodes)
 		if err != nil {
 			return nil, err
 		}
@@ -436,7 +436,7 @@ var ec2Actions = map[string]ec2Action{
 		passRoleCheck := func(roleARN string) error {
 			return c.passRole(roleARN)
 		}
-		return ec2spotinstanceapi.RequestSpotInstances(ctx, input, c.NATS, c.IAM, accountID, c.AZ, passRoleCheck, c.Quota, c.ExpectedNodes)
+		return ec2spotinstanceapi.RequestSpotInstances(ctx, input, c.NATS, c.clientTokenBucket, c.IAM, accountID, c.AZ, passRoleCheck, c.Quota, c.ExpectedNodes)
 	}),
 	"DescribeSpotInstanceRequests": ec2Handler(func(ctx context.Context, input *ec2.DescribeSpotInstanceRequestsInput, c *call, accountID string) (any, error) {
 		return ec2spotinstanceapi.DescribeSpotInstanceRequests(ctx, input, c.NATS, accountID)

spinifex/domains/ec2/awsapi/idem/idem.go

@@ -4,10 +4,15 @@
 package idem
 
 import (
+	"context"
+	"errors"
+	"log/slog"
 	"reflect"
 	"time"
 
+	"github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
 	"github.com/mulgadc/spinifex/spinifex/foundation/lifecycle/idempotency"
+	"github.com/nats-io/nats.go/jetstream"
 )
 
 const (
@@ -54,3 +59,45 @@ func TokenAndParams(input any) (token, paramHash string, ok bool) {
 	clone.FieldByName(tokenField).Set(reflect.Zero(field.Type()))
 	return field.Elem().String(), idempotency.ParamHash(clone.Addr().Interface()), true
 }
+
+// Bucket returns the KV bucket that holds the token records. It is called only
+// when a request carries a token, so tokenless requests never bind it.
+type Bucket func(ctx context.Context) (jetstream.KeyValue, error)
+
+// Run runs work under the input's ClientToken, storing its result under
+// namespace. Requests without a token run unwrapped, which is what AWS does:
+// the token is how a client opts in.
+func Run[Out any](
+	ctx context.Context,
+	bucket Bucket,
+	namespace, accountID string,
+	input any,
+	work func() (Out, error),
+) (Out, error) {
+	var zero Out
+	// Hashed before work runs, so work that mutates its input cannot change what
+	// a retry of the same token hashes to.
+	token, paramHash, ok := TokenAndParams(input)
+	if !ok {
+		return work()
+	}
+
+	kv, kerr := bucket(ctx)
+	if kerr != nil {
+		// Running anyway would create the duplicate the token was sent to
+		// prevent, so this fails rather than degrading to no idempotency.
+		slog.ErrorContext(ctx, "client-token store unavailable", "namespace", namespace, "err", kerr)
+		return zero, errors.New(awserrors.ErrorServerInternal)
+	}
+
+	store := idempotency.NewStore[Out](kv, namespace)
+	output, err := idempotency.Do(ctx, store, accountID, token, paramHash, work)
+	switch {
+	case errors.Is(err, idempotency.ErrParamMismatch):
+		return zero, errors.New(awserrors.ErrorIdempotentParameterMismatch)
+	case errors.Is(err, idempotency.ErrUnavailable):
+		slog.ErrorContext(ctx, "client-token replay unavailable", "namespace", namespace, "token", token, "err", err)
+		return zero, errors.New(awserrors.ErrorServerInternal)
+	}
+	return output, err
+}

spinifex/domains/ec2/awsapi/idem/idem_test.go

@@ -1,11 +1,15 @@
 package idem_test
 
 import (
+	"context"
+	"errors"
 	"testing"
 
 	"github.com/aws/aws-sdk-go/aws"
 	"github.com/aws/aws-sdk-go/service/ec2"
 	ec2idemapi "github.com/mulgadc/spinifex/spinifex/domains/ec2/awsapi/idem"
+	"github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
+	"github.com/nats-io/nats.go/jetstream"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 )
@@ -79,3 +83,21 @@ func TestTokenAndParams_LeavesTheInputIntact(t *testing.T) {
 	require.NotNil(t, input.ClientToken)
 	assert.Equal(t, "tok", *input.ClientToken)
 }
+
+// A token the store cannot honour fails the request rather than running it
+// unprotected; a request without a token never touches the store.
+func TestRun_BucketFailure(t *testing.T) {
+	t.Parallel()
+	broken := func(context.Context) (jetstream.KeyValue, error) { return nil, errors.New("no jetstream") }
+	ran := false
+	work := func() (string, error) { ran = true; return "ok", nil }
+
+	_, err := ec2idemapi.Run(t.Context(), broken, "", "acct",
+		&ec2.RunInstancesInput{ClientToken: aws.String("tok")}, work)
+	require.EqualError(t, err, awserrors.ErrorServerInternal)
+	assert.False(t, ran)
+
+	out, err := ec2idemapi.Run(t.Context(), broken, "", "acct", &ec2.RunInstancesInput{}, work)
+	require.NoError(t, err)
+	assert.Equal(t, "ok", out)
+}

spinifex/domains/ec2/awsapi/idempotency.go

@@ -4,7 +4,6 @@ import (
 	"context"
 	"errors"
 	"fmt"
-	"log/slog"
 
 	ec2idemapi "github.com/mulgadc/spinifex/spinifex/domains/ec2/awsapi/idem"
 	"github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
@@ -43,7 +42,9 @@ func ec2IdempotentHandler[In, Out any](handler func(ctx context.Context, input *
 			if !ok {
 				return nil, errors.New(awserrors.ErrorInternalError)
 			}
-			output, err := runEC2Idempotent(ctx, action, c, accountID, input, func() (Out, error) {
+			// Namespaced by action: one bucket holds every EC2 action's records, and
+			// without the prefix one action's payload could be decoded as another's Out.
+			output, err := ec2idemapi.Run(ctx, c.clientTokenBucket, action, accountID, input, func() (Out, error) {
 				return handler(ctx, input, c, accountID)
 			})
 			if err != nil {
@@ -53,43 +54,3 @@ func ec2IdempotentHandler[In, Out any](handler func(ctx context.Context, input *
 		},
 	}
 }
-
-// runEC2Idempotent runs work under the input's ClientToken. Requests without one
-// run unwrapped, which is what AWS does: the token is how a client opts in.
-func runEC2Idempotent[In, Out any](
-	ctx context.Context,
-	action string,
-	c *call,
-	accountID string,
-	input *In,
-	work func() (Out, error),
-) (Out, error) {
-	var zero Out
-	// Hashed before the handler runs, so a handler that mutates its input cannot
-	// change what a retry of the same token hashes to.
-	token, paramHash, ok := ec2idemapi.TokenAndParams(input)
-	if !ok {
-		return work()
-	}
-
-	kv, kerr := c.clientTokenBucket(ctx)
-	if kerr != nil {
-		// Running anyway would create the duplicate the token was sent to
-		// prevent, so this fails rather than degrading to no idempotency.
-		slog.ErrorContext(ctx, "client-token store unavailable", "action", action, "err", kerr)
-		return zero, errors.New(awserrors.ErrorServerInternal)
-	}
-
-	// Namespaced by action: one bucket holds every EC2 action's records, and
-	// without the prefix one action's payload could be decoded as another's Out.
-	store := idempotency.NewStore[Out](kv, action)
-	output, err := idempotency.Do(ctx, store, accountID, token, paramHash, work)
-	switch {
-	case errors.Is(err, idempotency.ErrParamMismatch):
-		return zero, errors.New(awserrors.ErrorIdempotentParameterMismatch)
-	case errors.Is(err, idempotency.ErrUnavailable):
-		slog.ErrorContext(ctx, "client-token replay unavailable", "action", action, "token", token, "err", err)
-		return zero, errors.New(awserrors.ErrorServerInternal)
-	}
-	return output, err
-}

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

@@ -0,0 +1,151 @@
+package instance
+
+import (
+	"context"
+	"encoding/json"
+	"log/slog"
+	"slices"
+
+	"github.com/aws/aws-sdk-go/aws"
+	"github.com/aws/aws-sdk-go/service/ec2"
+	awserrors "github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
+	natsmsg "github.com/mulgadc/spinifex/spinifex/foundation/messaging/nats"
+	"github.com/nats-io/nats.go"
+)
+
+// accountInstanceRecordsSubject serves every non-terminated instance record an
+// account owns from the shared store, whichever node holds the instance.
+const accountInstanceRecordsSubject = "ec2.DescribeAccountInstanceRecords"
+
+// DescribeAccountInstancesStrict lists every non-terminated instance the
+// account owns, or returns a retryable ServiceUnavailable. Existence comes from
+// the shared record store, so an instance no answering node holds is still
+// listed with its recorded state; node answers are joined on for live state.
+// Anything that could make the list shorter than the truth is an error: a
+// configured node that did not answer, a frame from a node outside nodeIDs, a
+// record the owner could not project, or a store that did not answer.
+func DescribeAccountInstancesStrict(ctx context.Context, natsConn *nats.Conn, nodeIDs []string, accountID string, opts ...DescribeOption) (*ec2.DescribeInstancesOutput, error) {
+	if len(nodeIDs) == 0 {
+		slog.ErrorContext(ctx, "DescribeInstances: strict listing refused, configured node set unknown")
+		return nil, incompleteListingError()
+	}
+	cfg := describeConfig{fanoutTimeout: defaultDescribeFanoutTimeout}
+	for _, opt := range opts {
+		opt(&cfg)
+	}
+
+	jsonData, err := json.Marshal(&ec2.DescribeInstancesInput{})
+	if err != nil {
+		return nil, awserrors.Errorf(awserrors.ErrorServerInternal, "marshal describe input: %v", err)
+	}
+
+	records, ok := queryInstanceBucket(ctx, natsConn, accountInstanceRecordsSubject, jsonData, accountID)
+	if !ok {
+		slog.WarnContext(ctx, "DescribeInstances: strict listing refused, instance record store did not answer")
+		return nil, incompleteListingError()
+	}
+
+	// Collect to the deadline with no responder grace, so a reply from a node
+	// outside the set that lands after the configured nodes is still seen.
+	frames, sum, err := natsmsg.Gather(ctx, natsConn, "ec2.DescribeInstances", jsonData, natsmsg.GatherOpts{
+		Timeout:            cfg.fanoutTimeout,
+		AccountID:          accountID,
+		ExpectedResponders: len(nodeIDs),
+		Mode:               natsmsg.CollectUntilDeadline,
+	})
+	if err != nil {
+		return nil, err
+	}
+	live, complete := judgeIdentityCompleteness(ctx, frames, sum, nodeIDs)
+	if stray := unconfiguredNodes(frames, sum, nodeIDs); len(stray) > 0 {
+		slog.WarnContext(ctx, "DescribeInstances: strict listing refused, frame from a node outside the configured set",
+			"unconfigured_nodes", stray)
+		return nil, incompleteListingError()
+	}
+	if !complete {
+		slog.WarnContext(ctx, "DescribeInstances: strict listing refused, fan-out incomplete", "configured_nodes", len(nodeIDs))
+		return nil, incompleteListingError()
+	}
+
+	joined, nodeOnly, malformed := joinLiveOntoRecords(records, live)
+	if malformed {
+		slog.WarnContext(ctx, "DescribeInstances: strict listing refused, a reservation or instance has no usable identity")
+		return nil, incompleteListingError()
+	}
+	if len(nodeOnly) > 0 {
+		slog.WarnContext(ctx, "DescribeInstances: strict listing refused, node reports instances with no record",
+			"instances", nodeOnly)
+		return nil, incompleteListingError()
+	}
+	return &ec2.DescribeInstancesOutput{Reservations: joined}, nil
+}
+
+// unconfiguredNodes returns the node IDs that answered, with a success or an
+// error, but are not in nodeIDs.
+func unconfiguredNodes(frames []natsmsg.Frame, sum natsmsg.Summary, nodeIDs []string) []string {
+	var stray []string
+	note := func(id string) {
+		if id != "" && !slices.Contains(nodeIDs, id) && !slices.Contains(stray, id) {
+			stray = append(stray, id)
+		}
+	}
+	for _, frame := range frames {
+		note(frame.NodeID)
+	}
+	for id := range sum.Responders {
+		note(id)
+	}
+	slices.Sort(stray)
+	return stray
+}
+
+// joinLiveOntoRecords starts from the recorded instances and replaces each with
+// the answering node's view of it. Instances only a node reports are returned
+// by ID: a lost record write can cause one, and it must not be dropped. A nil
+// reservation, instance or instance ID is reported as malformed, never skipped.
+func joinLiveOntoRecords(records, live []*ec2.Reservation) ([]*ec2.Reservation, []string, bool) {
+	type entry struct {
+		reservation *ec2.Reservation
+		instance    *ec2.Instance
+	}
+	byID := map[string]*entry{}
+	var order []string
+	malformed := false
+	add := func(reservations []*ec2.Reservation) {
+		for _, res := range reservations {
+			if res == nil {
+				malformed = true
+				continue
+			}
+			for _, inst := range res.Instances {
+				if inst == nil || aws.StringValue(inst.InstanceId) == "" {
+					malformed = true
+					continue
+				}
+				id := aws.StringValue(inst.InstanceId)
+				if existing, ok := byID[id]; ok {
+					existing.instance = inst
+					continue
+				}
+				byID[id] = &entry{reservation: res, instance: inst}
+				order = append(order, id)
+			}
+		}
+	}
+	add(records)
+	recorded := len(order)
+	add(live)
+	nodeOnly := slices.Clone(order[recorded:])
+
+	out := make([]*ec2.Reservation, 0, len(order))
+	for _, id := range order {
+		e := byID[id]
+		out = append(out, &ec2.Reservation{
+			ReservationId: e.reservation.ReservationId,
+			OwnerId:       e.reservation.OwnerId,
+			Groups:        e.reservation.Groups,
+			Instances:     []*ec2.Instance{e.instance},
+		})
+	}
+	return mergeReservationsByID(out), nodeOnly, malformed
+}

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

@@ -0,0 +1,229 @@
+package instance
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	"github.com/aws/aws-sdk-go/aws"
+	"github.com/aws/aws-sdk-go/service/ec2"
+	awserrors "github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
+	natsmsg "github.com/mulgadc/spinifex/spinifex/foundation/messaging/nats"
+	"github.com/nats-io/nats.go"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func strictInstance(id, state string) *ec2.Instance {
+	return &ec2.Instance{InstanceId: aws.String(id), State: &ec2.InstanceState{Name: aws.String(state)}}
+}
+
+func strictReservations(instances ...*ec2.Instance) *ec2.DescribeInstancesOutput {
+	out := &ec2.DescribeInstancesOutput{}
+	for _, inst := range instances {
+		out.Reservations = append(out.Reservations, &ec2.Reservation{
+			ReservationId: aws.String("r-" + aws.StringValue(inst.InstanceId)),
+			OwnerId:       aws.String("123456789012"),
+			Instances:     []*ec2.Instance{inst},
+		})
+	}
+	return out
+}
+
+func subscribeAccountRecords(t *testing.T, nc *nats.Conn, data []byte) {
+	t.Helper()
+	_, err := nc.Subscribe(accountInstanceRecordsSubject, func(msg *nats.Msg) { _ = msg.Respond(data) })
+	require.NoError(t, err)
+}
+
+const strictAccount = "123456789012"
+
+func strictList(nc *nats.Conn, nodes ...string) (*ec2.DescribeInstancesOutput, error) {
+	return DescribeAccountInstancesStrict(context.Background(), nc, nodes, strictAccount, WithFanoutTimeout(2*time.Second))
+}
+
+// Existence comes from the record store, so an instance no node holds is still
+// listed, with the state the record carries.
+func TestDescribeAccountInstancesStrict_UnheldRecordIsReturnedWithItsRecordedState(t *testing.T) {
+	t.Parallel()
+	_, nc := startTestNATSServer(t)
+	subscribeAccountRecords(t, nc, mustMarshalDescribeOutput(t, strictReservations(strictInstance("i-unheld", "stopped"))))
+	subscribeAsNode(t, nc, "ec2.DescribeInstances", "node-1", []byte(`{}`))
+	nc.Flush()
+
+	out, err := strictList(nc, "node-1")
+
+	require.NoError(t, err)
+	require.Len(t, out.Reservations, 1)
+	inst := out.Reservations[0].Instances[0]
+	assert.Equal(t, "i-unheld", aws.StringValue(inst.InstanceId))
+	assert.Equal(t, "stopped", aws.StringValue(inst.State.Name))
+}
+
+// A node's answer replaces the recorded view of the instance it holds.
+func TestDescribeAccountInstancesStrict_NodeAnswersAreJoinedOntoRecords(t *testing.T) {
+	t.Parallel()
+	_, nc := startTestNATSServer(t)
+	subscribeAccountRecords(t, nc, mustMarshalDescribeOutput(t, strictReservations(strictInstance("i-held", "pending"))))
+	subscribeAsNode(t, nc, "ec2.DescribeInstances", "node-1", mustMarshalDescribeOutput(t,
+		strictReservations(strictInstance("i-held", "running"))))
+	nc.Flush()
+
+	out, err := strictList(nc, "node-1")
+
+	require.NoError(t, err)
+	states := map[string]string{}
+	for _, res := range out.Reservations {
+		for _, inst := range res.Instances {
+			states[aws.StringValue(inst.InstanceId)] = aws.StringValue(inst.State.Name)
+		}
+	}
+	assert.Equal(t, map[string]string{"i-held": "running"}, states)
+}
+
+// An instance a node reports with no record can come from a lost record write.
+// Returning it or dropping it would both answer for an unproven account.
+func TestDescribeAccountInstancesStrict_NodeOnlyInstanceFails(t *testing.T) {
+	t.Parallel()
+	_, nc := startTestNATSServer(t)
+	subscribeAccountRecords(t, nc, mustMarshalDescribeOutput(t, strictReservations(strictInstance("i-held", "running"))))
+	subscribeAsNode(t, nc, "ec2.DescribeInstances", "node-1", mustMarshalDescribeOutput(t,
+		strictReservations(strictInstance("i-held", "running"), strictInstance("i-node-only", "running"))))
+	nc.Flush()
+
+	out, err := strictList(nc, "node-1")
+
+	require.Error(t, err)
+	assert.Nil(t, out)
+	code, ok := awserrors.ResolveErrorCode(err)
+	require.True(t, ok)
+	assert.Equal(t, awserrors.ErrorServiceUnavailable, code)
+}
+
+// A frame from a node outside the configured set means the expected set is
+// wrong, and a listing built on it could omit that node's instances.
+func TestDescribeAccountInstancesStrict_FrameFromUnconfiguredNodeFails(t *testing.T) {
+	t.Parallel()
+	_, nc := startTestNATSServer(t)
+	subscribeAccountRecords(t, nc, mustMarshalDescribeOutput(t, strictReservations()))
+	subscribeAsNode(t, nc, "ec2.DescribeInstances", "node-1", []byte(`{}`))
+	subscribeAsNode(t, nc, "ec2.DescribeInstances", "node-stray", mustMarshalDescribeOutput(t, strictReservations(strictInstance("i-on-stray", "running"))))
+	nc.Flush()
+
+	out, err := strictList(nc, "node-1")
+
+	assertIncompleteListing(t, out, err)
+}
+
+// A stray node that answers with an error is still a node outside the set, and
+// its instances are unaccounted for.
+func TestDescribeAccountInstancesStrict_ErrorFromUnconfiguredNodeFails(t *testing.T) {
+	t.Parallel()
+	_, nc := startTestNATSServer(t)
+	subscribeAccountRecords(t, nc, mustMarshalDescribeOutput(t, strictReservations()))
+	subscribeAsNode(t, nc, "ec2.DescribeInstances", "node-1", []byte(`{}`))
+	subscribeAsNode(t, nc, "ec2.DescribeInstances", "node-stray", awserrors.GenerateErrorPayload(awserrors.ErrorServerInternal))
+	nc.Flush()
+
+	out, err := strictList(nc, "node-1")
+
+	assertIncompleteListing(t, out, err)
+}
+
+// The configured nodes answer at once, so a reply that lands later but inside
+// the fan-out deadline must still be seen.
+func TestDescribeAccountInstancesStrict_LateFrameFromUnconfiguredNodeFails(t *testing.T) {
+	t.Parallel()
+	_, nc := startTestNATSServer(t)
+	subscribeAccountRecords(t, nc, mustMarshalDescribeOutput(t, strictReservations()))
+	subscribeAsNode(t, nc, "ec2.DescribeInstances", "node-1", []byte(`{}`))
+	late := mustMarshalDescribeOutput(t, strictReservations(strictInstance("i-on-stray", "running")))
+	_, err := nc.Subscribe("ec2.DescribeInstances", func(msg *nats.Msg) {
+		time.AfterFunc(600*time.Millisecond, func() {
+			reply := nats.NewMsg(msg.Reply)
+			reply.Data = late
+			reply.Header.Set(natsmsg.NodeIDHeader, "node-stray")
+			_ = msg.RespondMsg(reply)
+		})
+	})
+	require.NoError(t, err)
+	nc.Flush()
+
+	out, err := strictList(nc, "node-1")
+
+	assertIncompleteListing(t, out, err)
+}
+
+// A record with no instance ID cannot be listed or counted, so it must fail
+// the read rather than vanish from it.
+func TestDescribeAccountInstancesStrict_RecordWithoutInstanceIDFails(t *testing.T) {
+	t.Parallel()
+	_, nc := startTestNATSServer(t)
+	subscribeAccountRecords(t, nc, mustMarshalDescribeOutput(t, strictReservations(&ec2.Instance{State: &ec2.InstanceState{Name: aws.String("stopped")}})))
+	subscribeAsNode(t, nc, "ec2.DescribeInstances", "node-1", []byte(`{}`))
+	nc.Flush()
+
+	out, err := strictList(nc, "node-1")
+
+	assertIncompleteListing(t, out, err)
+}
+
+func TestDescribeAccountInstancesStrict_ConfiguredNodeSilentFails(t *testing.T) {
+	t.Parallel()
+	_, nc := startTestNATSServer(t)
+	subscribeAccountRecords(t, nc, mustMarshalDescribeOutput(t, strictReservations()))
+	subscribeAsNode(t, nc, "ec2.DescribeInstances", "node-1", []byte(`{}`))
+	nc.Flush()
+
+	out, err := strictList(nc, "node-1", "node-2")
+
+	assertIncompleteListing(t, out, err)
+}
+
+func TestDescribeAccountInstancesStrict_RecordStoreErrorFails(t *testing.T) {
+	t.Parallel()
+	_, nc := startTestNATSServer(t)
+	subscribeAccountRecords(t, nc, awserrors.GenerateErrorPayload(awserrors.ErrorServerInternal))
+	subscribeAsNode(t, nc, "ec2.DescribeInstances", "node-1", []byte(`{}`))
+	nc.Flush()
+
+	out, err := strictList(nc, "node-1")
+
+	assertIncompleteListing(t, out, err)
+}
+
+func TestDescribeAccountInstancesStrict_RecordStoreSilentFails(t *testing.T) {
+	t.Parallel()
+	_, nc := startTestNATSServer(t)
+	subscribeAsNode(t, nc, "ec2.DescribeInstances", "node-1", []byte(`{}`))
+	nc.Flush()
+
+	out, err := strictList(nc, "node-1")
+
+	assertIncompleteListing(t, out, err)
+}
+
+func TestDescribeAccountInstancesStrict_UnknownNodeSetFails(t *testing.T) {
+	t.Parallel()
+	_, nc := startTestNATSServer(t)
+	subscribeAccountRecords(t, nc, mustMarshalDescribeOutput(t, strictReservations()))
+	subscribeAsNode(t, nc, "ec2.DescribeInstances", "node-1", []byte(`{}`))
+	nc.Flush()
+
+	out, err := strictList(nc)
+
+	assertIncompleteListing(t, out, err)
+}
+
+func TestDescribeAccountInstancesStrict_UnattributedFrameFails(t *testing.T) {
+	t.Parallel()
+	_, nc := startTestNATSServer(t)
+	subscribeAccountRecords(t, nc, mustMarshalDescribeOutput(t, strictReservations()))
+	subscribeAsNode(t, nc, "ec2.DescribeInstances", "node-1", []byte(`{}`))
+	subscribeAsNode(t, nc, "ec2.DescribeInstances", "", []byte(`{}`))
+	nc.Flush()
+
+	out, err := strictList(nc, "node-1")
+
+	assertIncompleteListing(t, out, err)
+}

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

@@ -80,5 +80,5 @@ func DescribeInstanceAttribute(ctx context.Context, input *ec2.DescribeInstanceA
 	// terraform retries cleanly.
 	slog.WarnContext(ctx, "DescribeInstanceAttribute: instance absent or no responses",
 		"instance_id", *input.InstanceId, "received", sum.Received, "expected", expectedNodes)
-	return nil, errors.New(awserrors.ErrorInvalidInstanceIDNotFound)
+	return nil, awserrors.IDNotFound(awserrors.ErrorInvalidInstanceIDNotFound, "instance", *input.InstanceId)
 }

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

@@ -184,7 +184,7 @@ func TestDescribeInstanceAttribute_AllNodesNotFound(t *testing.T) {
 
 	_, err = DescribeInstanceAttribute(context.Background(), input, nc, 2, "123456789012")
 	require.Error(t, err)
-	assert.Equal(t, awserrors.ErrorInvalidInstanceIDNotFound, err.Error())
+	assert.EqualError(t, err, "The instance ID 'i-missing' does not exist: "+awserrors.ErrorInvalidInstanceIDNotFound)
 }
 
 // TestDescribeInstanceAttribute_ClientErrorPropagates ensures that a
@@ -254,7 +254,7 @@ func TestDescribeInstanceAttribute_NoResponders(t *testing.T) {
 	duration := time.Since(start)
 
 	require.Error(t, err)
-	assert.Equal(t, awserrors.ErrorInvalidInstanceIDNotFound, err.Error())
+	assert.EqualError(t, err, "The instance ID 'i-nowhere' does not exist: "+awserrors.ErrorInvalidInstanceIDNotFound)
 	// Must not block the caller past the 3 s deadline.
 	assert.Less(t, duration, 4*time.Second)
 }

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

@@ -4,6 +4,7 @@ import (
 	"context"
 	"encoding/json"
 	"fmt"
+	"github.com/mulgadc/spinifex/spinifex/foundation/aws/identifiers"
 	"github.com/mulgadc/spinifex/spinifex/foundation/messaging/nats"
 	"log/slog"
 	"sort"
@@ -74,12 +75,7 @@ func DescribeInstanceTypes(ctx context.Context, input *ec2.DescribeInstanceTypes
 		}
 	}
 
-	requestedTypes := make(map[string]bool)
-	for _, it := range input.InstanceTypes {
-		if it != nil {
-			requestedTypes[*it] = true
-		}
-	}
+	requestedTypes := identifiers.Set(input.InstanceTypes)
 
 	var finalInstanceTypes []*ec2.InstanceTypeInfo
 	if showCapacity {

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

@@ -126,7 +126,7 @@ func DescribeInstancesChecked(ctx context.Context, input *ec2.DescribeInstancesI
 		missing := missingInstanceIDs(reservations, input.InstanceIds)
 		if len(missing) > 0 {
 			if complete {
-				return nil, errors.New(awserrors.ErrorInvalidInstanceIDNotFound)
+				return nil, awserrors.IDsNotFound(awserrors.ErrorInvalidInstanceIDNotFound, "instance", missing)
 			}
 			// An incomplete sweep cannot prove absence, so a missing id is
 			// unresolved, not gone. NotFound would be a false negative and an

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

@@ -0,0 +1,63 @@
+package instance
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	"github.com/aws/aws-sdk-go/aws"
+	"github.com/aws/aws-sdk-go/service/ec2"
+	natsmsg "github.com/mulgadc/spinifex/spinifex/foundation/messaging/nats"
+	"github.com/nats-io/nats.go"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+// Pins today: the completeness judge merges the reservations of a frame from a
+// node outside the configured set and still reports the listing complete. The
+// stray node is never named, so a silent unconfigured node cannot be told from
+// one that does not exist.
+func TestCurrentBehaviour_JudgeIdentityCompleteness_FrameFromUnconfiguredNodeIsMerged(t *testing.T) {
+	t.Parallel()
+	stray := mustMarshalDescribeOutput(t, &ec2.DescribeInstancesOutput{
+		Reservations: []*ec2.Reservation{{ReservationId: aws.String("r-stray"), Instances: []*ec2.Instance{{InstanceId: aws.String("i-on-stray")}}}},
+	})
+	frames := []natsmsg.Frame{
+		{NodeID: "node-1", Data: []byte(`{}`)},
+		{NodeID: "node-stray", Data: stray},
+	}
+
+	reservations, complete := judgeIdentityCompleteness(context.Background(), frames, natsmsg.Summary{}, []string{"node-1"})
+
+	assert.True(t, complete, "CURRENT: a node outside nodeIDs does not make the listing incomplete")
+	require.Len(t, reservations, 1)
+	assert.Equal(t, "i-on-stray", *reservations[0].Instances[0].InstanceId,
+		"CURRENT: the unconfigured node's instances are merged into the result")
+}
+
+// Pins today: the fan-out counts every identified frame toward the expected
+// responders, so a stray frame can end collection before a configured node has
+// answered. The listing then refuses, which fails closed, but for a node that
+// was only slow.
+func TestCurrentBehaviour_DescribeAccountInstancesComplete_StrayFrameConsumesAResponderSlot(t *testing.T) {
+	t.Parallel()
+	_, nc := startTestNATSServer(t)
+	subscribeEmptyInstanceBuckets(t, nc)
+	subscribeAsNode(t, nc, "ec2.DescribeInstances", "node-1", []byte(`{}`))
+	subscribeAsNode(t, nc, "ec2.DescribeInstances", "node-stray", []byte(`{}`))
+	_, err := nc.Subscribe("ec2.DescribeInstances", func(msg *nats.Msg) {
+		time.AfterFunc(500*time.Millisecond, func() {
+			reply := nats.NewMsg(msg.Reply)
+			reply.Data = []byte(`{}`)
+			reply.Header.Set(natsmsg.NodeIDHeader, "node-2")
+			_ = msg.RespondMsg(reply)
+		})
+	})
+	require.NoError(t, err)
+	nc.Flush()
+
+	output, err := DescribeAccountInstancesComplete(context.Background(), nc, []string{"node-1", "node-2"}, "123456789012",
+		WithFanoutTimeout(3*time.Second))
+
+	assertIncompleteListing(t, output, err)
+}

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

@@ -448,11 +448,13 @@ func TestDescribeInstancesChecked_ExplicitIDNotFound_CompleteSweep(t *testing.T)
 	require.NoError(t, err)
 	nc.Flush()
 
-	input := &ec2.DescribeInstancesInput{InstanceIds: []*string{aws.String("i-doesnotexist0000000")}}
+	input := &ec2.DescribeInstancesInput{InstanceIds: aws.StringSlice([]string{"i-doesnotexist0000000", "i-real", "i-doesnotexist0000001"})}
 	_, err = DescribeInstancesChecked(context.Background(), input, nc, 1, nil, "123456789012")
 
-	require.Error(t, err)
-	assert.Equal(t, awserrors.ErrorInvalidInstanceIDNotFound, err.Error())
+	code, message, ok := awserrors.ResolveErrorDetail(err)
+	require.True(t, ok, "err = %v", err)
+	assert.Equal(t, awserrors.ErrorInvalidInstanceIDNotFound, code)
+	assert.Equal(t, "The instance IDs 'i-doesnotexist0000000, i-doesnotexist0000001' do not exist", message)
 }
 
 // TestDescribeInstances_ExplicitIDNotFound_StaysSilent guards the internal
@@ -624,8 +626,7 @@ func TestDescribeInstancesChecked_Identity_AllNodesAnswer_NotFoundAsserted(t *te
 	_, err := DescribeInstancesChecked(context.Background(), input, nc, 0, []string{"node-1", "node-2"}, "123456789012",
 		WithFanoutTimeout(2*time.Second))
 
-	require.Error(t, err)
-	assert.Equal(t, awserrors.ErrorInvalidInstanceIDNotFound, err.Error())
+	assert.True(t, awserrors.IsErrorCode(err, awserrors.ErrorInvalidInstanceIDNotFound), "err = %v", err)
 }
 
 // A node answering with a decodable error envelope never becomes a valid
@@ -684,7 +685,7 @@ func TestDescribeInstancesChecked_Identity_NilReservationsIsValidResponder(t *te
 		WithFanoutTimeout(2*time.Second))
 
 	require.Error(t, err, "a nil-Reservations node is a complete, valid responder, so the sweep is complete and NotFound must fire")
-	assert.Equal(t, awserrors.ErrorInvalidInstanceIDNotFound, err.Error())
+	assert.True(t, awserrors.IsErrorCode(err, awserrors.ErrorInvalidInstanceIDNotFound), "err = %v", err)
 }
 
 // An empty nodeIDs falls back to the legacy expectedNodes path, which the

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

@@ -47,7 +47,7 @@ func GetConsoleOutput(ctx context.Context, input *ec2.GetConsoleOutputInput, nat
 	if err != nil {
 		// No daemon subscription or timeout: stopped/terminated/non-existent instances all surface as NotFound.
 		if errors.Is(err, nats.ErrNoResponders) || errors.Is(err, nats.ErrTimeout) {
-			return nil, errors.New(awserrors.ErrorInvalidInstanceIDNotFound)
+			return nil, awserrors.IDNotFound(awserrors.ErrorInvalidInstanceIDNotFound, "instance", *input.InstanceId)
 		}
 		return nil, fmt.Errorf("failed to get console output: %w", err)
 	}

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

@@ -52,7 +52,7 @@ func GetPasswordData(ctx context.Context, input *ec2.GetPasswordDataInput, natsC
 	if err != nil {
 		// No daemon subscription or timeout: stopped/terminated/non-existent instances all surface as NotFound.
 		if errors.Is(err, nats.ErrNoResponders) || errors.Is(err, nats.ErrTimeout) {
-			return nil, errors.New(awserrors.ErrorInvalidInstanceIDNotFound)
+			return nil, awserrors.IDNotFound(awserrors.ErrorInvalidInstanceIDNotFound, "instance", *input.InstanceId)
 		}
 		return nil, fmt.Errorf("failed to get password data: %w", err)
 	}

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

@@ -120,7 +120,7 @@ func TestGetPasswordData_NoResponders(t *testing.T) {
 
 	require.Error(t, err)
 	assert.Nil(t, output)
-	assert.Equal(t, awserrors.ErrorInvalidInstanceIDNotFound, err.Error())
+	assert.EqualError(t, err, "The instance ID 'i-nosubscriber' does not exist: "+awserrors.ErrorInvalidInstanceIDNotFound)
 }
 
 func TestGetPasswordData_Timeout(t *testing.T) {
@@ -140,7 +140,7 @@ func TestGetPasswordData_Timeout(t *testing.T) {
 
 	require.Error(t, err)
 	assert.Nil(t, output)
-	assert.Equal(t, awserrors.ErrorInvalidInstanceIDNotFound, err.Error())
+	assert.EqualError(t, err, "The instance ID 'i-slow' does not exist: "+awserrors.ErrorInvalidInstanceIDNotFound)
 }
 
 func TestGetPasswordData_DaemonError(t *testing.T) {

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

@@ -82,7 +82,7 @@ func AssociateIamInstanceProfile(ctx context.Context, input *ec2.AssociateIamIns
 	assoc, err := natsmsg.NATSRequest[ec2.IamInstanceProfileAssociation](ctx, natsConn, subject, command, fanOutTimeout, accountID)
 	if err != nil {
 		if errors.Is(err, nats.ErrNoResponders) {
-			return nil, errors.New(awserrors.ErrorInvalidInstanceIDNotFound)
+			return nil, awserrors.IDNotFound(awserrors.ErrorInvalidInstanceIDNotFound, "instance", *input.InstanceId)
 		}
 		return nil, err
 	}

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

@@ -293,7 +293,7 @@ func TestAssociateIamInstanceProfile_NoResponders(t *testing.T) {
 	// a raw NATS timeout.
 	_, err := AssociateIamInstanceProfile(context.Background(), in, nc, svc, testGwAccountID, nil)
 	require.Error(t, err)
-	assert.Equal(t, awserrors.ErrorInvalidInstanceIDNotFound, err.Error())
+	assert.EqualError(t, err, "The instance ID 'i-no-daemon' does not exist: "+awserrors.ErrorInvalidInstanceIDNotFound)
 }
 
 func TestAssociateIamInstanceProfile_Success(t *testing.T) {

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

@@ -87,7 +87,7 @@ func RebootInstances(ctx context.Context, input *ec2.RebootInstancesInput, natsC
 			case state != bucketStateAbsent:
 				return nil, errors.New(awserrors.ErrorIncorrectInstanceState)
 			case queryOK:
-				return nil, errors.New(awserrors.ErrorInvalidInstanceIDNotFound)
+				return nil, awserrors.IDNotFound(awserrors.ErrorInvalidInstanceIDNotFound, "instance", instanceID)
 			default:
 				return nil, errors.New(awserrors.ErrorServerInternal)
 			}

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

@@ -88,7 +88,7 @@ func TestRebootInstances_NeverExistedIsNotFound(t *testing.T) {
 
 	_, err := RebootInstances(context.Background(), input, nc, "123456789012")
 	require.Error(t, err)
-	assert.Equal(t, awserrors.ErrorInvalidInstanceIDNotFound, err.Error())
+	assert.EqualError(t, err, "The instance ID 'i-d2d5fe795b6b57882' does not exist: "+awserrors.ErrorInvalidInstanceIDNotFound)
 }
 
 func TestRebootInstances_BucketsUnreachableIsServerError(t *testing.T) {

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

@@ -8,7 +8,6 @@ import (
 	"errors"
 	awsidentifiers "github.com/mulgadc/spinifex/spinifex/foundation/aws/identifiers"
 	"github.com/mulgadc/spinifex/spinifex/foundation/messaging/nats"
-	"log/slog"
 	"math"
 	"slices"
 	"strings"
@@ -16,6 +15,7 @@ import (
 
 	"github.com/aws/aws-sdk-go/aws"
 	"github.com/aws/aws-sdk-go/service/ec2"
+	ec2idemapi "github.com/mulgadc/spinifex/spinifex/domains/ec2/awsapi/idem"
 	ec2instance "github.com/mulgadc/spinifex/spinifex/domains/ec2/instance"
 	ec2key "github.com/mulgadc/spinifex/spinifex/domains/ec2/key"
 	ec2launchtemplate "github.com/mulgadc/spinifex/spinifex/domains/ec2/launchtemplate"
@@ -126,12 +126,13 @@ func hasPinnedPrivateIP(input *ec2.RunInstancesInput) bool {
 
 // RunInstances validates input, resolves any IAM instance profile (normalising
 // to ARN for daemons), enforces iam:PassRole, then dispatches via NATS.
-// When ClientToken is set, wraps the launch in idempotency via the KV store.
+// When ClientToken is set, wraps the launch in idempotency via tokenBucket,
+// which may be nil only when no ClientToken is supplied.
 // iamSvc may be nil only when no IamInstanceProfile is supplied.
 // passRoleCheck may be nil to skip PassRole enforcement.
 // launchQuotaCheck may be nil to skip quota enforcement; when present it runs
 // after validation and PassRole authorization, but before any launch dispatch.
-func RunInstances(ctx context.Context, input *ec2.RunInstancesInput, natsConn *nats.Conn, iamSvc InstanceProfiles, accountID string, passRoleCheck PassRoleChecker, launchQuotaCheck LaunchQuotaChecker, expectedNodes int) (reservation ec2.Reservation, err error) {
+func RunInstances(ctx context.Context, input *ec2.RunInstancesInput, natsConn *nats.Conn, tokenBucket ec2idemapi.Bucket, iamSvc InstanceProfiles, accountID string, passRoleCheck PassRoleChecker, launchQuotaCheck LaunchQuotaChecker, expectedNodes int) (reservation ec2.Reservation, err error) {
 	// Expand a referenced launch template before validation so validation, quota,
 	// and downstream routing all see the effective (post-merge) parameters.
 	if err = ExpandLaunchTemplate(ctx, natsConn, input, accountID); err != nil {
@@ -142,21 +143,10 @@ func RunInstances(ctx context.Context, input *ec2.RunInstancesInput, natsConn *n
 		return reservation, err
 	}
 
-	token := aws.StringValue(input.ClientToken)
-	if token == "" {
-		return runInstancesInner(ctx, input, natsConn, iamSvc, accountID, passRoleCheck, launchQuotaCheck, expectedNodes)
-	}
-
-	// ClientToken set: dedup concurrent/retried launches; store failure is fatal
-	// to avoid a double launch on retry.
-	store, serr := getClientTokenStore(ctx, natsConn)
-	if serr != nil {
-		slog.ErrorContext(ctx, "RunInstances: client-token store unavailable", "err", serr)
-		return reservation, errors.New(awserrors.ErrorServerInternal)
-	}
-	// Hash before any mutation so the same token+params always matches.
-	paramHash := clientTokenParamHash(input)
-	return runInstancesWithClientToken(ctx, store, accountID, token, paramHash, func() (ec2.Reservation, error) {
+	// Unnamespaced: every other EC2 action namespaces its keys by action, but
+	// RunInstances predates that, and re-keying live records would make a retry
+	// re-launch instead of replay. Hashed here, after template expansion.
+	return ec2idemapi.Run(ctx, tokenBucket, "", accountID, input, func() (ec2.Reservation, error) {
 		return runInstancesInner(ctx, input, natsConn, iamSvc, accountID, passRoleCheck, launchQuotaCheck, expectedNodes)
 	})
 }

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

@@ -347,7 +347,7 @@ func TestParseRunInstances(t *testing.T) {
 		t.Run(test.name, func(t *testing.T) {
 			t.Parallel()
 			// For validation tests, we can pass nil conn since validation happens before NATS call
-			response, err := RunInstances(context.Background(), test.input, nil, nil, "123456789012", nil, nil, 1)
+			response, err := RunInstances(context.Background(), test.input, nil, nil, nil, "123456789012", nil, nil, 1)
 
 			// Use assert to check if the error is as expected
 			assert.Equal(t, test.want, err)

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

@@ -4,6 +4,7 @@ import (
 	"context"
 	"encoding/json"
 	"errors"
+	"fmt"
 	"github.com/mulgadc/spinifex/spinifex/foundation/messaging/nats"
 	"log/slog"
 	"time"
@@ -102,7 +103,7 @@ func startLiveInstance(ctx context.Context, natsConn *nats.Conn, instanceID, acc
 
 	if responseError, parseErr := awserrors.ValidateErrorPayload(msg.Data); parseErr != nil {
 		slog.ErrorContext(ctx, "StartInstances: owner returned error", "instance_id", instanceID, "code", *responseError.Code)
-		return nil, false, startRefusalError(responseError)
+		return nil, false, startRefusalError(responseError, msg.Header)
 	}
 
 	slog.InfoContext(ctx, "StartInstances: restarted via owner node", "instance_id", instanceID, "response", string(msg.Data))
@@ -135,18 +136,25 @@ func startStoppedInstance(ctx context.Context, natsConn *nats.Conn, instanceID,
 
 	if responseError, parseErr := awserrors.ValidateErrorPayload(msg.Data); parseErr != nil {
 		slog.ErrorContext(ctx, "StartInstances: Daemon returned error", "instance_id", instanceID, "code", *responseError.Code)
-		return nil, startRefusalError(responseError)
+		return nil, startRefusalError(responseError, msg.Header)
 	}
 
 	slog.InfoContext(ctx, "StartInstances: Command sent successfully", "instance_id", instanceID, "response", string(msg.Data))
 	return newStateChange(instanceID, 0, "pending", 80, "stopped"), nil
 }
 
-// startRefusalError rebuilds the daemon's error, keeping the message it sent.
-// Reconstructing from the code alone renders the generic ErrorLookup text
-// instead, which replaces the name of the node holding the volume with
-// boilerplate about attaching volumes -- the one fact the operator needed.
-func startRefusalError(responseError ec2.ResponseError) error {
+// startRefusalError rebuilds the daemon's error, keeping the message it sent,
+// which names the node holding the volume, and wrapping ErrVolumeHeldElsewhere
+// when the reply's header says that is why the start was refused.
+func startRefusalError(responseError ec2.ResponseError, header nats.Header) error {
+	err := startRefusalCodeError(responseError)
+	if header.Get(natsmsg.VolumeHeldElsewhereHeader) == "true" {
+		return fmt.Errorf("%w: %w", ec2v1.ErrVolumeHeldElsewhere, err)
+	}
+	return err
+}
+
+func startRefusalCodeError(responseError ec2.ResponseError) error {
 	var code, message string
 	if responseError.Code != nil {
 		code = *responseError.Code

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

@@ -3,11 +3,14 @@ package instance
 import (
 	"context"
 	"encoding/json"
+	"errors"
 	"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/mulgadc/spinifex/spinifex/foundation/messaging/nats"
 	"github.com/nats-io/nats.go"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
@@ -161,3 +164,33 @@ func TestStartInstances_MixedSuccessAndFailure(t *testing.T) {
 	require.Error(t, err)
 	assert.Equal(t, "InvalidInstanceID.NotFound", err.Error())
 }
+
+// The AWS body says only IncorrectState, so the header is the one thing that
+// tells a held volume apart from any other refusal in that code.
+func TestStartInstances_VolumeHeldElsewhereHeader(t *testing.T) {
+	t.Parallel()
+	for name, held := range map[string]bool{"with header": true, "without header": false} {
+		t.Run(name, func(t *testing.T) {
+			t.Parallel()
+			_, nc := startTestNATSServer(t)
+			const id = "i-0abc0000000000002"
+			_, err := nc.Subscribe(ec2v1.InstanceCommandSubject(id), func(msg *nats.Msg) {
+				reply := nats.NewMsg(msg.Reply)
+				reply.Data = awserrors.GenerateErrorPayloadWithMessage(awserrors.ErrorIncorrectState, "volume vol-1 is published to node-a")
+				if held {
+					reply.Header.Set(natsmsg.VolumeHeldElsewhereHeader, "true")
+				}
+				_ = msg.RespondMsg(reply)
+			})
+			require.NoError(t, err)
+
+			_, err = StartInstances(t.Context(), &ec2.StartInstancesInput{InstanceIds: []*string{aws.String(id)}}, nc, "123456789012")
+			require.Error(t, err)
+			assert.Equal(t, held, errors.Is(err, ec2v1.ErrVolumeHeldElsewhere))
+			code, message, ok := awserrors.ResolveErrorDetail(err)
+			require.True(t, ok)
+			assert.Equal(t, awserrors.ErrorIncorrectState, code)
+			assert.Equal(t, "volume vol-1 is published to node-a", message)
+		})
+	}
+}