fix(ec2): log a held-volume start refusal at Warn with the instance ID

A start refused because another node holds a volume is returned as
IncorrectState, a client error, yet was logged at Error without the
instance it concerned. It is now logged at Warn, and every vmMgr.Start
failure log names the instance.

.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: |

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"`

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

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()) }
 }

spinifex/domains/acm/awsapi/registration.go

@@ -3,14 +3,11 @@ 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/nats-io/nats.go"
@@ -66,7 +63,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 +78,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 +101,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
@@ -144,16 +127,3 @@ func (d Deps) dispatch(w http.ResponseWriter, inv dispatch.Invocation) error {
 	WriteJSONResponse(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/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/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)
+		})
+	}
+}

spinifex/domains/ec2/awsapi/registration.go

@@ -1,8 +1,6 @@
 package awsapi
 
 import (
-	"net/http"
-
 	"github.com/mulgadc/spinifex/spinifex/ingress/aws/dispatch"
 )
 
@@ -11,7 +9,7 @@ import (
 func NewRegistration(deps Deps) dispatch.Registration {
 	return dispatch.Registration{
 		Service:       ServiceName,
-		ResolveAction: resolveAction,
+		ResolveAction: dispatch.ResolveQueryAction,
 		Dispatch:      NewHandler(deps).Dispatch,
 		Errors:        dispatch.ErrorEnvelopeXML,
 		Traits:        dispatch.ServiceTraits{XMLErrorShape: dispatch.XMLErrorShapeEC2},
@@ -24,13 +22,3 @@ func NewRegistration(deps Deps) dispatch.Registration {
 func OperationInventory() dispatch.Inventory {
 	return dispatch.Inventory{Registered: ActionNames()}
 }
-
-// resolveAction is the one Query Action parser telemetry and dispatch share.
-// It leaves the body readable for the dispatcher and answers "" on any failure.
-func resolveAction(r *http.Request) string {
-	args, err := readQueryArgs(r)
-	if err != nil {
-		return ""
-	}
-	return args["Action"]
-}

spinifex/domains/ec2/awsapi/registration_test.go

@@ -1,11 +1,6 @@
-//test:in-package — drives the unexported resolveAction with the registration.
-
 package awsapi
 
 import (
-	"net/http"
-	"net/http/httptest"
-	"strings"
 	"testing"
 
 	"github.com/mulgadc/spinifex/spinifex/ingress/aws/dispatch"
@@ -27,29 +22,3 @@ func TestNewRegistration_DeclaresTheEC2Service(t *testing.T) {
 	b := dispatch.NewBuilder()
 	require.NoError(t, b.Register(reg))
 }
-
-func TestResolveAction(t *testing.T) {
-	post := func(body string) *http.Request {
-		r := httptest.NewRequest(http.MethodPost, "/", strings.NewReader(body))
-		r.Header.Set("Content-Type", "application/x-www-form-urlencoded")
-		return r
-	}
-
-	assert.Equal(t, "DescribeVpcs", resolveAction(post("Action=DescribeVpcs&Version=2016-11-15")))
-	assert.Empty(t, resolveAction(post("Version=2016-11-15")))
-
-	// The body stays readable for the dispatcher after the action is resolved.
-	r := post("Action=DescribeVpcs")
-	require.Equal(t, "DescribeVpcs", resolveAction(r))
-	assert.Equal(t, "DescribeVpcs", resolveAction(r))
-}
-
-func TestResolveAction_UnreadableBody(t *testing.T) {
-	r := httptest.NewRequest(http.MethodPost, "/", errReader{})
-	r.Header.Set("Content-Type", "application/x-www-form-urlencoded")
-	assert.Empty(t, resolveAction(r))
-}
-
-type errReader struct{}
-
-func (errReader) Read([]byte) (int, error) { return 0, http.ErrBodyReadAfterClose }

spinifex/domains/ec2/awsapi/request.go

@@ -1,9 +1,7 @@
 package awsapi
 
 import (
-	"bytes"
 	"errors"
-	"io"
 	"log/slog"
 	"maps"
 	"net/http"
@@ -12,7 +10,6 @@ import (
 
 	"github.com/aws/aws-sdk-go/service/ec2"
 	"github.com/mulgadc/bluebottle/pkg/iampolicy"
-	"github.com/mulgadc/bluebottle/pkg/sigv4"
 	ec2instanceapi "github.com/mulgadc/spinifex/spinifex/domains/ec2/awsapi/instance"
 	"github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
 	"github.com/mulgadc/spinifex/spinifex/ingress/aws/dispatch"
@@ -26,25 +23,11 @@ const (
 	ec2ServicePrincipal = "ec2.amazonaws.com"
 )
 
-// readQueryArgs parses the Query body and rewinds it, so the same bytes can be
-// read again. Any read or parse failure is a malformed query string.
-func readQueryArgs(r *http.Request) (map[string]string, error) {
-	body, err := io.ReadAll(io.LimitReader(r.Body, sigv4.MaxPayloadLen+1))
-	if err != nil {
-		return nil, err
-	}
-	r.Body = io.NopCloser(bytes.NewReader(body))
-	if int64(len(body)) > sigv4.MaxPayloadLen {
-		return nil, errors.New(awserrors.ErrorRequestEntityTooLarge)
-	}
-	return query.ParseArgs(string(body))
-}
-
 // Dispatch parses, authorizes and runs one EC2 request, then writes the XML
 // response. A returned error is an awserrors code the gateway renders.
 func (h *Handler) Dispatch(w http.ResponseWriter, inv dispatch.Invocation) error {
 	r := inv.Request
-	queryArgs, err := readQueryArgs(r)
+	queryArgs, err := dispatch.ReadArgs(r)
 	if err != nil {
 		slog.Debug("EC2: malformed query string", "err", err)
 		return errors.New(awserrors.ErrorMalformedQueryString)

spinifex/domains/ec2/image/drain_test.go

@@ -1,140 +0,0 @@
-package image
-
-import (
-	"context"
-	"encoding/json"
-	"net/http"
-	"net/http/httptest"
-	"testing"
-	"time"
-
-	"github.com/mulgadc/spinifex/contracts/ec2/v1"
-	"github.com/mulgadc/spinifex/internal/testkit"
-	"github.com/mulgadc/spinifex/spinifex/bootstrap/config"
-	"github.com/mulgadc/spinifex/spinifex/domains/ec2/volumestate"
-	"github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
-	"github.com/mulgadc/spinifex/spinifex/providers/objectstore"
-	"github.com/nats-io/nats.go"
-	"github.com/stretchr/testify/assert"
-	"github.com/stretchr/testify/require"
-)
-
-const drainImageInstanceID = "i-image-drain-host"
-
-// setupDrainImageService mirrors domains/ec2/snapshot's setupDrainService: a
-// reachable (but rejecting) Predastore host so a drained volume falls through
-// into the real snapshot path and fails fast, a data dir with no local drain
-// socket, and a NATS connection to route the drain to the hosting node.
-func setupDrainImageService(t *testing.T) (*ImageServiceImpl, *objectstore.MemoryObjectStore, *nats.Conn) {
-	t.Helper()
-
-	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
-		w.WriteHeader(http.StatusForbidden)
-	}))
-	t.Cleanup(srv.Close)
-
-	_, nc := testutil.StartTestNATS(t)
-	store := objectstore.NewMemoryObjectStore()
-	cfg := &config.Config{
-		DataDir: testutil.SocketTempDir(t),
-		Predastore: config.PredastoreConfig{
-			Host:   srv.URL,
-			Bucket: testBucket,
-		},
-	}
-	svc := NewImageServiceImplWithConfig(cfg, store, nc)
-	return svc, store, nc
-}
-
-func seedImageVolumeAttachment(t *testing.T, store *objectstore.MemoryObjectStore, volumeID, state, instanceID string) {
-	t.Helper()
-	require.NoError(t, volumestate.Write(context.Background(), store, testBucket, volumeID, volumestate.Record{
-		State:            state,
-		AttachedInstance: instanceID,
-	}))
-}
-
-func drainImageResponder(t *testing.T, nc *nats.Conn, instanceID string, reply []byte) chan *nats.Msg {
-	t.Helper()
-
-	got := make(chan *nats.Msg, 4)
-	sub, err := nc.Subscribe("ec2.cmd."+instanceID, func(msg *nats.Msg) {
-		got <- msg
-		_ = msg.Respond(reply)
-	})
-	require.NoError(t, err)
-	t.Cleanup(func() { _ = sub.Unsubscribe() })
-
-	return got
-}
-
-func drainedImageAck(t *testing.T, volumeID string) []byte {
-	t.Helper()
-	data, err := json.Marshal(ec2v1.DrainVolumeResponse{VolumeID: volumeID, Status: ec2v1.DrainVolumeStatusDrained})
-	require.NoError(t, err)
-	return data
-}
-
-func awaitImageDrainCommand(t *testing.T, got chan *nats.Msg) ec2v1.EC2InstanceCommand {
-	t.Helper()
-	select {
-	case msg := <-got:
-		var command ec2v1.EC2InstanceCommand
-		require.NoError(t, json.Unmarshal(msg.Data, &command))
-		return command
-	case <-time.After(2 * time.Second):
-		t.Fatal("the node hosting the volume never received a drain command")
-		return ec2v1.EC2InstanceCommand{}
-	}
-}
-
-// CreateImageFromInstance's IsRunning path must drain the volume's hosting node
-// before it reads the live checkpoint — the same guarantee ec2.CreateSnapshot
-// has. Without this, a checkpoint predating a guest-triggered GPT rewrite
-// (e.g. growpart on first boot) can be captured silently.
-func TestSnapshotRunningVolume_DrainsAttachedVolume(t *testing.T) {
-	svc, store, nc := setupDrainImageService(t)
-	createTestVolumeConfig(t, store, "vol-image-drain", 10)
-	seedImageVolumeAttachment(t, store, "vol-image-drain", "in-use", drainImageInstanceID)
-	got := drainImageResponder(t, nc, drainImageInstanceID, drainedImageAck(t, "vol-image-drain"))
-
-	err := svc.snapshotRunningVolume("vol-image-drain", "snap-image-drain", testAccountID)
-	// The rejecting Predastore stub fails the snapshot itself; what matters
-	// here is that the drain reached the hosting node first.
-	require.Error(t, err)
-
-	command := awaitImageDrainCommand(t, got)
-	assert.True(t, command.Attributes.DrainVolume)
-	assert.Equal(t, drainImageInstanceID, command.ID)
-	require.NotNil(t, command.DrainVolumeData)
-	assert.Equal(t, "vol-image-drain", command.DrainVolumeData.VolumeID)
-}
-
-// An attached volume whose host reports the drain failed must fail
-// CreateImage rather than register an AMI built from a stale checkpoint.
-func TestSnapshotRunningVolume_UndrainableAttachedVolumeFails(t *testing.T) {
-	svc, store, nc := setupDrainImageService(t)
-	createTestVolumeConfig(t, store, "vol-image-nodrain", 10)
-	seedImageVolumeAttachment(t, store, "vol-image-nodrain", "in-use", drainImageInstanceID)
-	drainImageResponder(t, nc, drainImageInstanceID,
-		[]byte(`{"Code":"`+awserrors.ErrorServerInternal+`","Message":"drain failed"}`))
-
-	err := svc.snapshotRunningVolume("vol-image-nodrain", "snap-image-nodrain", testAccountID)
-	require.Error(t, err)
-	assert.Equal(t, awserrors.ErrorServerInternal, err.Error())
-}
-
-// An unattached (stopped-instance) volume takes the no-drain path: the
-// checkpoint Close() left behind is already current.
-func TestSnapshotRunningVolume_AvailableSkipsDrain(t *testing.T) {
-	svc, store, _ := setupDrainImageService(t)
-	createTestVolumeConfig(t, store, "vol-image-available", 10)
-	seedImageVolumeAttachment(t, store, "vol-image-available", "available", "")
-	svc.natsConn = nil
-
-	// No responder exists and the socket is never dialled: proceeding straight
-	// to the (rejecting) snapshot path proves the drain was skipped, not stuck.
-	err := svc.snapshotRunningVolume("vol-image-available", "snap-image-available", testAccountID)
-	require.Error(t, err)
-	assert.Equal(t, awserrors.ErrorServerInternal, err.Error())
-}

spinifex/domains/ec2/image/provider_snapshot_test.go

@@ -2,16 +2,20 @@ package image
 
 import (
 	"context"
+	"sync/atomic"
 	"testing"
 	"time"
 
 	"github.com/aws/aws-sdk-go/aws"
 	"github.com/aws/aws-sdk-go/service/ec2"
+	"github.com/mulgadc/spinifex/contracts/ec2/v1"
+	"github.com/mulgadc/spinifex/internal/testkit"
 	"github.com/mulgadc/spinifex/spinifex/bootstrap/config"
 	"github.com/mulgadc/spinifex/spinifex/domains/ec2/ebs/metadata"
 	"github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
 	"github.com/mulgadc/spinifex/spinifex/providers/ebs"
 	"github.com/mulgadc/spinifex/spinifex/providers/objectstore"
+	"github.com/nats-io/nats.go"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 )
@@ -96,15 +100,15 @@ func TestGetVolumeMetadata_Provider_UnknownVolumeIsNotFound(t *testing.T) {
 	assert.Equal(t, awserrors.ErrorInvalidVolumeNotFound, err.Error())
 }
 
-// TestSnapshotStoppedVolume_Provider_CreatesThroughProvider locks that a
-// stopped-volume snapshot under a provider is created via provider.CreateSnapshot
+// TestSnapshotVolume_Provider_CreatesThroughProvider locks that a
+// root-volume snapshot under a provider is created via provider.CreateSnapshot
 // and never falls through to building an embedded engine — the Predastore host
 // is a closed port, so any fallback would fail with connection refused.
-func TestSnapshotStoppedVolume_Provider_CreatesThroughProvider(t *testing.T) {
+func TestSnapshotVolume_Provider_CreatesThroughProvider(t *testing.T) {
 	svc, _, provider := setupProviderSnapshotImageService(t)
 	doc := seedProviderVolume(t, svc, provider, "vol-stopped001", 8)
 
-	err := svc.snapshotStoppedVolume(doc.VolumeID, "snap-stopped001", testAccountID)
+	err := svc.snapshotVolume(doc.VolumeID, "snap-stopped001", testAccountID)
 	require.NoError(t, err, "the provider path must not fall through to the embedded engine against the closed Predastore port")
 
 	// CreateSnapshot is idempotent on (SnapshotID, VolumeID): re-issuing it
@@ -118,22 +122,22 @@ func TestSnapshotStoppedVolume_Provider_CreatesThroughProvider(t *testing.T) {
 	assert.Equal(t, doc.VolumeID, snap.SourceVolumeID)
 }
 
-// TestSnapshotRunningVolume_Provider_PassesProviderHandleAsVolumeHandle locks
+// TestSnapshotVolume_Provider_PassesProviderHandleAsVolumeHandle locks
 // that the ebsmetadata document's ProviderHandle is threaded through as
 // CreateSnapshotRequest.VolumeHandle. A wrong handle must be rejected by the
 // provider (MemoryProvider treats an empty VolumeHandle as "skip the check",
 // so only a mismatched non-empty value proves the real value is forwarded).
-func TestSnapshotRunningVolume_Provider_PassesProviderHandleAsVolumeHandle(t *testing.T) {
+func TestSnapshotVolume_Provider_PassesProviderHandleAsVolumeHandle(t *testing.T) {
 	svc, _, provider := setupProviderSnapshotImageService(t)
 	doc := seedProviderVolume(t, svc, provider, "vol-running001", 8)
 
-	err := svc.snapshotRunningVolume(doc.VolumeID, "snap-running001", testAccountID)
+	err := svc.snapshotVolume(doc.VolumeID, "snap-running001", testAccountID)
 	require.NoError(t, err, "the correct provider handle must let CreateSnapshot succeed")
 
 	doc.ProviderHandle = "wrong-handle"
 	require.NoError(t, svc.MetadataStore().PutVolume(context.Background(), doc))
 
-	err = svc.snapshotRunningVolume(doc.VolumeID, "snap-running002", testAccountID)
+	err = svc.snapshotVolume(doc.VolumeID, "snap-running002", testAccountID)
 	require.Error(t, err, "a mismatched provider handle must be rejected, proving the value came from the document")
 	assert.Equal(t, awserrors.ErrorServerInternal, err.Error())
 }
@@ -222,3 +226,98 @@ func TestCreateImageFromInstance_SetsEnaSupportFromInstance(t *testing.T) {
 	require.NotNil(t, ami.EnaSupport)
 	assert.True(t, *ami.EnaSupport)
 }
+
+// createUnavailableProvider refuses every CreateSnapshot as retry-later.
+type createUnavailableProvider struct {
+	ebsprovider.EBSProvider
+}
+
+func (p *createUnavailableProvider) CreateSnapshot(context.Context, ebsprovider.CreateSnapshotRequest) (*ebsprovider.Snapshot, error) {
+	return nil, &ebsprovider.ProviderError{Code: ebsprovider.ErrorCodeUnavailable, Message: "held by node-2"}
+}
+
+// TestSnapshotVolume_Provider_UnavailableIsRetryLater locks that the
+// CreateImage snapshot surfaces a provider's retry-later as ServiceUnavailable.
+func TestSnapshotVolume_Provider_UnavailableIsRetryLater(t *testing.T) {
+	svc, _, provider := setupProviderSnapshotImageService(t)
+	doc := seedProviderVolume(t, svc, provider, "vol-stopped-busy", 8)
+	svc.SetEBSProvider(&createUnavailableProvider{EBSProvider: provider})
+
+	err := svc.snapshotVolume(doc.VolumeID, "snap-stopped-busy", testAccountID)
+	require.Error(t, err)
+	assert.Equal(t, awserrors.ErrorServiceUnavailable, err.Error())
+}
+
+// createInUseProvider refuses every CreateSnapshot as volume in use.
+type createInUseProvider struct {
+	ebsprovider.EBSProvider
+}
+
+func (p *createInUseProvider) CreateSnapshot(context.Context, ebsprovider.CreateSnapshotRequest) (*ebsprovider.Snapshot, error) {
+	return nil, &ebsprovider.ProviderError{Code: ebsprovider.ErrorCodeVolumeInUse, Message: "published"}
+}
+
+// TestSnapshotVolume_Provider_VolumeInUseIsIncorrectState locks that the
+// CreateImage snapshot tells the caller to detach a volume the provider refuses as in use.
+func TestSnapshotVolume_Provider_VolumeInUseIsIncorrectState(t *testing.T) {
+	svc, _, provider := setupProviderSnapshotImageService(t)
+	doc := seedProviderVolume(t, svc, provider, "vol-attached-inuse", 8)
+	svc.SetEBSProvider(&createInUseProvider{EBSProvider: provider})
+
+	err := svc.snapshotVolume(doc.VolumeID, "snap-attached-inuse", testAccountID)
+
+	code, message, ok := awserrors.ResolveErrorDetail(err)
+	require.True(t, ok, "unregistered error: %v", err)
+	assert.Equal(t, awserrors.ErrorIncorrectState, code)
+	assert.Contains(t, message, doc.VolumeID)
+	assert.Contains(t, message, "Detach")
+}
+
+// countingSnapshotProvider counts the CreateSnapshot calls it forwards.
+type countingSnapshotProvider struct {
+	ebsprovider.EBSProvider
+
+	creates atomic.Int32
+}
+
+func (p *countingSnapshotProvider) CreateSnapshot(ctx context.Context, req ebsprovider.CreateSnapshotRequest) (*ebsprovider.Snapshot, error) {
+	p.creates.Add(1)
+	return p.EBSProvider.CreateSnapshot(ctx, req)
+}
+
+// An image of a running instance is snapshotted by the provider, which drains
+// a buffered volume itself, so CreateImage neither dials the drain socket nor
+// asks the instance's host to drain.
+func TestCreateImageFromInstance_RunningInstanceLeavesTheDrainToTheProvider(t *testing.T) {
+	const instanceID = "i-running-drain"
+	svc, _, provider := setupProviderSnapshotImageService(t)
+	_, nc := testutil.StartTestNATS(t)
+	svc.natsConn = nc
+	svc.config.DataDir = testutil.SocketTempDir(t)
+	counting := &countingSnapshotProvider{EBSProvider: provider}
+	svc.SetEBSProvider(counting)
+
+	doc := seedProviderVolume(t, svc, provider, "vol-running-drain", 8)
+	doc.State, doc.AttachedInstance = "in-use", instanceID
+	require.NoError(t, svc.MetadataStore().PutVolume(context.Background(), doc))
+	drains := testutil.StartDrainSocket(t, svc.config.DataDir, doc.VolumeID, "OK\n")
+	var commands atomic.Int32
+	sub, err := nc.Subscribe(ec2v1.InstanceCommandSubject(instanceID), func(msg *nats.Msg) {
+		commands.Add(1)
+		_ = msg.Respond([]byte(`{"Code":"InternalError","Message":"no drain socket"}`))
+	})
+	require.NoError(t, err)
+	t.Cleanup(func() { _ = sub.Unsubscribe() })
+
+	out, err := svc.CreateImageFromInstance(CreateImageParams{
+		Input:        &ec2.CreateImageInput{InstanceId: aws.String(instanceID), Name: aws.String("running-ami")},
+		RootVolumeID: doc.VolumeID,
+		IsRunning:    true,
+	}, testAccountID)
+	require.NoError(t, err)
+	require.NotNil(t, out.ImageId)
+
+	assert.Equal(t, int32(1), counting.creates.Load(), "the provider must take the snapshot")
+	assert.Zero(t, drains(), "CreateImage must not dial the drain socket")
+	assert.Zero(t, commands.Load(), "CreateImage must not ask the instance's host to drain")
+}

spinifex/domains/ec2/image/service_impl.go

@@ -38,7 +38,7 @@ type CreateImageParams struct {
 	Input         *ec2.CreateImageInput
 	RootVolumeID  string
 	SourceImageID string
-	IsRunning     bool // true = use live checkpoint (instance still running), false = use numbered checkpoint from Close
+	IsRunning     bool // logged only: the provider's CreateSnapshot drains a live volume itself
 }
 
 // ImageServiceImpl handles AMI image operations with S3 storage.
@@ -69,9 +69,7 @@ func (s *ImageServiceImpl) MetadataStore() *ebsmetadata.Store {
 	return s.metadata
 }
 
-// NewImageServiceImpl creates a new daemon-side image service. natsConn is used
-// to drain a running instance's volume (routed to the node hosting it) before
-// CreateImageFromInstance reads its live checkpoint.
+// NewImageServiceImpl creates a new daemon-side image service.
 func NewImageServiceImpl(cfg *config.Config, natsConn *nats.Conn) *ImageServiceImpl {
 	store := objectstore.NewS3ObjectStoreFromConfig(
 		cfg.Predastore.Host,
@@ -99,8 +97,7 @@ func NewImageServiceImplWithStore(store objectstore.ObjectStore, bucketName stri
 }
 
 // NewImageServiceImplWithConfig creates an image service with an explicit
-// config and NATS connection (for testing the running-volume drain path,
-// which needs config.DataDir/Predastore and a NATS connection to route to).
+// config, object store and NATS connection (for testing).
 func NewImageServiceImplWithConfig(cfg *config.Config, store objectstore.ObjectStore, natsConn *nats.Conn) *ImageServiceImpl {
 	return &ImageServiceImpl{
 		config:     cfg,
@@ -447,15 +444,9 @@ func (s *ImageServiceImpl) CreateImageFromInstance(params CreateImageParams, acc
 		"rootVolumeId", params.RootVolumeID, "amiId", amiID, "snapshotId", snapshotID,
 		"isRunning", params.IsRunning)
 
-	// Step 1: Snapshot root volume (live via NATS or offline from S3)
-	var snapshotErr error
-	if params.IsRunning {
-		snapshotErr = s.snapshotRunningVolume(params.RootVolumeID, snapshotID, accountID)
-	} else {
-		snapshotErr = s.snapshotStoppedVolume(params.RootVolumeID, snapshotID, accountID)
-	}
-	if snapshotErr != nil {
-		return nil, snapshotErr
+	// Step 1: Snapshot root volume
+	if err := s.snapshotVolume(params.RootVolumeID, snapshotID, accountID); err != nil {
+		return nil, err
 	}
 
 	// Step 2: Read source volume config for size
@@ -521,51 +512,21 @@ func (s *ImageServiceImpl) CreateImageFromInstance(params CreateImageParams, acc
 	}, nil
 }
 
-// snapshotRunningVolume creates a snapshot of a running instance's volume by
-// draining it to the node hosting it (so any writes buffered there reach S3),
-// then reading the live checkpoint written by nbdkit on every NBD Flush. This
-// shares domains/ec2/snapshot's DrainVolume with ec2.CreateSnapshot so the two
-// live-snapshot paths cannot diverge on when a drain is required.
-func (s *ImageServiceImpl) snapshotRunningVolume(volumeID, snapshotID, accountID string) error {
+// snapshotVolume snapshots the root volume through the EBS provider, whether
+// or not the instance is running. A provider that buffers writes drains the
+// volume inside its own CreateSnapshot, so no drain is issued here.
+func (s *ImageServiceImpl) snapshotVolume(volumeID, snapshotID, accountID string) error {
 	if s.config == nil {
 		return errors.New(awserrors.ErrorServerInternal)
 	}
 
 	volMeta, err := s.getVolumeMetadata(context.Background(), accountID, volumeID)
 	if err != nil {
-		slog.Error("snapshotRunningVolume: failed to read volume metadata", "volumeId", volumeID, "err", err)
+		slog.Error("snapshotVolume: failed to read volume metadata", "volumeId", volumeID, "err", err)
 		return errors.New(awserrors.ErrorServerInternal)
 	}
 
-	// Flush the writes still buffered by whichever node serves the volume, so
-	// the snapshot below is current rather than an arbitrarily stale one
-	// predating e.g. a guest-triggered GPT rewrite (growpart on first boot). An
-	// attached volume that cannot be drained fails here rather than silently
-	// producing an image from a stale checkpoint.
-	if err := ec2snapshot.DrainVolume(context.Background(), s.config, s.store, s.natsConn, volumeID, volMeta.State, volMeta.AttachedInstance, accountID); err != nil {
-		slog.Error("snapshotRunningVolume: drain failed", "volumeId", volumeID, "err", err)
-		return errors.New(awserrors.ErrorServerInternal)
-	}
-
-	return s.createSnapshot(context.Background(), volMeta, snapshotID, "snapshotRunningVolume")
-}
-
-// snapshotStoppedVolume creates a snapshot of a detached volume. No drain: the
-// volume is stopped/detached, so nothing is buffered on a serving node.
-// viperblockd's handleCreateSnapshot prefers a live mounted engine over opening
-// a second one, so it handles the mounted case itself.
-func (s *ImageServiceImpl) snapshotStoppedVolume(volumeID, snapshotID, accountID string) error {
-	if s.config == nil {
-		return errors.New(awserrors.ErrorServerInternal)
-	}
-
-	volMeta, err := s.getVolumeMetadata(context.Background(), accountID, volumeID)
-	if err != nil {
-		slog.Error("snapshotStoppedVolume: failed to read volume metadata", "volumeId", volumeID, "err", err)
-		return errors.New(awserrors.ErrorServerInternal)
-	}
-
-	return s.createSnapshot(context.Background(), volMeta, snapshotID, "snapshotStoppedVolume")
+	return s.createSnapshot(context.Background(), volMeta, snapshotID, "snapshotVolume")
 }
 
 // createSnapshot delegates snapshot creation for volMeta to the EBS provider.
@@ -580,7 +541,7 @@ func (s *ImageServiceImpl) createSnapshot(ctx context.Context, volMeta ebsmetada
 	})
 	if err != nil || created == nil {
 		slog.ErrorContext(ctx, "provider CreateSnapshot failed", "op", op, "volumeId", volMeta.VolumeID, "snapshotId", snapshotID, "err", err)
-		return errors.New(awserrors.ErrorServerInternal)
+		return ec2snapshot.ProviderCreateError(volMeta.VolumeID, err)
 	}
 	slog.InfoContext(ctx, "snapshot created", "op", op, "volumeId", volMeta.VolumeID, "snapshotId", snapshotID)
 	return nil

spinifex/domains/ec2/image/service_impl_test.go

@@ -2551,56 +2551,19 @@ func TestResetImageAttribute_MissingParameters(t *testing.T) {
 	assert.Equal(t, awserrors.ErrorMissingParameter, err.Error())
 }
 
-func TestSnapshotRunningVolume_VolumeConfigMissing(t *testing.T) {
+func TestSnapshotVolume_VolumeConfigMissing(t *testing.T) {
 	store := objectstore.NewMemoryObjectStore()
 	svc := &ImageServiceImpl{
 		config:     &config.Config{},
 		store:      store,
 		bucketName: testBucket,
 	}
-	err := svc.snapshotRunningVolume("vol-missing", "snap-r1", testAccountID)
+	err := svc.snapshotVolume("vol-missing", "snap-s1", testAccountID)
 	require.Error(t, err)
 	assert.Equal(t, awserrors.ErrorServerInternal, err.Error())
 }
 
-func TestSnapshotRunningVolume_BackendInitFails(t *testing.T) {
-	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
-		w.WriteHeader(http.StatusForbidden)
-	}))
-	defer srv.Close()
-
-	store := objectstore.NewMemoryObjectStore()
-	svc := &ImageServiceImpl{
-		config: &config.Config{
-			Predastore: config.PredastoreConfig{
-				Host:   srv.URL,
-				Region: "us-east-1",
-				Bucket: testBucket,
-			},
-		},
-		store:      store,
-		bucketName: testBucket,
-	}
-	createTestVolumeConfig(t, store, "vol-running1", 10)
-
-	err := svc.snapshotRunningVolume("vol-running1", "snap-r2", testAccountID)
-	require.Error(t, err)
-	assert.Equal(t, awserrors.ErrorServerInternal, err.Error())
-}
-
-func TestSnapshotStoppedVolume_VolumeConfigMissing(t *testing.T) {
-	store := objectstore.NewMemoryObjectStore()
-	svc := &ImageServiceImpl{
-		config:     &config.Config{},
-		store:      store,
-		bucketName: testBucket,
-	}
-	err := svc.snapshotStoppedVolume("vol-missing", "snap-s1", testAccountID)
-	require.Error(t, err)
-	assert.Equal(t, awserrors.ErrorServerInternal, err.Error())
-}
-
-func TestSnapshotStoppedVolume_BackendInitFails(t *testing.T) {
+func TestSnapshotVolume_BackendInitFails(t *testing.T) {
 	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
 		w.WriteHeader(http.StatusForbidden)
 	}))
@@ -2620,7 +2583,7 @@ func TestSnapshotStoppedVolume_BackendInitFails(t *testing.T) {
 	}
 	createTestVolumeConfig(t, store, "vol-stopped1", 10)
 
-	err := svc.snapshotStoppedVolume("vol-stopped1", "snap-s2", testAccountID)
+	err := svc.snapshotVolume("vol-stopped1", "snap-s2", testAccountID)
 	require.Error(t, err)
 	assert.Equal(t, awserrors.ErrorServerInternal, err.Error())
 }
@@ -2646,60 +2609,7 @@ func imageS3MockServer(t *testing.T) *httptest.Server {
 	}))
 }
 
-func TestSnapshotRunningVolume_EncryptionKeyLoadError(t *testing.T) {
-	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
-		w.WriteHeader(http.StatusOK)
-	}))
-	defer srv.Close()
-
-	store := objectstore.NewMemoryObjectStore()
-	svc := &ImageServiceImpl{
-		config: &config.Config{
-			Predastore: config.PredastoreConfig{
-				Host:   srv.URL,
-				Region: "us-east-1",
-				Bucket: testBucket,
-			},
-			Viperblock: config.ViperblockConfig{
-				EncryptionKeyFile: "/nonexistent-image-run-key.bin",
-			},
-		},
-		store:      store,
-		bucketName: testBucket,
-	}
-	createTestVolumeConfig(t, store, "vol-running-enckey", 10)
-
-	err := svc.snapshotRunningVolume("vol-running-enckey", "snap-run-enckey", testAccountID)
-	require.Error(t, err)
-	assert.Equal(t, awserrors.ErrorServerInternal, err.Error())
-}
-
-func TestSnapshotRunningVolume_LoadStateFails(t *testing.T) {
-	srv := imageS3MockServer(t)
-	defer srv.Close()
-
-	store := objectstore.NewMemoryObjectStore()
-	svc := &ImageServiceImpl{
-		config: &config.Config{
-			Predastore: config.PredastoreConfig{
-				Host:      srv.URL,
-				Region:    "us-east-1",
-				Bucket:    testBucket,
-				AccessKey: "test-key",
-				SecretKey: "test-secret",
-			},
-		},
-		store:      store,
-		bucketName: testBucket,
-	}
-	createTestVolumeConfig(t, store, "vol-running-lsf", 10)
-
-	err := svc.snapshotRunningVolume("vol-running-lsf", "snap-run-lsf", testAccountID)
-	require.Error(t, err)
-	assert.Equal(t, awserrors.ErrorServerInternal, err.Error())
-}
-
-func TestSnapshotStoppedVolume_EncryptionKeyLoadError(t *testing.T) {
+func TestSnapshotVolume_EncryptionKeyLoadError(t *testing.T) {
 	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
 		w.WriteHeader(http.StatusOK)
 	}))
@@ -2722,12 +2632,12 @@ func TestSnapshotStoppedVolume_EncryptionKeyLoadError(t *testing.T) {
 	}
 	createTestVolumeConfig(t, store, "vol-stopped-enckey", 10)
 
-	err := svc.snapshotStoppedVolume("vol-stopped-enckey", "snap-stop-enckey", testAccountID)
+	err := svc.snapshotVolume("vol-stopped-enckey", "snap-stop-enckey", testAccountID)
 	require.Error(t, err)
 	assert.Equal(t, awserrors.ErrorServerInternal, err.Error())
 }
 
-func TestSnapshotStoppedVolume_LoadStateFails(t *testing.T) {
+func TestSnapshotVolume_LoadStateFails(t *testing.T) {
 	srv := imageS3MockServer(t)
 	defer srv.Close()
 
@@ -2747,7 +2657,7 @@ func TestSnapshotStoppedVolume_LoadStateFails(t *testing.T) {
 	}
 	createTestVolumeConfig(t, store, "vol-stopped-lsf", 10)
 
-	err := svc.snapshotStoppedVolume("vol-stopped-lsf", "snap-stop-lsf", testAccountID)
+	err := svc.snapshotVolume("vol-stopped-lsf", "snap-stop-lsf", testAccountID)
 	require.Error(t, err)
 	assert.Equal(t, awserrors.ErrorServerInternal, err.Error())
 }

spinifex/domains/ec2/instance/service_impl.go

@@ -1237,7 +1237,12 @@ func (s *InstanceServiceImpl) StartInstance(ctx context.Context, instance *vm.VM
 				"instanceId", command.ID, "err", err)
 			return errors.New(awserrors.ErrorInvalidInstanceIDNotFound)
 		default:
-			slog.ErrorContext(ctx, "StartInstance: vmMgr.Start failed", "err", err)
+			// A volume held elsewhere is the caller's 400, not a fault on this node.
+			if refusal := volumeHeldElsewhereError(err); refusal != nil {
+				slog.WarnContext(ctx, "StartInstance: vmMgr.Start failed", "instanceId", command.ID, "err", err)
+				return refusal
+			}
+			slog.ErrorContext(ctx, "StartInstance: vmMgr.Start failed", "instanceId", command.ID, "err", err)
 			return errors.New(awserrors.ErrorServerInternal)
 		}
 	}
@@ -2838,40 +2843,15 @@ func (s *InstanceServiceImpl) RecoverInstance(ctx context.Context, instance *vm.
 	return s.launchClaimedInstance(ctx, instance, undo)
 }
 
-// volumeExclusionMarkers are the refusals that mean another node is writing this
-// volume right now, or is still live and holds its unsealed writes. A previous
-// owner that has stopped heartbeating is opened over with a warning instead. The
-// text arrives having crossed NATS from viperblockd.
-var volumeExclusionMarkers = []string{
-	ec2v1.VolumeHeldElsewhereText,
-}
-
-// ErrVolumeHeldElsewhere reports a launch refused because another node still
-// holds one of the instance's volume leases.
-//
-// Recovery treats it differently from every other failure — it clears on the
-// lease's own clock rather than needing anything fixed — so it has to be
-// recognisable with errors.Is rather than inferred from text that has already
-// crossed NATS once.
-var ErrVolumeHeldElsewhere = errors.New("a volume is leased by another node")
-
-// volumeHeldElsewhereError converts a mount refused because another node holds
-// the volume into a client-facing error carrying that reason. Returns nil for
-// anything else. Without it the caller gets ServerInternal and is told to
-// retry, which cannot succeed while the holder is down, and the text naming the
-// holder never leaves the node's log.
+// volumeHeldElsewhereError converts a mount the EBS provider refused with its
+// held-elsewhere code into IncorrectState naming the holder, wrapping
+// ec2v1.ErrVolumeHeldElsewhere, and returns nil for anything else.
 func volumeHeldElsewhereError(err error) error {
-	if err == nil {
+	if !errors.Is(err, ebsprovider.ErrVolumeHeldElsewhere) {
 		return nil
 	}
-	msg := err.Error()
-	for _, marker := range volumeExclusionMarkers {
-		if strings.Contains(msg, marker) {
-			return fmt.Errorf("%w: %w", ErrVolumeHeldElsewhere,
-				awserrors.Errorf(awserrors.ErrorIncorrectState, "%s", msg))
-		}
-	}
-	return nil
+	return fmt.Errorf("%w: %w", ec2v1.ErrVolumeHeldElsewhere,
+		awserrors.Errorf(awserrors.ErrorIncorrectState, "%s", err.Error()))
 }
 
 // restoreClaimedStoppedInstance writes instance back to the shared

spinifex/domains/ec2/instance/volume_held_elsewhere_test.go

@@ -4,50 +4,19 @@ package instance
 //would widen the handler API for a test rather than for a caller.
 
 import (
+	"context"
 	"errors"
 	"fmt"
 	"testing"
 
+	"github.com/mulgadc/spinifex/contracts/ec2/v1"
 	"github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
+	"github.com/mulgadc/spinifex/spinifex/providers/ebs"
+	"github.com/mulgadc/spinifex/spinifex/runtime/compute/vm"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 )
 
-// TestVolumeHeldElsewhere_NamesTheHolder covers both refusals that mean the
-// data is on another node. An operator who gets ServerInternal is told to retry
-// something that cannot succeed while that node is down, so the node name and
-// the reason have to survive the trip out of the daemon.
-func TestVolumeHeldElsewhere_NamesTheHolder(t *testing.T) {
-	cases := []struct {
-		name string
-		err  error
-	}{
-		{
-			name: "live holder, refused by the lease",
-			err: errors.New("failed to mount volume: volume is leased by another owner: " +
-				"held by node-b since 2026-09-02T17:04:39Z"),
-		},
-	}
-
-	for _, tc := range cases {
-		t.Run(tc.name, func(t *testing.T) {
-			got := volumeHeldElsewhereError(tc.err)
-			require.Error(t, got, "a volume held elsewhere must not fall through to ServerInternal")
-
-			code := awserrors.ValidErrorCodeFromError(got)
-			assert.NotEqual(t, awserrors.ErrorServerInternal, code,
-				"ServerInternal reads as a transient fault and invites a retry that cannot succeed")
-			assert.Contains(t, got.Error(), "node-b", "the caller has to learn where the data is")
-
-			assert.ErrorIs(t, got, ErrVolumeHeldElsewhere,
-				"recovery retries this on a shorter clock than anything else, so it has to be matchable")
-			resolved, _, ok := awserrors.ResolveErrorDetail(got)
-			require.True(t, ok, "the sentinel must not hide the code from the layer that returns it")
-			assert.Equal(t, awserrors.ErrorIncorrectState, resolved)
-		})
-	}
-}
-
 // TestVolumeHeldElsewhere_IgnoresUnrelatedFailures guards the other direction:
 // a genuine internal fault must keep reporting as one rather than being
 // relabelled as somebody else's volume.
@@ -60,4 +29,76 @@ func TestVolumeHeldElsewhere_IgnoresUnrelatedFailures(t *testing.T) {
 	// than refused, so this text is not an exclusion and must not become one.
 	assert.NoError(t, volumeHeldElsewhereError(errors.New(
 		"took over from node-b, which held unsealed writes since 2026-09-02T17:04:39Z")))
+
+	// Only the provider's code classifies a refusal, so its wording alone does not.
+	assert.NoError(t, volumeHeldElsewhereError(errors.New(
+		"failed to mount volume: "+ec2v1.VolumeHeldElsewhereText+": held by node-b")))
+}
+
+// A provider is free to word its refusal however it likes; the held-elsewhere
+// code is what says another node has the volume.
+func TestVolumeHeldElsewhere_MatchesTheProviderCodeWhateverTheWording(t *testing.T) {
+	refusal := fmt.Errorf("publish volume vol-1: %w", &ebsprovider.ProviderError{
+		Code:    ebsprovider.ErrorCodeVolumeHeldElsewhere,
+		Message: "volume vol-1 is published to node-a",
+	})
+
+	got := volumeHeldElsewhereError(refusal)
+	require.Error(t, got, "a held-elsewhere code must be classified even without viperblockd's text")
+	assert.ErrorIs(t, got, ec2v1.ErrVolumeHeldElsewhere)
+	assert.Contains(t, got.Error(), "node-a", "the caller has to learn where the data is")
+	code, _, ok := awserrors.ResolveErrorDetail(got)
+	require.True(t, ok)
+	assert.Equal(t, awserrors.ErrorIncorrectState, code)
+}
+
+// volume_in_use is a read-only mismatch, not another node's claim, so it must
+// not be relabelled as a volume held elsewhere even though the sentinels nest.
+func TestVolumeHeldElsewhere_VolumeInUseAloneIsNotHeldElsewhere(t *testing.T) {
+	mismatch := fmt.Errorf("publish volume vol-1: %w", &ebsprovider.ProviderError{
+		Code:    ebsprovider.ErrorCodeVolumeInUse,
+		Message: "volume vol-1 is exported read-write, not read-only",
+	})
+	assert.NoError(t, volumeHeldElsewhereError(mismatch))
+}
+
+// heldElsewhereMounter refuses every mount the way the contract mounter
+// reports a provider's held-elsewhere code, in wording that is not viperblockd's.
+type heldElsewhereMounter struct{ raceVolumeMounter }
+
+func (heldElsewhereMounter) Mount(context.Context, *vm.VM) error {
+	return fmt.Errorf("publish volume vol-1: %w", &ebsprovider.ProviderError{
+		Code:    ebsprovider.ErrorCodeVolumeHeldElsewhere,
+		Message: "volume vol-1 is published to node-a",
+	})
+}
+
+// Both start paths a caller can reach must hand back the classified refusal:
+// the stopped-instance claim (ec2.start) and the live owner (ec2.cmd), which
+// the EKS reconciler tries first.
+func TestStartPaths_ReportAVolumeHeldElsewhereByCode(t *testing.T) {
+	t.Setenv("XDG_RUNTIME_DIR", t.TempDir())
+	types, _ := defaultPrepareInstanceTypes()
+
+	t.Run("claimed relaunch", func(t *testing.T) {
+		mgr := vm.NewManager()
+		mgr.SetDeps(vm.Deps{NodeID: "node-b", VolumeMounter: heldElsewhereMounter{}})
+		svc := &InstanceServiceImpl{resourceMgr: &fakeResourceCapacityProvider{instanceTypes: types}, vmMgr: mgr}
+
+		err := svc.launchClaimedInstance(t.Context(), &vm.VM{ID: "i-held-claim", AccountID: "acc", InstanceType: "t3.micro", Status: vm.StatePending}, func() {})
+		assert.ErrorIs(t, err, ec2v1.ErrVolumeHeldElsewhere)
+		assert.Equal(t, awserrors.ErrorIncorrectState, awserrors.ValidErrorCodeFromError(err))
+	})
+
+	t.Run("live owner", func(t *testing.T) {
+		const id = "i-held-live"
+		mgr := mgrWith(map[string]*vm.VM{id: {ID: id, Status: vm.StateStopped, InstanceType: "t3.micro"}})
+		mgr.SetDeps(vm.Deps{NodeID: "node-b", VolumeMounter: heldElsewhereMounter{}})
+		v, _ := mgr.Get(id)
+		svc := &InstanceServiceImpl{resourceMgr: &fakeResourceCapacityProvider{instanceTypes: types}, vmMgr: mgr}
+
+		err := svc.StartInstance(t.Context(), v, ec2v1.EC2InstanceCommand{ID: id})
+		assert.ErrorIs(t, err, ec2v1.ErrVolumeHeldElsewhere)
+		assert.Equal(t, awserrors.ErrorIncorrectState, awserrors.ValidErrorCodeFromError(err))
+	})
 }

spinifex/domains/ec2/launchtemplate/account_cleanup.go

@@ -130,3 +130,51 @@ func tallyOrphan(outcome orphanOutcome, removed, failed, unsettled *int) {
 	case orphanGone, orphanOwned:
 	}
 }
+
+// AccountTemplateDeletes is what one DeleteLiveTemplates run did, by template
+// ID. It is the owner's account of its attempts, not proof of absence.
+type AccountTemplateDeletes struct {
+	DeletedTemplateIDs []string
+	FailedTemplateIDs  []string
+	// Referenced are live templates left because referenced named them, with its reason.
+	Referenced map[string]string
+}
+
+// DeleteLiveTemplates deletes the account's live templates through the same
+// owner delete as DeleteLaunchTemplate, except those referenced names. A header
+// it cannot read or decode is left for the owner scan to report. A listing
+// failure is returned and nothing is deleted.
+func (s *LaunchTemplateServiceImpl) DeleteLiveTemplates(ctx context.Context, accountID string, referenced func(id, name string) string) (*AccountTemplateDeletes, error) {
+	keys, err := kvutil.Keys(ctx, s.kv)
+	if err != nil && !errors.Is(err, jetstream.ErrNoKeysFound) {
+		return nil, fmt.Errorf("list launch-template keys: %w", err)
+	}
+	prefix := accountID + "."
+	out := &AccountTemplateDeletes{Referenced: map[string]string{}}
+	for _, key := range keys {
+		if !isHeaderKey(key, prefix) {
+			continue
+		}
+		entry, err := s.kv.Get(ctx, key)
+		if err != nil {
+			continue
+		}
+		var header LaunchTemplateHeader
+		if json.Unmarshal(entry.Value(), &header) != nil || prefix+header.LaunchTemplateId != key {
+			continue
+		}
+		if reason := referenced(header.LaunchTemplateId, header.LaunchTemplateName); reason != "" {
+			out.Referenced[header.LaunchTemplateId] = reason
+			continue
+		}
+		if err := s.deleteTemplate(ctx, accountID, &header, entry); err != nil {
+			slog.WarnContext(ctx, "Launch-template account cleanup: live template delete failed", "launchTemplateId", header.LaunchTemplateId, "err", err)
+			out.FailedTemplateIDs = append(out.FailedTemplateIDs, header.LaunchTemplateId)
+			continue
+		}
+		out.DeletedTemplateIDs = append(out.DeletedTemplateIDs, header.LaunchTemplateId)
+	}
+	slog.InfoContext(ctx, "Launch-template account live delete completed", "accountID", accountID,
+		"deleted", len(out.DeletedTemplateIDs), "failed", len(out.FailedTemplateIDs), "referenced", len(out.Referenced))
+	return out, nil
+}

spinifex/domains/ec2/snapshot/drain_test.go

@@ -2,12 +2,8 @@ package snapshot
 
 import (
 	"context"
-	"encoding/json"
-	"github.com/mulgadc/spinifex/spinifex/foundation/messaging/nats"
-	"net/http"
-	"net/http/httptest"
+	"sync/atomic"
 	"testing"
-	"time"
 
 	"github.com/aws/aws-sdk-go/aws"
 	"github.com/aws/aws-sdk-go/service/ec2"
@@ -15,8 +11,6 @@ import (
 	"github.com/mulgadc/spinifex/internal/testkit"
 	"github.com/mulgadc/spinifex/spinifex/bootstrap/config"
 	"github.com/mulgadc/spinifex/spinifex/domains/ec2/ebs/metadata"
-	"github.com/mulgadc/spinifex/spinifex/domains/ec2/volumestate"
-	"github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
 	"github.com/mulgadc/spinifex/spinifex/providers/ebs"
 	"github.com/mulgadc/spinifex/spinifex/providers/objectstore"
 	"github.com/nats-io/nats.go"
@@ -24,321 +18,78 @@ import (
 	"github.com/stretchr/testify/require"
 )
 
-const drainInstanceID = "i-drain-host"
+const attachedInstanceID = "i-attached-host"
 
-// setupDrainService returns a snapshot service configured like a daemon that
-// answers ec2.CreateSnapshot: a reachable Predastore host (so drains are not
-// skipped as metadata-only), a data dir with no drain socket in it, and a NATS
-// connection. The Predastore stub rejects everything, so a snapshot that gets
-// past the drain fails fast rather than hanging on a dead address.
-func setupDrainService(t *testing.T) (*SnapshotServiceImpl, *objectstore.MemoryObjectStore, *nats.Conn) {
+// countingSnapshotProvider counts the CreateSnapshot calls it forwards.
+type countingSnapshotProvider struct {
+	ebsprovider.EBSProvider
+
+	creates atomic.Int32
+}
+
+func (p *countingSnapshotProvider) CreateSnapshot(ctx context.Context, req ebsprovider.CreateSnapshotRequest) (*ebsprovider.Snapshot, error) {
+	p.creates.Add(1)
+	return p.EBSProvider.CreateSnapshot(ctx, req)
+}
+
+// setupAttachedSnapshot is a viperblockd node answering CreateSnapshot for a
+// volume attached to a running instance, with a memory provider behind it.
+func setupAttachedSnapshot(t *testing.T, volumeID string) (*SnapshotServiceImpl, *countingSnapshotProvider, *nats.Conn) {
 	t.Helper()
 
-	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
-		w.WriteHeader(http.StatusForbidden)
-	}))
-	t.Cleanup(srv.Close)
-
 	_, nc := testutil.StartTestNATS(t)
-	store := objectstore.NewMemoryObjectStore()
 	cfg := &config.Config{
-		DataDir: testutil.SocketTempDir(t),
-		Predastore: config.PredastoreConfig{
-			Host:   srv.URL,
-			Bucket: "test-bucket",
-		},
+		DataDir:    testutil.SocketTempDir(t),
+		EBS:        config.EBSConfig{Provider: config.EBSProviderViperblockd},
+		Predastore: config.PredastoreConfig{Host: "127.0.0.1:1", Bucket: "test-bucket"},
 	}
-	return NewSnapshotServiceImplWithStore(cfg, store, nc), store, nc
-}
+	svc := NewSnapshotServiceImplWithStore(cfg, objectstore.NewMemoryObjectStore(), nc)
+	provider := &countingSnapshotProvider{EBSProvider: ebsprovider.NewMemoryProvider(ebsprovider.Capabilities{})}
+	svc.SetEBSProvider(provider)
 
-// seedVolumeAttachment writes the control-plane state.json that decides whether
-// a snapshot must drain.
-func seedVolumeAttachment(t *testing.T, store *objectstore.MemoryObjectStore, volumeID, state, instanceID string) {
-	t.Helper()
-	require.NoError(t, volumestate.Write(context.Background(), store, "test-bucket", volumeID, volumestate.Record{
-		State:            state,
-		AttachedInstance: instanceID,
-	}))
-}
-
-// drainResponder answers the per-instance command subject the way the node
-// hosting the instance does, and hands each request it received to the test.
-func drainResponder(t *testing.T, nc *nats.Conn, instanceID string, reply []byte) chan *nats.Msg {
-	t.Helper()
-
-	got := make(chan *nats.Msg, 4)
-	sub, err := nc.Subscribe("ec2.cmd."+instanceID, func(msg *nats.Msg) {
-		got <- msg
-		_ = msg.Respond(reply)
-	})
-	require.NoError(t, err)
-	t.Cleanup(func() { _ = sub.Unsubscribe() })
-
-	return got
-}
-
-// drainedAck is the reply of a node whose drain reached S3.
-func drainedAck(t *testing.T, volumeID string) []byte {
-	t.Helper()
-	data, err := json.Marshal(ec2v1.DrainVolumeResponse{VolumeID: volumeID, Status: ec2v1.DrainVolumeStatusDrained})
-	require.NoError(t, err)
-	return data
-}
-
-// requireReturnsWithin fails the test if fn has not returned successfully by
-// limit, so a drain that blocks on a socket it should not have dialed reports
-// as a failure rather than as a slow test.
-func requireReturnsWithin(t *testing.T, limit time.Duration, fn func() error) {
-	t.Helper()
-
-	done := make(chan error, 1)
-	go func() { done <- fn() }()
-
-	select {
-	case err := <-done:
-		require.NoError(t, err)
-	case <-time.After(limit):
-		t.Fatalf("drain did not return within %s", limit)
-	}
-}
-
-// awaitDrainCommand returns the command the hosting node received, failing the
-// test if none arrives.
-func awaitDrainCommand(t *testing.T, got chan *nats.Msg) ec2v1.EC2InstanceCommand {
-	t.Helper()
-	select {
-	case msg := <-got:
-		var command ec2v1.EC2InstanceCommand
-		require.NoError(t, json.Unmarshal(msg.Data, &command))
-		assert.Equal(t, testAccountID, msg.Header.Get(natsmsg.AccountIDHeader))
-		return command
-	case <-time.After(2 * time.Second):
-		t.Fatal("the node hosting the volume never received a drain command")
-		return ec2v1.EC2InstanceCommand{}
-	}
-}
-
-// An attached volume is drained on the node hosting its instance, which is the
-// only node with the socket — the node answering CreateSnapshot usually is not.
-func TestDrainVolume_AttachedRoutesToHostingNode(t *testing.T) {
-	svc, store, nc := setupDrainService(t)
-	seedVolumeAttachment(t, store, "vol-attached", "in-use", drainInstanceID)
-	got := drainResponder(t, nc, drainInstanceID, drainedAck(t, "vol-attached"))
-
-	require.NoError(t, svc.drainVolume(context.Background(), "vol-attached", "", "", testAccountID))
-
-	command := awaitDrainCommand(t, got)
-	assert.True(t, command.Attributes.DrainVolume)
-	assert.Equal(t, drainInstanceID, command.ID)
-	require.NotNil(t, command.DrainVolumeData)
-	assert.Equal(t, "vol-attached", command.DrainVolumeData.VolumeID)
-}
-
-// No responder on the command subject means no node is running the instance,
-// which is what a stopped instance looks like: stop tears down both the plugin
-// and the subscription but deliberately leaves a boot volume attached. Failing
-// here would make every stopped instance's root volume unsnapshottable.
-func TestDrainVolume_AttachedToStoppedInstanceTakesStoppedPath(t *testing.T) {
-	svc, store, _ := setupDrainService(t)
-	seedVolumeAttachment(t, store, "vol-no-host", "in-use", drainInstanceID)
-
-	require.NoError(t, svc.drainVolume(context.Background(), "vol-no-host", "", "", testAccountID))
-}
-
-// A host that still holds the instance but reports it not running has nothing
-// to flush, so the sealed checkpoint stands. This is the host-drain stop, which
-// unmounts the volumes but keeps the subscription.
-func TestDrainVolume_NotRunningAckTakesStoppedPath(t *testing.T) {
-	svc, store, nc := setupDrainService(t)
-	seedVolumeAttachment(t, store, "vol-not-running", "in-use", drainInstanceID)
-	ack, err := json.Marshal(ec2v1.DrainVolumeResponse{
-		VolumeID: "vol-not-running", Status: ec2v1.DrainVolumeStatusNotRunning,
-	})
-	require.NoError(t, err)
-	got := drainResponder(t, nc, drainInstanceID, ack)
-
-	require.NoError(t, svc.drainVolume(context.Background(), "vol-not-running", "", "", testAccountID))
-	assert.True(t, awaitDrainCommand(t, got).Attributes.DrainVolume)
-}
-
-// A hosting node that reports the drain failed (wedged plugin, socket gone
-// mid-restart) fails the snapshot rather than falling through.
-func TestDrainVolume_AttachedWithFailedAckFails(t *testing.T) {
-	svc, store, nc := setupDrainService(t)
-	seedVolumeAttachment(t, store, "vol-drain-err", "in-use", drainInstanceID)
-	drainResponder(t, nc, drainInstanceID,
-		[]byte(`{"Code":"`+awserrors.ErrorServerInternal+`","Message":"drain failed"}`))
-
-	err := svc.drainVolume(context.Background(), "vol-drain-err", "", "", testAccountID)
-	require.Error(t, err)
-	assert.Contains(t, err.Error(), "drain failed")
-}
-
-// An ack that is not the drained ack is not evidence the writes reached S3.
-func TestDrainVolume_AttachedWithUnexpectedAckFails(t *testing.T) {
-	svc, store, nc := setupDrainService(t)
-	seedVolumeAttachment(t, store, "vol-odd-ack", "in-use", drainInstanceID)
-	drainResponder(t, nc, drainInstanceID, []byte(`{}`))
-
-	err := svc.drainVolume(context.Background(), "vol-odd-ack", "", "", testAccountID)
-	require.Error(t, err)
-	assert.Contains(t, err.Error(), "unexpected ack")
-}
-
-// Nothing writes to an unattached volume, so the checkpoint its Close() left
-// behind is current and no drain is attempted — not even locally. Dialing the
-// socket is not a probe: it makes the plugin run a full flush, so an available
-// volume must be decided from the record without touching it.
-func TestDrainVolume_AvailableDoesNotDialLocalSocket(t *testing.T) {
-	svc, store, _ := setupDrainService(t)
-	seedVolumeAttachment(t, store, "vol-available", "available", "")
-
-	// A socket that never answers: a dial would park here until the read
-	// deadline, so returning promptly is what proves none was made.
-	testutil.StartBlockingDrainSocket(t, svc.config.DataDir, "vol-available", "OK\n")
-
-	// A nil connection proves no command was issued: routing one would error.
-	svc.natsConn = nil
-
-	requireReturnsWithin(t, 2*time.Second, func() error {
-		return svc.drainVolume(context.Background(), "vol-available", "", "", testAccountID)
-	})
-}
-
-// A volume recorded as in-use but with no instance (drifted state) has no node
-// to address, so it takes the stopped path rather than failing every snapshot.
-func TestDrainVolume_InUseWithoutInstanceTakesStoppedPath(t *testing.T) {
-	svc, store, _ := setupDrainService(t)
-	seedVolumeAttachment(t, store, "vol-no-instance", "in-use", "")
-	svc.natsConn = nil
-
-	require.NoError(t, svc.drainVolume(context.Background(), "vol-no-instance", "", "", testAccountID))
-}
-
-// When the volume is served by this node the socket is local, so the drain
-// short-circuits without a hop. This is every single-node deployment.
-func TestDrainVolume_LocalSocketShortCircuits(t *testing.T) {
-	svc, store, _ := setupDrainService(t)
-	seedVolumeAttachment(t, store, "vol-local", "in-use", drainInstanceID)
-	testutil.StartDrainSocket(t, svc.config.DataDir, "vol-local", "OK\n")
-
-	// A nil connection proves the local ack was enough: no responder exists for
-	// drainInstanceID, so a routed command would fail.
-	svc.natsConn = nil
-
-	require.NoError(t, svc.drainVolume(context.Background(), "vol-local", "", "", testAccountID))
-}
-
-// A socket that answers without acking means this node does serve the volume
-// and the flush failed. Routing that onward would come back to this same node
-// and re-run the drain that has already failed, so it fails here instead.
-func TestDrainVolume_LocalSocketErrorAckFailsWithoutRouting(t *testing.T) {
-	svc, store, nc := setupDrainService(t)
-	seedVolumeAttachment(t, store, "vol-local-err", "in-use", drainInstanceID)
-	testutil.StartDrainSocket(t, svc.config.DataDir, "vol-local-err", "ERR\n")
-	got := drainResponder(t, nc, drainInstanceID, drainedAck(t, "vol-local-err"))
-
-	err := svc.drainVolume(context.Background(), "vol-local-err", "", "", testAccountID)
-	require.Error(t, err)
-
-	assert.Empty(t, got, "a local drain that failed must not be re-issued to the host node")
-}
-
-// state.json is the control-plane-owned attachment record; the copy inside
-// config.json is rewritten by the live NBD plugin from its stale in-memory
-// state, so a volume reading "available" there can still be attached.
-func TestDrainVolume_StateJSONOverridesConfigJSON(t *testing.T) {
-	svc, store, nc := setupDrainService(t)
-	seedVolumeAttachment(t, store, "vol-stale-config", "in-use", drainInstanceID)
-	got := drainResponder(t, nc, drainInstanceID, drainedAck(t, "vol-stale-config"))
-
-	// "available" mirrors the stale State a live NBD plugin leaves in
-	// config.json; drainVolume must not trust it (see the test's own doc above).
-	require.NoError(t, svc.drainVolume(context.Background(), "vol-stale-config", "available", "", testAccountID))
-
-	assert.True(t, awaitDrainCommand(t, got).Attributes.DrainVolume)
-}
-
-// Volumes predating the state.json split have no state object, so the state
-// embedded in config.json is all there is to decide from.
-func TestDrainVolume_FallsBackToConfigJSONWhenNoStateObject(t *testing.T) {
-	svc, _, nc := setupDrainService(t)
-	got := drainResponder(t, nc, drainInstanceID, drainedAck(t, "vol-legacy"))
-
-	// The state/instance embedded in a legacy config.json, with no state.json to overlay it.
-	require.NoError(t, svc.drainVolume(context.Background(), "vol-legacy", "in-use", drainInstanceID, testAccountID))
-
-	assert.True(t, awaitDrainCommand(t, got).Attributes.DrainVolume)
-}
-
-// Without Predastore the snapshot is metadata-only and never reads a live
-// checkpoint, so there is nothing for a drain to make current.
-func TestDrainVolume_MetadataOnlySkipsDrain(t *testing.T) {
-	svc, store, _ := setupDrainService(t)
-	seedVolumeAttachment(t, store, "vol-meta-only", "in-use", drainInstanceID)
-	svc.config.Predastore.Host = ""
-	svc.natsConn = nil
-
-	require.NoError(t, svc.drainVolume(context.Background(), "vol-meta-only", "", "", testAccountID))
-}
-
-// putProviderVolume seeds both the control-plane document CreateSnapshot's
-// provider branch reads for attachment state, and the provider's own record,
-// mirroring what CreateVolume's provider branch leaves behind.
-func putProviderVolume(t *testing.T, svc *SnapshotServiceImpl, provider ebsprovider.EBSProvider, volumeID, state, instanceID string) {
-	t.Helper()
-	require.NoError(t, svc.metadata.PutVolume(context.Background(), ebsmetadata.Volume{
-		VolumeID: volumeID, TenantID: testAccountID, CapacityGiB: 4,
-		State: state, AttachedInstance: instanceID, AvailabilityZone: "us-east-1a",
-		ProviderHandle: "memory://volume/" + volumeID,
-	}))
 	_, err := provider.CreateVolume(context.Background(), ebsprovider.CreateVolumeRequest{
 		Versioned: ebsprovider.NewVersioned(), VolumeID: volumeID,
 		CapacityRange:    ebsprovider.CapacityRange{RequiredBytes: 4 * 1024 * 1024 * 1024},
 		AvailabilityZone: "us-east-1a",
 	})
 	require.NoError(t, err)
+	require.NoError(t, svc.metadata.PutVolume(context.Background(), ebsmetadata.Volume{
+		VolumeID: volumeID, TenantID: testAccountID, CapacityGiB: 4,
+		State: "in-use", AttachedInstance: attachedInstanceID, AvailabilityZone: "us-east-1a",
+	}))
+	return svc, provider, nc
 }
 
-// The provider branch used to skip the drain entirely, silently snapshotting a
-// stale checkpoint of an attached, actively-written volume. It must drain the
-// same way the legacy branch does.
-func TestCreateSnapshot_Provider_DrainsAttachedVolume(t *testing.T) {
-	svc, _, nc := setupDrainService(t)
-	provider := ebsprovider.NewMemoryProvider(ebsprovider.Capabilities{})
-	svc.SetEBSProvider(provider)
-	putProviderVolume(t, svc, provider, "vol-provider-drain", "in-use", drainInstanceID)
-
-	got := drainResponder(t, nc, drainInstanceID, drainedAck(t, "vol-provider-drain"))
+// The provider drains a buffered volume inside its own CreateSnapshot, so the
+// domain must not dial the drain socket itself: that would drain twice.
+func TestCreateSnapshot_AttachedVolumeOpensNoDrainSocket(t *testing.T) {
+	svc, provider, _ := setupAttachedSnapshot(t, "vol-attached-local")
+	drains := testutil.StartDrainSocket(t, svc.config.DataDir, "vol-attached-local", "OK\n")
 
 	_, err := svc.CreateSnapshot(context.Background(),
-		&ec2.CreateSnapshotInput{VolumeId: aws.String("vol-provider-drain")}, testAccountID)
+		&ec2.CreateSnapshotInput{VolumeId: aws.String("vol-attached-local")}, testAccountID)
 	require.NoError(t, err)
 
-	command := awaitDrainCommand(t, got)
-	assert.True(t, command.Attributes.DrainVolume)
-	assert.Equal(t, "vol-provider-drain", command.DrainVolumeData.VolumeID)
+	assert.Zero(t, drains(), "the domain must leave the drain to the provider")
+	assert.Equal(t, int32(1), provider.creates.Load())
 }
 
-// A provider-branch snapshot whose drain fails must fail closed rather than
-// return a snapshot of stale data: the provider must never be asked to
-// snapshot, and no snapshot metadata may be left behind.
-func TestCreateSnapshot_Provider_UndrainableAttachedVolumeFails(t *testing.T) {
-	svc, _, nc := setupDrainService(t)
-	provider := ebsprovider.NewMemoryProvider(ebsprovider.Capabilities{})
-	svc.SetEBSProvider(provider)
-	putProviderVolume(t, svc, provider, "vol-provider-nodrain", "in-use", drainInstanceID)
-
-	got := drainResponder(t, nc, drainInstanceID,
-		[]byte(`{"Code":"`+awserrors.ErrorServerInternal+`","Message":"drain failed"}`))
-
-	_, err := svc.CreateSnapshot(context.Background(),
-		&ec2.CreateSnapshotInput{VolumeId: aws.String("vol-provider-nodrain")}, testAccountID)
-	require.Error(t, err)
-	assert.Equal(t, awserrors.ErrorServerInternal, err.Error())
-	assert.True(t, awaitDrainCommand(t, got).Attributes.DrainVolume)
-
-	out, err := svc.DescribeSnapshots(context.Background(), &ec2.DescribeSnapshotsInput{}, testAccountID)
+// With no drain socket on this node, an attached volume is snapshotted by the
+// provider without asking the instance's host for a drain.
+func TestCreateSnapshot_AttachedVolumeSendsItsHostNoDrain(t *testing.T) {
+	svc, provider, nc := setupAttachedSnapshot(t, "vol-attached-remote")
+	var commands atomic.Int32
+	sub, err := nc.Subscribe(ec2v1.InstanceCommandSubject(attachedInstanceID), func(msg *nats.Msg) {
+		commands.Add(1)
+		_ = msg.Respond([]byte(`{"Code":"InternalError","Message":"no drain socket"}`))
+	})
 	require.NoError(t, err)
-	assert.Empty(t, out.Snapshots, "a failed drain must not leave a snapshot behind")
+	t.Cleanup(func() { _ = sub.Unsubscribe() })
+
+	snap, err := svc.CreateSnapshot(context.Background(),
+		&ec2.CreateSnapshotInput{VolumeId: aws.String("vol-attached-remote")}, testAccountID)
+	require.NoError(t, err)
+	assert.Equal(t, "vol-attached-remote", aws.StringValue(snap.VolumeId))
+	assert.Equal(t, int32(1), provider.creates.Load())
+	assert.Zero(t, commands.Load(), "the instance's host must not be asked to drain")
 }

spinifex/domains/ec2/snapshot/service_impl.go

@@ -5,10 +5,7 @@ import (
 	"errors"
 	"fmt"
 	"github.com/mulgadc/spinifex/spinifex/domains/ec2/tagmirror"
-	"github.com/mulgadc/spinifex/spinifex/foundation/messaging/nats"
 	"log/slog"
-	"net"
-	"path/filepath"
 	"strconv"
 	"strings"
 	"sync"
@@ -18,10 +15,8 @@ import (
 	"github.com/aws/aws-sdk-go/service/ec2"
 	"github.com/aws/aws-sdk-go/service/s3"
 	"github.com/mulgadc/bluebottle/pkg/safecast"
-	"github.com/mulgadc/spinifex/contracts/ec2/v1"
 	"github.com/mulgadc/spinifex/spinifex/bootstrap/config"
 	"github.com/mulgadc/spinifex/spinifex/domains/ec2/ebs/metadata"
-	"github.com/mulgadc/spinifex/spinifex/domains/ec2/volumestate"
 	"github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
 	awsfilters "github.com/mulgadc/spinifex/spinifex/foundation/aws/filters"
 	awsidentifiers "github.com/mulgadc/spinifex/spinifex/foundation/aws/identifiers"
@@ -209,19 +204,12 @@ func (s *SnapshotServiceImpl) CreateSnapshot(ctx context.Context, input *ec2.Cre
 		return nil, errors.New(awserrors.ErrorIncorrectState)
 	}
 
-	// Flush the writes buffered by whichever node serves the volume, so the
-	// provider reads a current checkpoint. An attached volume that cannot be
-	// drained fails here rather than silently snapshotting stale data.
-	if err := s.drainVolume(ctx, volumeID, volume.State, volume.AttachedInstance, accountID); err != nil {
-		slog.ErrorContext(ctx, "CreateSnapshot: drain failed", "volumeId", volumeID, "snapshotId", snapshotID, "err", err)
-		return nil, errors.New(awserrors.ErrorServerInternal)
-	}
-
 	created, err := s.provider.CreateSnapshot(ctx, ebsprovider.CreateSnapshotRequest{
 		Versioned: ebsprovider.NewVersioned(), SnapshotID: snapshotID, VolumeID: volumeID, VolumeHandle: volume.ProviderHandle,
 	})
 	if err != nil || created == nil {
-		return nil, errors.New(awserrors.ErrorServerInternal)
+		slog.ErrorContext(ctx, "CreateSnapshot: provider CreateSnapshot failed", "volumeId", volumeID, "snapshotId", snapshotID, "err", err)
+		return nil, ProviderCreateError(volumeID, err)
 	}
 	now := time.Now()
 	snapshotCfg := ebsmetadata.Snapshot{
@@ -259,177 +247,6 @@ func (s *SnapshotServiceImpl) CreateSnapshot(ctx context.Context, input *ec2.Cre
 	return snapshotConfigToEC2(snapshotCfg), nil
 }
 
-const (
-	// drainDialTimeout bounds the connect to the local drain socket. The socket
-	// is either being served on this node or it is not, so a slow connect means
-	// a wedged plugin rather than a busy one.
-	drainDialTimeout = time.Second
-
-	// drainAckTimeout bounds DrainToBackend on the hosting node: it flushes the
-	// WAL and the live checkpoint to S3, so it scales with the dirty set rather
-	// than with a fixed unit of work.
-	drainAckTimeout = 30 * time.Second
-
-	// drainRequestTimeout bounds the ec2.cmd round-trip. It exceeds
-	// drainAckTimeout so a slow drain surfaces as the hosting node's error
-	// instead of an opaque NATS timeout here.
-	drainRequestTimeout = 35 * time.Second
-)
-
-// ErrNoDrainSocket reports that this node has no drain socket for the volume,
-// which is how a node learns it does not serve it. It is deliberately distinct
-// from a socket that answered badly: only the former is worth routing onward,
-// and conflating the two makes a node re-run a drain that has already failed.
-var ErrNoDrainSocket = errors.New("no drain socket on this node")
-
-// DrainVolumeSocket asks the NBD plugin serving volumeID on this node to flush
-// its WAL chunks and live checkpoint to S3, then waits for the ack. The plugin
-// creates the socket under {dataDir}/viperblock/{volumeID}/, so this only
-// succeeds on the node hosting the volume; the daemon calls it when a drain
-// command is routed to it.
-func DrainVolumeSocket(dataDir, volumeID string) error {
-	sockPath := filepath.Join(dataDir, "viperblock", volumeID, "snapshot.sock")
-	conn, err := net.DialTimeout("unix", sockPath, drainDialTimeout)
-	if err != nil {
-		return fmt.Errorf("%w: dial %s: %w", ErrNoDrainSocket, sockPath, err)
-	}
-	defer conn.Close()
-
-	_ = conn.SetDeadline(time.Now().Add(drainAckTimeout))
-	buf := make([]byte, 8)
-	n, err := conn.Read(buf)
-	if err != nil {
-		return fmt.Errorf("read drain ack for %s: %w", volumeID, err)
-	}
-	if !strings.HasPrefix(string(buf[:n]), "OK") {
-		return fmt.Errorf("drain of %s did not ack OK: %q", volumeID, string(buf[:n]))
-	}
-	return nil
-}
-
-// drainVolume flushes the volume's in-flight writes to S3 before the snapshot
-// reads the live checkpoint from there.
-func (s *SnapshotServiceImpl) drainVolume(ctx context.Context, volumeID string, fallbackState, fallbackInstance string, accountID string) error {
-	return DrainVolume(ctx, s.config, s.store, s.natsConn, volumeID, fallbackState, fallbackInstance, accountID)
-}
-
-// DrainVolume flushes the volume's in-flight writes to S3 before a live
-// checkpoint is read from there. Shared by ec2.CreateSnapshot and
-// ec2.CreateImage's running-instance path (domains/ec2/image) so the two
-// live-snapshot call sites cannot diverge on when a drain is required.
-//
-// The drain socket lives on the node serving the volume, but both callers are
-// queue-grouped across every node, so the node answering the request usually
-// is not that node. Whether a drain is required is therefore decided from the
-// volume's attachment record, never from whether the socket happens to be
-// local: an attached volume is being written to right now, and reading its
-// checkpoint without draining silently captures an older one.
-func DrainVolume(ctx context.Context, cfg *config.Config, store objectstore.ObjectStore, natsConn *nats.Conn, volumeID string, fallbackState, fallbackInstance string, accountID string) error {
-	// A metadata-only snapshot never reads the live checkpoint, so there is
-	// nothing for a drain to make current.
-	if cfg == nil || cfg.Predastore.Host == "" {
-		return nil
-	}
-
-	state, instanceID, err := volumeAttachment(ctx, store, cfg.Predastore.Bucket, volumeID, fallbackState, fallbackInstance)
-	if err != nil {
-		return err
-	}
-
-	// Nothing is writing to an unattached volume: the checkpoint its Close()
-	// left behind is the current one. Decide this before touching the socket —
-	// dialing it is not a probe, it makes the plugin run a full flush.
-	if state != "in-use" || instanceID == "" {
-		slog.InfoContext(ctx, "DrainVolume: volume not attached, snapshotting the checkpoint left by Close (stopped instance path)",
-			"volumeId", volumeID, "state", state, "attachedInstance", instanceID)
-		return nil
-	}
-
-	// Fast path: the volume is served by this node, so no hop is needed. This is
-	// every single-node deployment, and the node hosting the instance on a
-	// cluster.
-	localErr := DrainVolumeSocket(cfg.DataDir, volumeID)
-	if localErr == nil {
-		return nil
-	}
-
-	// A socket that answered but did not ack means this node does serve the
-	// volume and the flush itself failed. Routing that onward would land back on
-	// this same node and repeat the drain that has already failed.
-	if !errors.Is(localErr, ErrNoDrainSocket) {
-		return fmt.Errorf("drain volume %s on this node: %w", volumeID, localErr)
-	}
-
-	slog.InfoContext(ctx, "DrainVolume: volume attached, routing drain to the node hosting it",
-		"volumeId", volumeID, "instanceId", instanceID)
-	return drainOnHostNode(ctx, natsConn, volumeID, instanceID, accountID)
-}
-
-// volumeAttachment returns the volume's authoritative state and attached
-// instance. state.json is control-plane-owned and authoritative; the fallback
-// pair (passed in from the caller's own volume record) is rewritten by the
-// live NBD plugin from its stale in-memory state and is only used for volumes
-// predating the split.
-func volumeAttachment(ctx context.Context, store objectstore.ObjectStore, bucket, volumeID string, fallbackState, fallbackInstance string) (state, instanceID string, err error) {
-	rec, found, err := volumestate.Read(ctx, store, bucket, volumeID)
-	if err != nil {
-		return "", "", fmt.Errorf("read volume state for %s: %w", volumeID, err)
-	}
-	if found {
-		return rec.State, rec.AttachedInstance, nil
-	}
-	return fallbackState, fallbackInstance, nil
-}
-
-// drainOnHostNode issues the drain on the node hosting instanceID. Only that
-// node subscribes ec2.cmd.{instanceID}, so the command is self-routing and no
-// volume-to-node resolution is needed here.
-//
-// An attachment record is not proof of a live writer: stop deliberately leaves
-// a boot volume attached (runtime/daemon/vm_adapters.go's volumeMounterAdapter.Unmount)
-// while tearing down both the NBD plugin and this subscription. The two
-// not-running signals below are therefore the stopped-instance path, not a
-// failure — treating them as one would make every stopped instance's root
-// volume permanently unsnapshottable.
-func drainOnHostNode(ctx context.Context, natsConn *nats.Conn, volumeID, instanceID, accountID string) error {
-	command := ec2v1.EC2InstanceCommand{
-		ID:              instanceID,
-		Attributes:      ec2v1.EC2CommandAttributes{DrainVolume: true},
-		DrainVolumeData: &ec2v1.DrainVolumeData{VolumeID: volumeID},
-	}
-
-	resp, err := natsmsg.NATSRequest[ec2v1.DrainVolumeResponse](ctx, natsConn,
-		ec2v1.InstanceCommandSubject(instanceID), command, drainRequestTimeout, accountID)
-	if err != nil {
-		// No subscriber at all: the instance runs nowhere in the cluster, so it
-		// was stopped (stop migrates it to shared KV before unsubscribing) and
-		// its volumes were sealed by Close. Warn rather than Info because the
-		// same shape appears if the hosting node has lost NATS while its VM
-		// keeps writing, which yields a stale snapshot as it did before routing
-		// existed. A live host that answers and fails is still a hard error.
-		if errors.Is(err, nats.ErrNoResponders) {
-			slog.WarnContext(ctx, "drainVolume: no node hosts the instance, treating the volume as stopped and snapshotting its sealed checkpoint",
-				"volumeId", volumeID, "instanceId", instanceID)
-			return nil
-		}
-		return fmt.Errorf("drain volume %s on the node hosting %s: %w", volumeID, instanceID, err)
-	}
-
-	switch resp.Status {
-	case ec2v1.DrainVolumeStatusDrained:
-		return nil
-	// The host still holds the instance but it is not running: nothing is
-	// writing, so there is nothing to flush. This is a host-drain stop, which
-	// keeps the VM (and this subscription) in place after unmounting.
-	case ec2v1.DrainVolumeStatusNotRunning:
-		slog.InfoContext(ctx, "drainVolume: instance is not running on its host, snapshotting the volume's sealed checkpoint",
-			"volumeId", volumeID, "instanceId", instanceID)
-		return nil
-	default:
-		return fmt.Errorf("drain volume %s on the node hosting %s: unexpected ack %q", volumeID, instanceID, resp.Status)
-	}
-}
-
 // describeSnapshotsValidFilters defines the set of filter names accepted by DescribeSnapshots.
 var describeSnapshotsValidFilters = map[string]bool{
 	"snapshot-id": true,
@@ -740,6 +557,21 @@ func unreadableRecordError(ctx context.Context, snapshotID, accountID, kind stri
 		"Cannot confirm that snapshot '%s' is not in use: %d %s record(s) could not be read", snapshotID, len(unreadable), kind)
 }
 
+// ProviderCreateError maps a provider CreateSnapshot failure of volumeID to
+// the code a client can act on: retry-later for a provider that cannot answer
+// now, IncorrectState for one that needs the volume detached, else internal.
+func ProviderCreateError(volumeID string, err error) error {
+	switch {
+	case errors.Is(err, ebsprovider.ErrUnavailable):
+		return errors.New(awserrors.ErrorServiceUnavailable)
+	case errors.Is(err, ebsprovider.ErrVolumeInUse):
+		return awserrors.Errorf(awserrors.ErrorIncorrectState,
+			"The volume '%s' is in use. Detach it before creating a snapshot.", volumeID)
+	default:
+		return errors.New(awserrors.ErrorServerInternal)
+	}
+}
+
 // providerDeleteError maps a provider DeleteSnapshot failure to the code a
 // client can act on. A transport failure is retryable: the delete is idempotent.
 func providerDeleteError(snapshotID string, err error) error {

spinifex/domains/ec2/snapshot/service_impl_test.go

@@ -1519,3 +1519,55 @@ func TestDropEmptySnapshotRefsDoesNotDropAConcurrentAdd(t *testing.T) {
 		assert.True(t, has, "iteration %d: the concurrent add was dropped by the empty-key delete", i)
 	}
 }
+
+// createFailingProvider fails only CreateSnapshot, with err.
+type createFailingProvider struct {
+	ebsprovider.EBSProvider
+
+	err error
+}
+
+func (p *createFailingProvider) CreateSnapshot(context.Context, ebsprovider.CreateSnapshotRequest) (*ebsprovider.Snapshot, error) {
+	return nil, p.err
+}
+
+// A provider that cannot snapshot a volume while it is attached refuses it
+// as in use, which the client can act on by detaching the volume first.
+func TestCreateSnapshot_ProviderVolumeInUseIsIncorrectState(t *testing.T) {
+	svc, store := setupTestSnapshotService(t)
+	createTestVolume(t, svc, store, "vol-create-inuse", 10)
+	svc.SetEBSProvider(&createFailingProvider{EBSProvider: svc.provider,
+		err: &ebsprovider.ProviderError{Code: ebsprovider.ErrorCodeVolumeInUse, Message: "published"}})
+
+	_, err := svc.CreateSnapshot(context.Background(), &ec2.CreateSnapshotInput{VolumeId: aws.String("vol-create-inuse")}, testAccountID)
+
+	code, message := clientErrorDetail(t, err)
+	assert.Equal(t, awserrors.ErrorIncorrectState, code)
+	assert.Contains(t, message, "vol-create-inuse")
+	assert.Contains(t, message, "Detach")
+}
+
+// A provider that cannot answer now, such as one refusing a volume another
+// node holds, is retry-later for the client; any other failure is internal.
+func TestCreateSnapshot_ProviderFailureMapsToSpecificError(t *testing.T) {
+	cases := []struct {
+		name     string
+		err      error
+		wantCode string
+	}{
+		{"provider unavailable", &ebsprovider.ProviderError{Code: ebsprovider.ErrorCodeUnavailable, Message: "held by node-2"}, awserrors.ErrorServiceUnavailable},
+		{"internal failure", &ebsprovider.ProviderError{Code: ebsprovider.ErrorCodeInternal, Message: "boom"}, awserrors.ErrorServerInternal},
+		{"transport failure", errors.New("connection reset"), awserrors.ErrorServerInternal},
+	}
+	for _, tc := range cases {
+		t.Run(tc.name, func(t *testing.T) {
+			svc, store := setupTestSnapshotService(t)
+			createTestVolume(t, svc, store, "vol-create-fail", 10)
+			svc.SetEBSProvider(&createFailingProvider{EBSProvider: svc.provider, err: tc.err})
+
+			_, err := svc.CreateSnapshot(context.Background(), &ec2.CreateSnapshotInput{VolumeId: aws.String("vol-create-fail")}, testAccountID)
+			require.Error(t, err)
+			assert.Equal(t, tc.wantCode, err.Error())
+		})
+	}
+}

spinifex/domains/ec2/volumestate/volumestate.go

@@ -1,78 +0,0 @@
-// Package volumestate reads and writes the control-plane-owned attachment
-// state for an EBS volume, persisted to a per-volume state.json object.
-//
-// It is kept out of config.json because config.json is rewritten by the live
-// nbdkit VB on every SaveState (clobbering any State the control plane wrote
-// there) and is a sealed object for encrypted volumes (a second writer reuses
-// the AES-GCM nonce). state.json is plaintext, viperblock never touches it, so
-// the control plane is its single writer. Readers outside the volume service
-// (the snapshot path) need the same record, which is why it lives here.
-package volumestate
-
-import (
-	"bytes"
-	"context"
-	"encoding/json"
-	"fmt"
-	"io"
-	"time"
-
-	"github.com/aws/aws-sdk-go/aws"
-	"github.com/aws/aws-sdk-go/service/s3"
-	"github.com/mulgadc/spinifex/spinifex/providers/objectstore"
-)
-
-// Record is the control-plane-owned attachment state of a volume.
-type Record struct {
-	State            string    `json:"state"`
-	AttachedInstance string    `json:"attachedInstance"`
-	DeviceName       string    `json:"deviceName"`
-	AttachedAt       time.Time `json:"attachedAt"`
-}
-
-// Key is the S3 key for a volume's control-plane state object.
-func Key(volumeID string) string { return volumeID + "/state.json" }
-
-// Write persists the attachment state, replacing any previous record.
-func Write(ctx context.Context, store objectstore.ObjectStore, bucket, volumeID string, rec Record) error {
-	data, err := json.Marshal(rec)
-	if err != nil {
-		return fmt.Errorf("marshal volume state: %w", err)
-	}
-	_, err = store.PutObject(ctx, &s3.PutObjectInput{
-		Bucket: aws.String(bucket),
-		Key:    aws.String(Key(volumeID)),
-		Body:   bytes.NewReader(data),
-	})
-	if err != nil {
-		return fmt.Errorf("failed to write volume state to S3: %w", err)
-	}
-	return nil
-}
-
-// Read returns the attachment state. found=false with a nil error means the
-// object is absent (a volume predating the state.json split), in which case the
-// caller falls back to the State embedded in config.json.
-func Read(ctx context.Context, store objectstore.ObjectStore, bucket, volumeID string) (Record, bool, error) {
-	getResult, err := store.GetObject(ctx, &s3.GetObjectInput{
-		Bucket: aws.String(bucket),
-		Key:    aws.String(Key(volumeID)),
-	})
-	if err != nil {
-		if objectstore.IsNoSuchKeyError(err) {
-			return Record{}, false, nil
-		}
-		return Record{}, false, fmt.Errorf("failed to get volume state: %w", err)
-	}
-	defer getResult.Body.Close()
-
-	body, err := io.ReadAll(getResult.Body)
-	if err != nil {
-		return Record{}, false, fmt.Errorf("failed to read volume state body: %w", err)
-	}
-	var rec Record
-	if err := json.Unmarshal(body, &rec); err != nil {
-		return Record{}, false, fmt.Errorf("failed to unmarshal volume state: %w", err)
-	}
-	return rec, true, nil
-}

spinifex/domains/ec2/vpc/external_ipam.go

@@ -122,7 +122,7 @@ func (m *ExternalIPAM) AllocateIP(ctx context.Context, region, az, purpose, allo
 	if m == nil {
 		return "", "", fmt.Errorf("external IPAM unavailable on this node: %w", errors.New(awserrors.ErrorInsufficientAddressCapacity))
 	}
-	pool := m.findPool(region, az)
+	pool := external.FindPool(m.pools, region, az)
 	if pool == nil {
 		return "", "", fmt.Errorf("no external pool available for region=%q az=%q: %w", region, az, errors.New(awserrors.ErrorInsufficientAddressCapacity))
 	}
@@ -221,30 +221,6 @@ func (m *ExternalIPAM) allocatorFor(poolName string) (external.Allocator, error)
 	return m.static, nil
 }
 
-// findPool returns the best pool for the given region/AZ using the same
-// fallback order as topology.go: AZ-scoped → region-scoped → unscoped.
-func (m *ExternalIPAM) findPool(region, az string) *external.ExternalPoolConfig {
-	for i := range m.pools {
-		p := &m.pools[i]
-		if p.AZ != "" && p.AZ == az && p.Region == region {
-			return p
-		}
-	}
-	for i := range m.pools {
-		p := &m.pools[i]
-		if p.AZ == "" && p.Region != "" && p.Region == region {
-			return p
-		}
-	}
-	for i := range m.pools {
-		p := &m.pools[i]
-		if p.Region == "" && p.AZ == "" {
-			return p
-		}
-	}
-	return nil
-}
-
 // filterStatic returns only the static pools (Source unset or "static").
 func filterStatic(pools []external.ExternalPoolConfig) []external.ExternalPoolConfig {
 	out := make([]external.ExternalPoolConfig, 0, len(pools))

spinifex/domains/ecr/auth/rotator_test.go

@@ -99,7 +99,7 @@ func TestRotateOnce_MintsAndSwaps(t *testing.T) {
 	r.rotateOnce(t.Context(), time.Now().UTC())
 
 	// A new active key took over.
-	assert.NotEqual(t, key.Kid, issuer.ActiveKid(), "issuer should sign with the rotated key")
+	assert.NotEqual(t, key.Kid, activeKid(issuer), "issuer should sign with the rotated key")
 
 	// Both old and new tokens verify (previous key retained).
 	_, err = verifier.Verify(oldTok)
@@ -142,7 +142,7 @@ func TestRotateOnce_PrunesExpiredKey(t *testing.T) {
 	require.NoError(t, err)
 	require.Len(t, metas, 1, "the older key should be pruned")
 	assert.Equal(t, second.Kid, active.Kid, "newest key remains active")
-	assert.Equal(t, second.Kid, issuer.ActiveKid())
+	assert.Equal(t, second.Kid, activeKid(issuer))
 }
 
 // TestRotator_RunRotatesThenStops drives the scheduler loop on a fast interval
@@ -169,7 +169,7 @@ func TestRotator_RunRotatesThenStops(t *testing.T) {
 	stopped := make(chan struct{})
 	go func() { r.Run(ctx); close(stopped) }()
 
-	require.Eventually(t, func() bool { return issuer.ActiveKid() != key.Kid },
+	require.Eventually(t, func() bool { return activeKid(issuer) != key.Kid },
 		2*time.Second, 5*time.Millisecond, "rotator should mint a new active key")
 
 	cancel()
@@ -218,3 +218,10 @@ func TestIssuerVerifier_ConcurrentSwap(t *testing.T) {
 	}
 	<-done
 }
+
+// activeKid reads the kid the issuer currently signs with.
+func activeKid(i *Issuer) string {
+	i.mu.RLock()
+	defer i.mu.RUnlock()
+	return i.key.Kid
+}

spinifex/domains/ecr/auth/token.go

@@ -91,16 +91,6 @@ func (i *Issuer) SetActiveKey(key *SigningKey) {
 	i.mu.Unlock()
 }
 
-// ActiveKid returns the kid of the current signing key, or "" if none.
-func (i *Issuer) ActiveKid() string {
-	i.mu.RLock()
-	defer i.mu.RUnlock()
-	if i.key == nil {
-		return ""
-	}
-	return i.key.Kid
-}
-
 // Mint returns a signed ES256 JWT for p and its expiry. The kid header lets any
 // verifier resolve the public key from the shared signing-key set.
 func (i *Issuer) Mint(p Principal) (token string, expiresAt time.Time, err error) {

spinifex/domains/ecr/awsapi/registration.go

@@ -2,14 +2,12 @@ package awsapi
 
 import (
 	"errors"
-	"io"
 	"log/slog"
 	"maps"
 	"net/http"
 	"slices"
 	"strings"
 
-	"github.com/mulgadc/bluebottle/pkg/sigv4"
 	ecrauth "github.com/mulgadc/spinifex/spinifex/domains/ecr/auth"
 	"github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
 	"github.com/mulgadc/spinifex/spinifex/ingress/aws/dispatch"
@@ -34,7 +32,7 @@ type Deps struct {
 func NewRegistration(deps Deps) dispatch.Registration {
 	return dispatch.Registration{
 		Service:       ServiceName,
-		ResolveAction: resolveAction,
+		ResolveAction: dispatch.ResolveTargetAction,
 		Dispatch:      deps.dispatch,
 		Errors:        dispatch.ErrorEnvelopeJSON,
 		Inventory:     OperationInventory(),
@@ -55,25 +53,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)
 	}
@@ -92,7 +76,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.Error("ECR dispatch: failed to read body", "err", err)
 		return err
@@ -178,19 +162,6 @@ func roleIDOf(actor dispatch.Actor) string {
 	return roleID
 }
 
-// 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
-}
-
 func without(values, excluded []string) []string {
 	return slices.DeleteFunc(slices.Clone(values), func(value string) bool {
 		return slices.Contains(excluded, value)

spinifex/domains/ecr/awsapi/registration_test.go

@@ -172,18 +172,6 @@ var composedCases = []struct{ action, body string }{
 	{"DeleteRepository", `{"repositoryName":"app"}`},
 }
 
-func TestActionFromTarget(t *testing.T) {
-	for target, want := range map[string]string{
-		TargetPrefix + ".CreateRepository": "CreateRepository",
-		"a.b.ListRepositories":             "ListRepositories",
-		"GetAuthorizationToken":            "GetAuthorizationToken",
-		TargetPrefix + ".":                 "",
-		"":                                 "",
-	} {
-		assert.Equal(t, want, actionFromTarget(target), "target %q", target)
-	}
-}
-
 // Every composed action must be in the Actions namespace, or dispatch rejects
 // it as InvalidAction before its resources are ever scoped.
 func TestComposedActionsAreInTheNamespace(t *testing.T) {

spinifex/domains/ecr/storage.go

@@ -28,42 +28,22 @@ func AccountBucket(accountID string) string {
 // shapes every ECR storage path must use. Blobs are pooled per account for
 // dedup; manifests and tags are indexed per repository.
 const (
-	// repoMetaKeyFmt is the repo config object (scan-on-push, tag mutability).
-	repoMetaKeyFmt = "repos/%s/_meta.json"
-	// tagKeyFmt maps a tag to a digest pointer (body is "sha256:...").
-	tagKeyFmt = "repos/%s/manifests/by-tag/%s"
 	// manifestKeyFmt holds manifest JSON keyed by its own digest.
 	manifestKeyFmt = "repos/%s/manifests/by-digest/%s"
-	// indexKeyFmt is the rolling tag->digest index used by ListImages.
-	indexKeyFmt = "repos/%s/manifests/index.json"
 	// blobKeyFmt is the account-wide blob pool, two-char hex shard.
 	blobKeyFmt = "blobs/%s/%s"
-	// uploadKeyFmt holds in-progress chunked upload bytes.
-	uploadKeyFmt = "uploads/%s"
 	// uploadChunkKeyFmt holds a single revision's assembled bytes under a unique
 	// per-write token, so a CAS-winning PATCH always points at exactly the bytes
 	// it hashed (a losing concurrent write lands on a different, orphaned token).
 	uploadChunkKeyFmt = "uploads/%s/%s"
 )
 
-// RepoMetaKey returns the key for a repository's config object.
-func RepoMetaKey(repo string) string { return fmt.Sprintf(repoMetaKeyFmt, repo) }
-
-// TagKey returns the key for a tag's digest pointer.
-func TagKey(repo, tag string) string { return fmt.Sprintf(tagKeyFmt, repo, tag) }
-
 // ManifestKey returns the key for a manifest stored by its digest. The digest
 // is sanitized via DigestToken so the object key carries no ':' separator.
 func ManifestKey(repo, digest string) string {
 	return fmt.Sprintf(manifestKeyFmt, repo, DigestToken(digest))
 }
 
-// IndexKey returns the key for a repository's tag->digest index.
-func IndexKey(repo string) string { return fmt.Sprintf(indexKeyFmt, repo) }
-
-// UploadKey returns the key for an in-progress upload's bytes.
-func UploadKey(uploadID string) string { return fmt.Sprintf(uploadKeyFmt, uploadID) }
-
 // UploadChunkKey returns the key for one revision's assembled upload bytes,
 // addressed by a unique per-write token.
 func UploadChunkKey(uploadID, token string) string {

spinifex/domains/ecr/storage_test.go

@@ -17,11 +17,7 @@ func TestKeyHelpers(t *testing.T) {
 		got  string
 		want string
 	}{
-		{"repo meta", RepoMetaKey("team/app"), "repos/team/app/_meta.json"},
-		{"tag", TagKey("team/app", "v1"), "repos/team/app/manifests/by-tag/v1"},
 		{"manifest", ManifestKey("team/app", "sha256:abcd"), "repos/team/app/manifests/by-digest/sha256-abcd"},
-		{"index", IndexKey("team/app"), "repos/team/app/manifests/index.json"},
-		{"upload", UploadKey("u-1"), "uploads/u-1"},
 	}
 	for _, c := range cases {
 		if c.got != c.want {

spinifex/domains/ecs/awsapi/registration.go

@@ -2,14 +2,11 @@ package awsapi
 
 import (
 	"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/foundation/aws/passrole"
 	"github.com/mulgadc/spinifex/spinifex/ingress/aws/dispatch"
@@ -35,7 +32,7 @@ type Deps struct {
 func NewRegistration(deps Deps) dispatch.Registration {
 	return dispatch.Registration{
 		Service:       ServiceName,
-		ResolveAction: resolveAction,
+		ResolveAction: dispatch.ResolveTargetAction,
 		Dispatch:      deps.dispatch,
 		Errors:        dispatch.ErrorEnvelopeJSON,
 		Inventory:     OperationInventory(),
@@ -50,35 +47,11 @@ func OperationInventory() dispatch.Inventory {
 	}
 }
 
-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
-}
-
-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
-}
-
 // dispatch resolves, scopes and authorizes the request, 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)
 	}
@@ -99,7 +72,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(), "ECS dispatch: failed to read body", "err", err)
 		return err

spinifex/domains/ecs/awsapi/registration_target_test.go

@@ -1,13 +0,0 @@
-package awsapi
-
-import (
-	"testing"
-
-	"github.com/stretchr/testify/assert"
-)
-
-func TestActionFromTarget(t *testing.T) {
-	assert.Equal(t, "RunTask", actionFromTarget("AmazonEC2ContainerServiceV20141113.RunTask"))
-	assert.Equal(t, "CreateCluster", actionFromTarget("CreateCluster"))
-	assert.Empty(t, actionFromTarget(""))
-}

spinifex/domains/ecs/capacity_providers.go

@@ -3,6 +3,7 @@ package ecs
 import (
 	"context"
 	"errors"
+	"github.com/mulgadc/spinifex/spinifex/domains/ecs/taskdefinition"
 	"time"
 
 	"github.com/aws/aws-sdk-go/aws"
@@ -23,7 +24,7 @@ func (s *Service) PutClusterCapacityProviders(ctx context.Context, input *ecs.Pu
 		return nil, err
 	}
 	var rec ClusterRecord
-	found, err := getJSON(ctx, kv, ClusterMetaKey(cluster), &rec)
+	found, err := taskdefinition.GetJSON(ctx, kv, ClusterMetaKey(cluster), &rec)
 	if err != nil {
 		return nil, err
 	}
@@ -33,7 +34,7 @@ func (s *Service) PutClusterCapacityProviders(ctx context.Context, input *ecs.Pu
 
 	rec.CapacityProviders = awsStringSlice(input.CapacityProviders)
 	rec.DefaultCapacityProviderStrategy = strategyFromAWS(input.DefaultCapacityProviderStrategy)
-	if err := putJSON(ctx, kv, ClusterMetaKey(cluster), &rec); err != nil {
+	if err := taskdefinition.PutJSON(ctx, kv, ClusterMetaKey(cluster), &rec); err != nil {
 		return nil, err
 	}
 	return &ecs.PutClusterCapacityProvidersOutput{Cluster: rec.toAWS()}, nil
@@ -53,7 +54,7 @@ func (s *Service) CreateCapacityProvider(ctx context.Context, input *ecs.CreateC
 	}
 
 	var existing CapacityProviderRecord
-	found, err := getJSON(ctx, kv, CapacityProviderKey(name), &existing)
+	found, err := taskdefinition.GetJSON(ctx, kv, CapacityProviderKey(name), &existing)
 	if err != nil {
 		return nil, err
 	}
@@ -69,7 +70,7 @@ func (s *Service) CreateCapacityProvider(ctx context.Context, input *ecs.CreateC
 		Tags:                     tagsToMap(input.Tags),
 		CreatedAt:                time.Now().UTC(),
 	}
-	if err := putJSON(ctx, kv, CapacityProviderKey(name), &rec); err != nil {
+	if err := taskdefinition.PutJSON(ctx, kv, CapacityProviderKey(name), &rec); err != nil {
 		return nil, err
 	}
 	return &ecs.CreateCapacityProviderOutput{CapacityProvider: rec.toAWS()}, nil
@@ -86,13 +87,13 @@ func (s *Service) DescribeCapacityProviders(ctx context.Context, input *ecs.Desc
 
 	names := awsStringSlice(input.CapacityProviders)
 	if len(names) == 0 {
-		keys, kerr := keysWithPrefix(ctx, kv, CapacityProvidersPrefix())
+		keys, kerr := taskdefinition.KeysWithPrefix(ctx, kv, CapacityProvidersPrefix())
 		if kerr != nil {
 			return nil, kerr
 		}
 		for _, k := range keys {
 			var rec CapacityProviderRecord
-			ok, gerr := getJSON(ctx, kv, k, &rec)
+			ok, gerr := taskdefinition.GetJSON(ctx, kv, k, &rec)
 			if gerr != nil {
 				return nil, gerr
 			}
@@ -106,7 +107,7 @@ func (s *Service) DescribeCapacityProviders(ctx context.Context, input *ecs.Desc
 	for _, ref := range names {
 		name := CapacityProviderShortName(ref)
 		var rec CapacityProviderRecord
-		ok, gerr := getJSON(ctx, kv, CapacityProviderKey(name), &rec)
+		ok, gerr := taskdefinition.GetJSON(ctx, kv, CapacityProviderKey(name), &rec)
 		if gerr != nil {
 			return nil, gerr
 		}
@@ -132,7 +133,7 @@ func (s *Service) DeleteCapacityProvider(ctx context.Context, input *ecs.DeleteC
 		return nil, err
 	}
 	var rec CapacityProviderRecord
-	found, err := getJSON(ctx, kv, CapacityProviderKey(name), &rec)
+	found, err := taskdefinition.GetJSON(ctx, kv, CapacityProviderKey(name), &rec)
 	if err != nil {
 		return nil, err
 	}

spinifex/domains/ecs/cluster_teardown.go

@@ -3,6 +3,7 @@ package ecs
 import (
 	"context"
 	"errors"
+	"github.com/mulgadc/spinifex/spinifex/domains/ecs/taskdefinition"
 	"strings"
 
 	"github.com/aws/aws-sdk-go/aws"
@@ -34,7 +35,7 @@ func (s *Service) DeleteCluster(ctx context.Context, input *ecs.DeleteClusterInp
 		return nil, err
 	}
 	var rec ClusterRecord
-	found, err := getJSON(ctx, kv, ClusterMetaKey(cluster), &rec)
+	found, err := taskdefinition.GetJSON(ctx, kv, ClusterMetaKey(cluster), &rec)
 	if err != nil {
 		return nil, err
 	}
@@ -71,7 +72,7 @@ func (s *Service) DeregisterContainerInstance(ctx context.Context, input *ecs.De
 		return nil, err
 	}
 	var rec InstanceRecord
-	found, err := getJSON(ctx, kv, InstanceKey(cluster, id), &rec)
+	found, err := taskdefinition.GetJSON(ctx, kv, InstanceKey(cluster, id), &rec)
 	if err != nil {
 		return nil, err
 	}
@@ -124,7 +125,7 @@ func (s *Service) UpdateContainerInstancesState(ctx context.Context, input *ecs.
 	for _, ref := range awsStringSlice(input.ContainerInstances) {
 		id := ContainerInstanceShortID(ref)
 		var rec InstanceRecord
-		found, gerr := getJSON(ctx, kv, InstanceKey(cluster, id), &rec)
+		found, gerr := taskdefinition.GetJSON(ctx, kv, InstanceKey(cluster, id), &rec)
 		if gerr != nil {
 			return nil, gerr
 		}
@@ -134,7 +135,7 @@ func (s *Service) UpdateContainerInstancesState(ctx context.Context, input *ecs.
 		}
 		rec.Status = status
 		rec.Reaped = false
-		if perr := putJSON(ctx, kv, InstanceKey(cluster, id), &rec); perr != nil {
+		if perr := taskdefinition.PutJSON(ctx, kv, InstanceKey(cluster, id), &rec); perr != nil {
 			return nil, perr
 		}
 		if status == InstanceStatusDraining {

spinifex/domains/ecs/cluster_teardown_test.go

@@ -28,7 +28,7 @@ func TestService_DeleteCluster_CascadesAndSweeps(t *testing.T) {
 	list, err := svc.ListClusters(context.Background(), &ecs.ListClustersInput{}, testAccountID)
 	require.NoError(t, err)
 	assert.Empty(t, list.ClusterArns)
-	keys, err := keysWithPrefix(t.Context(), kv, clusterKeyPrefix("web"))
+	keys, err := taskdefinition.KeysWithPrefix(t.Context(), kv, clusterKeyPrefix("web"))
 	require.NoError(t, err)
 	assert.Empty(t, keys)
 }

spinifex/domains/ecs/deployments_test.go

@@ -19,7 +19,7 @@ import (
 func reloadService(t *testing.T, kv jetstream.KeyValue, cluster, name string) *ServiceRecord {
 	t.Helper()
 	var rec ServiceRecord
-	found, err := getJSON(t.Context(), kv, ServiceKey(cluster, name), &rec)
+	found, err := taskdefinition.GetJSON(t.Context(), kv, ServiceKey(cluster, name), &rec)
 	require.NoError(t, err)
 	require.True(t, found)
 	return &rec
@@ -329,7 +329,7 @@ func TestDeployment_LegacyServiceSynthesizesPrimary(t *testing.T) {
 		DesiredCount: 1, Status: ServiceStatusActive,
 		SchedulingStrategy: SchedulingStrategyReplica, DeploymentID: "legacy-dep",
 	}
-	require.NoError(t, putJSON(t.Context(), kv, ServiceKey("web", "legacy"), rec))
+	require.NoError(t, taskdefinition.PutJSON(t.Context(), kv, ServiceKey("web", "legacy"), rec))
 
 	require.NoError(t, svc.reconcileService(context.Background(), kv, testAccountID, rec))
 	reloaded := reloadService(t, kv, "web", "legacy")
@@ -388,7 +388,7 @@ func TestDeployment_RepeatedStartFailuresBackOffWithoutTheCircuitBreaker(t *test
 	// Once it passes, launching resumes: the backoff is a brake, not a stop.
 	rec = reloadService(t, kv, "web", "web")
 	rec.primaryDeployment().NextLaunchAt = time.Now().UTC().Add(-time.Second)
-	require.NoError(t, putJSON(t.Context(), kv, ServiceKey("web", "web"), rec))
+	require.NoError(t, taskdefinition.PutJSON(t.Context(), kv, ServiceKey("web", "web"), rec))
 	require.NoError(t, svc.reconcileService(context.Background(), kv, testAccountID, reloadService(t, kv, "web", "web")))
 
 	assert.Positive(t, liveServiceTasks(t, svc, kv, "web", "web"), "backoff never released the deployment")

spinifex/domains/ecs/eip_test.go

@@ -2,6 +2,7 @@ package ecs
 
 import (
 	"context"
+	"github.com/mulgadc/spinifex/spinifex/domains/ecs/taskdefinition"
 	"testing"
 
 	"github.com/stretchr/testify/assert"
@@ -37,14 +38,14 @@ func TestAssignTaskPublicIP_EnabledAssociatesAndPersists(t *testing.T) {
 	eips := &stubEIPs{publicIP: "192.0.2.50", allocID: "eipalloc-1"}
 	svc.eips = eips
 
-	require.NoError(t, putJSON(t.Context(), kv, ServiceKey("web", "web"), &ServiceRecord{
+	require.NoError(t, taskdefinition.PutJSON(t.Context(), kv, ServiceKey("web", "web"), &ServiceRecord{
 		Name: "web", Cluster: "web", AssignPublicIP: "ENABLED",
 	}))
 	task := &TaskRecord{
 		TaskID: "t-1", Cluster: "web", Group: serviceTaskGroup("web"),
 		ENIID: "eni-1", ENIPrivateIP: "172.31.0.8",
 	}
-	require.NoError(t, putJSON(t.Context(), kv, TaskKey("web", "t-1"), task))
+	require.NoError(t, taskdefinition.PutJSON(t.Context(), kv, TaskKey("web", "t-1"), task))
 
 	svc.assignTaskPublicIP(context.Background(), kv, testAccountID, task)
 
@@ -53,7 +54,7 @@ func TestAssignTaskPublicIP_EnabledAssociatesAndPersists(t *testing.T) {
 	assert.Equal(t, "eipalloc-1", task.ENIEIPAllocationID)
 
 	var persisted TaskRecord
-	found, err := getJSON(t.Context(), kv, TaskKey("web", "t-1"), &persisted)
+	found, err := taskdefinition.GetJSON(t.Context(), kv, TaskKey("web", "t-1"), &persisted)
 	require.NoError(t, err)
 	require.True(t, found)
 	assert.Equal(t, "192.0.2.50", persisted.ENIPublicIP)
@@ -64,7 +65,7 @@ func TestAssignTaskPublicIP_DisabledNoOp(t *testing.T) {
 	eips := &stubEIPs{publicIP: "192.0.2.50", allocID: "eipalloc-1"}
 	svc.eips = eips
 
-	require.NoError(t, putJSON(t.Context(), kv, ServiceKey("web", "web"), &ServiceRecord{
+	require.NoError(t, taskdefinition.PutJSON(t.Context(), kv, ServiceKey("web", "web"), &ServiceRecord{
 		Name: "web", Cluster: "web", // AssignPublicIP unset
 	}))
 	task := &TaskRecord{

spinifex/domains/ecs/eni_test.go

@@ -189,14 +189,14 @@ func singleTaskRecord(t *testing.T, svc *Service, cluster string) (TaskRecord, b
 	t.Helper()
 	kv, err := svc.bucket(context.Background(), testAccountID)
 	require.NoError(t, err)
-	keys, err := keysWithPrefix(context.Background(), kv, TasksPrefix(cluster))
+	keys, err := taskdefinition.KeysWithPrefix(context.Background(), kv, TasksPrefix(cluster))
 	require.NoError(t, err)
 	require.LessOrEqual(t, len(keys), 1, "test expects at most one task record")
 	if len(keys) == 0 {
 		return TaskRecord{}, false
 	}
 	var rec TaskRecord
-	found, err := getJSON(context.Background(), kv, keys[0], &rec)
+	found, err := taskdefinition.GetJSON(context.Background(), kv, keys[0], &rec)
 	require.NoError(t, err)
 	return rec, found
 }
@@ -317,10 +317,10 @@ func TestReaper_Awsvpc_ReleasesENI(t *testing.T) {
 	kv, err := svc.bucket(t.Context(), testAccountID)
 	require.NoError(t, err)
 	var rec InstanceRecord
-	_, err = getJSON(t.Context(), kv, InstanceKey("web", "i-1"), &rec)
+	_, err = taskdefinition.GetJSON(t.Context(), kv, InstanceKey("web", "i-1"), &rec)
 	require.NoError(t, err)
 	rec.LastSeen = time.Now().UTC().Add(-2 * heartbeatTimeout)
-	require.NoError(t, putJSON(t.Context(), kv, InstanceKey("web", "i-1"), &rec))
+	require.NoError(t, taskdefinition.PutJSON(t.Context(), kv, InstanceKey("web", "i-1"), &rec))
 
 	sc := NewScheduler(nc, svc, "test-holder")
 	sc.reapBucket(context.Background(), kv, testAccountID, time.Now().UTC())

spinifex/domains/ecs/export_test.go

@@ -15,8 +15,6 @@ const TestAccountID = testAccountID
 var (
 	NewTestService   = newTestService
 	RegisterInstance = registerInstance
-	PutJSON          = putJSON
-	GetJSON          = getJSON
 	ServiceTaskGroup = serviceTaskGroup
 )
 

spinifex/domains/ecs/gpu_report.go

@@ -1,6 +1,10 @@
 package ecs
 
-import "context"
+import (
+	"context"
+
+	"github.com/mulgadc/spinifex/spinifex/domains/ecs/taskdefinition"
+)
 
 // ReportTaskGPUInput is the agent's per-task GPU device-assignment report
 // (agent → gateway). Internal agent↔gateway shape, not an AWS SDK action:
@@ -36,7 +40,7 @@ func (s *Service) ReportTaskGPU(ctx context.Context, input *ReportTaskGPUInput,
 		return nil, err
 	}
 	var task TaskRecord
-	found, err := getJSON(ctx, kv, TaskKey(cluster, taskID), &task)
+	found, err := taskdefinition.GetJSON(ctx, kv, TaskKey(cluster, taskID), &task)
 	if err != nil {
 		return nil, err
 	}
@@ -55,7 +59,7 @@ func (s *Service) ReportTaskGPU(ctx context.Context, input *ReportTaskGPUInput,
 		}
 	}
 	if changed {
-		if err := putJSON(ctx, kv, TaskKey(cluster, taskID), &task); err != nil {
+		if err := taskdefinition.PutJSON(ctx, kv, TaskKey(cluster, taskID), &task); err != nil {
 			return nil, err
 		}
 	}

spinifex/domains/ecs/gpu_report_test.go

@@ -2,6 +2,7 @@ package ecs
 
 import (
 	"context"
+	"github.com/mulgadc/spinifex/spinifex/domains/ecs/taskdefinition"
 	"testing"
 
 	"github.com/aws/aws-sdk-go/aws"
@@ -17,7 +18,7 @@ func TestReportTaskGPU_MergesOntoTaskContainers(t *testing.T) {
 	svc, _ := newTestService(t)
 	kv, err := svc.bucket(t.Context(), testAccountID)
 	require.NoError(t, err)
-	require.NoError(t, putJSON(t.Context(), kv, TaskKey("web", "t-1"), &TaskRecord{
+	require.NoError(t, taskdefinition.PutJSON(t.Context(), kv, TaskKey("web", "t-1"), &TaskRecord{
 		TaskID: "t-1", Cluster: "web",
 		Containers: []ContainerState{
 			{Name: "web", Status: "RUNNING"},
@@ -33,7 +34,7 @@ func TestReportTaskGPU_MergesOntoTaskContainers(t *testing.T) {
 	assert.Equal(t, "OK", out.Acknowledgment)
 
 	var rec TaskRecord
-	found, err := getJSON(t.Context(), kv, TaskKey("web", "t-1"), &rec)
+	found, err := taskdefinition.GetJSON(t.Context(), kv, TaskKey("web", "t-1"), &rec)
 	require.NoError(t, err)
 	require.True(t, found)
 	for _, c := range rec.Containers {
@@ -63,7 +64,7 @@ func TestDescribeTasks_GPUIDsPopulatedFromReport(t *testing.T) {
 	svc, _ := newTestService(t)
 	kv, err := svc.bucket(t.Context(), testAccountID)
 	require.NoError(t, err)
-	require.NoError(t, putJSON(t.Context(), kv, TaskKey("web", "t-1"), &TaskRecord{
+	require.NoError(t, taskdefinition.PutJSON(t.Context(), kv, TaskKey("web", "t-1"), &TaskRecord{
 		TaskID: "t-1", ARN: TaskARN(svc.region, testAccountID, "web", "t-1"), Cluster: "web",
 		Containers: []ContainerState{{Name: "trainer", Status: "RUNNING", GPUIDs: []string{"GPU-aaa", "GPU-bbb"}}},
 	}))
@@ -85,7 +86,7 @@ func TestRecordTaskState_PreservesGPUIDsAcrossStoppedReport(t *testing.T) {
 	svc, _ := newTestService(t)
 	kv, err := svc.bucket(t.Context(), testAccountID)
 	require.NoError(t, err)
-	require.NoError(t, putJSON(t.Context(), kv, TaskKey("web", "t-1"), &TaskRecord{
+	require.NoError(t, taskdefinition.PutJSON(t.Context(), kv, TaskKey("web", "t-1"), &TaskRecord{
 		TaskID: "t-1", ARN: TaskARN(svc.region, testAccountID, "web", "t-1"), Cluster: "web",
 		ContainerInstanceID: "i-1",
 		Containers:          []ContainerState{{Name: "trainer", Status: "PENDING"}},
@@ -99,7 +100,7 @@ func TestRecordTaskState_PreservesGPUIDsAcrossStoppedReport(t *testing.T) {
 	}))
 
 	var rec TaskRecord
-	found, err := getJSON(t.Context(), kv, TaskKey("web", "t-1"), &rec)
+	found, err := taskdefinition.GetJSON(t.Context(), kv, TaskKey("web", "t-1"), &rec)
 	require.NoError(t, err)
 	require.True(t, found)
 	require.Len(t, rec.Containers, 1)
@@ -114,7 +115,7 @@ func TestRecordTaskState_PreservesGPUIDsAcrossStoppedReport(t *testing.T) {
 		Containers: []ecsv1.ContainerStatus{{Name: "trainer", Status: ecsv1.TaskStatusStopped, ExitCode: &exit}},
 	}))
 
-	found, err = getJSON(t.Context(), kv, TaskKey("web", "t-1"), &rec)
+	found, err = taskdefinition.GetJSON(t.Context(), kv, TaskKey("web", "t-1"), &rec)
 	require.NoError(t, err)
 	require.True(t, found)
 	require.Len(t, rec.Containers, 1)
@@ -137,7 +138,7 @@ func TestRecordTaskState_ReportedGPUIDsOverwritePinned(t *testing.T) {
 	svc, _ := newTestService(t)
 	kv, err := svc.bucket(t.Context(), testAccountID)
 	require.NoError(t, err)
-	require.NoError(t, putJSON(t.Context(), kv, TaskKey("web", "t-1"), &TaskRecord{
+	require.NoError(t, taskdefinition.PutJSON(t.Context(), kv, TaskKey("web", "t-1"), &TaskRecord{
 		TaskID: "t-1", ARN: TaskARN(svc.region, testAccountID, "web", "t-1"), Cluster: "web",
 		ContainerInstanceID: "i-1",
 		Containers:          []ContainerState{{Name: "trainer", Status: "RUNNING", GPUIDs: []string{"GPU-aaa"}}},
@@ -150,7 +151,7 @@ func TestRecordTaskState_ReportedGPUIDsOverwritePinned(t *testing.T) {
 	}))
 
 	var rec TaskRecord
-	found, err := getJSON(t.Context(), kv, TaskKey("web", "t-1"), &rec)
+	found, err := taskdefinition.GetJSON(t.Context(), kv, TaskKey("web", "t-1"), &rec)
 	require.NoError(t, err)
 	require.True(t, found)
 	require.Len(t, rec.Containers, 1)
@@ -163,7 +164,7 @@ func TestRecordTaskState_NonGPUContainerStaysEmpty(t *testing.T) {
 	svc, _ := newTestService(t)
 	kv, err := svc.bucket(t.Context(), testAccountID)
 	require.NoError(t, err)
-	require.NoError(t, putJSON(t.Context(), kv, TaskKey("web", "t-1"), &TaskRecord{
+	require.NoError(t, taskdefinition.PutJSON(t.Context(), kv, TaskKey("web", "t-1"), &TaskRecord{
 		TaskID: "t-1", ARN: TaskARN(svc.region, testAccountID, "web", "t-1"), Cluster: "web",
 		ContainerInstanceID: "i-1",
 		Containers:          []ContainerState{{Name: "web", Status: "PENDING"}},
@@ -183,7 +184,7 @@ func TestRecordTaskState_NonGPUContainerStaysEmpty(t *testing.T) {
 	}))
 
 	var rec TaskRecord
-	found, err := getJSON(t.Context(), kv, TaskKey("web", "t-1"), &rec)
+	found, err := taskdefinition.GetJSON(t.Context(), kv, TaskKey("web", "t-1"), &rec)
 	require.NoError(t, err)
 	require.True(t, found)
 	require.Len(t, rec.Containers, 1)

spinifex/domains/ecs/instance_recovery_test.go

@@ -2,6 +2,7 @@ package ecs
 
 import (
 	"context"
+	"github.com/mulgadc/spinifex/spinifex/domains/ecs/taskdefinition"
 	"testing"
 	"time"
 
@@ -27,7 +28,7 @@ func instanceStatus(t *testing.T, svc *Service, cluster, id string) InstanceReco
 	kv, err := svc.bucket(t.Context(), testAccountID)
 	require.NoError(t, err)
 	var rec InstanceRecord
-	found, err := getJSON(t.Context(), kv, InstanceKey(cluster, id), &rec)
+	found, err := taskdefinition.GetJSON(t.Context(), kv, InstanceKey(cluster, id), &rec)
 	require.NoError(t, err)
 	require.True(t, found)
 	return rec
@@ -46,7 +47,7 @@ func TestRegister_RestoresReapedInstanceToActive(t *testing.T) {
 	require.NoError(t, err)
 	rec := instanceStatus(t, svc, "web", "i-1")
 	rec.LastSeen = time.Now().UTC().Add(-2 * heartbeatTimeout)
-	require.NoError(t, putJSON(t.Context(), kv, InstanceKey("web", "i-1"), &rec))
+	require.NoError(t, taskdefinition.PutJSON(t.Context(), kv, InstanceKey("web", "i-1"), &rec))
 
 	NewScheduler(nc, svc, "test-holder").reapBucket(context.Background(), kv, testAccountID, time.Now().UTC())
 
@@ -86,7 +87,7 @@ func TestReaper_ReleasesCapacityAndRecoversClean(t *testing.T) {
 	kv, err := svc.bucket(t.Context(), testAccountID)
 	require.NoError(t, err)
 	seeded.LastSeen = time.Now().UTC().Add(-2 * heartbeatTimeout)
-	require.NoError(t, putJSON(t.Context(), kv, InstanceKey("web", "i-1"), &seeded))
+	require.NoError(t, taskdefinition.PutJSON(t.Context(), kv, InstanceKey("web", "i-1"), &seeded))
 
 	NewScheduler(nc, svc, "test-holder").reapBucket(context.Background(), kv, testAccountID, time.Now().UTC())
 

spinifex/domains/ecs/instances.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"encoding/json"
 	"errors"
+	"github.com/mulgadc/spinifex/spinifex/domains/ecs/taskdefinition"
 	"log/slog"
 	"slices"
 	"time"
@@ -17,14 +18,14 @@ import (
 
 // listInstanceRecords returns all container-instance records in a cluster.
 func (s *Service) listInstanceRecords(ctx context.Context, kv jetstream.KeyValue, cluster string) ([]InstanceRecord, error) {
-	keys, err := keysWithPrefix(ctx, kv, InstancesPrefix(cluster))
+	keys, err := taskdefinition.KeysWithPrefix(ctx, kv, InstancesPrefix(cluster))
 	if err != nil {
 		return nil, err
 	}
 	out := make([]InstanceRecord, 0, len(keys))
 	for _, k := range keys {
 		var rec InstanceRecord
-		found, err := getJSON(ctx, kv, k, &rec)
+		found, err := taskdefinition.GetJSON(ctx, kv, k, &rec)
 		if err != nil {
 			return nil, err
 		}
@@ -138,7 +139,7 @@ func (s *Service) DescribeContainerInstances(ctx context.Context, input *ecs.Des
 	for _, ref := range awsStringSlice(input.ContainerInstances) {
 		id := ContainerInstanceShortID(ref)
 		var rec InstanceRecord
-		found, err := getJSON(ctx, kv, InstanceKey(cluster, id), &rec)
+		found, err := taskdefinition.GetJSON(ctx, kv, InstanceKey(cluster, id), &rec)
 		if err != nil {
 			return nil, err
 		}
@@ -194,7 +195,7 @@ func (s *Service) ListContainerInstances(ctx context.Context, input *ecs.ListCon
 // it back. Used by both the AWS register path and the bus register handler.
 func (s *Service) upsertInstance(ctx context.Context, kv jetstream.KeyValue, accountID, cluster, instanceID string, mutate func(*InstanceRecord)) (*InstanceRecord, error) {
 	var rec InstanceRecord
-	found, err := getJSON(ctx, kv, InstanceKey(cluster, instanceID), &rec)
+	found, err := taskdefinition.GetJSON(ctx, kv, InstanceKey(cluster, instanceID), &rec)
 	if err != nil {
 		return nil, err
 	}
@@ -212,7 +213,7 @@ func (s *Service) upsertInstance(ctx context.Context, kv jetstream.KeyValue, acc
 	if mutate != nil {
 		mutate(&rec)
 	}
-	if err := putJSON(ctx, kv, InstanceKey(cluster, instanceID), &rec); err != nil {
+	if err := taskdefinition.PutJSON(ctx, kv, InstanceKey(cluster, instanceID), &rec); err != nil {
 		return nil, err
 	}
 	return &rec, nil
@@ -345,7 +346,7 @@ func (s *Service) recordHeartbeat(ctx context.Context, msg *ecsv1.Heartbeat) err
 		return err
 	}
 	var rec InstanceRecord
-	found, err := getJSON(ctx, kv, InstanceKey(msg.ClusterName, msg.InstanceID), &rec)
+	found, err := taskdefinition.GetJSON(ctx, kv, InstanceKey(msg.ClusterName, msg.InstanceID), &rec)
 	if err != nil || !found {
 		return err
 	}
@@ -353,7 +354,7 @@ func (s *Service) recordHeartbeat(ctx context.Context, msg *ecsv1.Heartbeat) err
 	if msg.Status != "" {
 		rec.Status = msg.Status
 	}
-	return putJSON(ctx, kv, InstanceKey(msg.ClusterName, msg.InstanceID), &rec)
+	return taskdefinition.PutJSON(ctx, kv, InstanceKey(msg.ClusterName, msg.InstanceID), &rec)
 }
 
 // recordTaskState applies an agent task-state report: it updates the task record
@@ -364,7 +365,7 @@ func (s *Service) recordTaskState(ctx context.Context, msg *ecsv1.TaskState) err
 		return err
 	}
 	var task TaskRecord
-	found, err := getJSON(ctx, kv, TaskKey(msg.ClusterName, msg.TaskID), &task)
+	found, err := taskdefinition.GetJSON(ctx, kv, TaskKey(msg.ClusterName, msg.TaskID), &task)
 	if err != nil || !found {
 		return err
 	}
@@ -406,7 +407,7 @@ func (s *Service) recordTaskState(ctx context.Context, msg *ecsv1.TaskState) err
 		// filtering on desired status would otherwise see it as still wanted.
 		task.DesiredStatus = TaskStatusStopped
 	}
-	if err := putJSON(ctx, kv, TaskKey(msg.ClusterName, msg.TaskID), &task); err != nil {
+	if err := taskdefinition.PutJSON(ctx, kv, TaskKey(msg.ClusterName, msg.TaskID), &task); err != nil {
 		return err
 	}
 
@@ -430,7 +431,7 @@ func (s *Service) recordTaskState(ctx context.Context, msg *ecsv1.TaskState) err
 		s.reclaimAssignInbox(ctx, kv, msg.ClusterName, task.ContainerInstanceID, msg.TaskID)
 		s.reclaimStopInbox(ctx, kv, msg.ClusterName, task.ContainerInstanceID, msg.TaskID)
 		s.reclaimTaskENI(ctx, msg.AccountID, &task)
-		if perr := putJSON(ctx, kv, TaskKey(msg.ClusterName, msg.TaskID), &task); perr != nil {
+		if perr := taskdefinition.PutJSON(ctx, kv, TaskKey(msg.ClusterName, msg.TaskID), &task); perr != nil {
 			slog.ErrorContext(ctx, "ECS task STOPPED: persist after EIP release failed", "task", msg.TaskID, "err", perr)
 		}
 		return s.releaseReservation(ctx, kv, msg.ClusterName, task.ContainerInstanceID, msg.TaskID, task.ReservedCPU, task.ReservedMemoryMiB, task.GPU, eniReservationFor(task.NetworkMode))
@@ -485,7 +486,7 @@ func (s *Service) recordDeploymentFailure(ctx context.Context, kv jetstream.KeyV
 		return
 	}
 	var svc ServiceRecord
-	found, err := getJSON(ctx, kv, ServiceKey(cluster, name), &svc)
+	found, err := taskdefinition.GetJSON(ctx, kv, ServiceKey(cluster, name), &svc)
 	if err != nil || !found {
 		return
 	}
@@ -493,7 +494,7 @@ func (s *Service) recordDeploymentFailure(ctx context.Context, kv jetstream.KeyV
 		if svc.Deployments[i].ID == depID {
 			svc.Deployments[i].FailedTasks++
 			svc.Deployments[i].UpdatedAt = time.Now().UTC()
-			if perr := putJSON(ctx, kv, ServiceKey(cluster, name), &svc); perr != nil {
+			if perr := taskdefinition.PutJSON(ctx, kv, ServiceKey(cluster, name), &svc); perr != nil {
 				slog.ErrorContext(ctx, "ECS deployment failure accounting: persist failed", "service", name, "err", perr)
 			}
 			return

spinifex/domains/ecs/poll.go

@@ -2,6 +2,7 @@ package ecs
 
 import (
 	"context"
+	"github.com/mulgadc/spinifex/spinifex/domains/ecs/taskdefinition"
 
 	ecsv1 "github.com/mulgadc/spinifex/contracts/ecs/v1"
 	"github.com/nats-io/nats.go/jetstream"
@@ -55,14 +56,14 @@ func (s *Service) PollAssignments(ctx context.Context, input *PollAssignmentsInp
 		_ = kv.Delete(ctx, StopKey(cluster, instanceID, TaskShortID(taskID)))
 	}
 
-	keys, err := keysWithPrefix(ctx, kv, AssignmentsPrefix(cluster, instanceID))
+	keys, err := taskdefinition.KeysWithPrefix(ctx, kv, AssignmentsPrefix(cluster, instanceID))
 	if err != nil {
 		return nil, err
 	}
 	out := &PollAssignmentsOutput{}
 	for _, key := range keys {
 		var as ecsv1.Assign
-		found, err := getJSON(ctx, kv, key, &as)
+		found, err := taskdefinition.GetJSON(ctx, kv, key, &as)
 		if err != nil {
 			return nil, err
 		}
@@ -71,13 +72,13 @@ func (s *Service) PollAssignments(ctx context.Context, input *PollAssignmentsInp
 		}
 	}
 
-	stopKeys, err := keysWithPrefix(ctx, kv, StopsPrefix(cluster, instanceID))
+	stopKeys, err := taskdefinition.KeysWithPrefix(ctx, kv, StopsPrefix(cluster, instanceID))
 	if err != nil {
 		return nil, err
 	}
 	for _, key := range stopKeys {
 		var sd ecsv1.StopDirective
-		found, err := getJSON(ctx, kv, key, &sd)
+		found, err := taskdefinition.GetJSON(ctx, kv, key, &sd)
 		if err != nil {
 			return nil, err
 		}

spinifex/domains/ecs/poll_test.go

@@ -2,6 +2,7 @@ package ecs
 
 import (
 	"context"
+	"github.com/mulgadc/spinifex/spinifex/domains/ecs/taskdefinition"
 	"testing"
 
 	ecsv1 "github.com/mulgadc/spinifex/contracts/ecs/v1"
@@ -16,9 +17,9 @@ import (
 func TestPollAssignments_DrainsStopsAndAcks(t *testing.T) {
 	svc, _, kv := serviceTestRig(t)
 
-	require.NoError(t, putJSON(t.Context(), kv, AssignmentKey("web", "i-1", "t-1"),
+	require.NoError(t, taskdefinition.PutJSON(t.Context(), kv, AssignmentKey("web", "i-1", "t-1"),
 		&ecsv1.Assign{TaskID: "t-1", ClusterName: "web", InstanceID: "i-1"}))
-	require.NoError(t, putJSON(t.Context(), kv, StopKey("web", "i-1", "t-2"),
+	require.NoError(t, taskdefinition.PutJSON(t.Context(), kv, StopKey("web", "i-1", "t-2"),
 		&ecsv1.StopDirective{TaskID: "t-2", Reason: "drain"}))
 
 	out, err := svc.PollAssignments(context.Background(), &PollAssignmentsInput{
@@ -41,7 +42,7 @@ func TestPollAssignments_DrainsStopsAndAcks(t *testing.T) {
 	assert.Empty(t, out.Assignments)
 	assert.Empty(t, out.Stops)
 
-	found, err := getJSON(t.Context(), kv, StopKey("web", "i-1", "t-2"), &ecsv1.StopDirective{})
+	found, err := taskdefinition.GetJSON(t.Context(), kv, StopKey("web", "i-1", "t-2"), &ecsv1.StopDirective{})
 	require.NoError(t, err)
 	assert.False(t, found)
 }

spinifex/domains/ecs/revisit_test.go

@@ -7,6 +7,7 @@ package ecs
 import (
 	"context"
 	"encoding/json"
+	"github.com/mulgadc/spinifex/spinifex/domains/ecs/taskdefinition"
 	"testing"
 	"time"
 
@@ -60,11 +61,11 @@ func runOneTask(t *testing.T, svc *Service, kv jetstream.KeyValue) string {
 		Cluster: aws.String("web"), TaskDefinition: aws.String("app"), Count: aws.Int64(1),
 	}, testAccountID)
 	require.NoError(t, err)
-	keys, err := keysWithPrefix(t.Context(), kv, TasksPrefix("web"))
+	keys, err := taskdefinition.KeysWithPrefix(t.Context(), kv, TasksPrefix("web"))
 	require.NoError(t, err)
 	require.Len(t, keys, 1)
 	var task TaskRecord
-	found, err := getJSON(t.Context(), kv, keys[0], &task)
+	found, err := taskdefinition.GetJSON(t.Context(), kv, keys[0], &task)
 	require.NoError(t, err)
 	require.True(t, found)
 	return task.TaskID
@@ -128,11 +129,11 @@ func TestScheduler_TheReaperAsksForTheSoonestHeartbeatWindow(t *testing.T) {
 	now := time.Now().UTC()
 
 	var inst InstanceRecord
-	found, err := getJSON(t.Context(), kv, InstanceKey("web", "i-1"), &inst)
+	found, err := taskdefinition.GetJSON(t.Context(), kv, InstanceKey("web", "i-1"), &inst)
 	require.NoError(t, err)
 	require.True(t, found)
 	inst.LastSeen = now.Add(-30 * time.Second)
-	require.NoError(t, putJSON(t.Context(), kv, InstanceKey("web", "i-1"), &inst))
+	require.NoError(t, taskdefinition.PutJSON(t.Context(), kv, InstanceKey("web", "i-1"), &inst))
 
 	due, err := sc.reap(t.Context())
 	require.NoError(t, err)
@@ -142,12 +143,12 @@ func TestScheduler_TheReaperAsksForTheSoonestHeartbeatWindow(t *testing.T) {
 	// Past the window the reaper acts, and with nothing live left it asks for
 	// nothing: the resync is the only thing that need bring it back.
 	inst.LastSeen = now.Add(-2 * heartbeatTimeout)
-	require.NoError(t, putJSON(t.Context(), kv, InstanceKey("web", "i-1"), &inst))
+	require.NoError(t, taskdefinition.PutJSON(t.Context(), kv, InstanceKey("web", "i-1"), &inst))
 	due, err = sc.reap(t.Context())
 	require.NoError(t, err)
 	assert.Zero(t, due)
 
-	found, err = getJSON(t.Context(), kv, InstanceKey("web", "i-1"), &inst)
+	found, err = taskdefinition.GetJSON(t.Context(), kv, InstanceKey("web", "i-1"), &inst)
 	require.NoError(t, err)
 	require.True(t, found)
 	assert.Equal(t, InstanceStatusDraining, inst.Status, "an instance past its window is still reaped")
@@ -195,17 +196,17 @@ func TestScheduler_TheTimerPassReturnsTheSoonestOfItsJobs(t *testing.T) {
 	// A live instance due in 10s, and a stopped task due much later: the pass must
 	// come back for the sooner of the two.
 	var inst InstanceRecord
-	found, err := getJSON(t.Context(), kv, InstanceKey("web", "i-1"), &inst)
+	found, err := taskdefinition.GetJSON(t.Context(), kv, InstanceKey("web", "i-1"), &inst)
 	require.NoError(t, err)
 	require.True(t, found)
 	inst.LastSeen = now.Add(10*time.Second - heartbeatTimeout)
-	require.NoError(t, putJSON(t.Context(), kv, InstanceKey("web", "i-1"), &inst))
+	require.NoError(t, taskdefinition.PutJSON(t.Context(), kv, InstanceKey("web", "i-1"), &inst))
 
 	stopped := TaskRecord{
 		TaskID: "t-1", Cluster: "web", ARN: TaskARN(testRegion, testAccountID, "web", "t-1"),
 		LastStatus: TaskStatusStopped, DesiredStatus: TaskStatusStopped, StoppedAt: now,
 	}
-	require.NoError(t, putJSON(t.Context(), kv, TaskKey("web", "t-1"), &stopped))
+	require.NoError(t, taskdefinition.PutJSON(t.Context(), kv, TaskKey("web", "t-1"), &stopped))
 
 	reapDue, err := sc.reap(t.Context())
 	require.NoError(t, err)
@@ -231,11 +232,11 @@ func TestScheduler_AnInFlightServiceAsksToBeRevisited(t *testing.T) {
 	require.NoError(t, err)
 	assert.Equal(t, reconcileInterval, due)
 
-	keys, err := keysWithPrefix(t.Context(), kv, TasksPrefix("web"))
+	keys, err := taskdefinition.KeysWithPrefix(t.Context(), kv, TasksPrefix("web"))
 	require.NoError(t, err)
 	require.Len(t, keys, 1)
 	var task TaskRecord
-	found, err := getJSON(t.Context(), kv, keys[0], &task)
+	found, err := taskdefinition.GetJSON(t.Context(), kv, keys[0], &task)
 	require.NoError(t, err)
 	require.True(t, found)
 	require.NoError(t, svc.recordTaskState(t.Context(), &ecsv1.TaskState{

spinifex/domains/ecs/scheduler.go

@@ -5,6 +5,7 @@ import (
 	"encoding/json"
 	"errors"
 	"fmt"
+	"github.com/mulgadc/spinifex/spinifex/domains/ecs/taskdefinition"
 	"log/slog"
 	"strings"
 	"sync"
@@ -362,7 +363,7 @@ func (sc *Scheduler) reap(ctx context.Context) (time.Duration, error) {
 // reapBucket reaps one account's disconnected instances and reports when the
 // soonest instance it left alone falls due.
 func (sc *Scheduler) reapBucket(ctx context.Context, kv jetstream.KeyValue, accountID string, now time.Time) time.Duration {
-	keys, err := keysWithPrefix(ctx, kv, "clusters/")
+	keys, err := taskdefinition.KeysWithPrefix(ctx, kv, "clusters/")
 	if err != nil {
 		return 0
 	}
@@ -372,7 +373,7 @@ func (sc *Scheduler) reapBucket(ctx context.Context, kv jetstream.KeyValue, acco
 			continue
 		}
 		var inst InstanceRecord
-		found, err := getJSON(ctx, kv, k, &inst)
+		found, err := taskdefinition.GetJSON(ctx, kv, k, &inst)
 		if err != nil || !found {
 			continue
 		}
@@ -387,7 +388,7 @@ func (sc *Scheduler) reapBucket(ctx context.Context, kv jetstream.KeyValue, acco
 			"lastSeen", inst.LastSeen.Format(time.RFC3339))
 		inst.Status = InstanceStatusDraining
 		inst.Reaped = true
-		if perr := putJSON(ctx, kv, k, &inst); perr != nil {
+		if perr := taskdefinition.PutJSON(ctx, kv, k, &inst); perr != nil {
 			continue
 		}
 		sc.stopInstanceTasks(ctx, kv, accountID, inst.Cluster, inst.InstanceID)
@@ -398,13 +399,13 @@ func (sc *Scheduler) reapBucket(ctx context.Context, kv jetstream.KeyValue, acco
 // stopInstanceTasks transitions a reaped instance's non-stopped tasks to STOPPED
 // and reclaims each awsvpc task's ENI (leak guard for a dead agent).
 func (sc *Scheduler) stopInstanceTasks(ctx context.Context, kv jetstream.KeyValue, accountID, cluster, instanceID string) {
-	keys, err := keysWithPrefix(ctx, kv, TasksPrefix(cluster))
+	keys, err := taskdefinition.KeysWithPrefix(ctx, kv, TasksPrefix(cluster))
 	if err != nil {
 		return
 	}
 	for _, k := range keys {
 		var task TaskRecord
-		found, err := getJSON(ctx, kv, k, &task)
+		found, err := taskdefinition.GetJSON(ctx, kv, k, &task)
 		if err != nil || !found {
 			continue
 		}

spinifex/domains/ecs/service.go

@@ -5,9 +5,7 @@ package ecs
 
 import (
 	"context"
-	"encoding/json"
 	"errors"
-	"fmt"
 	"log/slog"
 	"maps"
 	"slices"
@@ -168,39 +166,12 @@ func (s *Service) bucket(ctx context.Context, accountID string) (jetstream.KeyVa
 	return GetOrCreateAccountBucket(ctx, js, accountID)
 }
 
-// getJSON reads key into out. Returns (false, nil) when the key is absent.
-func getJSON(ctx context.Context, kv jetstream.KeyValue, key string, out any) (bool, error) {
-	entry, err := kv.Get(ctx, key)
-	if errors.Is(err, jetstream.ErrKeyNotFound) {
-		return false, nil
-	}
-	if err != nil {
-		return false, err
-	}
-	if err := json.Unmarshal(entry.Value(), out); err != nil {
-		return false, fmt.Errorf("unmarshal %s: %w", key, err)
-	}
-	return true, nil
-}
-
-// putJSON marshals v and writes it at key.
-func putJSON(ctx context.Context, kv jetstream.KeyValue, key string, v any) error {
-	data, err := json.Marshal(v)
-	if err != nil {
-		return err
-	}
-	if _, err := kv.Put(ctx, key, data); err != nil {
-		return fmt.Errorf("put %s: %w", key, err)
-	}
-	return nil
-}
-
 // deleteKeysWithPrefix deletes every KV key under prefix. Used by DeleteCluster
 // to sweep a cluster's whole key subtree (meta/instances/tasks/services/
 // assignments) after its tasks are stopped. Best-effort per key; the first
 // delete error is returned after attempting the rest.
 func deleteKeysWithPrefix(ctx context.Context, kv jetstream.KeyValue, prefix string) error {
-	keys, err := keysWithPrefix(ctx, kv, prefix)
+	keys, err := taskdefinition.KeysWithPrefix(ctx, kv, prefix)
 	if err != nil {
 		return err
 	}
@@ -213,25 +184,6 @@ func deleteKeysWithPrefix(ctx context.Context, kv jetstream.KeyValue, prefix str
 	return firstErr
 }
 
-// keysWithPrefix returns all KV keys under prefix.
-func keysWithPrefix(ctx context.Context, kv jetstream.KeyValue, prefix string) ([]string, error) {
-	keys, err := kv.Keys(ctx)
-	if errors.Is(err, jetstream.ErrNoKeysFound) {
-		return nil, nil
-	}
-	if err != nil {
-		return nil, err
-	}
-	out := make([]string, 0, len(keys))
-	for _, k := range keys {
-		if strings.HasPrefix(k, prefix) {
-			out = append(out, k)
-		}
-	}
-	slices.Sort(out)
-	return out, nil
-}
-
 // --- Cluster ---
 
 // CreateCluster persists a cluster meta record. Idempotent: re-creating an
@@ -247,7 +199,7 @@ func (s *Service) CreateCluster(ctx context.Context, input *ecs.CreateClusterInp
 	}
 
 	var rec ClusterRecord
-	found, err := getJSON(ctx, kv, ClusterMetaKey(name), &rec)
+	found, err := taskdefinition.GetJSON(ctx, kv, ClusterMetaKey(name), &rec)
 	if err != nil {
 		return nil, err
 	}
@@ -269,7 +221,7 @@ func (s *Service) CreateCluster(ctx context.Context, input *ecs.CreateClusterInp
 				rec.Tags[aws.StringValue(t.Key)] = aws.StringValue(t.Value)
 			}
 		}
-		if err := putJSON(ctx, kv, ClusterMetaKey(name), &rec); err != nil {
+		if err := taskdefinition.PutJSON(ctx, kv, ClusterMetaKey(name), &rec); err != nil {
 			return nil, err
 		}
 	}
@@ -292,7 +244,7 @@ func (s *Service) DescribeClusters(ctx context.Context, input *ecs.DescribeClust
 	for _, name := range names {
 		name = ClusterShortName(name)
 		var rec ClusterRecord
-		found, err := getJSON(ctx, kv, ClusterMetaKey(name), &rec)
+		found, err := taskdefinition.GetJSON(ctx, kv, ClusterMetaKey(name), &rec)
 		if err != nil {
 			return nil, err
 		}
@@ -358,7 +310,7 @@ func (s *Service) ListClusters(ctx context.Context, _ *ecs.ListClustersInput, ac
 	if err != nil {
 		return nil, err
 	}
-	keys, err := keysWithPrefix(ctx, kv, "clusters/")
+	keys, err := taskdefinition.KeysWithPrefix(ctx, kv, "clusters/")
 	if err != nil {
 		return nil, err
 	}
@@ -368,7 +320,7 @@ func (s *Service) ListClusters(ctx context.Context, _ *ecs.ListClustersInput, ac
 			continue
 		}
 		var rec ClusterRecord
-		found, err := getJSON(ctx, kv, k, &rec)
+		found, err := taskdefinition.GetJSON(ctx, kv, k, &rec)
 		if err != nil {
 			return nil, err
 		}

spinifex/domains/ecs/service_test.go

@@ -378,7 +378,7 @@ func TestService_TaskCredIDIsMintedSecret(t *testing.T) {
 		assert.Equal(t, as.CredID, second[taskID].CredID, "re-delivered assign changed credential ID")
 
 		var rec TaskRecord
-		found, gerr := getJSON(ctx, kv, TaskKey("web", taskID), &rec)
+		found, gerr := taskdefinition.GetJSON(ctx, kv, TaskKey("web", taskID), &rec)
 		require.NoError(t, gerr)
 		require.True(t, found)
 		assert.Equal(t, as.CredID, rec.CredID)
@@ -558,10 +558,10 @@ func TestScheduler_ReapBucket_StopsStaleInstanceTasks(t *testing.T) {
 	kv, err := svc.bucket(t.Context(), testAccountID)
 	require.NoError(t, err)
 	var rec InstanceRecord
-	_, err = getJSON(t.Context(), kv, InstanceKey("web", "i-1"), &rec)
+	_, err = taskdefinition.GetJSON(t.Context(), kv, InstanceKey("web", "i-1"), &rec)
 	require.NoError(t, err)
 	rec.LastSeen = time.Now().UTC().Add(-2 * heartbeatTimeout)
-	require.NoError(t, putJSON(t.Context(), kv, InstanceKey("web", "i-1"), &rec))
+	require.NoError(t, taskdefinition.PutJSON(t.Context(), kv, InstanceKey("web", "i-1"), &rec))
 
 	sc := NewScheduler(nc, svc, "test-holder")
 	sc.reapBucket(context.Background(), kv, testAccountID, time.Now().UTC())
@@ -721,7 +721,7 @@ func TestService_RunTask_GPU_ReservesOnPlacement(t *testing.T) {
 	kv, err := svc.bucket(t.Context(), testAccountID)
 	require.NoError(t, err)
 	var rec InstanceRecord
-	found, err := getJSON(t.Context(), kv, InstanceKey("web", "i-1"), &rec)
+	found, err := taskdefinition.GetJSON(t.Context(), kv, InstanceKey("web", "i-1"), &rec)
 	require.NoError(t, err)
 	require.True(t, found)
 	assert.Equal(t, 2, rec.TotalGPU)
@@ -732,7 +732,7 @@ func TestService_RunTask_GPU_ReservesOnPlacement(t *testing.T) {
 		AccountID: testAccountID, ClusterName: "web", InstanceID: "i-1", TaskID: taskID,
 		LastStatus: ecsv1.TaskStatusStopped, Reason: "exited",
 	}))
-	found, err = getJSON(t.Context(), kv, InstanceKey("web", "i-1"), &rec)
+	found, err = taskdefinition.GetJSON(t.Context(), kv, InstanceKey("web", "i-1"), &rec)
 	require.NoError(t, err)
 	require.True(t, found)
 	assert.Equal(t, 0, rec.ReservedGPU)

spinifex/domains/ecs/services.go

@@ -6,6 +6,7 @@ import (
 	"encoding/json"
 	"errors"
 	"fmt"
+	"github.com/mulgadc/spinifex/spinifex/domains/ecs/taskdefinition"
 	"log/slog"
 	"strings"
 	"time"
@@ -145,7 +146,7 @@ func (s *Service) CreateService(ctx context.Context, input *ecs.CreateServiceInp
 		return nil, err
 	}
 	var clusterRec ClusterRecord
-	found, err := getJSON(ctx, kv, ClusterMetaKey(cluster), &clusterRec)
+	found, err := taskdefinition.GetJSON(ctx, kv, ClusterMetaKey(cluster), &clusterRec)
 	if err != nil {
 		return nil, err
 	}
@@ -154,7 +155,7 @@ func (s *Service) CreateService(ctx context.Context, input *ecs.CreateServiceInp
 	}
 
 	var existing ServiceRecord
-	if ok, gerr := getJSON(ctx, kv, ServiceKey(cluster, name), &existing); gerr != nil {
+	if ok, gerr := taskdefinition.GetJSON(ctx, kv, ServiceKey(cluster, name), &existing); gerr != nil {
 		return nil, gerr
 	} else if ok && existing.Status == ServiceStatusActive {
 		return &ecs.CreateServiceOutput{Service: s.serviceToAWS(accountID, &existing)}, nil
@@ -195,7 +196,7 @@ func (s *Service) CreateService(ctx context.Context, input *ecs.CreateServiceInp
 	applyDeploymentConfig(&rec, input.DeploymentConfiguration)
 	rec.LastGoodTaskDefARN = taskDef.ARN
 	rec.Deployments = []Deployment{newPrimaryDeployment(rec.DeploymentID, taskDef, rec.DesiredCount)}
-	if err := putJSON(ctx, kv, ServiceKey(cluster, name), &rec); err != nil {
+	if err := taskdefinition.PutJSON(ctx, kv, ServiceKey(cluster, name), &rec); err != nil {
 		return nil, err
 	}
 
@@ -214,7 +215,7 @@ func (s *Service) UpdateService(ctx context.Context, input *ecs.UpdateServiceInp
 		return nil, err
 	}
 	var rec ServiceRecord
-	found, err := getJSON(ctx, kv, ServiceKey(cluster, name), &rec)
+	found, err := taskdefinition.GetJSON(ctx, kv, ServiceKey(cluster, name), &rec)
 	if err != nil {
 		return nil, err
 	}
@@ -253,7 +254,7 @@ func (s *Service) UpdateService(ctx context.Context, input *ecs.UpdateServiceInp
 		}
 	}
 	rec.UpdatedAt = time.Now().UTC()
-	if err := putJSON(ctx, kv, ServiceKey(cluster, name), &rec); err != nil {
+	if err := taskdefinition.PutJSON(ctx, kv, ServiceKey(cluster, name), &rec); err != nil {
 		return nil, err
 	}
 	if err := s.reconcileService(ctx, kv, accountID, &rec); err != nil {
@@ -272,7 +273,7 @@ func (s *Service) DeleteService(ctx context.Context, input *ecs.DeleteServiceInp
 		return nil, err
 	}
 	var rec ServiceRecord
-	found, err := getJSON(ctx, kv, ServiceKey(cluster, name), &rec)
+	found, err := taskdefinition.GetJSON(ctx, kv, ServiceKey(cluster, name), &rec)
 	if err != nil {
 		return nil, err
 	}
@@ -292,7 +293,7 @@ func (s *Service) DeleteService(ctx context.Context, input *ecs.DeleteServiceInp
 	rec.RunningCount = 0
 	rec.PendingCount = 0
 	rec.UpdatedAt = time.Now().UTC()
-	if err := putJSON(ctx, kv, ServiceKey(cluster, name), &rec); err != nil {
+	if err := taskdefinition.PutJSON(ctx, kv, ServiceKey(cluster, name), &rec); err != nil {
 		return nil, err
 	}
 	return &ecs.DeleteServiceOutput{Service: s.serviceToAWS(accountID, &rec)}, nil
@@ -309,7 +310,7 @@ func (s *Service) DescribeServices(ctx context.Context, input *ecs.DescribeServi
 	for _, ref := range awsStringSlice(input.Services) {
 		name := ServiceShortName(ref)
 		var rec ServiceRecord
-		found, gerr := getJSON(ctx, kv, ServiceKey(cluster, name), &rec)
+		found, gerr := taskdefinition.GetJSON(ctx, kv, ServiceKey(cluster, name), &rec)
 		if gerr != nil {
 			return nil, gerr
 		}
@@ -430,7 +431,7 @@ func persistServiceIfChanged(ctx context.Context, kv jetstream.KeyValue, svc *Se
 	if bytes.Equal(before, after) {
 		return nil
 	}
-	return putJSON(ctx, kv, ServiceKey(svc.Cluster, svc.Name), svc)
+	return taskdefinition.PutJSON(ctx, kv, ServiceKey(svc.Cluster, svc.Name), svc)
 }
 
 // reconcileAllServices reconciles every ACTIVE service in every ECS account
@@ -603,14 +604,14 @@ func (s *Service) listServiceTasks(ctx context.Context, kv jetstream.KeyValue, c
 
 // listTaskRecords returns every task record in a cluster.
 func (s *Service) listTaskRecords(ctx context.Context, kv jetstream.KeyValue, cluster string) ([]TaskRecord, error) {
-	keys, err := keysWithPrefix(ctx, kv, TasksPrefix(cluster))
+	keys, err := taskdefinition.KeysWithPrefix(ctx, kv, TasksPrefix(cluster))
 	if err != nil {
 		return nil, err
 	}
 	out := make([]TaskRecord, 0, len(keys))
 	for _, k := range keys {
 		var rec TaskRecord
-		found, err := getJSON(ctx, kv, k, &rec)
+		found, err := taskdefinition.GetJSON(ctx, kv, k, &rec)
 		if err != nil {
 			return nil, err
 		}
@@ -623,14 +624,14 @@ func (s *Service) listTaskRecords(ctx context.Context, kv jetstream.KeyValue, cl
 
 // listServiceRecords returns every service record in a cluster.
 func (s *Service) listServiceRecords(ctx context.Context, kv jetstream.KeyValue, cluster string) ([]ServiceRecord, error) {
-	keys, err := keysWithPrefix(ctx, kv, ServicesPrefix(cluster))
+	keys, err := taskdefinition.KeysWithPrefix(ctx, kv, ServicesPrefix(cluster))
 	if err != nil {
 		return nil, err
 	}
 	out := make([]ServiceRecord, 0, len(keys))
 	for _, k := range keys {
 		var rec ServiceRecord
-		found, err := getJSON(ctx, kv, k, &rec)
+		found, err := taskdefinition.GetJSON(ctx, kv, k, &rec)
 		if err != nil {
 			return nil, err
 		}
@@ -643,7 +644,7 @@ func (s *Service) listServiceRecords(ctx context.Context, kv jetstream.KeyValue,
 
 // allServiceRecords returns every service record across all clusters in a bucket.
 func (s *Service) allServiceRecords(ctx context.Context, kv jetstream.KeyValue) ([]ServiceRecord, error) {
-	keys, err := keysWithPrefix(ctx, kv, "clusters/")
+	keys, err := taskdefinition.KeysWithPrefix(ctx, kv, "clusters/")
 	if err != nil {
 		return nil, err
 	}
@@ -653,7 +654,7 @@ func (s *Service) allServiceRecords(ctx context.Context, kv jetstream.KeyValue)
 			continue
 		}
 		var rec ServiceRecord
-		found, err := getJSON(ctx, kv, k, &rec)
+		found, err := taskdefinition.GetJSON(ctx, kv, k, &rec)
 		if err != nil {
 			return nil, err
 		}

spinifex/domains/ecs/services_test.go

@@ -3,6 +3,7 @@ package ecs
 import (
 	"context"
 	"fmt"
+	"github.com/mulgadc/spinifex/spinifex/domains/ecs/taskdefinition"
 	"testing"
 	"time"
 
@@ -469,7 +470,7 @@ func TestService_StopTask_StopsAndDeregisters(t *testing.T) {
 		NetworkMode: NetworkModeAwsvpc, ENIPrivateIP: "10.0.1.9",
 		ReservedCPU: 64, ReservedMemoryMiB: 64,
 	}
-	require.NoError(t, putJSON(t.Context(), kv, TaskKey("web", "t-9"), task))
+	require.NoError(t, taskdefinition.PutJSON(t.Context(), kv, TaskKey("web", "t-9"), task))
 
 	// Cooperative stop: the task stays RUNNING with desiredStatus=STOPPED and a
 	// stop directive lands in the instance inbox; nothing is deregistered until the
@@ -483,7 +484,7 @@ func TestService_StopTask_StopsAndDeregisters(t *testing.T) {
 	assert.Empty(t, reg.deregistered)
 
 	var sd ecsv1.StopDirective
-	found, err := getJSON(t.Context(), kv, StopKey("web", "i-1", "t-9"), &sd)
+	found, err := taskdefinition.GetJSON(t.Context(), kv, StopKey("web", "i-1", "t-9"), &sd)
 	require.NoError(t, err)
 	require.True(t, found)
 	assert.Equal(t, "bye", sd.Reason)
@@ -494,7 +495,7 @@ func TestService_StopTask_StopsAndDeregisters(t *testing.T) {
 		TaskID: "t-9", LastStatus: ecsv1.TaskStatusStopped,
 	}))
 	assert.Equal(t, []string{key(tgARN, "10.0.1.9", 80)}, reg.deregistered)
-	found, err = getJSON(t.Context(), kv, StopKey("web", "i-1", "t-9"), &sd)
+	found, err = taskdefinition.GetJSON(t.Context(), kv, StopKey("web", "i-1", "t-9"), &sd)
 	require.NoError(t, err)
 	assert.False(t, found)
 
@@ -518,7 +519,7 @@ func TestService_StopTask_UnplacedForceStops(t *testing.T) {
 		TaskID: "t-np", Cluster: "web",
 		LastStatus: TaskStatusPending, DesiredStatus: TaskStatusRunning,
 	}
-	require.NoError(t, putJSON(t.Context(), kv, TaskKey("web", "t-np"), task))
+	require.NoError(t, taskdefinition.PutJSON(t.Context(), kv, TaskKey("web", "t-np"), task))
 
 	out, err := svc.StopTask(context.Background(), &ecs.StopTaskInput{
 		Cluster: aws.String("web"), Task: aws.String("t-np"), Reason: aws.String("cancel"),
@@ -550,7 +551,7 @@ func TestService_TargetGroup_RegisterOnRunningDeregisterOnStopped(t *testing.T)
 		ContainerInstanceID: "i-1", LastStatus: TaskStatusPending,
 		NetworkMode: NetworkModeAwsvpc, ENIPrivateIP: "10.0.1.5",
 	}
-	require.NoError(t, putJSON(t.Context(), kv, TaskKey("web", "t-1"), task))
+	require.NoError(t, taskdefinition.PutJSON(t.Context(), kv, TaskKey("web", "t-1"), task))
 
 	// RUNNING → register.
 	require.NoError(t, svc.recordTaskState(context.Background(), &ecsv1.TaskState{

spinifex/domains/ecs/store.go

@@ -107,12 +107,6 @@ func ServiceKey(cluster, serviceName string) string {
 	return ServicesPrefix(cluster) + serviceName
 }
 
-// LeaderLeaseKey returns the per-cluster scheduler leader-lease key in the shared
-// leader bucket (Q3).
-func LeaderLeaseKey(accountID, clusterName string) string {
-	return fmt.Sprintf("%s/%s", accountID, clusterName)
-}
-
 // CapacityProvidersPrefix returns the KV key prefix under which all
 // capacity-provider records live. Account-scoped (not cluster-scoped),
 // matching the AWS ARN shape; a cluster references providers by name.

spinifex/domains/ecs/store_test.go

@@ -124,8 +124,6 @@ func TestKeyPaths(t *testing.T) {
 
 	assert.Equal(t, "clusters/web/services/", ServicesPrefix("web"))
 	assert.Equal(t, "clusters/web/services/api", ServiceKey("web", "api"))
-
-	assert.Equal(t, "123456789012/web", LeaderLeaseKey("123456789012", "web"))
 }
 
 // Prefix helpers must be a true prefix of their per-record key so a KV

spinifex/domains/ecs/tagging_current_behaviour_test.go

@@ -2,6 +2,7 @@ package ecs
 
 import (
 	"context"
+	"github.com/mulgadc/spinifex/spinifex/domains/ecs/taskdefinition"
 	"strings"
 	"sync"
 	"testing"
@@ -57,11 +58,11 @@ func TestCurrentBehaviour_Tagging_ECSTagWriteRevertsInterleavedRecordWrite(t *te
 	key := ServiceKey("web", "api")
 	wrapped := &interleavingKV{KeyValue: kv, key: key, between: func() {
 		var rec ServiceRecord
-		found, gerr := getJSON(ctx, kv, key, &rec)
+		found, gerr := taskdefinition.GetJSON(ctx, kv, key, &rec)
 		require.NoError(t, gerr)
 		require.True(t, found)
 		rec.DesiredCount = 5
-		require.NoError(t, putJSON(ctx, kv, key, &rec))
+		require.NoError(t, taskdefinition.PutJSON(ctx, kv, key, &rec))
 	}}
 
 	err = svc.mutateResourceTags(ctx, wrapped, ecsResourceService, "web", "api", func(existing map[string]string) map[string]string {
@@ -74,7 +75,7 @@ func TestCurrentBehaviour_Tagging_ECSTagWriteRevertsInterleavedRecordWrite(t *te
 	require.NoError(t, err)
 
 	var final ServiceRecord
-	found, err := getJSON(ctx, kv, key, &final)
+	found, err := taskdefinition.GetJSON(ctx, kv, key, &final)
 	require.NoError(t, err)
 	require.True(t, found)
 	assert.Equal(t, 0, final.DesiredCount, "the interleaved DesiredCount=5 write is lost")

spinifex/domains/ecs/tags.go

@@ -78,7 +78,7 @@ func (s *Service) resourceTags(ctx context.Context, kv jetstream.KeyValue, kind
 	switch kind {
 	case ecsResourceCluster:
 		var rec ClusterRecord
-		found, err := getJSON(ctx, kv, ClusterMetaKey(id), &rec)
+		found, err := taskdefinition.GetJSON(ctx, kv, ClusterMetaKey(id), &rec)
 		if err != nil {
 			return nil, err
 		}
@@ -92,7 +92,7 @@ func (s *Service) resourceTags(ctx context.Context, kv jetstream.KeyValue, kind
 			return nil, errors.New(awserrors.ErrorECSInvalidParameter)
 		}
 		var rec taskdefinition.Record
-		found, err := getJSON(ctx, kv, taskdefinition.RevKey(family, rev), &rec)
+		found, err := taskdefinition.GetJSON(ctx, kv, taskdefinition.RevKey(family, rev), &rec)
 		if err != nil {
 			return nil, err
 		}
@@ -102,7 +102,7 @@ func (s *Service) resourceTags(ctx context.Context, kv jetstream.KeyValue, kind
 		return rec.Tags, nil
 	case ecsResourceService:
 		var rec ServiceRecord
-		found, err := getJSON(ctx, kv, ServiceKey(cluster, id), &rec)
+		found, err := taskdefinition.GetJSON(ctx, kv, ServiceKey(cluster, id), &rec)
 		if err != nil {
 			return nil, err
 		}
@@ -112,7 +112,7 @@ func (s *Service) resourceTags(ctx context.Context, kv jetstream.KeyValue, kind
 		return rec.Tags, nil
 	case ecsResourceTask:
 		var rec TaskRecord
-		found, err := getJSON(ctx, kv, TaskKey(cluster, id), &rec)
+		found, err := taskdefinition.GetJSON(ctx, kv, TaskKey(cluster, id), &rec)
 		if err != nil {
 			return nil, err
 		}
@@ -122,7 +122,7 @@ func (s *Service) resourceTags(ctx context.Context, kv jetstream.KeyValue, kind
 		return rec.Tags, nil
 	case ecsResourceContainerInstance:
 		var rec InstanceRecord
-		found, err := getJSON(ctx, kv, InstanceKey(cluster, id), &rec)
+		found, err := taskdefinition.GetJSON(ctx, kv, InstanceKey(cluster, id), &rec)
 		if err != nil {
 			return nil, err
 		}
@@ -142,7 +142,7 @@ func (s *Service) mutateResourceTags(ctx context.Context, kv jetstream.KeyValue,
 	switch kind {
 	case ecsResourceCluster:
 		var rec ClusterRecord
-		found, err := getJSON(ctx, kv, ClusterMetaKey(id), &rec)
+		found, err := taskdefinition.GetJSON(ctx, kv, ClusterMetaKey(id), &rec)
 		if err != nil {
 			return err
 		}
@@ -150,14 +150,14 @@ func (s *Service) mutateResourceTags(ctx context.Context, kv jetstream.KeyValue,
 			return errors.New(awserrors.ErrorECSClusterNotFound)
 		}
 		rec.Tags = mutate(rec.Tags)
-		return putJSON(ctx, kv, ClusterMetaKey(id), &rec)
+		return taskdefinition.PutJSON(ctx, kv, ClusterMetaKey(id), &rec)
 	case ecsResourceTaskDefinition:
 		family, rev := taskdefinition.ParseRef(id)
 		if family == "" || rev == 0 {
 			return errors.New(awserrors.ErrorECSInvalidParameter)
 		}
 		var rec taskdefinition.Record
-		found, err := getJSON(ctx, kv, taskdefinition.RevKey(family, rev), &rec)
+		found, err := taskdefinition.GetJSON(ctx, kv, taskdefinition.RevKey(family, rev), &rec)
 		if err != nil {
 			return err
 		}
@@ -165,10 +165,10 @@ func (s *Service) mutateResourceTags(ctx context.Context, kv jetstream.KeyValue,
 			return errors.New(awserrors.ErrorECSInvalidParameter)
 		}
 		rec.Tags = mutate(rec.Tags)
-		return putJSON(ctx, kv, taskdefinition.RevKey(family, rev), &rec)
+		return taskdefinition.PutJSON(ctx, kv, taskdefinition.RevKey(family, rev), &rec)
 	case ecsResourceService:
 		var rec ServiceRecord
-		found, err := getJSON(ctx, kv, ServiceKey(cluster, id), &rec)
+		found, err := taskdefinition.GetJSON(ctx, kv, ServiceKey(cluster, id), &rec)
 		if err != nil {
 			return err
 		}
@@ -176,10 +176,10 @@ func (s *Service) mutateResourceTags(ctx context.Context, kv jetstream.KeyValue,
 			return errors.New(awserrors.ErrorECSServiceNotFound)
 		}
 		rec.Tags = mutate(rec.Tags)
-		return putJSON(ctx, kv, ServiceKey(cluster, id), &rec)
+		return taskdefinition.PutJSON(ctx, kv, ServiceKey(cluster, id), &rec)
 	case ecsResourceTask:
 		var rec TaskRecord
-		found, err := getJSON(ctx, kv, TaskKey(cluster, id), &rec)
+		found, err := taskdefinition.GetJSON(ctx, kv, TaskKey(cluster, id), &rec)
 		if err != nil {
 			return err
 		}
@@ -187,10 +187,10 @@ func (s *Service) mutateResourceTags(ctx context.Context, kv jetstream.KeyValue,
 			return errors.New(awserrors.ErrorECSInvalidParameter)
 		}
 		rec.Tags = mutate(rec.Tags)
-		return putJSON(ctx, kv, TaskKey(cluster, id), &rec)
+		return taskdefinition.PutJSON(ctx, kv, TaskKey(cluster, id), &rec)
 	case ecsResourceContainerInstance:
 		var rec InstanceRecord
-		found, err := getJSON(ctx, kv, InstanceKey(cluster, id), &rec)
+		found, err := taskdefinition.GetJSON(ctx, kv, InstanceKey(cluster, id), &rec)
 		if err != nil {
 			return err
 		}
@@ -198,7 +198,7 @@ func (s *Service) mutateResourceTags(ctx context.Context, kv jetstream.KeyValue,
 			return errors.New(awserrors.ErrorECSInvalidParameter)
 		}
 		rec.Tags = mutate(rec.Tags)
-		return putJSON(ctx, kv, InstanceKey(cluster, id), &rec)
+		return taskdefinition.PutJSON(ctx, kv, InstanceKey(cluster, id), &rec)
 	default:
 		return errors.New(awserrors.ErrorECSInvalidParameter)
 	}

spinifex/domains/ecs/task_accounting_test.go

@@ -1,6 +1,7 @@
 package ecs_test
 
 import (
+	"github.com/mulgadc/spinifex/spinifex/domains/ecs/taskdefinition"
 	"testing"
 
 	"github.com/aws/aws-sdk-go/aws"
@@ -23,7 +24,7 @@ func seedTasks(t *testing.T, svc *ecsdomain.Service, cluster string, recs ...*ec
 	for _, r := range recs {
 		r.Cluster = cluster
 		r.ARN = ecsdomain.TaskARN(svc.Region(), ecsdomain.TestAccountID, cluster, r.TaskID)
-		require.NoError(t, ecsdomain.PutJSON(t.Context(), kv, ecsdomain.TaskKey(cluster, r.TaskID), r))
+		require.NoError(t, taskdefinition.PutJSON(t.Context(), kv, ecsdomain.TaskKey(cluster, r.TaskID), r))
 	}
 }
 
@@ -124,7 +125,7 @@ func TestRecordTaskState_SelfExitMovesDesiredStatusToStopped(t *testing.T) {
 	kv, err := svc.Bucket(t.Context(), ecsdomain.TestAccountID)
 	require.NoError(t, err)
 	var rec ecsdomain.TaskRecord
-	found, err := ecsdomain.GetJSON(t.Context(), kv, ecsdomain.TaskKey("web", "t-1"), &rec)
+	found, err := taskdefinition.GetJSON(t.Context(), kv, ecsdomain.TaskKey("web", "t-1"), &rec)
 	require.NoError(t, err)
 	require.True(t, found)
 	assert.Equal(t, ecsdomain.TaskStatusStopped, rec.DesiredStatus)

spinifex/domains/ecs/task_stop.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"encoding/json"
 	"errors"
+	"github.com/mulgadc/spinifex/spinifex/domains/ecs/taskdefinition"
 	"log/slog"
 	"strings"
 	"time"
@@ -33,7 +34,7 @@ func (s *Service) StopTask(ctx context.Context, input *ecs.StopTaskInput, accoun
 		return nil, err
 	}
 	var task TaskRecord
-	found, err := getJSON(ctx, kv, TaskKey(cluster, taskID), &task)
+	found, err := taskdefinition.GetJSON(ctx, kv, TaskKey(cluster, taskID), &task)
 	if err != nil {
 		return nil, err
 	}
@@ -57,7 +58,7 @@ func (s *Service) StartTask(ctx context.Context, input *ecs.StartTaskInput, acco
 		return nil, err
 	}
 	var clusterRec ClusterRecord
-	found, err := getJSON(ctx, kv, ClusterMetaKey(cluster), &clusterRec)
+	found, err := taskdefinition.GetJSON(ctx, kv, ClusterMetaKey(cluster), &clusterRec)
 	if err != nil {
 		return nil, err
 	}
@@ -101,7 +102,7 @@ func (s *Service) StartTask(ctx context.Context, input *ecs.StartTaskInput, acco
 				continue
 			}
 		}
-		if err := putJSON(ctx, kv, TaskKey(cluster, taskID), rec); err != nil {
+		if err := taskdefinition.PutJSON(ctx, kv, TaskKey(cluster, taskID), rec); err != nil {
 			return nil, err
 		}
 		if err := s.publishAssign(ctx, kv, accountID, cluster, inst.InstanceID, rec, taskDef); err != nil {
@@ -168,12 +169,12 @@ func (s *Service) requestStopTask(ctx context.Context, kv jetstream.KeyValue, ac
 		return
 	}
 	task.DesiredStatus = TaskStatusStopped
-	if perr := putJSON(ctx, kv, TaskKey(task.Cluster, task.TaskID), task); perr != nil {
+	if perr := taskdefinition.PutJSON(ctx, kv, TaskKey(task.Cluster, task.TaskID), task); perr != nil {
 		slog.ErrorContext(ctx, "ECS requestStopTask: persist failed", "task", task.TaskID, "err", perr)
 		return
 	}
 	sd := ecsv1.StopDirective{TaskID: task.TaskID, Reason: reason}
-	if perr := putJSON(ctx, kv, StopKey(task.Cluster, task.ContainerInstanceID, task.TaskID), &sd); perr != nil {
+	if perr := taskdefinition.PutJSON(ctx, kv, StopKey(task.Cluster, task.ContainerInstanceID, task.TaskID), &sd); perr != nil {
 		slog.ErrorContext(ctx, "ECS requestStopTask: post stop directive failed", "task", task.TaskID, "err", perr)
 	}
 }
@@ -196,7 +197,7 @@ func (s *Service) forceStopTask(ctx context.Context, kv jetstream.KeyValue, acco
 	// cleared fields land in this single write.
 	s.releaseTaskPublicIP(ctx, accountID, task)
 	s.reclaimTaskENI(ctx, accountID, task)
-	if perr := putJSON(ctx, kv, TaskKey(task.Cluster, task.TaskID), task); perr != nil {
+	if perr := taskdefinition.PutJSON(ctx, kv, TaskKey(task.Cluster, task.TaskID), task); perr != nil {
 		slog.ErrorContext(ctx, "ECS forceStopTask: persist failed", "task", task.TaskID, "err", perr)
 		return
 	}
@@ -238,7 +239,7 @@ func (s *Service) applyServiceTargets(ctx context.Context, kv jetstream.KeyValue
 		return
 	}
 	var svc ServiceRecord
-	found, err := getJSON(ctx, kv, ServiceKey(task.Cluster, name), &svc)
+	found, err := taskdefinition.GetJSON(ctx, kv, ServiceKey(task.Cluster, name), &svc)
 	if err != nil || !found || len(svc.LoadBalancers) == 0 {
 		return
 	}
@@ -270,7 +271,7 @@ func (s *Service) assignTaskPublicIP(ctx context.Context, kv jetstream.KeyValue,
 		return
 	}
 	var svc ServiceRecord
-	found, err := getJSON(ctx, kv, ServiceKey(task.Cluster, name), &svc)
+	found, err := taskdefinition.GetJSON(ctx, kv, ServiceKey(task.Cluster, name), &svc)
 	if err != nil || !found || !strings.EqualFold(svc.AssignPublicIP, "ENABLED") {
 		return
 	}
@@ -281,7 +282,7 @@ func (s *Service) assignTaskPublicIP(ctx context.Context, kv jetstream.KeyValue,
 	}
 	task.ENIPublicIP = publicIP
 	task.ENIEIPAllocationID = allocID
-	if perr := putJSON(ctx, kv, TaskKey(task.Cluster, task.TaskID), task); perr != nil {
+	if perr := taskdefinition.PutJSON(ctx, kv, TaskKey(task.Cluster, task.TaskID), task); perr != nil {
 		slog.ErrorContext(ctx, "ECS auto-EIP: persist public IP failed", "task", task.TaskID, "err", perr)
 	}
 }

spinifex/domains/ecs/taskdef_validation_test.go

@@ -3,6 +3,7 @@ package ecs
 import (
 	"context"
 	ecsv1 "github.com/mulgadc/spinifex/contracts/ecs/v1"
+	"github.com/mulgadc/spinifex/spinifex/domains/ecs/taskdefinition"
 	"testing"
 
 	"github.com/aws/aws-sdk-go/aws"
@@ -240,7 +241,7 @@ func TestRunTask_AssignCarriesGPU(t *testing.T) {
 	kv, err := svc.bucket(t.Context(), testAccountID)
 	require.NoError(t, err)
 	var rec TaskRecord
-	found, err := getJSON(t.Context(), kv, TaskKey("web", taskID), &rec)
+	found, err := taskdefinition.GetJSON(t.Context(), kv, TaskKey("web", taskID), &rec)
 	require.NoError(t, err)
 	require.True(t, found)
 	assert.Equal(t, 1, rec.GPU)

spinifex/domains/ecs/taskdefinition/owner.go

@@ -63,10 +63,10 @@ func (o *Owner) Register(ctx context.Context, kv jetstream.KeyValue, accountID s
 		RequiresCompatibilities: spec.RequiresCompatibilities,
 		RuntimePlatform:         spec.RuntimePlatform,
 	}
-	if err := putJSON(ctx, kv, RevKey(spec.Family, rev), rec); err != nil {
+	if err := PutJSON(ctx, kv, RevKey(spec.Family, rev), rec); err != nil {
 		return nil, err
 	}
-	if err := putJSON(ctx, kv, LatestRevKey(spec.Family), rev); err != nil {
+	if err := PutJSON(ctx, kv, LatestRevKey(spec.Family), rev); err != nil {
 		return nil, err
 	}
 	return rec, nil
@@ -75,7 +75,7 @@ func (o *Owner) Register(ctx context.Context, kv jetstream.KeyValue, accountID s
 // nextRevision reads the family's latest-rev and returns latest+1 (1 if absent).
 func nextRevision(ctx context.Context, kv jetstream.KeyValue, family string) (int, error) {
 	var latest int
-	found, err := getJSON(ctx, kv, LatestRevKey(family), &latest)
+	found, err := GetJSON(ctx, kv, LatestRevKey(family), &latest)
 	if err != nil {
 		return 0, err
 	}
@@ -88,7 +88,7 @@ func nextRevision(ctx context.Context, kv jetstream.KeyValue, family string) (in
 // Deregister marks one revision INACTIVE; the revision stays describable.
 func (o *Owner) Deregister(ctx context.Context, kv jetstream.KeyValue, family string, rev int) (*Record, error) {
 	var rec Record
-	found, err := getJSON(ctx, kv, RevKey(family, rev), &rec)
+	found, err := GetJSON(ctx, kv, RevKey(family, rev), &rec)
 	if err != nil {
 		return nil, err
 	}
@@ -100,7 +100,7 @@ func (o *Owner) Deregister(ctx context.Context, kv jetstream.KeyValue, family st
 		rec.DeregisteredAt = &now
 	}
 	rec.Status = StatusInactive
-	if err := putJSON(ctx, kv, RevKey(family, rev), &rec); err != nil {
+	if err := PutJSON(ctx, kv, RevKey(family, rev), &rec); err != nil {
 		return nil, err
 	}
 	return &rec, nil
@@ -115,7 +115,7 @@ func (o *Owner) Resolve(ctx context.Context, kv jetstream.KeyValue, ref string)
 	}
 	if rev == 0 {
 		var latest int
-		found, err := getJSON(ctx, kv, LatestRevKey(family), &latest)
+		found, err := GetJSON(ctx, kv, LatestRevKey(family), &latest)
 		if err != nil {
 			return nil, err
 		}
@@ -125,7 +125,7 @@ func (o *Owner) Resolve(ctx context.Context, kv jetstream.KeyValue, ref string)
 		rev = latest
 	}
 	var rec Record
-	found, err := getJSON(ctx, kv, RevKey(family, rev), &rec)
+	found, err := GetJSON(ctx, kv, RevKey(family, rev), &rec)
 	if err != nil {
 		return nil, err
 	}
@@ -142,7 +142,7 @@ func (o *Owner) List(ctx context.Context, kv jetstream.KeyValue, family, status
 	if family != "" {
 		prefix = RevsPrefix(family)
 	}
-	keys, err := keysWithPrefix(ctx, kv, prefix)
+	keys, err := KeysWithPrefix(ctx, kv, prefix)
 	if err != nil {
 		return nil, err
 	}
@@ -152,7 +152,7 @@ func (o *Owner) List(ctx context.Context, kv jetstream.KeyValue, family, status
 			continue
 		}
 		var rec Record
-		found, err := getJSON(ctx, kv, k, &rec)
+		found, err := GetJSON(ctx, kv, k, &rec)
 		if err != nil {
 			return nil, err
 		}
@@ -163,8 +163,8 @@ func (o *Owner) List(ctx context.Context, kv jetstream.KeyValue, family, status
 	return arns, nil
 }
 
-// getJSON reads key into out. Returns (false, nil) when the key is absent.
-func getJSON(ctx context.Context, kv jetstream.KeyValue, key string, out any) (bool, error) {
+// GetJSON reads key into out. Returns (false, nil) when the key is absent.
+func GetJSON(ctx context.Context, kv jetstream.KeyValue, key string, out any) (bool, error) {
 	entry, err := kv.Get(ctx, key)
 	if errors.Is(err, jetstream.ErrKeyNotFound) {
 		return false, nil
@@ -178,8 +178,8 @@ func getJSON(ctx context.Context, kv jetstream.KeyValue, key string, out any) (b
 	return true, nil
 }
 
-// putJSON marshals v and writes it at key.
-func putJSON(ctx context.Context, kv jetstream.KeyValue, key string, v any) error {
+// PutJSON marshals v and writes it at key.
+func PutJSON(ctx context.Context, kv jetstream.KeyValue, key string, v any) error {
 	data, err := json.Marshal(v)
 	if err != nil {
 		return err
@@ -190,8 +190,8 @@ func putJSON(ctx context.Context, kv jetstream.KeyValue, key string, v any) erro
 	return nil
 }
 
-// keysWithPrefix returns all KV keys under prefix, sorted.
-func keysWithPrefix(ctx context.Context, kv jetstream.KeyValue, prefix string) ([]string, error) {
+// KeysWithPrefix returns all KV keys under prefix, sorted.
+func KeysWithPrefix(ctx context.Context, kv jetstream.KeyValue, prefix string) ([]string, error) {
 	keys, err := kv.Keys(ctx)
 	if errors.Is(err, jetstream.ErrNoKeysFound) {
 		return nil, nil

spinifex/domains/ecs/tasks.go

@@ -34,7 +34,7 @@ func (s *Service) RunTask(ctx context.Context, input *ecs.RunTaskInput, accountI
 	}
 
 	var clusterRec ClusterRecord
-	found, err := getJSON(ctx, kv, ClusterMetaKey(cluster), &clusterRec)
+	found, err := taskdefinition.GetJSON(ctx, kv, ClusterMetaKey(cluster), &clusterRec)
 	if err != nil {
 		return nil, err
 	}
@@ -82,7 +82,7 @@ func (s *Service) RunTask(ctx context.Context, input *ecs.RunTaskInput, accountI
 			}
 		}
 
-		if err := putJSON(ctx, kv, TaskKey(cluster, taskID), rec); err != nil {
+		if err := taskdefinition.PutJSON(ctx, kv, TaskKey(cluster, taskID), rec); err != nil {
 			return nil, err
 		}
 		if err := s.publishAssign(ctx, kv, accountID, cluster, inst.InstanceID, rec, taskDef); err != nil {
@@ -166,7 +166,7 @@ func (s *Service) provisionTaskENI(ctx context.Context, kv jetstream.KeyValue, a
 			rec.DesiredStatus = TaskStatusStopped
 			rec.StoppedReason = fmt.Sprintf("ENI attach failed: %s", err)
 			rec.StoppedAt = time.Now().UTC()
-			if perr := putJSON(ctx, kv, TaskKey(cluster, rec.TaskID), rec); perr != nil {
+			if perr := taskdefinition.PutJSON(ctx, kv, TaskKey(cluster, rec.TaskID), rec); perr != nil {
 				slog.ErrorContext(ctx, "ECS RunTask: persist ENI rollback failed", "task", rec.TaskID, "err", perr)
 			}
 		}
@@ -271,7 +271,7 @@ func (s *Service) publishAssign(ctx context.Context, kv jetstream.KeyValue, acco
 	for _, c := range td.Containers {
 		msg.Containers = append(msg.Containers, containerToAssign(c))
 	}
-	return putJSON(ctx, kv, AssignmentKey(cluster, instanceID, rec.TaskID), &msg)
+	return taskdefinition.PutJSON(ctx, kv, AssignmentKey(cluster, instanceID, rec.TaskID), &msg)
 }
 
 // DescribeTasks returns task records for the named tasks in a cluster.
@@ -285,7 +285,7 @@ func (s *Service) DescribeTasks(ctx context.Context, input *ecs.DescribeTasksInp
 	for _, ref := range awsStringSlice(input.Tasks) {
 		taskID := ContainerInstanceShortID(ref)
 		var rec TaskRecord
-		found, err := getJSON(ctx, kv, TaskKey(cluster, taskID), &rec)
+		found, err := taskdefinition.GetJSON(ctx, kv, TaskKey(cluster, taskID), &rec)
 		if err != nil {
 			return nil, err
 		}
@@ -308,14 +308,14 @@ func (s *Service) ListTasks(ctx context.Context, input *ecs.ListTasksInput, acco
 		return nil, err
 	}
 	var clusterRec ClusterRecord
-	found, err := getJSON(ctx, kv, ClusterMetaKey(cluster), &clusterRec)
+	found, err := taskdefinition.GetJSON(ctx, kv, ClusterMetaKey(cluster), &clusterRec)
 	if err != nil {
 		return nil, err
 	}
 	if !found {
 		return nil, errors.New(awserrors.ErrorECSClusterNotFound)
 	}
-	keys, err := keysWithPrefix(ctx, kv, TasksPrefix(cluster))
+	keys, err := taskdefinition.KeysWithPrefix(ctx, kv, TasksPrefix(cluster))
 	if err != nil {
 		return nil, err
 	}
@@ -323,7 +323,7 @@ func (s *Service) ListTasks(ctx context.Context, input *ecs.ListTasksInput, acco
 	arns := make([]string, 0, len(keys))
 	for _, k := range keys {
 		var rec TaskRecord
-		found, err := getJSON(ctx, kv, k, &rec)
+		found, err := taskdefinition.GetJSON(ctx, kv, k, &rec)
 		if err != nil {
 			return nil, err
 		}

spinifex/domains/ecs/tasks_gc.go

@@ -2,6 +2,7 @@ package ecs
 
 import (
 	"context"
+	"github.com/mulgadc/spinifex/spinifex/domains/ecs/taskdefinition"
 	"log/slog"
 	"strings"
 	"time"
@@ -52,7 +53,7 @@ func (sc *Scheduler) sweepStoppedTasks(ctx context.Context) (time.Duration, erro
 // the ENI release for any that still owe one. A missing StoppedAt or an owed
 // (unreleased) ENI both block deletion; the latter until the release succeeds.
 func (s *Service) sweepStoppedBucket(ctx context.Context, kv jetstream.KeyValue, accountID string, now time.Time, retention time.Duration) (int, time.Duration, error) {
-	keys, err := keysWithPrefix(ctx, kv, "clusters/")
+	keys, err := taskdefinition.KeysWithPrefix(ctx, kv, "clusters/")
 	if err != nil {
 		return 0, 0, err
 	}
@@ -64,7 +65,7 @@ func (s *Service) sweepStoppedBucket(ctx context.Context, kv jetstream.KeyValue,
 			continue
 		}
 		var task TaskRecord
-		found, gerr := getJSON(ctx, kv, k, &task)
+		found, gerr := taskdefinition.GetJSON(ctx, kv, k, &task)
 		if gerr != nil || !found {
 			continue
 		}
@@ -116,7 +117,7 @@ func (s *Service) retryTaskENIRelease(ctx context.Context, kv jetstream.KeyValue
 		task.ENIReleaseAttempts++
 		task.ENIReleaseNextTry = now.Add(eniReleaseBackoff(task.ENIReleaseAttempts))
 	}
-	if perr := putJSON(ctx, kv, key, task); perr != nil {
+	if perr := taskdefinition.PutJSON(ctx, kv, key, task); perr != nil {
 		slog.Warn("ECS sweep: persist ENI retry state failed", "key", key, "err", perr)
 	}
 }

spinifex/domains/ecs/tasks_gc_test.go

@@ -3,6 +3,7 @@ package ecs
 import (
 	"errors"
 	"fmt"
+	"github.com/mulgadc/spinifex/spinifex/domains/ecs/taskdefinition"
 	"testing"
 	"time"
 
@@ -19,7 +20,7 @@ func TestSweepStoppedBucket_PrunesOnlyStale(t *testing.T) {
 			TaskID: id, Cluster: "web", ARN: TaskARN(testRegion, testAccountID, "web", id),
 			LastStatus: status, DesiredStatus: status, StoppedAt: stoppedAt,
 		}
-		require.NoError(t, putJSON(t.Context(), kv, TaskKey("web", id), &rec))
+		require.NoError(t, taskdefinition.PutJSON(t.Context(), kv, TaskKey("web", id), &rec))
 	}
 
 	put("stale", TaskStatusStopped, now.Add(-2*time.Hour))   // older than retention -> prune
@@ -36,7 +37,7 @@ func TestSweepStoppedBucket_PrunesOnlyStale(t *testing.T) {
 
 	exists := func(id string) bool {
 		var rec TaskRecord
-		found, gerr := getJSON(t.Context(), kv, TaskKey("web", id), &rec)
+		found, gerr := taskdefinition.GetJSON(t.Context(), kv, TaskKey("web", id), &rec)
 		require.NoError(t, gerr)
 		return found
 	}
@@ -58,7 +59,7 @@ func TestSweepStoppedBucket_RetriesOwedENI_Succeeds(t *testing.T) {
 		NetworkMode: NetworkModeAwsvpc, ENIID: "eni-1", ENIAttachmentID: "att-1",
 		StoppedAt: now.Add(-2 * time.Hour), // well past retention
 	}
-	require.NoError(t, putJSON(t.Context(), kv, TaskKey("web", "t-1"), &rec))
+	require.NoError(t, taskdefinition.PutJSON(t.Context(), kv, TaskKey("web", "t-1"), &rec))
 
 	pruned, due, err := svc.sweepStoppedBucket(t.Context(), kv, testAccountID, now, time.Hour)
 	require.NoError(t, err)
@@ -67,7 +68,7 @@ func TestSweepStoppedBucket_RetriesOwedENI_Succeeds(t *testing.T) {
 	assert.Equal(t, 1, eni.releaseCalls)
 
 	var got TaskRecord
-	found, gerr := getJSON(t.Context(), kv, TaskKey("web", "t-1"), &got)
+	found, gerr := taskdefinition.GetJSON(t.Context(), kv, TaskKey("web", "t-1"), &got)
 	require.NoError(t, gerr)
 	require.True(t, found)
 	assert.Equal(t, "eni-1", got.ENIID, "the identity survives release as the forensic record")
@@ -88,7 +89,7 @@ func TestSweepStoppedBucket_RetryFails_BacksOff(t *testing.T) {
 		NetworkMode: NetworkModeAwsvpc, ENIID: "eni-1",
 		StoppedAt: now.Add(-2 * time.Hour),
 	}
-	require.NoError(t, putJSON(t.Context(), kv, TaskKey("web", "t-1"), &rec))
+	require.NoError(t, taskdefinition.PutJSON(t.Context(), kv, TaskKey("web", "t-1"), &rec))
 
 	pruned, due, err := svc.sweepStoppedBucket(t.Context(), kv, testAccountID, now, time.Hour)
 	require.NoError(t, err)
@@ -96,7 +97,7 @@ func TestSweepStoppedBucket_RetryFails_BacksOff(t *testing.T) {
 	assert.Equal(t, eniReleaseBackoffBase, due)
 
 	var got TaskRecord
-	found, gerr := getJSON(t.Context(), kv, TaskKey("web", "t-1"), &got)
+	found, gerr := taskdefinition.GetJSON(t.Context(), kv, TaskKey("web", "t-1"), &got)
 	require.NoError(t, gerr)
 	require.True(t, found)
 	assert.Equal(t, "eni-1", got.ENIID, "a failed retry is still owed")
@@ -118,7 +119,7 @@ func TestSweepStoppedBucket_RetryNotDueYet_Skipped(t *testing.T) {
 		StoppedAt:         now.Add(-2 * time.Hour),
 		ENIReleaseNextTry: now.Add(10 * time.Minute),
 	}
-	require.NoError(t, putJSON(t.Context(), kv, TaskKey("web", "t-1"), &rec))
+	require.NoError(t, taskdefinition.PutJSON(t.Context(), kv, TaskKey("web", "t-1"), &rec))
 
 	pruned, due, err := svc.sweepStoppedBucket(t.Context(), kv, testAccountID, now, time.Hour)
 	require.NoError(t, err)
@@ -142,7 +143,7 @@ func TestSweepStoppedBucket_HonoursPerPassCap(t *testing.T) {
 			NetworkMode: NetworkModeAwsvpc, ENIID: "eni-" + id,
 			StoppedAt: now.Add(-2 * time.Hour),
 		}
-		require.NoError(t, putJSON(t.Context(), kv, TaskKey("web", id), &rec))
+		require.NoError(t, taskdefinition.PutJSON(t.Context(), kv, TaskKey("web", id), &rec))
 	}
 
 	_, due, err := svc.sweepStoppedBucket(t.Context(), kv, testAccountID, now, time.Hour)
@@ -173,7 +174,7 @@ func TestSweepStoppedBucket_CappedOverflow_AsksForSweepInterval(t *testing.T) {
 			NetworkMode: NetworkModeAwsvpc, ENIID: "eni-" + id,
 			StoppedAt: now.Add(-2 * time.Hour),
 		}
-		require.NoError(t, putJSON(t.Context(), kv, TaskKey("web", id), &rec))
+		require.NoError(t, taskdefinition.PutJSON(t.Context(), kv, TaskKey("web", id), &rec))
 	}
 
 	_, due, err := svc.sweepStoppedBucket(t.Context(), kv, testAccountID, now, time.Hour)
@@ -196,7 +197,7 @@ func TestSweepStoppedBucket_PrunesReleased_KeepsOwed(t *testing.T) {
 		NetworkMode: NetworkModeAwsvpc, ENIID: "eni-released", ENIReleased: true,
 		StoppedAt: now.Add(-2 * time.Hour), // well past retention
 	}
-	require.NoError(t, putJSON(t.Context(), kv, TaskKey("web", "released"), &released))
+	require.NoError(t, taskdefinition.PutJSON(t.Context(), kv, TaskKey("web", "released"), &released))
 
 	owed := TaskRecord{
 		TaskID: "owed", Cluster: "web", ARN: TaskARN(testRegion, testAccountID, "web", "owed"),
@@ -204,7 +205,7 @@ func TestSweepStoppedBucket_PrunesReleased_KeepsOwed(t *testing.T) {
 		NetworkMode: NetworkModeAwsvpc, ENIID: "eni-owed", ENIReleased: false,
 		StoppedAt: now.Add(-2 * time.Hour), // also well past retention, but still owed
 	}
-	require.NoError(t, putJSON(t.Context(), kv, TaskKey("web", "owed"), &owed))
+	require.NoError(t, taskdefinition.PutJSON(t.Context(), kv, TaskKey("web", "owed"), &owed))
 
 	pruned, _, err := svc.sweepStoppedBucket(t.Context(), kv, testAccountID, now, time.Hour)
 	require.NoError(t, err)
@@ -212,7 +213,7 @@ func TestSweepStoppedBucket_PrunesReleased_KeepsOwed(t *testing.T) {
 
 	exists := func(id string) bool {
 		var rec TaskRecord
-		found, gerr := getJSON(t.Context(), kv, TaskKey("web", id), &rec)
+		found, gerr := taskdefinition.GetJSON(t.Context(), kv, TaskKey("web", id), &rec)
 		require.NoError(t, gerr)
 		return found
 	}

spinifex/domains/eks/awsapi/registration.go

@@ -3,11 +3,9 @@ package awsapi
 import (
 	"encoding/json"
 	"errors"
-	"io"
 	"log/slog"
 	"net/http"
 
-	"github.com/mulgadc/bluebottle/pkg/sigv4"
 	"github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
 	"github.com/mulgadc/spinifex/spinifex/foundation/aws/passrole"
 	"github.com/mulgadc/spinifex/spinifex/ingress/aws/dispatch"
@@ -64,17 +62,6 @@ func CallerFromActor(actor dispatch.Actor) Caller {
 	}
 }
 
-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
-}
-
 // dispatch runs the action and renames the generic parameter error to the code
 // EKS declares. The internal routes keep theirs: the CP-VM agents match on it.
 func (d Deps) dispatch(w http.ResponseWriter, inv dispatch.Invocation) error {
@@ -139,7 +126,7 @@ func (d Deps) serve(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(), "EKS dispatch: failed to read body", "err", err)
 		return err

spinifex/domains/eks/cluster/cluster_reconciler.go

@@ -795,8 +795,7 @@ func retryableStartRefusal(err error) bool {
 	if code, ok := awserrors.ResolveErrorCode(err); ok && code == awserrors.ErrorInsufficientInstanceCapacity {
 		return true
 	}
-	_, message, _ := awserrors.ResolveErrorDetail(err)
-	return strings.Contains(message, ec2v1.VolumeHeldElsewhereText) || strings.Contains(err.Error(), ec2v1.VolumeHeldElsewhereText)
+	return errors.Is(err, ec2v1.ErrVolumeHeldElsewhere)
 }
 
 // maybeReformEtcdQuorum escalates a wedged HA control plane that the in-place

spinifex/domains/eks/cluster/cluster_reconciler_start_refusal_test.go

@@ -3,9 +3,11 @@ package cluster
 import (
 	"context"
 	"errors"
+	"fmt"
 	"testing"
 	"time"
 
+	"github.com/mulgadc/spinifex/contracts/ec2/v1"
 	"github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
@@ -30,12 +32,12 @@ func (c *refusingCPControl) StartInstance(context.Context, string) error {
 
 func (c *refusingCPControl) StopInstance(context.Context, string) error { return nil }
 
-// What the start path returns after crossing NATS: the code and the message the
-// owning node sent, with no sentinel left to match on.
+// What the start path returns after crossing NATS: StartInstances wraps the
+// sentinel back on when the owner's reply carries the held-elsewhere header.
 var (
 	errStartCapacity = errors.New(awserrors.ErrorInsufficientInstanceCapacity)
-	errStartHeld     = awserrors.Errorf(awserrors.ErrorIncorrectState, "%s",
-		"volume is leased by another owner: held by node-a since 2026-10-30T01:00:00Z")
+	errStartHeld     = fmt.Errorf("%w: %w", ec2v1.ErrVolumeHeldElsewhere,
+		awserrors.Errorf(awserrors.ErrorIncorrectState, "%s", "volume vol-1 is published to node-a"))
 )
 
 // A refusal that clears by itself must not use up the restart budget, or the
@@ -72,18 +74,30 @@ func TestMaybeRecoverCP_RetryableStartRefusalIsRetriedNotReplaced(t *testing.T)
 	}
 }
 
-// The cap still bounds the faults nothing says will clear.
+// The cap still bounds the faults nothing says will clear. IncorrectState
+// without the held-elsewhere sentinel is one, whatever its message says.
 func TestMaybeRecoverCP_UnknownStartFailureStillCountsAgainstTheBudget(t *testing.T) {
 	t.Parallel()
-	cp := &refusingCPControl{err: errors.New(awserrors.ErrorServerInternal)}
-	r := newRecoveryReconciler(t, cp)
-	r.degradedSince = time.Now().Add(-10 * time.Minute)
+	for name, refusal := range map[string]error{
+		"server internal": errors.New(awserrors.ErrorServerInternal),
+		"unrelated incorrect state": awserrors.Errorf(awserrors.ErrorIncorrectState, "%s",
+			"instance i-cp is not in a state from which it can be started"),
+		"lease text without the sentinel": awserrors.Errorf(awserrors.ErrorIncorrectState, "%s",
+			ec2v1.VolumeHeldElsewhereText+": held by node-a since 2026-10-30T01:00:00Z"),
+	} {
+		t.Run(name, func(t *testing.T) {
+			t.Parallel()
+			cp := &refusingCPControl{err: refusal}
+			r := newRecoveryReconciler(t, cp)
+			r.degradedSince = time.Now().Add(-10 * time.Minute)
 
-	for range r.maxRestartAttempts + 2 {
-		r.lastRestartAt = time.Time{}
-		r.maybeRecoverControlPlane(t.Context(), metaWithCP("i-cp"), "stale")
+			for range r.maxRestartAttempts + 2 {
+				r.lastRestartAt = time.Time{}
+				r.maybeRecoverControlPlane(t.Context(), metaWithCP("i-cp"), "stale")
+			}
+
+			require.Equal(t, r.maxRestartAttempts, cp.startCalls, "stops at the attempt cap")
+			assert.Equal(t, r.maxRestartAttempts, r.restartAttempts)
+		})
 	}
-
-	require.Equal(t, r.maxRestartAttempts, cp.startCalls, "stops at the attempt cap")
-	assert.Equal(t, r.maxRestartAttempts, r.restartAttempts)
 }

spinifex/domains/elbv2/awsapi/registration.go

@@ -1,17 +1,14 @@
 package awsapi
 
 import (
-	"bytes"
 	"context"
 	"errors"
-	"io"
 	"log/slog"
 	"maps"
 	"net/http"
 	"slices"
 
 	"github.com/aws/aws-sdk-go/service/elbv2"
-	"github.com/mulgadc/bluebottle/pkg/sigv4"
 	elbv2domain "github.com/mulgadc/spinifex/spinifex/domains/elbv2"
 	"github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
 	awsxml "github.com/mulgadc/spinifex/spinifex/foundation/aws/xml"
@@ -217,7 +214,7 @@ func NewRegistration(deps Deps) dispatch.Registration {
 	return dispatch.Registration{
 		Service:       ServiceName,
 		InventoryName: InventoryName,
-		ResolveAction: resolveAction,
+		ResolveAction: dispatch.ResolveQueryAction,
 		Dispatch:      deps.dispatch,
 		Errors:        dispatch.ErrorEnvelopeXML,
 		Inventory:     OperationInventory(),
@@ -235,35 +232,11 @@ func OperationInventory() dispatch.Inventory {
 	return dispatch.Inventory{Registered: ActionNames()}
 }
 
-// resolveAction is the one Query Action parser telemetry and dispatch share.
-// It leaves the body readable for the dispatcher and answers "" on any failure.
-func resolveAction(r *http.Request) string {
-	args, err := readQueryArgs(r)
-	if err != nil {
-		return ""
-	}
-	return args["Action"]
-}
-
-// readQueryArgs parses the Query body and rewinds it, so the same bytes can be
-// read again. Any read or parse failure is a malformed query string.
-func readQueryArgs(r *http.Request) (map[string]string, error) {
-	body, err := io.ReadAll(io.LimitReader(r.Body, sigv4.MaxPayloadLen+1))
-	if err != nil {
-		return nil, err
-	}
-	r.Body = io.NopCloser(bytes.NewReader(body))
-	if int64(len(body)) > sigv4.MaxPayloadLen {
-		return nil, errors.New(awserrors.ErrorRequestEntityTooLarge)
-	}
-	return query.ParseArgs(string(body))
-}
-
 // dispatch parses, scopes and authorizes the request from one parse 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
-	queryArgs, err := readQueryArgs(r)
+	queryArgs, err := dispatch.ReadArgs(r)
 	if err != nil {
 		slog.DebugContext(r.Context(), "ELBv2: malformed query string", "err", err)
 		return errors.New(awserrors.ErrorMalformedQueryString)

spinifex/domains/elbv2/awsapi/registration_test.go

@@ -3,10 +3,6 @@ package awsapi
 import (
 	"context"
 	"encoding/xml"
-	"io"
-	"net/http"
-	"net/http/httptest"
-	"strings"
 	"testing"
 
 	"github.com/aws/aws-sdk-go/service/elbv2"
@@ -31,22 +27,6 @@ func TestNewRegistration_RegistersUnderCredentialScopeAndInventoryName(t *testin
 	assert.Empty(t, inv.Unsupported)
 }
 
-func TestResolveAction_ReadsActionAndRewindsBody(t *testing.T) {
-	body := "Action=DescribeLoadBalancers&Version=2015-12-01"
-	r := httptest.NewRequest(http.MethodPost, "/", strings.NewReader(body))
-
-	assert.Equal(t, "DescribeLoadBalancers", resolveAction(r))
-
-	again, err := io.ReadAll(r.Body)
-	require.NoError(t, err)
-	assert.Equal(t, body, string(again), "the dispatcher must still see the whole body")
-}
-
-func TestResolveAction_UnparseableBodyIsEmpty(t *testing.T) {
-	r := httptest.NewRequest(http.MethodPost, "/", strings.NewReader("Action=%zz"))
-	assert.Empty(t, resolveAction(r))
-}
-
 // A registration with no quota reports the AWS default cap and enforces none.
 func TestDescribeAccountLimits_NoQuotaReportsDefault(t *testing.T) {
 	out, err := actions["DescribeAccountLimits"].dispatch(t.Context(), "DescribeAccountLimits",

spinifex/domains/elbv2/service_impl.go

@@ -3293,66 +3293,12 @@ func (s *ELBv2ServiceImpl) DescribeTags(ctx context.Context, input *elbv2.Descri
 		}
 		arn := *arnPtr
 
-		resourceType, err := elbv2ResourceTypeFromArn(arn)
+		h, notFoundError, found, err := s.resolveTaggable(ctx, arn)
 		if err != nil {
 			return nil, err
 		}
-
-		var (
-			tags          map[string]string
-			ownerAccount  string
-			notFoundError string
-			found         bool
-		)
-		switch resourceType {
-		case resourcearn.ELBv2LoadBalancer:
-			notFoundError = awserrors.ErrorELBv2LoadBalancerNotFound
-			lb, lbErr := s.store.GetLoadBalancerByArn(ctx, arn)
-			if lbErr != nil {
-				slog.ErrorContext(ctx, "DescribeTags: failed to get LB", "arn", arn, "err", lbErr)
-				return nil, errors.New(awserrors.ErrorServerInternal)
-			}
-			if lb != nil {
-				found = true
-				tags = lb.Tags
-				ownerAccount = lb.AccountID
-			}
-		case resourcearn.ELBv2TargetGroup:
-			notFoundError = awserrors.ErrorELBv2TargetGroupNotFound
-			tg, tgErr := s.store.GetTargetGroupByArn(ctx, arn)
-			if tgErr != nil {
-				slog.ErrorContext(ctx, "DescribeTags: failed to get target group", "arn", arn, "err", tgErr)
-				return nil, errors.New(awserrors.ErrorServerInternal)
-			}
-			if tg != nil {
-				found = true
-				tags = tg.Tags
-				ownerAccount = tg.AccountID
-			}
-		case resourcearn.ELBv2Listener:
-			notFoundError = awserrors.ErrorELBv2ListenerNotFound
-			l, lErr := s.store.GetListenerByArn(ctx, arn)
-			if lErr != nil {
-				slog.ErrorContext(ctx, "DescribeTags: failed to get listener", "arn", arn, "err", lErr)
-				return nil, errors.New(awserrors.ErrorServerInternal)
-			}
-			if l != nil {
-				found = true
-				tags = l.Tags
-				ownerAccount = l.AccountID
-			}
-		case resourcearn.ELBv2ListenerRule:
-			notFoundError = awserrors.ErrorELBv2RuleNotFound
-			r, rErr := s.store.GetRuleByArn(ctx, arn)
-			if rErr != nil {
-				slog.ErrorContext(ctx, "DescribeTags: failed to get rule", "arn", arn, "err", rErr)
-				return nil, errors.New(awserrors.ErrorServerInternal)
-			}
-			if r != nil {
-				found = true
-				tags = r.Tags
-				ownerAccount = r.AccountID
-			} else if lArn, isDefault := listenerArnFromDefaultRuleArn(arn); isDefault {
+		if !found {
+			if lArn, isDefault := listenerArnFromDefaultRuleArn(arn); isDefault {
 				// Synthetic default rule: not stored, carries no tags. Resolve via
 				// its parent listener so a controller's post-create rule-tag sync
 				// gets an empty TagDescription instead of an error.
@@ -3363,19 +3309,18 @@ func (s *ELBv2ServiceImpl) DescribeTags(ctx context.Context, input *elbv2.Descri
 				}
 				if l != nil {
 					found = true
-					tags = nil
-					ownerAccount = l.AccountID
+					h = taggable{owner: l.AccountID}
 				}
 			}
 		}
 
-		if !found || ownerAccount != accountID {
+		if !found || h.owner != accountID {
 			return nil, errors.New(notFoundError)
 		}
 
 		descriptions = append(descriptions, &elbv2.TagDescription{
 			ResourceArn: aws.String(arn),
-			Tags:        tagsMapToSDK(tags),
+			Tags:        tagsMapToSDK(h.tags),
 		})
 	}
 

spinifex/domains/elbv2/store.go

@@ -578,22 +578,6 @@ func (s *Store) PutRule(ctx context.Context, r *RuleRecord) error {
 	return err
 }
 
-// GetRule retrieves a rule by its short ID.
-func (s *Store) GetRule(ctx context.Context, ruleID string) (*RuleRecord, error) {
-	entry, err := s.kv.Get(ctx, KeyPrefixRule+ruleID)
-	if err != nil {
-		if errors.Is(err, jetstream.ErrKeyNotFound) {
-			return nil, nil
-		}
-		return nil, err
-	}
-	var r RuleRecord
-	if err := json.Unmarshal(entry.Value(), &r); err != nil {
-		return nil, fmt.Errorf("unmarshal rule: %w", err)
-	}
-	return &r, nil
-}
-
 // DeleteRule removes a rule by its short ID.
 func (s *Store) DeleteRule(ctx context.Context, ruleID string) error {
 	err := s.kv.Delete(ctx, KeyPrefixRule+ruleID)

spinifex/domains/iam/awsapi/registration.go

@@ -1,10 +1,8 @@
 package awsapi
 
 import (
-	"bytes"
 	"context"
 	"errors"
-	"io"
 	"log/slog"
 	"maps"
 	"net/http"
@@ -13,7 +11,6 @@ import (
 	"github.com/aws/aws-sdk-go/aws"
 	"github.com/aws/aws-sdk-go/service/iam"
 	"github.com/mulgadc/bluebottle/pkg/iampolicy"
-	"github.com/mulgadc/bluebottle/pkg/sigv4"
 	iamdomain "github.com/mulgadc/spinifex/spinifex/domains/iam"
 	"github.com/mulgadc/spinifex/spinifex/foundation/aws/arn"
 	"github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
@@ -376,7 +373,7 @@ var getRootUser = handler(func(accountID string, _ *iam.GetUserInput, d Deps) (a
 func NewRegistration(deps Deps) dispatch.Registration {
 	return dispatch.Registration{
 		Service:       ServiceName,
-		ResolveAction: resolveAction,
+		ResolveAction: dispatch.ResolveQueryAction,
 		Dispatch:      deps.dispatch,
 		Errors:        dispatch.ErrorEnvelopeXML,
 		Traits:        dispatch.ServiceTraits{XMLThrottleCode: awserrors.ErrorThrottling},
@@ -395,33 +392,11 @@ func OperationInventory() dispatch.Inventory {
 	return dispatch.Inventory{Registered: ActionNames()}
 }
 
-// resolveAction is the one Query Action parser telemetry and dispatch share.
-// It leaves the body readable for the dispatcher and answers "" on any failure.
-func resolveAction(r *http.Request) string {
-	args, err := readQueryArgs(r)
-	if err != nil {
-		return ""
-	}
-	return args["Action"]
-}
-
-func readQueryArgs(r *http.Request) (map[string]string, error) {
-	body, err := io.ReadAll(io.LimitReader(r.Body, sigv4.MaxPayloadLen+1))
-	if err != nil {
-		return nil, err
-	}
-	r.Body = io.NopCloser(bytes.NewReader(body))
-	if int64(len(body)) > sigv4.MaxPayloadLen {
-		return nil, errors.New(awserrors.ErrorRequestEntityTooLarge)
-	}
-	return query.ParseArgs(string(body))
-}
-
 // dispatch parses, scopes and authorizes the request from one parse 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
-	queryArgs, err := readQueryArgs(r)
+	queryArgs, err := dispatch.ReadArgs(r)
 	if err != nil {
 		slog.DebugContext(r.Context(), "IAM: malformed query string", "err", err)
 		return errors.New(awserrors.ErrorMalformedQueryString)

spinifex/domains/network/external/backoff/backoff.go

@@ -0,0 +1,33 @@
+// Package backoff holds the jitter and context-aware sleep shared by the
+// external-network allocators' retry and poll loops.
+package backoff
+
+import (
+	"context"
+	"math/rand/v2"
+	"time"
+)
+
+// Jitter returns a uniform offset in [-span, span), or 0 when span <= 0.
+func Jitter(span time.Duration) time.Duration {
+	if span <= 0 {
+		return 0
+	}
+	return time.Duration(rand.Int64N(int64(2*span))) - span //nolint:gosec // jitter, not a secret
+}
+
+// Sleep waits for d or until ctx is done, returning ctx.Err() in the latter
+// case. A non-positive d still reports an already-cancelled ctx.
+func Sleep(ctx context.Context, d time.Duration) error {
+	if d <= 0 {
+		return ctx.Err()
+	}
+	t := time.NewTimer(d)
+	defer t.Stop()
+	select {
+	case <-ctx.Done():
+		return ctx.Err()
+	case <-t.C:
+		return nil
+	}
+}

spinifex/domains/network/external/backoff/backoff_test.go

@@ -0,0 +1,30 @@
+package backoff
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func TestJitter_StaysWithinSpan(t *testing.T) {
+	assert.Zero(t, Jitter(0))
+	assert.Zero(t, Jitter(-time.Second))
+	for range 1000 {
+		j := Jitter(10 * time.Millisecond)
+		require.GreaterOrEqual(t, j, -10*time.Millisecond)
+		require.Less(t, j, 10*time.Millisecond)
+	}
+}
+
+func TestSleep_ReportsCancellation(t *testing.T) {
+	ctx, cancel := context.WithCancel(context.Background())
+	cancel()
+	require.ErrorIs(t, Sleep(ctx, 0), context.Canceled, "a zero wait must still see a cancelled ctx")
+	require.ErrorIs(t, Sleep(ctx, time.Hour), context.Canceled)
+
+	require.NoError(t, Sleep(context.Background(), 0))
+	require.NoError(t, Sleep(context.Background(), time.Millisecond))
+}

spinifex/domains/network/external/dhcp/kv.go

@@ -232,35 +232,8 @@ func fromWireEntry(w wireEntry) (Entry, error) {
 		RawOffer:    w.RawOffer,
 		RawACK:      w.RawACK,
 	}
-	if w.IP != "" {
-		lease.IP = net.ParseIP(w.IP)
-	}
-	if w.SubnetMask != "" {
-		mask, err := parseMask(w.SubnetMask)
-		if err != nil {
-			return Entry{}, fmt.Errorf("parse subnet_mask %q: %w", w.SubnetMask, err)
-		}
-		lease.SubnetMask = mask
-	}
-	for _, s := range w.Routers {
-		if ip := net.ParseIP(s); ip != nil {
-			lease.Routers = append(lease.Routers, ip)
-		}
-	}
-	for _, s := range w.DNS {
-		if ip := net.ParseIP(s); ip != nil {
-			lease.DNS = append(lease.DNS, ip)
-		}
-	}
-	if w.ServerID != "" {
-		lease.ServerID = net.ParseIP(w.ServerID)
-	}
-	if w.HWAddr != "" {
-		hw, err := net.ParseMAC(w.HWAddr)
-		if err != nil {
-			return Entry{}, fmt.Errorf("parse hwaddr %q: %w", w.HWAddr, err)
-		}
-		lease.HWAddr = hw
+	if err := decodeLeaseAddrs(lease, w.IP, w.SubnetMask, w.Routers, w.DNS, w.ServerID, w.HWAddr); err != nil {
+		return Entry{}, err
 	}
 	if !w.Expiry.IsZero() {
 		lease.LeaseDuration = w.Expiry.Sub(w.Acquired)

spinifex/domains/network/external/dhcp/lease_addrs_internal_test.go

@@ -0,0 +1,59 @@
+package dhcp
+
+import (
+	"net"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func TestWireAndKVDecodeSameLeaseAddrs(t *testing.T) {
+	wl := &wireLease{
+		IP:         "10.0.0.5",
+		SubnetMask: "255.255.255.0",
+		Routers:    []string{"10.0.0.1", "bogus"},
+		DNS:        []string{"1.1.1.1", "8.8.8.8"},
+		ServerID:   "10.0.0.2",
+		HWAddr:     "02:00:00:00:00:01",
+	}
+	we := wireEntry{
+		ClientID:   "c",
+		IP:         wl.IP,
+		SubnetMask: wl.SubnetMask,
+		Routers:    wl.Routers,
+		DNS:        wl.DNS,
+		ServerID:   wl.ServerID,
+		HWAddr:     wl.HWAddr,
+	}
+
+	fromWire, err := fromWireLease(wl)
+	require.NoError(t, err)
+	fromKV, err := fromWireEntry(we)
+	require.NoError(t, err)
+
+	for _, l := range []*Lease{fromWire, fromKV.Lease} {
+		assert.True(t, l.IP.Equal(net.ParseIP("10.0.0.5")))
+		assert.Equal(t, net.CIDRMask(24, 32), l.SubnetMask)
+		require.Len(t, l.Routers, 1, "unparseable router is dropped")
+		assert.True(t, l.Routers[0].Equal(net.ParseIP("10.0.0.1")))
+		assert.Len(t, l.DNS, 2)
+		assert.True(t, l.ServerID.Equal(net.ParseIP("10.0.0.2")))
+		assert.Equal(t, "02:00:00:00:00:01", l.HWAddr.String())
+	}
+}
+
+func TestWireAndKVRejectSameBadLeaseAddrs(t *testing.T) {
+	cases := map[string]struct{ mask, hw, want string }{
+		"bad mask":   {mask: "not-a-mask", want: `parse subnet_mask "not-a-mask"`},
+		"bad hwaddr": {hw: "zz", want: `parse hwaddr "zz"`},
+	}
+	for name, tc := range cases {
+		t.Run(name, func(t *testing.T) {
+			_, err := fromWireLease(&wireLease{SubnetMask: tc.mask, HWAddr: tc.hw})
+			require.ErrorContains(t, err, tc.want)
+			_, err = fromWireEntry(wireEntry{SubnetMask: tc.mask, HWAddr: tc.hw})
+			require.ErrorContains(t, err, tc.want)
+		})
+	}
+}

spinifex/domains/network/external/dhcp/manager.go

@@ -6,12 +6,12 @@ import (
 	"errors"
 	"fmt"
 	"log/slog"
-	"math/rand/v2"
 	"net"
 	"os"
 	"sync"
 	"time"
 
+	"github.com/mulgadc/spinifex/spinifex/domains/network/external/backoff"
 	"github.com/mulgadc/spinifex/spinifex/foundation/telemetry"
 
 	"github.com/nats-io/nats.go"
@@ -395,7 +395,7 @@ func (m *Manager) run(ctx context.Context, self *leaseLoop, e Entry, reaffirm bo
 					return
 				}
 				slog.Warn("dhcp manager: re-DORA after expiry failed; backing off", "client_id", e.Lease.ClientID, "err", err)
-				if !sleepWithCtx(ctx, postExpiryBackoff) {
+				if backoff.Sleep(ctx, postExpiryBackoff) != nil {
 					return
 				}
 			}
@@ -515,7 +515,7 @@ func (m *Manager) acquireWithBackoff(ctx context.Context, purpose string, req Ac
 		if remaining <= 0 {
 			break
 		}
-		attempt := base + jitter(acquireAttemptJitter)
+		attempt := base + backoff.Jitter(acquireAttemptJitter)
 		if attempt <= 0 {
 			attempt = base
 		}
@@ -882,38 +882,12 @@ func decodeHWAddr(s string) (net.HardwareAddr, error) {
 func sleepUntil(ctx context.Context, now func() time.Time, deadline time.Time) bool {
 	d := deadline.Sub(now())
 	if d > 0 {
-		d += jitter(renewJitter)
+		d += backoff.Jitter(renewJitter)
 		if d < 0 {
 			d = 0
 		}
 	} else {
 		d = 0
 	}
-	return sleepWithCtx(ctx, d)
-}
-
-func sleepWithCtx(ctx context.Context, d time.Duration) bool {
-	if d <= 0 {
-		select {
-		case <-ctx.Done():
-			return false
-		default:
-			return true
-		}
-	}
-	t := time.NewTimer(d)
-	defer t.Stop()
-	select {
-	case <-ctx.Done():
-		return false
-	case <-t.C:
-		return true
-	}
-}
-
-func jitter(span time.Duration) time.Duration {
-	if span <= 0 {
-		return 0
-	}
-	return time.Duration(rand.Int64N(int64(span*2))) - span //nolint:gosec // jitter, not cryptographic
+	return backoff.Sleep(ctx, d) == nil
 }

spinifex/domains/network/external/dhcp/manager_test.go

@@ -435,7 +435,7 @@ func TestNATSClientAcquireReleaseRoundtrip(t *testing.T) {
 	assert.Equal(t, "eip", got.Purpose)
 	assert.Equal(t, "default", got.PoolName)
 
-	require.NoError(t, client.RequestRelease(context.Background(), "eipalloc-A"))
+	require.NoError(t, client.RequestReleaseByIP(context.Background(), "default", lease.IP.String(), false))
 	_, err = store.Get(t.Context(), "eipalloc-A")
 	assert.ErrorIs(t, err, jetstream.ErrKeyNotFound)
 	assert.Equal(t, 1, fake.ReleaseCount())
@@ -676,8 +676,9 @@ func TestManagerReleaseUnknownClientIsNoop(t *testing.T) {
 			_ = s.Unsubscribe()
 		}
 	})
-	client := dhcp.NewNATSClient(nc, time.Second)
-	assert.NoError(t, client.RequestRelease(context.Background(), "never-existed"))
+	msg, err := nc.Request(dhcp.TopicRelease, []byte(`{"client_id":"never-existed"}`), time.Second)
+	require.NoError(t, err)
+	assert.JSONEq(t, `{}`, string(msg.Data))
 	assert.Equal(t, 0, fake.ReleaseCount())
 }
 

spinifex/domains/network/external/dhcp/nats_client.go

@@ -89,15 +89,6 @@ func (c *NATSClient) RequestAcquire(ctx context.Context, p AcquireParams) (*Leas
 	return fromWireLease(reply.Lease)
 }
 
-// RequestRelease issues an idempotent release RPC. Unknown client-ids
-// return nil (vpcd treats them as already-released).
-func (c *NATSClient) RequestRelease(ctx context.Context, clientID string) error {
-	if clientID == "" {
-		return errors.New("dhcp NATSClient: clientID required")
-	}
-	return c.requestRelease(ctx, releaseWireRequest{ClientID: clientID})
-}
-
 // RequestReleaseByIP releases the lease that holds (poolName, ip).
 // Used by DHCPPoolAllocator.Release, whose external.Allocator signature
 // only carries the IP — vpcd looks up the client-id from the lease store.

spinifex/domains/network/external/dhcp/wire.go

@@ -199,35 +199,44 @@ func fromWireLease(w *wireLease) (*Lease, error) {
 		RawOffer:      w.RawOffer,
 		RawACK:        w.RawACK,
 	}
-	if w.IP != "" {
-		l.IP = net.ParseIP(w.IP)
-	}
-	if w.SubnetMask != "" {
-		m, err := parseMask(w.SubnetMask)
-		if err != nil {
-			return nil, fmt.Errorf("subnet_mask: %w", err)
-		}
-		l.SubnetMask = m
-	}
-	for _, s := range w.Routers {
-		if ip := net.ParseIP(s); ip != nil {
-			l.Routers = append(l.Routers, ip)
-		}
-	}
-	for _, s := range w.DNS {
-		if ip := net.ParseIP(s); ip != nil {
-			l.DNS = append(l.DNS, ip)
-		}
-	}
-	if w.ServerID != "" {
-		l.ServerID = net.ParseIP(w.ServerID)
-	}
-	if w.HWAddr != "" {
-		hw, err := net.ParseMAC(w.HWAddr)
-		if err != nil {
-			return nil, fmt.Errorf("hwaddr: %w", err)
-		}
-		l.HWAddr = hw
+	if err := decodeLeaseAddrs(l, w.IP, w.SubnetMask, w.Routers, w.DNS, w.ServerID, w.HWAddr); err != nil {
+		return nil, err
 	}
 	return l, nil
 }
+
+// decodeLeaseAddrs parses the address fields shared by the wire and KV lease
+// formats into l. Unparseable routers and DNS servers are dropped.
+func decodeLeaseAddrs(l *Lease, ip, mask string, routers, dns []string, serverID, hw string) error {
+	if ip != "" {
+		l.IP = net.ParseIP(ip)
+	}
+	if mask != "" {
+		m, err := parseMask(mask)
+		if err != nil {
+			return fmt.Errorf("parse subnet_mask %q: %w", mask, err)
+		}
+		l.SubnetMask = m
+	}
+	for _, s := range routers {
+		if r := net.ParseIP(s); r != nil {
+			l.Routers = append(l.Routers, r)
+		}
+	}
+	for _, s := range dns {
+		if d := net.ParseIP(s); d != nil {
+			l.DNS = append(l.DNS, d)
+		}
+	}
+	if serverID != "" {
+		l.ServerID = net.ParseIP(serverID)
+	}
+	if hw != "" {
+		mac, err := net.ParseMAC(hw)
+		if err != nil {
+			return fmt.Errorf("parse hwaddr %q: %w", hw, err)
+		}
+		l.HWAddr = mac
+	}
+	return nil
+}

spinifex/domains/network/external/ocinet/allocator.go

@@ -27,12 +27,12 @@ import (
 	"errors"
 	"fmt"
 	"log/slog"
-	"math/rand/v2"
 	"net/netip"
 	"sync"
 	"time"
 
 	"github.com/mulgadc/spinifex/spinifex/domains/network/external"
+	"github.com/mulgadc/spinifex/spinifex/domains/network/external/backoff"
 	"github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
 	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvstore"
 	"github.com/mulgadc/spinifex/spinifex/providers/cloud/oci"
@@ -203,7 +203,7 @@ func New(client oci.Client, store Store, cfg Config) (*PoolAllocator, error) {
 		a.budget = assignBudget
 	}
 	if a.sleep == nil {
-		a.sleep = sleepCtx
+		a.sleep = backoff.Sleep
 	}
 	if a.now == nil {
 		a.now = time.Now
@@ -410,7 +410,7 @@ func (a *PoolAllocator) waitAssigned(ctx context.Context, pub oci.PublicIP) (oci
 	deadline := time.Now().Add(a.budget)
 	for i := 0; ; i++ {
 		wait := a.schedule[min(i, len(a.schedule)-1)]
-		if j := jitter(assignJitter); wait+j > 0 {
+		if j := backoff.Jitter(assignJitter); wait+j > 0 {
 			wait += j
 		}
 		if time.Now().Add(wait).After(deadline) {
@@ -501,26 +501,3 @@ func displayName(req external.AllocateRequest) string {
 		return DisplayNamePrefix + "unclaimed"
 	}
 }
-
-func jitter(span time.Duration) time.Duration {
-	if span <= 0 {
-		return 0
-	}
-	// Spreading the poll ladder, not generating a secret: math/rand is the
-	// right tool and crypto/rand would be cargo cult here.
-	return time.Duration(rand.Int64N(int64(2*span))) - span //nolint:gosec // jitter, not a secret
-}
-
-func sleepCtx(ctx context.Context, d time.Duration) error {
-	if d <= 0 {
-		return nil
-	}
-	t := time.NewTimer(d)
-	defer t.Stop()
-	select {
-	case <-ctx.Done():
-		return ctx.Err()
-	case <-t.C:
-		return nil
-	}
-}

spinifex/domains/network/external/ocinet/store.go

@@ -47,11 +47,6 @@ func NewKVStore(js jetstream.JetStream) *KVStore {
 	return &KVStore{store: kvstore.New[Record](js, bucketConfig())}
 }
 
-// NewKVStoreOver builds a store over an already-open bucket.
-func NewKVStoreOver(kv jetstream.KeyValue) *KVStore {
-	return &KVStore{store: kvstore.Over[Record](nil, kv, bucketConfig())}
-}
-
 // Mutate implements Store. Upsert rather than Mutate: the first allocation on
 // a pool is what creates its record, and Mutate reports ErrNotFound for an
 // absent key — so every pool would fail its own first allocation, after OCI

spinifex/domains/network/host/imds_bridge.go

@@ -65,13 +65,6 @@ func EnsureIMDSInputRule(ctx context.Context, r Runner) error {
 	return nil
 }
 
-// RemoveIMDSInputRule drops the INPUT accept; a missing rule is not an error.
-func RemoveIMDSInputRule(ctx context.Context, r Runner) {
-	if _, err := r.Run(ctx, "iptables", natRuleArgs("-D", "filter", "INPUT", imdsInputSpec)...); err != nil {
-		slog.Debug("host: IMDS input rule not present on delete", "err", err)
-	}
-}
-
 // imdsRemapComment tags the DNAT rules so an operator can find them and the
 // uninstaller can remove them.
 const imdsRemapComment = "spinifex-imds-remap"
@@ -118,25 +111,6 @@ func EnsureIMDSRemapRules(ctx context.Context, r Runner) error {
 	return nil
 }
 
-// RemoveIMDSRemapRules drops the DNATs; a missing rule is not an error.
-func RemoveIMDSRemapRules(ctx context.Context, r Runner) {
-	for _, spec := range imdsRemapSpecs() {
-		if _, err := r.Run(ctx, "iptables", natRuleArgs("-D", "nat", "PREROUTING", spec)...); err != nil {
-			slog.Debug("host: IMDS remap rule not present on delete", "err", err)
-		}
-	}
-}
-
-// RemoveIMDSBridge deletes the IMDS bridge and every flow on it. Idempotent:
-// --if-exists tolerates an already-absent bridge.
-func RemoveIMDSBridge(ctx context.Context, r Runner) error {
-	if _, err := r.Run(ctx, "ovs-vsctl", "--if-exists", "del-br", IMDSBridge); err != nil {
-		return fmt.Errorf("delete %s: %w", IMDSBridge, err)
-	}
-	slog.Debug("IMDS bridge removed", "bridge", IMDSBridge)
-	return nil
-}
-
 // installIMDSFlow adds one ovs-ofctl flow to IMDSBridge under cookie. spec is
 // the flow expression without the cookie (match + actions).
 func installIMDSFlow(ctx context.Context, r Runner, cookie, spec string) error {

spinifex/domains/network/host/imds_bridge_test.go

@@ -73,18 +73,6 @@ func TestEnsureIMDSBridgeAddBrError(t *testing.T) {
 	}
 }
 
-func TestRemoveIMDSBridge(t *testing.T) {
-	s := newStubRunner()
-	s.expect("ovs-vsctl", nil, nil)
-
-	if err := RemoveIMDSBridge(context.Background(), s); err != nil {
-		t.Fatalf("RemoveIMDSBridge: %v", err)
-	}
-	if !s.called("ovs-vsctl --if-exists del-br " + IMDSBridge) {
-		t.Errorf("expected del-br; calls: %v", s.calls)
-	}
-}
-
 func TestInstallIMDSFlow(t *testing.T) {
 	s := newStubRunner()
 	s.expect("ovs-ofctl", nil, nil)

spinifex/domains/network/host/natrules.go

@@ -77,15 +77,3 @@ func EnsureNATEgressRules(ctx context.Context, r Runner) error {
 	}
 	return nil
 }
-
-// RemoveNATEgressRules deletes the routed-NAT egress rules; missing rules are
-// not an error (teardown is idempotent).
-func RemoveNATEgressRules(ctx context.Context, r Runner) {
-	for _, rule := range natEgressRules {
-		if _, err := r.Run(ctx, "iptables", natRuleArgs("-D", rule.table, rule.chain, rule.spec)...); err != nil {
-			slog.Debug("host: NAT egress rule not present on delete", "table", rule.table, "chain", rule.chain, "err", err)
-		}
-	}
-	_, _ = r.Run(ctx, "iptables", "-t", "nat", "-F", EIPHairpinChain)
-	_, _ = r.Run(ctx, "iptables", "-t", "nat", "-X", EIPHairpinChain)
-}

spinifex/domains/network/host/natrules_test.go

@@ -80,17 +80,6 @@ func TestEnsureNATEgressRules_InsertFailure(t *testing.T) {
 	}
 }
 
-func TestRemoveNATEgressRules_IgnoresMissing(t *testing.T) {
-	r := newStubRunner()
-	r.expect("iptables -t nat -D", nil, fmt.Errorf("no match"))
-	r.expect("iptables -t filter -D", nil, nil)
-
-	RemoveNATEgressRules(context.Background(), r)
-	if !r.called("iptables -t nat -D POSTROUTING") {
-		t.Errorf("expected POSTROUTING delete attempt, got %v", r.calls)
-	}
-}
-
 // The hairpin jump must end above the MASQUERADE: both terminate in nat, so
 // whichever is first decides the source, and MASQUERADE's is exempt from DNAT.
 func TestEnsureNATEgressRules_HairpinJumpPrecedesMasquerade(t *testing.T) {

spinifex/domains/network/ovn/live.go

@@ -883,60 +883,24 @@ func (c *LiveClient) AddNAT(ctx context.Context, routerName string, nat *nbdb.NA
 
 // DeleteNAT removes every NAT rule matching (natType, logicalIP) from routerName.
 // Scoped to routerName.NAT — AWS CIDRs repeat across VPCs so the pair is not globally
-// unique. Deletes all matches so a duplicate row (e.g. minted by a racing reconcile)
-// is fully cleaned on teardown rather than leaking past the first delete.
+// unique. Deletes all matches so a racing reconcile's duplicate row does not leak.
 func (c *LiveClient) DeleteNAT(ctx context.Context, routerName string, natType, logicalIP string) error {
-	lr, err := c.GetLogicalRouter(ctx, routerName)
-	if err != nil {
-		return fmt.Errorf("get logical router for NAT delete: %w", err)
-	}
-	routerNATs := make(map[string]struct{}, len(lr.NAT))
-	for _, uuid := range lr.NAT {
-		routerNATs[uuid] = struct{}{}
-	}
-
-	var nats []nbdb.NAT
-	err = c.client.WhereCache(func(n *nbdb.NAT) bool {
-		if _, ok := routerNATs[n.UUID]; !ok {
-			return false
-		}
+	return c.deleteRouterNATs(ctx, routerName, func(n *nbdb.NAT) bool {
 		return n.Type == natType && n.LogicalIP == logicalIP
-	}).List(ctx, &nats)
-	if err != nil {
-		return fmt.Errorf("find NAT: %w", err)
-	}
-	if len(nats) == 0 {
-		return fmt.Errorf("NAT %s %s on %s: %w", natType, logicalIP, routerName, ErrNATNotFound)
-	}
-
-	var allOps []ovsdb.Operation
-	for i := range nats {
-		nat := &nats[i]
-		mutateOps, mErr := c.client.Where(lr).Mutate(lr, model.Mutation{
-			Field:   &lr.NAT,
-			Mutator: "delete",
-			Value:   []string{nat.UUID},
-		})
-		if mErr != nil {
-			return fmt.Errorf("mutate router NAT ops: %w", mErr)
-		}
-		deleteOps, dErr := c.client.Where(nat).Delete()
-		if dErr != nil {
-			return fmt.Errorf("delete NAT ops: %w", dErr)
-		}
-		allOps = append(allOps, mutateOps...)
-		allOps = append(allOps, deleteOps...)
-	}
-
-	if err := c.transactOps(ctx, allOps); err != nil {
-		return fmt.Errorf("delete NAT transact: %w", err)
-	}
-	return nil
+	}, fmt.Sprintf("NAT %s %s", natType, logicalIP))
 }
 
-// DeleteNATByExternalIP removes a NAT rule matching (natType, externalIP) from
+// DeleteNATByExternalIP removes every NAT rule matching (natType, externalIP) from
 // routerName only. Returns ErrNATNotFound if absent on this router.
 func (c *LiveClient) DeleteNATByExternalIP(ctx context.Context, routerName string, natType, externalIP string) error {
+	return c.deleteRouterNATs(ctx, routerName, func(n *nbdb.NAT) bool {
+		return n.Type == natType && n.ExternalIP == externalIP
+	}, fmt.Sprintf("NAT %s external_ip=%s", natType, externalIP))
+}
+
+// deleteRouterNATs removes every NAT on routerName matching pred in one transaction,
+// returning ErrNATNotFound (described by desc) when none match.
+func (c *LiveClient) deleteRouterNATs(ctx context.Context, routerName string, pred func(*nbdb.NAT) bool, desc string) error {
 	lr, err := c.GetLogicalRouter(ctx, routerName)
 	if err != nil {
 		return fmt.Errorf("get logical router for NAT delete: %w", err)
@@ -951,13 +915,13 @@ func (c *LiveClient) DeleteNATByExternalIP(ctx context.Context, routerName strin
 		if _, ok := routerNATs[n.UUID]; !ok {
 			return false
 		}
-		return n.Type == natType && n.ExternalIP == externalIP
+		return pred(n)
 	}).List(ctx, &nats)
 	if err != nil {
-		return fmt.Errorf("find NAT by external IP: %w", err)
+		return fmt.Errorf("find %s: %w", desc, err)
 	}
 	if len(nats) == 0 {
-		return fmt.Errorf("NAT %s external_ip=%s on %s: %w", natType, externalIP, routerName, ErrNATNotFound)
+		return fmt.Errorf("%s on %s: %w", desc, routerName, ErrNATNotFound)
 	}
 
 	var allOps []ovsdb.Operation
@@ -980,7 +944,7 @@ func (c *LiveClient) DeleteNATByExternalIP(ctx context.Context, routerName strin
 	}
 
 	if err := c.transactOps(ctx, allOps); err != nil {
-		return fmt.Errorf("delete NAT by external IP transact: %w", err)
+		return fmt.Errorf("delete %s transact: %w", desc, err)
 	}
 	return nil
 }
@@ -1168,32 +1132,18 @@ func (c *LiveClient) FindStaticRoute(ctx context.Context, routerName, ipPrefix s
 }
 
 func (c *LiveClient) DeleteStaticRoute(ctx context.Context, routerName string, ipPrefix string) error {
+	route, err := c.FindStaticRoute(ctx, routerName, ipPrefix)
+	if err != nil {
+		return err
+	}
+	if route == nil {
+		return fmt.Errorf("static route %s not found", ipPrefix)
+	}
+
 	lr, err := c.GetLogicalRouter(ctx, routerName)
 	if err != nil {
 		return fmt.Errorf("get logical router for route delete: %w", err)
 	}
-	// Scope the prefix match to this router's rows — every VPC router carries
-	// 0.0.0.0/0, so an unscoped cache match can grab another router's route.
-	owned := make(map[string]struct{}, len(lr.StaticRoutes))
-	for _, u := range lr.StaticRoutes {
-		owned[u] = struct{}{}
-	}
-	var routes []nbdb.LogicalRouterStaticRoute
-	err = c.client.WhereCache(func(r *nbdb.LogicalRouterStaticRoute) bool {
-		if r.IPPrefix != ipPrefix {
-			return false
-		}
-		_, ok := owned[r.UUID]
-		return ok
-	}).List(ctx, &routes)
-	if err != nil {
-		return fmt.Errorf("find static route: %w", err)
-	}
-	if len(routes) == 0 {
-		return fmt.Errorf("static route %s not found", ipPrefix)
-	}
-
-	route := &routes[0]
 
 	mutateOps, err := c.client.Where(lr).Mutate(lr, model.Mutation{
 		Field:   &lr.StaticRoutes,

spinifex/domains/network/policy/sg.go

@@ -17,19 +17,11 @@ type SGSpec struct {
 	EgressRules  []Rule
 }
 
-// SecurityGroupManager attaches ACL policy to an SG's port group. EnsureSG
-// and UpdateSG share replace-all semantics: each call replaces the full ACL
-// set (infra + tenant) in one OVSDB transaction.
+// SecurityGroupManager attaches ACL policy to an SG's port group.
 type SecurityGroupManager interface {
-	// EnsureSG sets the full ACL set; idempotent.
+	// EnsureSG replaces the full ACL set (infra + tenant) in one OVSDB
+	// transaction; idempotent, so it serves both create and update.
 	EnsureSG(ctx context.Context, sg SGSpec) error
-
-	// UpdateSG replaces the SG's ACL set. Semantically identical to EnsureSG.
-	UpdateSG(ctx context.Context, sg SGSpec) error
-
-	// DeleteSG clears every ACL on the SG's port group; the PG itself
-	// stays (topology.Manager owns the lifecycle).
-	DeleteSG(ctx context.Context, groupID string) error
 }
 
 // EgressPolicy is the platform egress-abuse policy applied to every guest port
@@ -62,18 +54,6 @@ func (m *sgManager) EnsureSG(ctx context.Context, sg SGSpec) error {
 	return m.applyACLs(ctx, sg)
 }
 
-func (m *sgManager) UpdateSG(ctx context.Context, sg SGSpec) error {
-	return m.applyACLs(ctx, sg)
-}
-
-func (m *sgManager) DeleteSG(ctx context.Context, groupID string) error {
-	pg := topology.SecurityGroupPortGroup(groupID)
-	if err := m.ovn.ClearACLs(ctx, pg); err != nil {
-		return fmt.Errorf("clear ACLs on %s: %w", pg, err)
-	}
-	return nil
-}
-
 // applyACLs atomically replaces infra + tenant ACLs in one OVSDB transaction.
 // ClearACLs+AddACLs split left zero-ACL gaps that defaulted traffic to drop.
 func (m *sgManager) applyACLs(ctx context.Context, sg SGSpec) error {

spinifex/domains/network/policy/sg_test.go

@@ -43,7 +43,7 @@ func TestSGManager_EnsureSG_AppliesInfraPlusTenantACLs(t *testing.T) {
 	assert.Len(t, storedPG.ACLs, 9)
 }
 
-func TestSGManager_UpdateSG_ReplacesACLSet(t *testing.T) {
+func TestSGManager_EnsureSG_ReplacesACLSet(t *testing.T) {
 	ctx := context.Background()
 	m, pg := newSGMockWithPG(t, "sg-web")
 	sg := NewSecurityGroupManager(m, EgressPolicy{})
@@ -56,7 +56,7 @@ func TestSGManager_UpdateSG_ReplacesACLSet(t *testing.T) {
 	assert.Equal(t, 7, firstCount) // 6 infra + 1 tenant
 
 	// Replace with a different set — wider ingress, no egress.
-	require.NoError(t, sg.UpdateSG(ctx, SGSpec{
+	require.NoError(t, sg.EnsureSG(ctx, SGSpec{
 		GroupID: "sg-web",
 		IngressRules: []Rule{
 			{IPProtocol: "tcp", FromPort: 80, ToPort: 80, CIDR: "0.0.0.0/0"},
@@ -67,24 +67,6 @@ func TestSGManager_UpdateSG_ReplacesACLSet(t *testing.T) {
 	assert.Len(t, m.PortGroups[pg].ACLs, 9) // 6 infra + 3 tenant
 }
 
-func TestSGManager_DeleteSG_ClearsACLsKeepsPortGroup(t *testing.T) {
-	ctx := context.Background()
-	m, pg := newSGMockWithPG(t, "sg-web")
-	sg := NewSecurityGroupManager(m, EgressPolicy{})
-
-	require.NoError(t, sg.EnsureSG(ctx, SGSpec{
-		GroupID:      "sg-web",
-		IngressRules: []Rule{{IPProtocol: "-1", CIDR: "0.0.0.0/0"}},
-	}))
-	assert.NotEmpty(t, m.PortGroups[pg].ACLs)
-
-	require.NoError(t, sg.DeleteSG(ctx, "sg-web"))
-	assert.Empty(t, m.PortGroups[pg].ACLs)
-	// Port group still present — L2 owns lifecycle.
-	_, exists := m.PortGroups[pg]
-	assert.True(t, exists)
-}
-
 func TestSGManager_EnsureSG_MissingPortGroup_ReturnsError(t *testing.T) {
 	ctx := context.Background()
 	m := mock.New()
@@ -95,9 +77,9 @@ func TestSGManager_EnsureSG_MissingPortGroup_ReturnsError(t *testing.T) {
 	assert.Contains(t, err.Error(), "port group")
 }
 
-// TestSGManager_UpdateSG_AtomicReplace asserts EnsureSG/UpdateSG use
+// TestSGManager_EnsureSG_AtomicReplace asserts repeated EnsureSG calls use
 // ReplaceACLs (single OVSDB transaction) and never call ClearACLs.
-func TestSGManager_UpdateSG_AtomicReplace(t *testing.T) {
+func TestSGManager_EnsureSG_AtomicReplace(t *testing.T) {
 	ctx := context.Background()
 	m, pg := newSGMockWithPG(t, "sg-web")
 	sg := NewSecurityGroupManager(m, EgressPolicy{})
@@ -106,14 +88,14 @@ func TestSGManager_UpdateSG_AtomicReplace(t *testing.T) {
 		GroupID:      "sg-web",
 		IngressRules: []Rule{{IPProtocol: "tcp", FromPort: 22, ToPort: 22, CIDR: "10.0.0.0/8"}},
 	}))
-	require.NoError(t, sg.UpdateSG(ctx, SGSpec{
+	require.NoError(t, sg.EnsureSG(ctx, SGSpec{
 		GroupID:      "sg-web",
 		IngressRules: []Rule{{IPProtocol: "tcp", FromPort: 80, ToPort: 80, CIDR: "0.0.0.0/0"}},
 	}))
 
-	assert.Equal(t, 2, m.ReplaceACLCalls, "EnsureSG+UpdateSG must call ReplaceACLs")
-	assert.Zero(t, m.ClearACLCalls, "EnsureSG/UpdateSG must NOT call ClearACLs")
-	assert.NotEmpty(t, m.PortGroups[pg].ACLs, "PG ACLs must never be empty after UpdateSG")
+	assert.Equal(t, 2, m.ReplaceACLCalls, "each EnsureSG must call ReplaceACLs")
+	assert.Zero(t, m.ClearACLCalls, "EnsureSG must NOT call ClearACLs")
+	assert.NotEmpty(t, m.PortGroups[pg].ACLs, "PG ACLs must never be empty after a replace")
 }
 
 func hasACLNamed(m *mock.Client, name string) bool {

spinifex/domains/network/subscribers/security.go

@@ -103,8 +103,8 @@ func (s *Subscriber) handleUpdateSG(msg *nats.Msg) {
 	}
 	ctx, cancel := context.WithTimeout(context.Background(), sgHandlerTimeout)
 	defer cancel()
-	if err := s.sg.UpdateSG(ctx, toSGSpec(evt)); err != nil {
-		slog.Error("subscribers: UpdateSG failed", "group_id", evt.GroupId, "err", err)
+	if err := s.sg.EnsureSG(ctx, toSGSpec(evt)); err != nil {
+		slog.Error("subscribers: vpc.update-sg EnsureSG failed", "group_id", evt.GroupId, "err", err)
 		respond(msg, err)
 		return
 	}

spinifex/domains/network/subscribers/security_description_test.go

@@ -1,5 +1,5 @@
 //test:in-package — projects a SecurityGroupEvent through the unexported
-//toSGSpec, which is how the vpc.update-sg handler reaches policy.UpdateSG.
+//toSGSpec, which is how the vpc.update-sg handler reaches policy.EnsureSG.
 
 package subscribers
 
@@ -19,7 +19,7 @@ import (
 	"github.com/mulgadc/spinifex/spinifex/domains/network/topology"
 )
 
-// aclFingerprints applies a vpc.update-sg payload through policy.UpdateSG and
+// aclFingerprints applies a vpc.update-sg payload through policy.EnsureSG and
 // returns the resulting ACL set as sorted, UUID-free strings.
 func aclFingerprints(t *testing.T, payload string) []string {
 	t.Helper()
@@ -30,7 +30,7 @@ func aclFingerprints(t *testing.T, payload string) []string {
 	pg := topology.SecurityGroupPortGroup(evt.GroupId)
 	require.NoError(t, m.CreatePortGroup(context.Background(), pg, nil))
 	sg := policy.NewSecurityGroupManager(m, policy.EgressPolicy{})
-	require.NoError(t, sg.UpdateSG(context.Background(), toSGSpec(evt)))
+	require.NoError(t, sg.EnsureSG(context.Background(), toSGSpec(evt)))
 
 	stored, ok := m.PortGroups[pg]
 	require.True(t, ok)

spinifex/domains/network/topology/names.go

@@ -32,15 +32,6 @@ func GatewaySwitchPort(vpcID string) string { return "gw-port-" + vpcID }
 // Only this binding carries the gateway-chassis claim; the LRP binding stays chassis-less.
 func GatewayChassisRedirectPort(vpcID string) string { return "cr-" + GatewayRouterPort(vpcID) }
 
-// ExternalSwitch is the OVN logical switch name for the legacy per-VPC external
-// switch. Retained only so the reconciler/teardown can identify and remove
-// pre-shared-switch deployments; new attaches use ExternalSwitchShared.
-func ExternalSwitch(vpcID string) string { return "ext-" + vpcID }
-
-// ExternalLocalnetPort is the legacy per-VPC localnet LSP name. Retained for
-// legacy cleanup only; see ExternalLocalnetPortShared.
-func ExternalLocalnetPort(vpcID string) string { return "ext-port-" + vpcID }
-
 // ExternalSwitchShared is the single shared external logical switch. Every VPC
 // gateway router port attaches here so only one untagged localnet exists per
 // physical network — multiple per-VPC localnets on one uplink collide on L2.

spinifex/domains/network/topology/names_test.go

@@ -16,8 +16,6 @@ func TestNameHelpers(t *testing.T) {
 		{"GatewayRouterPort", GatewayRouterPort("vpc-abc"), "gw-vpc-abc"},
 		{"GatewaySwitchPort", GatewaySwitchPort("vpc-abc"), "gw-port-vpc-abc"},
 		{"GatewayChassisRedirectPort", GatewayChassisRedirectPort("vpc-abc"), "cr-gw-vpc-abc"},
-		{"ExternalSwitch", ExternalSwitch("vpc-abc"), "ext-vpc-abc"},
-		{"ExternalLocalnetPort", ExternalLocalnetPort("vpc-abc"), "ext-port-vpc-abc"},
 		{"ExternalSwitchShared", ExternalSwitchShared(), "ext-shared"},
 		{"ExternalLocalnetPortShared", ExternalLocalnetPortShared(), "ext-port-shared"},
 		{"SecurityGroupPortGroup", SecurityGroupPortGroup("sg-abc-123"), "sg_abc_123"},