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

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

Makefile

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

internal/ecsgw/client.go

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

internal/ecsgw/client_test.go

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

internal/gwsign/signer.go

@@ -23,6 +23,12 @@ type Signer struct {
 	provider aws.CredentialsProvider
 }
 
+// New returns a Signer over provider. Wrap a provider that fetches remotely in
+// aws.NewCredentialsCache, since Sign retrieves on every call.
+func New(provider aws.CredentialsProvider) *Signer {
+	return &Signer{provider: provider}
+}
+
 // NewStatic returns a Signer using fixed long-lived keys (no session token).
 func NewStatic(accessKey, secretKey string) *Signer {
 	return &Signer{provider: credentials.NewStaticCredentialsProvider(accessKey, secretKey, "")}

internal/imdscreds/imdscreds.go

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

internal/imdscreds/imdscreds_test.go

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

internal/stsauth/stsauth.go

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

internal/stsauth/stsauth_test.go

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

internal/testkit/awserror.go

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

internal/testkit/nats.go

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

internal/testkit/nats_test.go

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

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

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

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

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

spinifex/agents/ecs/agent/agent.go

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

spinifex/agents/ecs/agent/config.go

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

spinifex/agents/ecs/agent/config_test.go

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

spinifex/agents/ecs/agent/controlplane.go

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

spinifex/agents/ecs/agent/controlplane_test.go

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

spinifex/agents/ecs/agent/credendpoint.go

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

spinifex/agents/ecs/agent/credendpoint_test.go

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

spinifex/agents/ecs/agent/credinject_test.go

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

spinifex/agents/ecs/agent/doc.go

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

spinifex/agents/ecs/agent/ecrresolver.go

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

spinifex/agents/ecs/agent/ecrresolver_test.go

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

spinifex/agents/ecs/credentials/imds.go

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

spinifex/agents/ecs/credentials/imds_test.go

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

spinifex/agents/ecs/credentials/provider.go

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

spinifex/agents/eks/konnectivitycert/cert.go

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

spinifex/agents/eks/podidentity/certs.go

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

spinifex/agents/eks/tokenwebhook/cert.go

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

spinifex/agents/eks/webhookcert/cert.go

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

spinifex/agents/eks/webhookcert/cert_test.go

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

spinifex/domains/acm/awsapi/certificates.go

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

spinifex/domains/acm/awsapi/certificates_test.go

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

spinifex/domains/acm/awsapi/handler.go

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

spinifex/domains/acm/awsapi/registration.go

@@ -10,6 +10,7 @@ import (
 
 	"github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
 	"github.com/mulgadc/spinifex/spinifex/ingress/aws/dispatch"
+	"github.com/mulgadc/spinifex/spinifex/ingress/aws/envelope"
 	"github.com/nats-io/nats.go"
 )
 
@@ -124,6 +125,6 @@ func (d Deps) dispatch(w http.ResponseWriter, inv dispatch.Invocation) error {
 		return err
 	}
 
-	WriteJSONResponse(w, output)
+	envelope.WriteJSONResult(w, output)
 	return nil
 }

spinifex/domains/admission/quota/override.go

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

spinifex/domains/admission/quota/seal.go

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

spinifex/domains/admission/quota/seal_test.go

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

spinifex/domains/ec2/account/seal.go

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

spinifex/domains/ec2/account/seal_test.go

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

spinifex/domains/ec2/accountinventory/compute.go

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

spinifex/domains/ec2/accountinventory/compute_test.go

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

spinifex/domains/ec2/accountinventory/inventory.go

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

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

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

spinifex/domains/ec2/awsapi/actions.go

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

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

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

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

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

spinifex/domains/ec2/awsapi/idempotency.go

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

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

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

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

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

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

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

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

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

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

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

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

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

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

@@ -1,80 +0,0 @@
-package instance
-
-import (
-	"context"
-	"errors"
-	"fmt"
-	"log/slog"
-	"sync"
-
-	"github.com/aws/aws-sdk-go/service/ec2"
-	ec2idemapi "github.com/mulgadc/spinifex/spinifex/domains/ec2/awsapi/idem"
-	"github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
-	"github.com/mulgadc/spinifex/spinifex/foundation/lifecycle/idempotency"
-	"github.com/nats-io/nats.go"
-	"github.com/nats-io/nats.go/jetstream"
-)
-
-// ClientTokenStore is the RunInstances idempotency store: the first caller owns
-// the launch, duplicates replay its reservation or poll for it.
-type ClientTokenStore = idempotency.Store[ec2.Reservation]
-
-var (
-	ctStore        *ClientTokenStore
-	ctOnce         sync.Once
-	errCTStoreInit error
-)
-
-// getClientTokenStore lazily initialises the process-wide client-token store via
-// sync.Once. Process-wide because RunInstances is a free function with no
-// instance to hang it on.
-func getClientTokenStore(ctx context.Context, nc *nats.Conn) (*ClientTokenStore, error) {
-	ctOnce.Do(func() {
-		js, err := jetstream.New(nc)
-		if err != nil {
-			errCTStoreInit = fmt.Errorf("clienttoken jetstream: %w", err)
-			return
-		}
-		// The bind happens once per process, so it must not inherit the first
-		// caller's cancellation: a client that disconnects mid-open would poison
-		// the store for every later launch. Deadline-free, so the open falls back
-		// to the JetStream API's own timeout.
-		ctStore, errCTStoreInit = newClientTokenStore(context.WithoutCancel(ctx), js)
-	})
-	return ctStore, errCTStoreInit
-}
-
-// newClientTokenStore takes the bucket unnamespaced. Every other EC2 action
-// namespaces its keys by action, but RunInstances predates that: re-keying live
-// records would make a retry re-launch instead of replay.
-func newClientTokenStore(ctx context.Context, js jetstream.JetStream) (*ClientTokenStore, error) {
-	return idempotency.OpenStore[ec2.Reservation](ctx, js, ec2idemapi.KVBucket, ec2idemapi.TTL)
-}
-
-// clientTokenParamHash hashes the request excluding ClientToken, so the same
-// params always produce the same hash. Must run before any input mutation.
-func clientTokenParamHash(input *ec2.RunInstancesInput) string {
-	clone := *input
-	clone.ClientToken = nil
-	return idempotency.ParamHash(&clone)
-}
-
-// runInstancesWithClientToken wraps a launch in ClientToken idempotency and maps
-// the token errors onto the AWS ones RunInstances returns. Extracted for
-// unit-testability.
-func runInstancesWithClientToken(
-	ctx context.Context,
-	store *ClientTokenStore,
-	accountID, token, paramHash string,
-	launch func() (ec2.Reservation, error),
-) (ec2.Reservation, error) {
-	res, err := idempotency.Do(ctx, store, accountID, token, paramHash, launch)
-	switch {
-	case errors.Is(err, idempotency.ErrParamMismatch):
-		return ec2.Reservation{}, errors.New(awserrors.ErrorIdempotentParameterMismatch)
-	case errors.Is(err, idempotency.ErrUnavailable):
-		slog.ErrorContext(ctx, "RunInstances: client-token claim failed", "token", token, "err", err)
-		return ec2.Reservation{}, errors.New(awserrors.ErrorServerInternal)
-	}
-	return res, err
-}

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

@@ -1,155 +1,55 @@
 package instance
 
 import (
-	"errors"
+	"context"
 	"testing"
+	"time"
 
 	"github.com/aws/aws-sdk-go/aws"
 	"github.com/aws/aws-sdk-go/service/ec2"
 	"github.com/mulgadc/spinifex/internal/testkit"
 	"github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
+	"github.com/mulgadc/spinifex/spinifex/foundation/lifecycle/idempotency"
+	"github.com/nats-io/nats.go/jetstream"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 )
 
 const ctTestAccount = "111122223333"
 
-func newTestClientTokenStore(t *testing.T) *ClientTokenStore {
-	t.Helper()
+// RunInstances records predate per-action namespacing: they live at the bare
+// account/token key, hashed over the request minus its token. A record written
+// that way must still replay, or a retry across an upgrade would re-launch.
+func TestRunInstances_ClientTokenReplaysUnnamespacedRecord(t *testing.T) {
+	t.Parallel()
 	_, nc, _ := testutil.StartTestJetStream(t)
-	store, err := newClientTokenStore(t.Context(), testutil.NewJetStream(t, nc))
+	kv, err := idempotency.OpenBucket(t.Context(), testutil.NewJetStream(t, nc), "ec2-tokens", time.Minute)
 	require.NoError(t, err)
-	return store
-}
+	bucket := func(context.Context) (jetstream.KeyValue, error) { return kv, nil }
 
-// The claim/replay/abort mechanics live in spinifex/foundation/lifecycle/idempotency and are tested
-// there. What is EC2's own is the param hash, the launch orchestration around
-// the store, and the AWS error a mismatch maps onto.
+	const tok = "rt-1"
+	input := defaults
+	input.ClientToken = aws.String(tok)
+	unTokened := input
+	unTokened.ClientToken = nil
+	hash := idempotency.ParamHash(&unTokened)
 
-// clientTokenParamHash ignores ClientToken (same params, different token →
-// same hash) but reflects a real parameter change.
-func TestClientTokenParamHash_IgnoresTokenReflectsParams(t *testing.T) {
-	t.Parallel()
-	base := &ec2.RunInstancesInput{
-		ImageId:      aws.String("ami-123"),
-		InstanceType: aws.String("t3.micro"),
-		MinCount:     aws.Int64(1),
-		MaxCount:     aws.Int64(1),
-		ClientToken:  aws.String("tok-A"),
-	}
-	sameParamsDiffToken := *base
-	sameParamsDiffToken.ClientToken = aws.String("tok-B")
-	diffParams := *base
-	diffParams.InstanceType = aws.String("t3.large")
-
-	assert.Equal(t, clientTokenParamHash(base), clientTokenParamHash(&sameParamsDiffToken),
-		"token must not affect the param hash")
-	assert.NotEqual(t, clientTokenParamHash(base), clientTokenParamHash(&diffParams),
-		"a real param change must change the hash")
-}
-
-// The hash is taken from a copy, so hashing must not clear the caller's token.
-func TestClientTokenParamHash_LeavesTheInputAlone(t *testing.T) {
-	t.Parallel()
-	input := &ec2.RunInstancesInput{ImageId: aws.String("ami-123"), ClientToken: aws.String("tok-A")}
-
-	clientTokenParamHash(input)
-
-	assert.Equal(t, "tok-A", aws.StringValue(input.ClientToken),
-		"hashing must not strip the token the launch still needs")
-}
-
-// A completed token replays its reservation and must NOT invoke the launcher.
-func TestRunInstancesWithClientToken_ReplaySkipsLaunch(t *testing.T) {
-	t.Parallel()
-	store := newTestClientTokenStore(t)
-	const tok, hash = "rt-1", "h"
+	store := idempotency.NewStore[ec2.Reservation](kv, "")
 	_, owned, err := store.Claim(t.Context(), ctTestAccount, tok, hash)
 	require.NoError(t, err)
 	require.True(t, owned)
 	require.NoError(t, store.Finalize(t.Context(), ctTestAccount, tok, hash, ec2.Reservation{ReservationId: aws.String("r-x")}))
+	_, err = kv.Get(t.Context(), idempotency.Key(ctTestAccount, tok))
+	require.NoError(t, err, "record must sit at the unprefixed key")
 
-	launched := false
-	res, err := runInstancesWithClientToken(t.Context(), store, ctTestAccount, tok, hash, func() (ec2.Reservation, error) {
-		launched = true
-		return ec2.Reservation{}, nil
-	})
+	// A nil NATS connection: replay must not reach the launch path.
+	replay := input
+	res, err := RunInstances(t.Context(), &replay, nil, bucket, nil, ctTestAccount, nil, nil, 1)
 	require.NoError(t, err)
-	assert.False(t, launched, "replay must not launch")
 	assert.Equal(t, "r-x", aws.StringValue(res.ReservationId))
-}
 
-// The owner launches once and finalizes; a duplicate replays without launching.
-func TestRunInstancesWithClientToken_OwnerLaunchesOnceThenReplay(t *testing.T) {
-	t.Parallel()
-	store := newTestClientTokenStore(t)
-	const tok, hash = "rt-2", "h"
-	launches := 0
-	launch := func() (ec2.Reservation, error) {
-		launches++
-		return ec2.Reservation{ReservationId: aws.String("r-own")}, nil
-	}
-
-	res, err := runInstancesWithClientToken(t.Context(), store, ctTestAccount, tok, hash, launch)
-	require.NoError(t, err)
-	assert.Equal(t, "r-own", aws.StringValue(res.ReservationId))
-
-	res2, err := runInstancesWithClientToken(t.Context(), store, ctTestAccount, tok, hash, launch)
-	require.NoError(t, err)
-	assert.Equal(t, "r-own", aws.StringValue(res2.ReservationId))
-	assert.Equal(t, 1, launches, "duplicate must replay, not relaunch")
-}
-
-// A launch failure aborts the token so a retry re-launches.
-func TestRunInstancesWithClientToken_LaunchFailureAborts(t *testing.T) {
-	t.Parallel()
-	store := newTestClientTokenStore(t)
-	const tok, hash = "rt-3", "h"
-
-	_, err := runInstancesWithClientToken(t.Context(), store, ctTestAccount, tok, hash, func() (ec2.Reservation, error) {
-		return ec2.Reservation{}, errors.New("no capacity")
-	})
-	require.Error(t, err)
-
-	relaunched := false
-	res, err := runInstancesWithClientToken(t.Context(), store, ctTestAccount, tok, hash, func() (ec2.Reservation, error) {
-		relaunched = true
-		return ec2.Reservation{ReservationId: aws.String("r-retry")}, nil
-	})
-	require.NoError(t, err)
-	assert.True(t, relaunched, "after abort the retry must launch")
-	assert.Equal(t, "r-retry", aws.StringValue(res.ReservationId))
-}
-
-// Token reuse with different params maps to the AWS IdempotentParameterMismatch
-// error code and never launches.
-func TestRunInstancesWithClientToken_ParamMismatchMapsAWSError(t *testing.T) {
-	t.Parallel()
-	store := newTestClientTokenStore(t)
-	const tok = "rt-4"
-	_, owned, err := store.Claim(t.Context(), ctTestAccount, tok, "hA")
-	require.NoError(t, err)
-	require.True(t, owned)
-	require.NoError(t, store.Finalize(t.Context(), ctTestAccount, tok, "hA", ec2.Reservation{ReservationId: aws.String("r")}))
-
-	launched := false
-	_, err = runInstancesWithClientToken(t.Context(), store, ctTestAccount, tok, "hB", func() (ec2.Reservation, error) {
-		launched = true
-		return ec2.Reservation{}, nil
-	})
+	changed := input
+	changed.InstanceType = aws.String("t3.large")
+	_, err = RunInstances(t.Context(), &changed, nil, bucket, nil, ctTestAccount, nil, nil, 1)
 	require.EqualError(t, err, awserrors.ErrorIdempotentParameterMismatch)
-	assert.False(t, launched)
-}
-
-// getClientTokenStore binds the process-wide store once and returns the same
-// instance on subsequent calls.
-func TestGetClientTokenStore_BindsOnce(t *testing.T) {
-	t.Parallel()
-	_, nc, _ := testutil.StartTestJetStream(t)
-	s1, err := getClientTokenStore(t.Context(), nc)
-	require.NoError(t, err)
-	require.NotNil(t, s1)
-	s2, err := getClientTokenStore(t.Context(), nc)
-	require.NoError(t, err)
-	assert.Same(t, s1, s2, "store binds once")
 }

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

@@ -830,7 +830,7 @@ func TestRunInstances_SingleInstanceDistributes(t *testing.T) {
 		MaxCount:     aws.Int64(1),
 	}
 
-	reservation, err := RunInstances(context.Background(), input, nc, nil, "test-account", nil, nil, 1)
+	reservation, err := RunInstances(context.Background(), input, nc, nil, nil, "test-account", nil, nil, 1)
 	require.NoError(t, err)
 	assert.Len(t, reservation.Instances, 1)
 	assert.Equal(t, "i-single", aws.StringValue(reservation.Instances[0].InstanceId))
@@ -1383,7 +1383,7 @@ func TestRunInstances_ClusterPlacementGroupRouting(t *testing.T) {
 		},
 	}
 
-	reservation, err := RunInstances(context.Background(), input, nc, nil, "test-account", nil, nil, 1)
+	reservation, err := RunInstances(context.Background(), input, nc, nil, nil, "test-account", nil, nil, 1)
 	require.NoError(t, err)
 	assert.Len(t, reservation.Instances, 2)
 }
@@ -1432,7 +1432,7 @@ func TestRunInstances_MultiInstanceUsesDistribution(t *testing.T) {
 		MaxCount:     aws.Int64(2),
 	}
 
-	reservation, err := RunInstances(context.Background(), input, nc, nil, "test-account", nil, nil, 1)
+	reservation, err := RunInstances(context.Background(), input, nc, nil, nil, "test-account", nil, nil, 1)
 	require.NoError(t, err)
 	assert.Len(t, reservation.Instances, 2)
 	assert.True(t, statusQueried, "multi-instance launch should query node status")

spinifex/domains/ec2/awsapi/spotinstance/RequestSpotInstances.go

@@ -14,6 +14,7 @@ import (
 
 	"github.com/aws/aws-sdk-go/aws"
 	"github.com/aws/aws-sdk-go/service/ec2"
+	ec2idemapi "github.com/mulgadc/spinifex/spinifex/domains/ec2/awsapi/idem"
 	ec2instanceapi "github.com/mulgadc/spinifex/spinifex/domains/ec2/awsapi/instance"
 	ec2spotinstance "github.com/mulgadc/spinifex/spinifex/domains/ec2/spotinstance"
 	"github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
@@ -62,7 +63,7 @@ func ValidateRequestSpotInstancesInput(input *ec2.RequestSpotInstancesInput) err
 // ClientToken passed through, same vCPU quota gate), then builds and persists one
 // active/fulfilled SpotInstanceRequest per launched instance. On a launch failure
 // (including InsufficientInstanceCapacity) it returns the error and persists nothing.
-func RequestSpotInstances(ctx context.Context, input *ec2.RequestSpotInstancesInput, natsConn *nats.Conn, iamSvc ec2instanceapi.InstanceProfiles, accountID, az string, passRoleCheck ec2instanceapi.PassRoleChecker, quota LaunchQuota, expectedNodes int) (ec2.RequestSpotInstancesOutput, error) {
+func RequestSpotInstances(ctx context.Context, input *ec2.RequestSpotInstancesInput, natsConn *nats.Conn, tokenBucket ec2idemapi.Bucket, iamSvc ec2instanceapi.InstanceProfiles, accountID, az string, passRoleCheck ec2instanceapi.PassRoleChecker, quota LaunchQuota, expectedNodes int) (ec2.RequestSpotInstancesOutput, error) {
 	var output ec2.RequestSpotInstancesOutput
 
 	if err := ValidateRequestSpotInstancesInput(input); err != nil {
@@ -80,7 +81,7 @@ func RequestSpotInstances(ctx context.Context, input *ec2.RequestSpotInstancesIn
 
 	// RunInstances normalises runInput in place (e.g. instance profile to ARN),
 	// so the launch spec echoed back is built from runInput afterwards.
-	reservation, err := ec2instanceapi.RunInstances(ctx, runInput, natsConn, iamSvc, accountID, passRoleCheck, launchQuotaCheck, expectedNodes)
+	reservation, err := ec2instanceapi.RunInstances(ctx, runInput, natsConn, tokenBucket, iamSvc, accountID, passRoleCheck, launchQuotaCheck, expectedNodes)
 	if err != nil {
 		return output, err
 	}

spinifex/domains/ec2/awsapi/spotinstance/quota_gate_order_test.go

@@ -125,7 +125,7 @@ func TestRequestSpotInstances_QuotaGateRunsBeforeDispatchAndLaunch(t *testing.T)
 		InstanceCount: aws.Int64(1),
 	}
 
-	_, err = RequestSpotInstances(context.Background(), input, nc, nil, "test-account", "ap-southeast-2a", nil, quota, 1)
+	_, err = RequestSpotInstances(context.Background(), input, nc, nil, nil, "test-account", "ap-southeast-2a", nil, quota, 1)
 	require.NoError(t, err)
 
 	quotaIdx := order.indexOf("quota-check")

spinifex/domains/ec2/eigw/service_impl.go

@@ -158,12 +158,7 @@ var describeEIGWValidFilters = map[string]bool{
 func (s *EgressOnlyIGWServiceImpl) DescribeEgressOnlyInternetGateways(ctx context.Context, input *ec2.DescribeEgressOnlyInternetGatewaysInput, accountID string) (*ec2.DescribeEgressOnlyInternetGatewaysOutput, error) {
 	var egressOnlyIGWs []*ec2.EgressOnlyInternetGateway
 
-	eigwIDs := make(map[string]bool)
-	for _, id := range input.EgressOnlyInternetGatewayIds {
-		if id != nil {
-			eigwIDs[*id] = true
-		}
-	}
+	eigwIDs := awsidentifiers.Set(input.EgressOnlyInternetGatewayIds)
 
 	parsedFilters, err := awsfilters.ParseFilters(input.Filters, describeEIGWValidFilters)
 	if err != nil {

spinifex/domains/ec2/eip/service_impl.go

@@ -572,19 +572,9 @@ var describeAddressesValidFilters = map[string]bool{
 
 // DescribeAddresses lists Elastic IPs with optional filtering by allocation ID.
 func (s *EIPServiceImpl) DescribeAddresses(ctx context.Context, input *ec2.DescribeAddressesInput, accountID string) (*ec2.DescribeAddressesOutput, error) {
-	allocIDs := make(map[string]bool)
-	for _, id := range input.AllocationIds {
-		if id != nil {
-			allocIDs[*id] = true
-		}
-	}
+	allocIDs := awsidentifiers.Set(input.AllocationIds)
 
-	publicIPs := make(map[string]bool)
-	for _, ip := range input.PublicIps {
-		if ip != nil {
-			publicIPs[*ip] = true
-		}
-	}
+	publicIPs := awsidentifiers.Set(input.PublicIps)
 
 	parsedFilters, err := awsfilters.ParseFilters(input.Filters, describeAddressesValidFilters)
 	if err != nil {

spinifex/domains/ec2/guestmetadata/imdsv1.go

@@ -3,7 +3,6 @@ package guestmetadata
 import (
 	"context"
 	"log/slog"
-	"sync"
 	"time"
 
 	"github.com/aws/aws-sdk-go/service/ec2"
@@ -14,50 +13,6 @@ import (
 // little and keeps the untokened path off the NATS fan-out.
 const v1AllowTTL = 30 * time.Second
 
-type v1AllowEntry struct {
-	allowed bool
-	expires time.Time
-}
-
-// v1AllowCache memoises whether an instance permits IMDSv1. It exists for
-// amplification control: the untokened path is reachable without credentials, so
-// resolving HttpTokens per request would let any guest drive unbounded
-// DescribeInstances fan-outs.
-type v1AllowCache struct {
-	mu      sync.Mutex
-	entries map[string]v1AllowEntry
-}
-
-func newV1AllowCache() *v1AllowCache {
-	return &v1AllowCache{entries: make(map[string]v1AllowEntry)}
-}
-
-func (c *v1AllowCache) get(instanceID string, now time.Time) (bool, bool) {
-	c.mu.Lock()
-	defer c.mu.Unlock()
-	e, ok := c.entries[instanceID]
-	if !ok || now.After(e.expires) {
-		return false, false
-	}
-	return e.allowed, true
-}
-
-func (c *v1AllowCache) put(instanceID string, allowed bool, now time.Time) {
-	c.mu.Lock()
-	defer c.mu.Unlock()
-	c.entries[instanceID] = v1AllowEntry{allowed: allowed, expires: now.Add(v1AllowTTL)}
-}
-
-func (c *v1AllowCache) sweep(now time.Time) {
-	c.mu.Lock()
-	defer c.mu.Unlock()
-	for id, e := range c.entries {
-		if now.After(e.expires) {
-			delete(c.entries, id)
-		}
-	}
-}
-
 // imdsV1Allowed reports whether this ENI's instance was launched or modified
 // with HttpTokens=optional, which is the only case where an untokened GET is
 // served. Defaults closed: an unattached ENI, a lookup failure or a missing
@@ -66,7 +21,7 @@ func (s *IMDSServiceImpl) imdsV1Allowed(ctx context.Context, eni *eniFacts) bool
 	if eni.instanceID == "" {
 		return false
 	}
-	if allowed, ok := s.v1Allow.get(eni.instanceID, s.now()); ok {
+	if allowed, ok := s.v1Allow.get(eni.instanceID); ok {
 		return allowed
 	}
 
@@ -79,6 +34,6 @@ func (s *IMDSServiceImpl) imdsV1Allowed(ctx context.Context, eni *eniFacts) bool
 
 	// Cached even on failure: without that, a guest could replay a failing
 	// lookup to keep the fan-out running.
-	s.v1Allow.put(eni.instanceID, allowed, s.now())
+	s.v1Allow.put(eni.instanceID, allowed)
 	return allowed
 }

spinifex/domains/ec2/guestmetadata/imdsv1_test.go

@@ -5,7 +5,6 @@ import (
 	"errors"
 	"net/http"
 	"testing"
-	"time"
 
 	"github.com/aws/aws-sdk-go/service/ec2"
 	"github.com/stretchr/testify/assert"
@@ -107,36 +106,6 @@ func TestIMDSv1_LookupFailureDeniesAndCaches(t *testing.T) {
 	assert.Equal(t, 1, res.calls, "a failing lookup must not be retried per request")
 }
 
-// An expired entry is re-resolved, so flipping HttpTokens takes effect without a
-// vpcd restart.
-func TestIMDSv1_CacheExpires(t *testing.T) {
-	c := newV1AllowCache()
-	now := time.Unix(1_700_000_000, 0).UTC()
-
-	c.put("i-1", true, now)
-	allowed, ok := c.get("i-1", now.Add(v1AllowTTL-time.Second))
-	require.True(t, ok)
-	assert.True(t, allowed)
-
-	_, ok = c.get("i-1", now.Add(v1AllowTTL+time.Second))
-	assert.False(t, ok, "entry must expire after the TTL")
-}
-
-// sweep drops expired entries so the map cannot grow without bound across
-// instance churn.
-func TestIMDSv1_CacheSweep(t *testing.T) {
-	c := newV1AllowCache()
-	now := time.Unix(1_700_000_000, 0).UTC()
-
-	c.put("i-old", true, now)
-	c.put("i-new", true, now.Add(v1AllowTTL))
-	c.sweep(now.Add(v1AllowTTL + time.Second))
-
-	_, ok := c.get("i-old", now.Add(v1AllowTTL+time.Second))
-	assert.False(t, ok)
-	assert.Len(t, c.entries, 1, "only the live entry survives the sweep")
-}
-
 // An ENI with no attached instance denies without attempting a lookup.
 func TestIMDSv1_UnattachedENIDenies(t *testing.T) {
 	res := v1Resolver(ec2.HttpTokensStateOptional)

spinifex/domains/ec2/guestmetadata/metadata_test.go

@@ -150,10 +150,12 @@ func withTapConn(h http.Handler, eni *eniFacts, setTTL connTTLSetter) http.Handl
 // clock. The per-tap responder is left nil — the HTTP handler is exercised directly.
 func newTestService(res eniResolver, fIAM profileLookup, assumer stsAssumer) (*IMDSServiceImpl, time.Time) {
 	now := time.Unix(1_700_000_000, 0).UTC()
+	v1Allow := newTTLCache[bool](v1AllowTTL)
+	v1Allow.now = func() time.Time { return now }
 	return &IMDSServiceImpl{
 		resolver:       res,
 		tokens:         newTokenStore(),
-		v1Allow:        newV1AllowCache(),
+		v1Allow:        v1Allow,
 		creds:          newCredCache(assumer),
 		iam:            fIAM,
 		roleMiss:       newRoleMissLogger(func() time.Time { return now }),

spinifex/domains/ec2/guestmetadata/nats_clients.go

@@ -151,3 +151,14 @@ func (c *ttlCache[V]) put(key string, v V) {
 	defer c.mu.Unlock()
 	c.m[key] = ttlCacheEntry[V]{val: v, expiresAt: c.now().Add(c.ttl)}
 }
+
+// sweep evicts every entry that has expired as of now.
+func (c *ttlCache[V]) sweep(now time.Time) {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	for k, e := range c.m {
+		if now.After(e.expiresAt) {
+			delete(c.m, k)
+		}
+	}
+}

spinifex/domains/ec2/guestmetadata/nats_clients_test.go

@@ -197,3 +197,20 @@ func TestTTLCache_ExpiresEntries(t *testing.T) {
 	_, ok = c.get("k")
 	assert.False(t, ok)
 }
+
+// sweep drops expired entries so the map cannot grow without bound across
+// instance churn.
+func TestTTLCache_SweepDropsExpired(t *testing.T) {
+	now := time.Unix(1700000000, 0)
+	c := newTTLCache[bool](time.Minute)
+	c.now = func() time.Time { return now }
+
+	c.put("old", true)
+	now = now.Add(time.Minute)
+	c.put("new", true)
+	c.sweep(now.Add(time.Second))
+
+	assert.Len(t, c.m, 1, "only the live entry survives the sweep")
+	_, ok := c.m["new"]
+	assert.True(t, ok)
+}

spinifex/domains/ec2/guestmetadata/service.go

@@ -65,9 +65,11 @@ type eniResolver interface {
 // responder per local primary-ENI tap, each serving the shared mux with the
 // tap's ENI identity.
 type IMDSServiceImpl struct {
-	resolver       eniResolver
-	tokens         *tokenStore
-	v1Allow        *v1AllowCache
+	resolver eniResolver
+	tokens   *tokenStore
+	// v1Allow memoises HttpTokens per instance: the untokened path needs no
+	// credentials, so resolving it per request would allow unbounded fan-outs.
+	v1Allow        *ttlCache[bool]
 	creds          *credCache
 	iam            profileLookup
 	roleMiss       *roleMissLogger
@@ -169,7 +171,7 @@ func NewIMDSServiceImpl(ctx context.Context, natsConn *nats.Conn, sts stsAssumer
 			lookup:   &localInstanceLookup{local: localState, records: records},
 		},
 		tokens:         newTokenStore(),
-		v1Allow:        newV1AllowCache(),
+		v1Allow:        newTTLCache[bool](v1AllowTTL),
 		creds:          newCredCache(sts),
 		iam:            iamSvc,
 		roleMiss:       newRoleMissLogger(time.Now),

spinifex/domains/ec2/igw/service_impl.go

@@ -219,12 +219,7 @@ func (s *IGWServiceImpl) DescribeInternetGateways(ctx context.Context, input *ec
 		return nil, err
 	}
 
-	igwIDs := make(map[string]bool)
-	for _, id := range input.InternetGatewayIds {
-		if id != nil {
-			igwIDs[*id] = true
-		}
-	}
+	igwIDs := awsidentifiers.Set(input.InternetGatewayIds)
 
 	parsedFilters, err := awsfilters.ParseFilters(input.Filters, describeIGWValidFilters)
 	if err != nil {

spinifex/domains/ec2/igw/service_impl_test.go

@@ -174,14 +174,14 @@ func TestDeleteInternetGateway_WhileAttached(t *testing.T) {
 	_, err = svc.DeleteInternetGateway(context.Background(), &ec2.DeleteInternetGatewayInput{
 		InternetGatewayId: aws.String(igwID),
 	}, testAccountID)
-	requireAWSError(t, err, awserrors.ErrorDependencyViolation, "The internetGateway '"+igwID+"' has dependencies and cannot be deleted.")
+	testutil.RequireAWSError(t, err, awserrors.ErrorDependencyViolation, "The internetGateway '"+igwID+"' has dependencies and cannot be deleted.")
 
 	// Attaching again names the gateway and the network it is on
 	_, err = svc.AttachInternetGateway(context.Background(), &ec2.AttachInternetGatewayInput{
 		InternetGatewayId: aws.String(igwID),
 		VpcId:             aws.String("vpc-test123"),
 	}, testAccountID)
-	requireAWSError(t, err, awserrors.ErrorResourceAlreadyAssociated, "resource "+igwID+" is already attached to network vpc-test123")
+	testutil.RequireAWSError(t, err, awserrors.ErrorResourceAlreadyAssociated, "resource "+igwID+" is already attached to network vpc-test123")
 }
 
 func TestInternetGateway_NotFoundNamesTheID(t *testing.T) {
@@ -191,20 +191,12 @@ func TestInternetGateway_NotFoundNamesTheID(t *testing.T) {
 	_, err := svc.DescribeInternetGateways(context.Background(), &ec2.DescribeInternetGatewaysInput{
 		InternetGatewayIds: []*string{aws.String("igw-0000000000000dead")},
 	}, testAccountID)
-	requireAWSError(t, err, awserrors.ErrorInvalidInternetGatewayIDNotFound, "The internetGateway ID 'igw-0000000000000dead' does not exist")
+	testutil.RequireAWSError(t, err, awserrors.ErrorInvalidInternetGatewayIDNotFound, "The internetGateway ID 'igw-0000000000000dead' does not exist")
 
 	_, err = svc.DeleteInternetGateway(context.Background(), &ec2.DeleteInternetGatewayInput{
 		InternetGatewayId: aws.String("igw-0000000000000dead"),
 	}, testAccountID)
-	requireAWSError(t, err, awserrors.ErrorInvalidInternetGatewayIDNotFound, "The internetGateway ID 'igw-0000000000000dead' does not exist")
-}
-
-func requireAWSError(t *testing.T, err error, code, message string) {
-	t.Helper()
-	got, msg, ok := awserrors.ResolveErrorDetail(err)
-	require.True(t, ok, "error %v carries no AWS code", err)
-	assert.Equal(t, code, got)
-	assert.Equal(t, message, msg)
+	testutil.RequireAWSError(t, err, awserrors.ErrorInvalidInternetGatewayIDNotFound, "The internetGateway ID 'igw-0000000000000dead' does not exist")
 }
 
 func TestDescribeInternetGateways_All(t *testing.T) {

spinifex/domains/ec2/instance/describe_account_records_test.go

@@ -0,0 +1,89 @@
+package instance
+
+import (
+	"context"
+	"errors"
+	"testing"
+
+	"github.com/aws/aws-sdk-go/service/ec2"
+	"github.com/mulgadc/spinifex/spinifex/bootstrap/config"
+	awserrors "github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
+	"github.com/mulgadc/spinifex/spinifex/runtime/compute/vm"
+	vmmock "github.com/mulgadc/spinifex/spinifex/runtime/compute/vm/mock"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func recordsService(store *vmmock.StateStore) *InstanceServiceImpl {
+	return &InstanceServiceImpl{vmMgr: mgrWith(map[string]*vm.VM{}), stoppedStore: store, config: &config.Config{}}
+}
+
+func recordIDs(out *ec2.DescribeInstancesOutput) map[string]string {
+	ids := map[string]string{}
+	for _, res := range out.Reservations {
+		for _, inst := range res.Instances {
+			ids[*inst.InstanceId] = *inst.State.Name
+		}
+	}
+	return ids
+}
+
+// The record read answers for instances the stopped listing leaves out: a
+// running or drain-stopped record that no node need be holding.
+func TestDescribeAccountInstanceRecords_ReturnsEveryNonTerminatedRecordOfTheAccount(t *testing.T) {
+	owner := "111122223333"
+	store := &vmmock.StateStore{
+		Stopped: map[string]*vm.VM{"i-stopped": nilRecordVM("i-stopped", owner, true, true, vm.StateStopped)},
+		Saved: map[string]map[string]*vm.VM{"node-gone": {
+			"i-running": nilRecordVM("i-running", owner, true, true, vm.StateRunning),
+			"i-other":   nilRecordVM("i-other", "999999999999", true, true, vm.StateRunning),
+		}},
+	}
+
+	out, err := recordsService(store).DescribeAccountInstanceRecords(context.Background(), &ec2.DescribeInstancesInput{}, owner)
+
+	require.NoError(t, err)
+	assert.Equal(t, map[string]string{"i-stopped": "stopped", "i-running": "running"}, recordIDs(out),
+		"the other account's record is not returned")
+}
+
+// A record that cannot be projected fails the read; it is never skipped, since
+// the caller takes existence from this answer.
+func TestDescribeAccountInstanceRecords_NilReservationOrInstanceFails(t *testing.T) {
+	owner := "111122223333"
+	for name, bad := range map[string]*vm.VM{
+		"nil reservation": nilRecordVM("i-bad", owner, false, true, vm.StateRunning),
+		"nil instance":    nilRecordVM("i-bad", owner, true, false, vm.StateRunning),
+	} {
+		t.Run(name, func(t *testing.T) {
+			store := &vmmock.StateStore{
+				Stopped: map[string]*vm.VM{"i-good": nilRecordVM("i-good", owner, true, true, vm.StateStopped)},
+				Saved:   map[string]map[string]*vm.VM{"node-1": {"i-bad": bad}},
+			}
+
+			out, err := recordsService(store).DescribeAccountInstanceRecords(context.Background(), &ec2.DescribeInstancesInput{}, owner)
+
+			require.Error(t, err)
+			assert.Nil(t, out, "no shorter list is returned")
+			assert.Equal(t, awserrors.ErrorServerInternal, err.Error())
+		})
+	}
+}
+
+func TestDescribeAccountInstanceRecords_StoreErrorFails(t *testing.T) {
+	store := &vmmock.StateStore{ListLiveErr: errors.New("kv down")}
+
+	out, err := recordsService(store).DescribeAccountInstanceRecords(context.Background(), &ec2.DescribeInstancesInput{}, "111122223333")
+
+	require.Error(t, err)
+	assert.Nil(t, out)
+}
+
+func TestDescribeAccountInstanceRecords_NoStoreFails(t *testing.T) {
+	svc := &InstanceServiceImpl{vmMgr: mgrWith(map[string]*vm.VM{}), config: &config.Config{}}
+
+	out, err := svc.DescribeAccountInstanceRecords(context.Background(), &ec2.DescribeInstancesInput{}, "111122223333")
+
+	require.Error(t, err)
+	assert.Nil(t, out)
+}

spinifex/domains/ec2/instance/describe_nil_records_current_behaviour_test.go

@@ -0,0 +1,64 @@
+package instance
+
+import (
+	"context"
+	"testing"
+
+	"github.com/aws/aws-sdk-go/aws"
+	"github.com/aws/aws-sdk-go/service/ec2"
+	"github.com/mulgadc/spinifex/spinifex/bootstrap/config"
+	"github.com/mulgadc/spinifex/spinifex/runtime/compute/vm"
+	vmmock "github.com/mulgadc/spinifex/spinifex/runtime/compute/vm/mock"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func nilRecordVM(id, owner string, withReservation, withInstance bool, status vm.InstanceState) *vm.VM {
+	v := &vm.VM{ID: id, InstanceType: "t3.micro", Status: status, AccountID: owner}
+	if withReservation {
+		v.Reservation = &ec2.Reservation{ReservationId: aws.String("r-" + id), OwnerId: aws.String(owner)}
+	}
+	if withInstance {
+		v.Instance = &ec2.Instance{InstanceId: aws.String(id), InstanceType: aws.String("t3.micro")}
+	}
+	return v
+}
+
+// Pins today: the node path drops an instance whose Reservation or Instance is
+// nil without a log line or an error, so it reads as absent from a listing the
+// caller may treat as complete.
+func TestCurrentBehaviour_DescribeInstances_NodePathSkipsNilReservationAndInstanceSilently(t *testing.T) {
+	owner := "111122223333"
+	vms := map[string]*vm.VM{
+		"i-good":    nilRecordVM("i-good", owner, true, true, vm.StateRunning),
+		"i-no-res":  nilRecordVM("i-no-res", owner, false, true, vm.StateRunning),
+		"i-no-inst": nilRecordVM("i-no-inst", owner, true, false, vm.StateRunning),
+	}
+	svc := &InstanceServiceImpl{vmMgr: mgrWith(vms), config: &config.Config{}}
+
+	out, err := svc.DescribeInstances(context.Background(), &ec2.DescribeInstancesInput{}, owner)
+
+	require.NoError(t, err, "CURRENT: no error for the two skipped instances")
+	require.Len(t, out.Reservations, 1)
+	require.Len(t, out.Reservations[0].Instances, 1)
+	assert.Equal(t, "i-good", *out.Reservations[0].Instances[0].InstanceId, "CURRENT: only the complete record is returned")
+}
+
+// Pins today: the KV path skips the same records, with a warning, and still
+// answers success.
+func TestCurrentBehaviour_DescribeStoppedInstances_KVPathSkipsNilReservationAndInstance(t *testing.T) {
+	owner := "111122223333"
+	store := &vmmock.StateStore{Stopped: map[string]*vm.VM{
+		"i-good":    nilRecordVM("i-good", owner, true, true, vm.StateStopped),
+		"i-no-res":  nilRecordVM("i-no-res", owner, false, true, vm.StateStopped),
+		"i-no-inst": nilRecordVM("i-no-inst", owner, true, false, vm.StateStopped),
+	}}
+	svc := &InstanceServiceImpl{vmMgr: mgrWith(map[string]*vm.VM{}), stoppedStore: store, config: &config.Config{}}
+
+	out, err := svc.DescribeStoppedInstances(context.Background(), &ec2.DescribeInstancesInput{}, owner)
+
+	require.NoError(t, err, "CURRENT: no error for the two skipped instances")
+	require.Len(t, out.Reservations, 1)
+	require.Len(t, out.Reservations[0].Instances, 1)
+	assert.Equal(t, "i-good", *out.Reservations[0].Instances[0].InstanceId, "CURRENT: only the complete record is returned")
+}

spinifex/domains/ec2/instance/instance_filter_test.go

@@ -164,6 +164,20 @@ func TestInstanceMatchesFilters_TagKey(t *testing.T) {
 	assert.False(t, instanceMatchesFilters(inst, ic, filters))
 }
 
+// A tag launched without a value carries an empty value in AWS, so tag-key
+// and an empty tag:Key filter still find it.
+func TestInstanceMatchesFilters_NilValueTag(t *testing.T) {
+	inst := &vm.VM{ID: "i-abc"}
+	ic := &ec2.Instance{
+		InstanceId: aws.String("i-abc"),
+		State:      &ec2.InstanceState{Name: aws.String("running")},
+		Tags:       []*ec2.Tag{{Key: aws.String("Name")}},
+	}
+	assert.True(t, instanceMatchesFilters(inst, ic, map[string][]string{"tag-key": {"Name"}}))
+	assert.True(t, instanceMatchesFilters(inst, ic, map[string][]string{"tag:Name": {""}}))
+	assert.False(t, instanceMatchesFilters(inst, ic, map[string][]string{"tag-key": {"Other"}}))
+}
+
 func TestInstanceMatchesFilters_TagValue(t *testing.T) {
 	inst := &vm.VM{ID: "i-abc"}
 	ic := &ec2.Instance{

spinifex/domains/ec2/instance/service.go

@@ -106,6 +106,7 @@ type GPUClaimer interface {
 type StoppedInstanceStore interface {
 	LoadStoppedInstance(instanceID string) (*vm.VM, error)
 	ListStoppedInstances() ([]*vm.VM, error)
+	ListNonTerminatedInstances() ([]*vm.VM, error)
 	ListTerminatedInstances() ([]*vm.VM, error)
 	WriteStoppedInstance(instanceID string, instance *vm.VM) error
 	DeleteStoppedInstance(instanceID string) error

spinifex/domains/ec2/instance/service_impl.go

@@ -1507,7 +1507,7 @@ func (s *InstanceServiceImpl) ReplaceIamProfileAssociation(ctx context.Context,
 // the caller. Supports "instance-id" and "state" filters; unknown filters return
 // InvalidParameterValue.
 func (s *InstanceServiceImpl) DescribeIamProfileAssociations(ctx context.Context, input *ec2.DescribeIamInstanceProfileAssociationsInput, accountID string) (*ec2.DescribeIamInstanceProfileAssociationsOutput, error) {
-	assocFilter := stringPtrSliceToSet(input.AssociationIds)
+	assocFilter := awsidentifiers.Set(input.AssociationIds)
 	instFilter := make(map[string]bool)
 	stateFilter := make(map[string]bool)
 	for _, f := range input.Filters {
@@ -1580,19 +1580,6 @@ func (s *InstanceServiceImpl) findInstanceByAssociationID(associationID, account
 	return owner, true
 }
 
-func stringPtrSliceToSet(in []*string) map[string]bool {
-	if len(in) == 0 {
-		return nil
-	}
-	out := make(map[string]bool, len(in))
-	for _, s := range in {
-		if s != nil && *s != "" {
-			out[*s] = true
-		}
-	}
-	return out
-}
-
 func (s *InstanceServiceImpl) GenerateVolumes(ctx context.Context, input *ec2.RunInstancesInput, instance *vm.VM) ([]VolumeInfo, error) {
 	p := parseVolumeParams(input)
 	if input.ImageId != nil {
@@ -2035,8 +2022,7 @@ func IsInstanceVisibleToCaller(accountID string, v *vm.VM) bool {
 // instanceMatchesFilters checks whether a VM + its built ec2.Instance copy satisfy all parsed awsfilters.
 func instanceMatchesFilters(inst *vm.VM, ic *ec2.Instance, filters map[string][]string) bool {
 	for name, values := range filters {
-		if strings.HasPrefix(name, "tag:") {
-			// tag:Key filters are handled after all field awsfilters.
+		if awsfilters.IsTagFilter(name) {
 			continue
 		}
 
@@ -2075,16 +2061,6 @@ func instanceMatchesFilters(inst *vm.VM, ic *ec2.Instance, filters map[string][]
 				return false
 			}
 			continue
-		case "tag-key":
-			if !matchTagKey(ic.Tags, values) {
-				return false
-			}
-			continue
-		case "tag-value":
-			if !matchTagValue(ic.Tags, values) {
-				return false
-			}
-			continue
 		default:
 			// Filter name passed ParseFilters but has no case — treat as non-match
 			// to avoid silently ignoring it.
@@ -2096,7 +2072,6 @@ func instanceMatchesFilters(inst *vm.VM, ic *ec2.Instance, filters map[string][]
 		}
 	}
 
-	// Check tag:Key filters via the instance's Tag slice.
 	tags := awsfilters.EC2TagsToMap(ic.Tags)
 	return awsfilters.MatchesTags(filters, tags)
 }
@@ -2126,20 +2101,6 @@ func matchNICAnyPrivateIP(nics []*ec2.InstanceNetworkInterface, values []string)
 	})
 }
 
-// matchTagKey returns true if any tag key on the resource matches any of the filter values.
-func matchTagKey(tags []*ec2.Tag, values []string) bool {
-	return slices.ContainsFunc(tags, func(t *ec2.Tag) bool {
-		return t.Key != nil && awsfilters.MatchesAny(values, *t.Key)
-	})
-}
-
-// matchTagValue returns true if any tag value on the resource matches any of the filter values.
-func matchTagValue(tags []*ec2.Tag, values []string) bool {
-	return slices.ContainsFunc(tags, func(t *ec2.Tag) bool {
-		return t.Value != nil && awsfilters.MatchesAny(values, *t.Value)
-	})
-}
-
 // DescribeInstances returns instances on this node visible to the caller's account.
 func (s *InstanceServiceImpl) DescribeInstances(ctx context.Context, input *ec2.DescribeInstancesInput, accountID string) (*ec2.DescribeInstancesOutput, error) {
 	slog.InfoContext(ctx, "Processing DescribeInstances request from this node", "accountID", accountID)
@@ -2273,7 +2234,18 @@ func (s *InstanceServiceImpl) DescribeStoppedInstances(ctx context.Context, inpu
 	if s.stoppedStore == nil {
 		return nil, errors.New(awserrors.ErrorServerInternal)
 	}
-	return s.describeInstancesFromKV(ctx, input, accountID, s.stoppedStore.ListStoppedInstances, 80, "stopped", "DescribeStoppedInstances")
+	return s.describeInstancesFromKV(ctx, input, accountID, s.stoppedStore.ListStoppedInstances, 80, "stopped", "DescribeStoppedInstances", false)
+}
+
+// DescribeAccountInstanceRecords returns every non-terminated instance record
+// the account owns, whichever node holds it. It is the existence source for the
+// strict account inventory, so a record it cannot project fails the call rather
+// than being skipped.
+func (s *InstanceServiceImpl) DescribeAccountInstanceRecords(ctx context.Context, input *ec2.DescribeInstancesInput, accountID string) (*ec2.DescribeInstancesOutput, error) {
+	if s.stoppedStore == nil {
+		return nil, errors.New(awserrors.ErrorServerInternal)
+	}
+	return s.describeInstancesFromKV(ctx, input, accountID, s.stoppedStore.ListNonTerminatedInstances, 0, "pending", "DescribeAccountInstanceRecords", true)
 }
 
 // DescribeTerminatedInstances returns terminated instances from the terminated KV bucket.
@@ -2281,10 +2253,10 @@ func (s *InstanceServiceImpl) DescribeTerminatedInstances(ctx context.Context, i
 	if s.stoppedStore == nil {
 		return nil, errors.New(awserrors.ErrorServerInternal)
 	}
-	return s.describeInstancesFromKV(ctx, input, accountID, s.stoppedStore.ListTerminatedInstances, 48, "terminated", "DescribeTerminatedInstances")
+	return s.describeInstancesFromKV(ctx, input, accountID, s.stoppedStore.ListTerminatedInstances, 48, "terminated", "DescribeTerminatedInstances", false)
 }
 
-func (s *InstanceServiceImpl) describeInstancesFromKV(ctx context.Context, input *ec2.DescribeInstancesInput, accountID string, listFn func() ([]*vm.VM, error), fallbackCode int64, fallbackName, opName string) (*ec2.DescribeInstancesOutput, error) {
+func (s *InstanceServiceImpl) describeInstancesFromKV(ctx context.Context, input *ec2.DescribeInstancesInput, accountID string, listFn func() ([]*vm.VM, error), fallbackCode int64, fallbackName, opName string, strict bool) (*ec2.DescribeInstancesOutput, error) {
 	instanceIDFilter, err := ParseInstanceIDFilter(input.InstanceIds)
 	if err != nil {
 		return nil, err
@@ -2312,6 +2284,11 @@ func (s *InstanceServiceImpl) describeInstancesFromKV(ctx context.Context, input
 			continue
 		}
 		if instance.Reservation == nil || instance.Instance == nil {
+			if strict {
+				slog.ErrorContext(ctx, opName+": record has nil Reservation/Instance, refusing a listing that would omit it",
+					"instanceId", instance.ID)
+				return nil, errors.New(awserrors.ErrorServerInternal)
+			}
 			slog.WarnContext(ctx, opName+": skipping instance with nil Reservation/Instance (data integrity issue)",
 				"instanceId", instance.ID)
 			continue

spinifex/domains/ec2/instance/service_impl_test.go

@@ -1038,7 +1038,7 @@ func TestDescribeInstancesFromKV_FilterExcludesOnlyInstance_ReturnsZeroReservati
 	input := &ec2.DescribeInstancesInput{
 		Filters: []*ec2.Filter{{Name: aws.String("instance-type"), Values: []*string{aws.String("t3.nano")}}},
 	}
-	out, err := svc.describeInstancesFromKV(context.Background(), input, awsidentifiers.GlobalAccountID, listFn, 80, "stopped", "DescribeStoppedInstances")
+	out, err := svc.describeInstancesFromKV(context.Background(), input, awsidentifiers.GlobalAccountID, listFn, 80, "stopped", "DescribeStoppedInstances", false)
 	require.NoError(t, err)
 	assert.Empty(t, out.Reservations, "a reservation with every instance filtered out must not appear in the response")
 }

spinifex/domains/ec2/launchtemplate/service_impl.go

@@ -516,7 +516,7 @@ func (s *LaunchTemplateServiceImpl) commitHeader(ctx context.Context, op, accoun
 		if err == nil {
 			return nil
 		}
-		if !isRevisionConflict(err) {
+		if !kvutil.IsRevisionConflict(err) {
 			slog.ErrorContext(ctx, op+": header commit failed", "launchTemplateId", ltID, "err", err)
 			return err
 		}
@@ -541,11 +541,6 @@ func (s *LaunchTemplateServiceImpl) commitHeader(ctx context.Context, op, accoun
 	return errors.New("header commit retries exhausted")
 }
 
-// isRevisionConflict reports a CAS write that lost to a concurrent writer.
-func isRevisionConflict(err error) bool {
-	return errors.Is(err, jetstream.ErrKeyRevisionMismatch) || errors.Is(err, jetstream.ErrKeyExists)
-}
-
 // abandonVersion removes a body whose header commit failed. A failure is
 // logged and leaves a headerless body, which no ID, name or launch resolves.
 func (s *LaunchTemplateServiceImpl) abandonVersion(ctx context.Context, accountID, ltID string, n int64) {
@@ -616,7 +611,7 @@ func (s *LaunchTemplateServiceImpl) setDefaultVersion(ctx context.Context, accou
 		if err == nil {
 			return header, nil
 		}
-		if !isRevisionConflict(err) {
+		if !kvutil.IsRevisionConflict(err) {
 			slog.ErrorContext(ctx, "ModifyLaunchTemplate: header update failed", "launchTemplateId", header.LaunchTemplateId, "err", err)
 			return nil, errors.New(awserrors.ErrorServerInternal)
 		}
@@ -882,8 +877,8 @@ func (s *LaunchTemplateServiceImpl) DescribeLaunchTemplates(ctx context.Context,
 		}
 	}
 
-	nameSet := stringSet(input.LaunchTemplateNames)
-	idSet := stringSet(input.LaunchTemplateIds)
+	nameSet := awsidentifiers.Set(input.LaunchTemplateNames)
+	idSet := awsidentifiers.Set(input.LaunchTemplateIds)
 
 	prefix := accountID + "."
 	keys, err := s.kv.Keys(ctx)
@@ -944,7 +939,7 @@ func (s *LaunchTemplateServiceImpl) DescribeLaunchTemplates(ctx context.Context,
 
 func templateMatchesFilters(h *LaunchTemplateHeader, filters map[string][]string) bool {
 	for name, values := range filters {
-		if strings.HasPrefix(name, "tag:") {
+		if awsfilters.IsTagFilter(name) {
 			continue
 		}
 		switch name {
@@ -960,17 +955,6 @@ func templateMatchesFilters(h *LaunchTemplateHeader, filters map[string][]string
 			if !awsfilters.MatchesAny(values, h.CreateTime.Format(time.RFC3339)) {
 				return false
 			}
-		case "tag-key":
-			matched := false
-			for k := range h.Tags {
-				if awsfilters.MatchesAny(values, k) {
-					matched = true
-					break
-				}
-			}
-			if !matched {
-				return false
-			}
 		default:
 			return false
 		}
@@ -1177,19 +1161,6 @@ func isHeaderKey(k, prefix string) bool {
 	return strings.HasPrefix(rest, "lt-") && !strings.Contains(rest, ".")
 }
 
-func stringSet(in []*string) map[string]bool {
-	if len(in) == 0 {
-		return nil
-	}
-	set := make(map[string]bool, len(in))
-	for _, s := range in {
-		if s != nil {
-			set[*s] = true
-		}
-	}
-	return set
-}
-
 // validateTemplateID enforces the lt- id prefix so a syntactically invalid id
 // returns Malformed rather than NotFound, matching image/volume id handling.
 func validateTemplateID(id string) error {

spinifex/domains/ec2/launchtemplate/service_impl_test.go

@@ -178,6 +178,12 @@ func TestDescribeLaunchTemplates_TagFilter(t *testing.T) {
 	}, testAccountID)
 	require.NoError(t, err)
 	assert.Empty(t, out.LaunchTemplates)
+
+	out, err = svc.DescribeLaunchTemplates(context.Background(), &ec2.DescribeLaunchTemplatesInput{
+		Filters: []*ec2.Filter{{Name: aws.String("tag-key"), Values: []*string{aws.String("env")}}},
+	}, testAccountID)
+	require.NoError(t, err)
+	assert.Len(t, out.LaunchTemplates, 1)
 }
 
 func TestDescribeLaunchTemplates_AccountIsolation(t *testing.T) {

spinifex/domains/ec2/natgw/lifecycle_invariants_test.go

@@ -3,6 +3,7 @@ package natgw
 import (
 	"encoding/json"
 	"fmt"
+	testutil "github.com/mulgadc/spinifex/internal/testkit"
 	"testing"
 
 	"github.com/aws/aws-sdk-go/aws"
@@ -40,7 +41,7 @@ func TestRLC3_NatGatewayDeletesWhileRouted(t *testing.T) {
 		svc, js := setupTestServiceJS(t)
 		natgwID := createTestNatGateway(t, svc)
 
-		seedKV(t, js, kvBucketRouteTables, map[string][]byte{
+		testutil.SeedKV(t, js, kvBucketRouteTables, map[string][]byte{
 			kvutil.AccountKey(testAccountID, "rtb-routed"): fmt.Appendf(nil,
 				`{"route_table_id":"rtb-routed","vpc_id":"vpc-test1","routes":[{"destination_cidr_block":"0.0.0.0/0","nat_gateway_id":%q}]}`, natgwID),
 		})

spinifex/domains/ec2/natgw/service_impl.go

@@ -238,7 +238,7 @@ func (s *NatGatewayServiceImpl) DeleteNatGateway(ctx context.Context, input *ec2
 
 // kvBucketRouteTables is the route-table KV bucket. Duplicated as a literal
 // (not imported from the routetable package) because routetable imports this
-// package for PublishAddEvent — importing it back would be a cycle.
+// package for KVBucketNatGateways — importing it back would be a cycle.
 const kvBucketRouteTables = "spinifex-vpc-route-tables"
 
 // publishDeleteEventsForNatGateway tears down the SNAT for every private subnet
@@ -368,12 +368,7 @@ func (s *NatGatewayServiceImpl) DescribeNatGateways(ctx context.Context, input *
 		return nil, err
 	}
 
-	natgwIDs := make(map[string]bool)
-	for _, id := range input.NatGatewayIds {
-		if id != nil {
-			natgwIDs[*id] = true
-		}
-	}
+	natgwIDs := awsidentifiers.Set(input.NatGatewayIds)
 
 	prefix := accountID + "."
 	keys, err := s.natgwKV.Keys(ctx)
@@ -470,17 +465,6 @@ func natgwMatchesFilters(record *NatGatewayRecord, filters map[string][]string)
 	return awsfilters.MatchesTags(filters, record.Tags)
 }
 
-// PublishAddEvent publishes a vpc.add-nat-gateway event for vpcd to create the SNAT rule.
-// Called by the route table service when CreateRoute targets a NAT GW.
-func (s *NatGatewayServiceImpl) PublishAddEvent(vpcId, natGatewayId, publicIp, subnetCidr string) {
-	projection.New(s.natsConn).AddNATGateway(networkv1.NATGatewayEvent{
-		VpcId:        vpcId,
-		NatGatewayId: natGatewayId,
-		PublicIp:     publicIp,
-		SubnetCidr:   subnetCidr,
-	})
-}
-
 // PublishDeleteEvent publishes a vpc.delete-nat-gateway event for vpcd to remove the SNAT rule.
 func (s *NatGatewayServiceImpl) PublishDeleteEvent(vpcId, natGatewayId, publicIp, subnetCidr string) {
 	projection.New(s.natsConn).RemoveNATGateway(networkv1.NATGatewayEvent{
@@ -491,23 +475,6 @@ func (s *NatGatewayServiceImpl) PublishDeleteEvent(vpcId, natGatewayId, publicIp
 	})
 }
 
-// GetNatGateway retrieves a NAT Gateway record by ID.
-func (s *NatGatewayServiceImpl) GetNatGateway(accountID, natgwID string) (*NatGatewayRecord, error) {
-	ctx := context.Background()
-	entry, err := s.natgwKV.Get(ctx, kvutil.AccountKey(accountID, natgwID))
-	if err != nil {
-		if errors.Is(err, jetstream.ErrKeyNotFound) {
-			return nil, errors.New(awserrors.ErrorInvalidNatGatewayIDNotFound)
-		}
-		return nil, errors.New(awserrors.ErrorServerInternal)
-	}
-	var record NatGatewayRecord
-	if err := json.Unmarshal(entry.Value(), &record); err != nil {
-		return nil, errors.New(awserrors.ErrorServerInternal)
-	}
-	return &record, nil
-}
-
 func recordToEC2(record *NatGatewayRecord) *ec2.NatGateway {
 	ngw := &ec2.NatGateway{
 		NatGatewayId:     aws.String(record.NatGatewayId),

spinifex/domains/ec2/natgw/service_impl_test.go

@@ -30,17 +30,17 @@ func setupTestServiceJS(t *testing.T) (*NatGatewayServiceImpl, jetstream.JetStre
 	js := testutil.NewJetStream(t, nc)
 
 	// Seed VPC KV
-	seedKV(t, js, ec2vpc.KVBucketVPCs, map[string][]byte{
+	testutil.SeedKV(t, js, ec2vpc.KVBucketVPCs, map[string][]byte{
 		kvutil.AccountKey(testAccountID, "vpc-test1"): []byte(`{"vpc_id":"vpc-test1","cidr_block":"10.0.0.0/16","state":"available"}`),
 	})
 
 	// Seed subnet KV (public subnet)
-	seedKV(t, js, ec2vpc.KVBucketSubnets, map[string][]byte{
+	testutil.SeedKV(t, js, ec2vpc.KVBucketSubnets, map[string][]byte{
 		kvutil.AccountKey(testAccountID, "subnet-pub1"): []byte(`{"subnet_id":"subnet-pub1","vpc_id":"vpc-test1","cidr_block":"10.0.1.0/24","state":"available","map_public_ip_on_launch":true}`),
 	})
 
 	// Seed EIP KV
-	seedKV(t, js, ec2eip.KVBucketEIPs, map[string][]byte{
+	testutil.SeedKV(t, js, ec2eip.KVBucketEIPs, map[string][]byte{
 		kvutil.AccountKey(testAccountID, "eipalloc-test1"): []byte(`{"allocation_id":"eipalloc-test1","public_ip":"203.0.113.50","state":"allocated"}`),
 		kvutil.AccountKey(testAccountID, "eipalloc-used"):  []byte(`{"allocation_id":"eipalloc-used","public_ip":"203.0.113.51","state":"associated","association_id":"eipassoc-xxx"}`),
 	})
@@ -50,17 +50,6 @@ func setupTestServiceJS(t *testing.T) (*NatGatewayServiceImpl, jetstream.JetStre
 	return svc, js
 }
 
-func seedKV(t *testing.T, js jetstream.JetStream, bucket string, entries map[string][]byte) jetstream.KeyValue {
-	t.Helper()
-	kv, err := js.CreateOrUpdateKeyValue(t.Context(), jetstream.KeyValueConfig{Bucket: bucket, History: 1})
-	require.NoError(t, err)
-	for key, value := range entries {
-		_, err := kv.Put(t.Context(), key, value)
-		require.NoError(t, err)
-	}
-	return kv
-}
-
 func TestDeletedBucketTTL(t *testing.T) {
 	_, js := setupTestServiceJS(t)
 	kv, err := js.KeyValue(t.Context(), KVBucketDeletedNatGateways)
@@ -181,8 +170,8 @@ func TestDeleteNatGateway(t *testing.T) {
 	assert.Equal(t, natgwID, *deleteOut.NatGatewayId)
 
 	// Should be gone
-	_, err = svc.GetNatGateway(testAccountID, natgwID)
-	assert.EqualError(t, err, awserrors.ErrorInvalidNatGatewayIDNotFound)
+	_, err = svc.natgwKV.Get(t.Context(), kvutil.AccountKey(testAccountID, natgwID))
+	assert.ErrorIs(t, err, jetstream.ErrKeyNotFound)
 }
 
 func TestDescribeNatGateways(t *testing.T) {
@@ -245,28 +234,6 @@ func TestDescribeNatGateways_FilterByState(t *testing.T) {
 	assert.Empty(t, out.NatGateways)
 }
 
-func TestGetNatGateway(t *testing.T) {
-	svc := setupTestService(t)
-	createOut, err := svc.CreateNatGateway(t.Context(), &ec2.CreateNatGatewayInput{
-		SubnetId:     aws.String("subnet-pub1"),
-		AllocationId: aws.String("eipalloc-test1"),
-	}, testAccountID)
-	require.NoError(t, err)
-	natgwID := *createOut.NatGateway.NatGatewayId
-
-	record, err := svc.GetNatGateway(testAccountID, natgwID)
-	require.NoError(t, err)
-	assert.Equal(t, natgwID, record.NatGatewayId)
-	assert.Equal(t, "203.0.113.50", record.PublicIp)
-	assert.Equal(t, "eipalloc-test1", record.AllocationId)
-}
-
-func TestGetNatGateway_NotFound(t *testing.T) {
-	svc := setupTestService(t)
-	_, err := svc.GetNatGateway(testAccountID, "nat-nope")
-	assert.EqualError(t, err, awserrors.ErrorInvalidNatGatewayIDNotFound)
-}
-
 func TestCreateNatGateway_MissingParams(t *testing.T) {
 	svc := setupTestService(t)
 

spinifex/domains/ec2/placementgroup/service_impl.go

@@ -227,18 +227,8 @@ func (s *PlacementGroupServiceImpl) DescribePlacementGroups(ctx context.Context,
 	}
 
 	// Build filter maps for GroupNames/GroupIds parameters
-	nameSet := make(map[string]bool)
-	for _, name := range input.GroupNames {
-		if name != nil {
-			nameSet[*name] = true
-		}
-	}
-	idSet := make(map[string]bool)
-	for _, id := range input.GroupIds {
-		if id != nil {
-			idSet[*id] = true
-		}
-	}
+	nameSet := awsidentifiers.Set(input.GroupNames)
+	idSet := awsidentifiers.Set(input.GroupIds)
 
 	prefix := accountID + "."
 	keys, err := s.kv.Keys(ctx)

spinifex/domains/ec2/routetable/service_impl.go

@@ -615,12 +615,7 @@ func (s *RouteTableServiceImpl) DescribeRouteTables(ctx context.Context, input *
 		return nil, err
 	}
 
-	rtbIDs := make(map[string]bool)
-	for _, id := range input.RouteTableIds {
-		if id != nil {
-			rtbIDs[*id] = true
-		}
-	}
+	rtbIDs := awsidentifiers.Set(input.RouteTableIds)
 
 	parsedFilters, err := awsfilters.ParseFilters(input.Filters, describeRouteTablesValidFilters)
 	if err != nil {

spinifex/domains/ec2/routetable/service_impl_test.go

@@ -15,7 +15,6 @@ import (
 	"github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
 	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
 	"github.com/nats-io/nats.go"
-	"github.com/nats-io/nats.go/jetstream"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 )
@@ -28,23 +27,23 @@ func setupTestService(t *testing.T) *RouteTableServiceImpl {
 	js := testutil.NewJetStream(t, nc)
 
 	// Seed VPC KV
-	seedKV(t, js, ec2vpc.KVBucketVPCs, map[string][]byte{
+	testutil.SeedKV(t, js, ec2vpc.KVBucketVPCs, map[string][]byte{
 		kvutil.AccountKey(testAccountID, "vpc-test1"): []byte(`{"vpc_id":"vpc-test1","cidr_block":"10.0.0.0/16","state":"available"}`),
 	})
 
 	// Seed IGW KV (attached to vpc-test1)
-	seedKV(t, js, ec2igw.KVBucketIGW, map[string][]byte{
+	testutil.SeedKV(t, js, ec2igw.KVBucketIGW, map[string][]byte{
 		kvutil.AccountKey(testAccountID, "igw-test1"): []byte(`{"internet_gateway_id":"igw-test1","vpc_id":"vpc-test1","state":"available"}`),
 	})
 
 	// Seed subnet KV
-	seedKV(t, js, ec2vpc.KVBucketSubnets, map[string][]byte{
+	testutil.SeedKV(t, js, ec2vpc.KVBucketSubnets, map[string][]byte{
 		kvutil.AccountKey(testAccountID, "subnet-test1"): []byte(`{"subnet_id":"subnet-test1","vpc_id":"vpc-test1","cidr_block":"10.0.1.0/24","state":"available"}`),
 		kvutil.AccountKey(testAccountID, "subnet-priv1"): []byte(`{"subnet_id":"subnet-priv1","vpc_id":"vpc-test1","cidr_block":"10.0.11.0/24","state":"available"}`),
 	})
 
 	// Seed NAT Gateway KV
-	seedKV(t, js, ec2natgw.KVBucketNatGateways, map[string][]byte{
+	testutil.SeedKV(t, js, ec2natgw.KVBucketNatGateways, map[string][]byte{
 		kvutil.AccountKey(testAccountID, "nat-test1"): []byte(`{"nat_gateway_id":"nat-test1","vpc_id":"vpc-test1","subnet_id":"subnet-test1","public_ip":"192.168.1.100","state":"available"}`),
 	})
 
@@ -79,17 +78,6 @@ func drainEvents(t *testing.T, svc *RouteTableServiceImpl, sub *nats.Subscriptio
 	}
 }
 
-func seedKV(t *testing.T, js jetstream.JetStream, bucket string, entries map[string][]byte) jetstream.KeyValue {
-	t.Helper()
-	kv, err := js.CreateOrUpdateKeyValue(t.Context(), jetstream.KeyValueConfig{Bucket: bucket, History: 1})
-	require.NoError(t, err)
-	for key, value := range entries {
-		_, err := kv.Put(t.Context(), key, value)
-		require.NoError(t, err)
-	}
-	return kv
-}
-
 func createTestRtb(t *testing.T, svc *RouteTableServiceImpl) string {
 	t.Helper()
 	out, err := svc.CreateRouteTable(t.Context(), &ec2.CreateRouteTableInput{
@@ -192,7 +180,7 @@ func TestCreateRouteTable_VpcNotFound(t *testing.T) {
 	_, err := svc.CreateRouteTable(t.Context(), &ec2.CreateRouteTableInput{
 		VpcId: aws.String("vpc-nonexistent"),
 	}, testAccountID)
-	requireAWSError(t, err, awserrors.ErrorInvalidVpcIDNotFound, "The vpc ID 'vpc-nonexistent' does not exist")
+	testutil.RequireAWSError(t, err, awserrors.ErrorInvalidVpcIDNotFound, "The vpc ID 'vpc-nonexistent' does not exist")
 }
 
 func TestDeleteRouteTable(t *testing.T) {
@@ -207,7 +195,7 @@ func TestDeleteRouteTable(t *testing.T) {
 
 	// Should be gone
 	_, err = svc.getRouteTable(t.Context(), testAccountID, rtbID)
-	requireAWSError(t, err, awserrors.ErrorInvalidRouteTableIDNotFound, "The routeTable ID '"+rtbID+"' does not exist")
+	testutil.RequireAWSError(t, err, awserrors.ErrorInvalidRouteTableIDNotFound, "The routeTable ID '"+rtbID+"' does not exist")
 }
 
 func TestDeleteRouteTable_Main(t *testing.T) {
@@ -219,7 +207,7 @@ func TestDeleteRouteTable_Main(t *testing.T) {
 	_, err = svc.DeleteRouteTable(t.Context(), &ec2.DeleteRouteTableInput{
 		RouteTableId: aws.String(record.RouteTableId),
 	}, testAccountID)
-	requireAWSError(t, err, awserrors.ErrorDependencyViolation, "The routeTable '"+record.RouteTableId+"' has dependencies and cannot be deleted.")
+	testutil.RequireAWSError(t, err, awserrors.ErrorDependencyViolation, "The routeTable '"+record.RouteTableId+"' has dependencies and cannot be deleted.")
 }
 
 func TestDeleteRouteTable_WithAssociations(t *testing.T) {
@@ -238,7 +226,7 @@ func TestDeleteRouteTable_WithAssociations(t *testing.T) {
 	_, err = svc.DeleteRouteTable(t.Context(), &ec2.DeleteRouteTableInput{
 		RouteTableId: aws.String(rtbID),
 	}, testAccountID)
-	requireAWSError(t, err, awserrors.ErrorDependencyViolation, "The routeTable '"+rtbID+"' has dependencies and cannot be deleted.")
+	testutil.RequireAWSError(t, err, awserrors.ErrorDependencyViolation, "The routeTable '"+rtbID+"' has dependencies and cannot be deleted.")
 }
 
 func TestDescribeRouteTables(t *testing.T) {
@@ -427,7 +415,7 @@ func TestCreateRoute_DuplicateDestination(t *testing.T) {
 		DestinationCidrBlock: aws.String("0.0.0.0/0"),
 		NatGatewayId:         aws.String("nat-test1"),
 	}, testAccountID)
-	requireAWSError(t, err, awserrors.ErrorRouteAlreadyExists, "The route identified by 0.0.0.0/0 already exists.")
+	testutil.RequireAWSError(t, err, awserrors.ErrorRouteAlreadyExists, "The route identified by 0.0.0.0/0 already exists.")
 
 	// AWS checks the target before the destination: "local" is no gateway
 	_, err = svc.CreateRoute(t.Context(), &ec2.CreateRouteInput{
@@ -435,7 +423,7 @@ func TestCreateRoute_DuplicateDestination(t *testing.T) {
 		DestinationCidrBlock: aws.String("10.0.0.0/16"),
 		GatewayId:            aws.String("local"),
 	}, testAccountID)
-	requireAWSError(t, err, awserrors.ErrorInvalidGatewayIDNotFound, "The gateway ID 'local' does not exist")
+	testutil.RequireAWSError(t, err, awserrors.ErrorInvalidGatewayIDNotFound, "The gateway ID 'local' does not exist")
 
 	// A gateway route to the VPC's CIDR, or inside it, is refused before the
 	// existing local route is considered
@@ -445,7 +433,7 @@ func TestCreateRoute_DuplicateDestination(t *testing.T) {
 			DestinationCidrBlock: aws.String(dest),
 			GatewayId:            aws.String("igw-test1"),
 		}, testAccountID)
-		requireAWSError(t, err, awserrors.ErrorInvalidParameterValue,
+		testutil.RequireAWSError(t, err, awserrors.ErrorInvalidParameterValue,
 			"The destination CIDR block "+dest+" is equal to or more specific than one of this VPC's CIDR blocks. This route can target only an interface or an instance.")
 	}
 
@@ -502,7 +490,7 @@ func TestDeleteRoute_NotFound(t *testing.T) {
 		RouteTableId:         aws.String(rtbID),
 		DestinationCidrBlock: aws.String("192.168.0.0/16"),
 	}, testAccountID)
-	requireAWSError(t, err, awserrors.ErrorInvalidRouteNotFound, "no route with destination-cidr-block 192.168.0.0/16 in route table "+rtbID)
+	testutil.RequireAWSError(t, err, awserrors.ErrorInvalidRouteNotFound, "no route with destination-cidr-block 192.168.0.0/16 in route table "+rtbID)
 }
 
 func TestReplaceRoute(t *testing.T) {
@@ -526,14 +514,6 @@ func TestReplaceRoute(t *testing.T) {
 	require.NoError(t, err)
 }
 
-func requireAWSError(t *testing.T, err error, code, message string) {
-	t.Helper()
-	got, msg, ok := awserrors.ResolveErrorDetail(err)
-	require.True(t, ok, "error %v carries no AWS code", err)
-	assert.Equal(t, code, got)
-	assert.Equal(t, message, msg)
-}
-
 func TestCreateRoute_MissingGateway(t *testing.T) {
 	t.Parallel()
 	svc := setupTestService(t)
@@ -544,7 +524,7 @@ func TestCreateRoute_MissingGateway(t *testing.T) {
 		DestinationCidrBlock: aws.String("0.0.0.0/0"),
 		GatewayId:            aws.String("igw-0000000000000dead"),
 	}, testAccountID)
-	requireAWSError(t, err, awserrors.ErrorInvalidGatewayIDNotFound, "The gateway ID 'igw-0000000000000dead' does not exist")
+	testutil.RequireAWSError(t, err, awserrors.ErrorInvalidGatewayIDNotFound, "The gateway ID 'igw-0000000000000dead' does not exist")
 }
 
 func TestReplaceRoute_MissingRouteOrGateway(t *testing.T) {
@@ -557,7 +537,7 @@ func TestReplaceRoute_MissingRouteOrGateway(t *testing.T) {
 		DestinationCidrBlock: aws.String("198.51.100.0/24"),
 		GatewayId:            aws.String("igw-test1"),
 	}, testAccountID)
-	requireAWSError(t, err, awserrors.ErrorInvalidParameterValue,
+	testutil.RequireAWSError(t, err, awserrors.ErrorInvalidParameterValue,
 		"There is no route defined for '198.51.100.0/24' in the route table. Use CreateRoute instead.")
 
 	// AWS reports the missing gateway ahead of the missing route.
@@ -566,7 +546,7 @@ func TestReplaceRoute_MissingRouteOrGateway(t *testing.T) {
 		DestinationCidrBlock: aws.String("198.51.100.0/24"),
 		GatewayId:            aws.String("igw-0000000000000dead"),
 	}, testAccountID)
-	requireAWSError(t, err, awserrors.ErrorInvalidGatewayIDNotFound, "The gateway ID 'igw-0000000000000dead' does not exist")
+	testutil.RequireAWSError(t, err, awserrors.ErrorInvalidGatewayIDNotFound, "The gateway ID 'igw-0000000000000dead' does not exist")
 }
 
 func TestAssociateRouteTable(t *testing.T) {

spinifex/domains/ec2/spotinstance/account_scan.go

@@ -0,0 +1,81 @@
+package spotinstance
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"strings"
+
+	"github.com/aws/aws-sdk-go/aws"
+	"github.com/nats-io/nats.go/jetstream"
+
+	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
+)
+
+// AccountSpotRequest is one persisted request in an account scan.
+type AccountSpotRequest struct {
+	RequestID string
+	State     string
+}
+
+// AccountScan lists every request an account holds in both buckets.
+type AccountScan struct {
+	Active   []AccountSpotRequest
+	Terminal []AccountSpotRequest
+}
+
+// ScanAccount lists the account's requests, or fails. Unlike
+// DescribeSpotInstanceRequests it never skips a record it cannot read: a
+// shorter list would be taken as proof that nothing is left.
+func (s *SpotInstanceServiceImpl) ScanAccount(ctx context.Context, accountID string) (*AccountScan, error) {
+	if accountID == "" {
+		return nil, errors.New("spot request scan needs an account ID")
+	}
+	active, err := scanBucket(ctx, s.activeKV, accountID)
+	if err != nil {
+		return nil, fmt.Errorf("scan active spot requests: %w", err)
+	}
+	terminal, err := scanBucket(ctx, s.terminalKV, accountID)
+	if err != nil {
+		return nil, fmt.Errorf("scan terminal spot requests: %w", err)
+	}
+	return &AccountScan{Active: active, Terminal: terminal}, nil
+}
+
+func scanBucket(ctx context.Context, kv jetstream.KeyValue, accountID string) ([]AccountSpotRequest, error) {
+	keys, err := kv.Keys(ctx)
+	if err != nil {
+		if errors.Is(err, jetstream.ErrNoKeysFound) {
+			return nil, nil
+		}
+		return nil, err
+	}
+	prefix := accountID + "."
+	var out []AccountSpotRequest
+	for _, key := range keys {
+		if key == kvutil.VersionKey || !strings.HasPrefix(key, prefix) {
+			continue
+		}
+		entry, err := kv.Get(ctx, key)
+		if err != nil {
+			// Expired from the terminal bucket between listing and read.
+			if errors.Is(err, jetstream.ErrKeyNotFound) {
+				continue
+			}
+			return nil, fmt.Errorf("read %s: %w", key, err)
+		}
+		var record SpotRequestRecord
+		if err := json.Unmarshal(entry.Value(), &record); err != nil {
+			return nil, fmt.Errorf("decode %s: %w", key, err)
+		}
+		if record.Request == nil || aws.StringValue(record.Request.SpotInstanceRequestId) == "" {
+			return nil, fmt.Errorf("record %s carries no request", key)
+		}
+		out = append(out, AccountSpotRequest{
+			RequestID: aws.StringValue(record.Request.SpotInstanceRequestId),
+			State:     aws.StringValue(record.Request.State),
+		})
+	}
+	return out, nil
+}

spinifex/domains/ec2/spotinstance/account_scan_test.go

@@ -0,0 +1,84 @@
+package spotinstance
+
+import (
+	"context"
+	"errors"
+	"testing"
+
+	"github.com/aws/aws-sdk-go/aws"
+	"github.com/aws/aws-sdk-go/service/ec2"
+	"github.com/nats-io/nats.go/jetstream"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+
+	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
+)
+
+func TestScanAccountListsActiveAndTerminalForOneAccount(t *testing.T) {
+	svc := setupTestService(t)
+	putSIRs(t, svc, makeSIR("sir-aaa", "i-aaa"), makeSIR("sir-bbb", "i-bbb"))
+	_, err := svc.CancelSpotInstanceRequests(t.Context(), &ec2.CancelSpotInstanceRequestsInput{
+		SpotInstanceRequestIds: aws.StringSlice([]string{"sir-bbb"}),
+	}, testAccountID)
+	require.NoError(t, err)
+	_, err = svc.PutSpotInstanceRequests(t.Context(), &PutSpotRequestsInput{Requests: []*ec2.SpotInstanceRequest{makeSIR("sir-other", "i-o")}}, "999999999999")
+	require.NoError(t, err)
+
+	scan, err := svc.ScanAccount(t.Context(), testAccountID)
+	require.NoError(t, err)
+	require.Len(t, scan.Active, 1)
+	assert.Equal(t, "sir-aaa", scan.Active[0].RequestID)
+	require.Len(t, scan.Terminal, 1)
+	assert.Equal(t, "sir-bbb", scan.Terminal[0].RequestID)
+}
+
+func TestScanAccountFailsOnAnUndecodableRecord(t *testing.T) {
+	svc := setupTestService(t)
+	putSIRs(t, svc, makeSIR("sir-aaa", "i-aaa"))
+	_, err := svc.activeKV.Put(t.Context(), kvutil.AccountKey(testAccountID, "sir-bad"), []byte("{not json"))
+	require.NoError(t, err)
+
+	scan, err := svc.ScanAccount(t.Context(), testAccountID)
+	require.Error(t, err)
+	assert.Nil(t, scan)
+}
+
+func TestScanAccountFailsOnARecordWithoutARequest(t *testing.T) {
+	svc := setupTestService(t)
+	_, err := svc.terminalKV.Put(t.Context(), kvutil.AccountKey(testAccountID, "sir-empty"), []byte(`{"account_id":"x"}`))
+	require.NoError(t, err)
+
+	_, err = svc.ScanAccount(t.Context(), testAccountID)
+	require.Error(t, err)
+}
+
+func TestScanAccountOfAnEmptyAccountIsEmptyNotAnError(t *testing.T) {
+	svc := setupTestService(t)
+	scan, err := svc.ScanAccount(t.Context(), testAccountID)
+	require.NoError(t, err)
+	assert.Empty(t, scan.Active)
+	assert.Empty(t, scan.Terminal)
+}
+
+type failingGetKV struct {
+	jetstream.KeyValue
+
+	key string
+}
+
+func (k failingGetKV) Get(ctx context.Context, key string) (jetstream.KeyValueEntry, error) {
+	if key == k.key {
+		return nil, errors.New("injected read failure")
+	}
+	return k.KeyValue.Get(ctx, key)
+}
+
+func TestScanAccountFailsWhenARecordCannotBeRead(t *testing.T) {
+	svc := setupTestService(t)
+	putSIRs(t, svc, makeSIR("sir-aaa", "i-aaa"), makeSIR("sir-bbb", "i-bbb"))
+	svc.activeKV = failingGetKV{KeyValue: svc.activeKV, key: kvutil.AccountKey(testAccountID, "sir-bbb")}
+
+	scan, err := svc.ScanAccount(t.Context(), testAccountID)
+	require.Error(t, err)
+	assert.Nil(t, scan)
+}

spinifex/domains/ec2/spotinstance/service_impl.go

@@ -5,8 +5,8 @@ import (
 	"encoding/json"
 	"errors"
 	"fmt"
+	"github.com/mulgadc/spinifex/spinifex/foundation/aws/identifiers"
 	"log/slog"
-	"slices"
 	"strings"
 	"time"
 
@@ -121,12 +121,7 @@ func (s *SpotInstanceServiceImpl) DescribeSpotInstanceRequests(ctx context.Conte
 		return nil, err
 	}
 
-	idSet := make(map[string]bool)
-	for _, id := range input.SpotInstanceRequestIds {
-		if id != nil {
-			idSet[*id] = true
-		}
-	}
+	idSet := identifiers.Set(input.SpotInstanceRequestIds)
 
 	active, err := s.listRequests(ctx, s.activeKV, accountID)
 	if err != nil {
@@ -310,7 +305,7 @@ func setSpotStatus(req *ec2.SpotInstanceRequest, code, message string) {
 }
 
 // sirMatchesFilters reports whether a request matches all non-tag awsfilters.
-// tag:* filters are handled separately via awsfilters.MatchesTags.
+// Tag filters are handled separately via awsfilters.MatchesTags.
 func sirMatchesFilters(req *ec2.SpotInstanceRequest, filters map[string][]string) bool {
 	for name, values := range filters {
 		switch {
@@ -346,11 +341,7 @@ func sirMatchesFilters(req *ec2.SpotInstanceRequest, filters map[string][]string
 			if !awsfilters.MatchesAny(values, aws.StringValue(launchSpec(req).KeyName)) {
 				return false
 			}
-		case name == "tag-key":
-			if !sirMatchesTagKey(req.Tags, values) {
-				return false
-			}
-		case strings.HasPrefix(name, "tag:"):
+		case awsfilters.IsTagFilter(name):
 			// Handled by awsfilters.MatchesTags.
 		default:
 			return false
@@ -359,12 +350,6 @@ func sirMatchesFilters(req *ec2.SpotInstanceRequest, filters map[string][]string
 	return true
 }
 
-func sirMatchesTagKey(tags []*ec2.Tag, values []string) bool {
-	return slices.ContainsFunc(tags, func(t *ec2.Tag) bool {
-		return t.Key != nil && awsfilters.MatchesAny(values, *t.Key)
-	})
-}
-
 // launchSpec returns the request's launch specification, or an empty one when
 // absent, so field reads in filter matching need no nil guard.
 func launchSpec(req *ec2.SpotInstanceRequest) *ec2.LaunchSpecification {

spinifex/domains/ec2/vpc/GetSecurityGroupsForVpc_test.go

@@ -189,6 +189,44 @@ func TestGetSecurityGroupsForVpc_PagesEveryGroupOnce(t *testing.T) {
 	assert.Len(t, seen, len(expected), "a group was reported on more than one page")
 }
 
+func TestGetSecurityGroupsForVpc_DeleteBetweenPagesSkipsNothing(t *testing.T) {
+	t.Parallel()
+	svc := setupTestVPCService(t)
+	vpcID := createTestVPC(t, svc, "10.0.0.0/16")
+	for _, name := range []string{"grp-a", "grp-b", "grp-c", "grp-d", "grp-e", "grp-f", "grp-g"} {
+		createTestSG(t, svc, vpcID, name)
+	}
+
+	first, err := svc.GetSecurityGroupsForVpc(context.Background(), &ec2.GetSecurityGroupsForVpcInput{
+		VpcId:      aws.String(vpcID),
+		MaxResults: aws.Int64(5),
+	}, testAccountID)
+	require.NoError(t, err)
+	require.NotNil(t, first.NextToken)
+
+	var deleted string
+	for _, g := range first.SecurityGroupForVpcs {
+		if aws.StringValue(g.GroupName) != "default" {
+			deleted = aws.StringValue(g.GroupId)
+			break
+		}
+	}
+	_, err = svc.DeleteSecurityGroup(context.Background(), &ec2.DeleteSecurityGroupInput{GroupId: aws.String(deleted)}, testAccountID)
+	require.NoError(t, err)
+
+	second, err := svc.GetSecurityGroupsForVpc(context.Background(), &ec2.GetSecurityGroupsForVpcInput{
+		VpcId:      aws.String(vpcID),
+		MaxResults: aws.Int64(5),
+		NextToken:  first.NextToken,
+	}, testAccountID)
+	require.NoError(t, err)
+	assert.Nil(t, second.NextToken)
+
+	seen := append(groupIDsForVpc(first), groupIDsForVpc(second)...)
+	expected := append(describedGroupIDsForVpc(t, svc, vpcID), deleted)
+	assert.ElementsMatch(t, expected, seen)
+}
+
 func TestGetSecurityGroupsForVpc_MalformedNextToken(t *testing.T) {
 	t.Parallel()
 	svc := setupTestVPCService(t)
@@ -199,7 +237,7 @@ func TestGetSecurityGroupsForVpc_MalformedNextToken(t *testing.T) {
 		NextToken: aws.String("not-a-token"),
 	}, testAccountID)
 	require.Error(t, err)
-	assert.Equal(t, awserrors.ErrorInvalidParameterValue, err.Error())
+	assert.True(t, awserrors.IsErrorCode(err, awserrors.ErrorInvalidParameterValue), err.Error())
 }
 
 func TestGetSecurityGroupsForVpc_Filters(t *testing.T) {

spinifex/domains/ec2/vpc/eni_abandoned_test.go

@@ -2,6 +2,7 @@ package vpc
 
 import (
 	"context"
+	testutil "github.com/mulgadc/spinifex/internal/testkit"
 	"testing"
 	"time"
 
@@ -137,7 +138,7 @@ func TestAbandonedENIBlocksSecurityGroupDelete(t *testing.T) {
 		GroupId: aws.String(sgID),
 	}, testAccountID)
 	require.Error(t, err)
-	requireAWSError(t, err, awserrors.ErrorDependencyViolation, "resource "+sgID+" has a dependent object")
+	testutil.RequireAWSError(t, err, awserrors.ErrorDependencyViolation, "resource "+sgID+" has a dependent object")
 
 	orphans, err := svc.ListAbandonedInstanceENIs(context.Background(), 15*time.Minute)
 	require.NoError(t, err)

spinifex/domains/ec2/vpc/eni_test.go

@@ -497,7 +497,7 @@ func TestDescribeNetworkInterfaces_NotFound(t *testing.T) {
 	_, err := svc.DescribeNetworkInterfaces(context.Background(), &ec2.DescribeNetworkInterfacesInput{
 		NetworkInterfaceIds: []*string{aws.String("eni-0000000000000dead")},
 	}, testAccountID)
-	requireAWSError(t, err, awserrors.ErrorInvalidNetworkInterfaceIDNotFound, "The networkInterface ID 'eni-0000000000000dead' does not exist")
+	testutil.RequireAWSError(t, err, awserrors.ErrorInvalidNetworkInterfaceIDNotFound, "The networkInterface ID 'eni-0000000000000dead' does not exist")
 }
 
 func TestDescribeNetworkInterfaces_MalformedID(t *testing.T) {
@@ -511,9 +511,9 @@ func TestDescribeNetworkInterfaces_MalformedID(t *testing.T) {
 	}
 
 	for _, id := range []string{"eni-xyz", "eni-", "eni-ABC"} {
-		requireAWSError(t, describe(id), awserrors.ErrorInvalidNetworkInterfaceIdMalformed, fmt.Sprintf("Invalid id: %q", id))
+		testutil.RequireAWSError(t, describe(id), awserrors.ErrorInvalidNetworkInterfaceIdMalformed, fmt.Sprintf("Invalid id: %q", id))
 	}
-	requireAWSError(t, describe("foo"), awserrors.ErrorInvalidNetworkInterfaceIdMalformed, `Invalid id: "foo" (expecting "eni-...")`)
+	testutil.RequireAWSError(t, describe("foo"), awserrors.ErrorInvalidNetworkInterfaceIdMalformed, `Invalid id: "foo" (expecting "eni-...")`)
 	// Unlike a group ID, any length of lowercase hex is merely unknown.
 	for _, id := range []string{"eni-12345", "eni-000000000000000000001"} {
 		requireAWSCode(t, describe(id), awserrors.ErrorInvalidNetworkInterfaceIDNotFound)

spinifex/domains/ec2/vpc/external_ipam.go

@@ -12,19 +12,6 @@ import (
 	"github.com/nats-io/nats.go/jetstream"
 )
 
-// Backward-compatible aliases so callers (daemon, EIP, EC2 handlers, tests)
-// keep importing record + allocation shapes from this package while the
-// allocator implementation lives in network/external.
-type (
-	ExternalIPAllocation = external.ExternalIPAllocation
-	ExternalIPAMRecord   = external.PoolRecord
-)
-
-const (
-	KVBucketExternalIPAM        = external.KVBucketStaticPool
-	KVBucketExternalIPAMVersion = external.KVBucketStaticPoolVersion
-)
-
 // ExternalIPAM is the AWS-facing entry point for external IP allocation,
 // dispatching per pool name to the static allocator or to whatever the wiring
 // installed for that pool.
@@ -199,11 +186,11 @@ func (m *ExternalIPAM) BindOwner(ctx context.Context, poolName, ip, eniID string
 
 // GetPoolRecord returns the current IPAM record for a pool. DHCP-sourced
 // pools have no static record — the per-AZ lease bucket is authoritative.
-func (m *ExternalIPAM) GetPoolRecord(poolName string) (*ExternalIPAMRecord, error) {
+func (m *ExternalIPAM) GetPoolRecord(poolName string) (*external.PoolRecord, error) {
 	return m.getPoolRecord(context.Background(), poolName)
 }
 
-func (m *ExternalIPAM) getPoolRecord(ctx context.Context, poolName string) (*ExternalIPAMRecord, error) {
+func (m *ExternalIPAM) getPoolRecord(ctx context.Context, poolName string) (*external.PoolRecord, error) {
 	if m.static == nil {
 		return nil, fmt.Errorf("pool record unavailable: no static allocator")
 	}

spinifex/domains/ec2/vpc/purpose.go

@@ -4,28 +4,8 @@ package vpc
 // clusters can reclaim / audit by (owner, purpose). The constants match the
 // JSON values written to KV records.
 const (
-	PurposeENIPrimary    = "eni-primary"
-	PurposeENISecondary  = "eni-secondary"
-	PurposeENIPublic     = "eni-public" // auto-assigned public IP
-	PurposeEIP           = "eip"
-	PurposeNATGWExternal = "natgw-external"
-	PurposeIGWLRP        = "igw-lrp"
-	// Future-reserved values; defined now so call sites and migrations can
-	// reference them without a follow-up rename.
-	PurposeEdgeLRExternal = "edge-lr-external"
-	PurposeNodeMgmt       = "node-mgmt"       // currently .env
-	PurposeSubnetRouter   = "subnet-router"   // .1
-	PurposeSubnetDNS      = "subnet-dns"      // .2
-	PurposeSubnetReserved = "subnet-reserved" // .0, .3, .last
-	// PurposeUnknown tags records the migration could not classify (e.g.
-	// orphan IPs with no owning ENIRecord). Surfaces in logs for triage.
-	PurposeUnknown = "unknown"
+	PurposeENIPrimary = "eni-primary"
+	PurposeENIPublic  = "eni-public" // auto-assigned public IP
+	PurposeEIP        = "eip"
+	PurposeIGWLRP     = "igw-lrp"
 )
-
-// LegacyExternalTypeToPurpose maps the old ExternalIPAllocation.Type values
-// onto the new Purpose enum. Used by the v1→v2 external-IPAM migration.
-var LegacyExternalTypeToPurpose = map[string]string{
-	"gateway":     PurposeIGWLRP,
-	"auto_assign": PurposeENIPublic,
-	"elastic_ip":  PurposeEIP,
-}

spinifex/domains/ec2/vpc/security_group.go

@@ -538,12 +538,7 @@ func (s *VPCServiceImpl) DescribeSecurityGroups(ctx context.Context, input *ec2.
 		return nil, err
 	}
 
-	groupIDs := make(map[string]bool)
-	for _, id := range input.GroupIds {
-		if id != nil {
-			groupIDs[*id] = true
-		}
-	}
+	groupIDs := awsidentifiers.Set(input.GroupIds)
 
 	parsedFilters, err := awsfilters.ParseFilters(input.Filters, describeSecurityGroupsValidFilters)
 	if err != nil {
@@ -675,9 +670,11 @@ var getSecurityGroupsForVpcValidFilters = map[string]bool{
 	"primary-vpc-id": true,
 }
 
-// defaultSGForVpcPerPage caps a request that names no MaxResults, matching the
-// model's ceiling.
-const defaultSGForVpcPerPage = 1000
+var getSecurityGroupsForVpcPaging = paging.EC2{
+	MaxResults: 1000,
+	TooLarge:   "Value ( %d ) for parameter maxResults is invalid. Expecting a value smaller than 1000.",
+	TooSmall:   "Value ( %d ) for parameter maxResults is invalid. Expecting a value greater than 5.",
+}
 
 // GetSecurityGroupsForVpc lists the security groups an interface in the named
 // VPC can be associated with. It is the same account-prefixed scan
@@ -690,6 +687,11 @@ func (s *VPCServiceImpl) GetSecurityGroupsForVpc(ctx context.Context, input *ec2
 	}
 	vpcID := *input.VpcId
 
+	pageReq, err := getSecurityGroupsForVpcPaging.Parse(input.MaxResults, input.NextToken, 0)
+	if err != nil {
+		return nil, err
+	}
+
 	// An unknown VPC is not-found rather than an empty list: an empty list is
 	// indistinguishable from a VPC that genuinely holds no groups, and would
 	// send the caller on to create an interface in a VPC that does not exist.
@@ -742,17 +744,7 @@ func (s *VPCServiceImpl) GetSecurityGroupsForVpc(ctx context.Context, input *ec2
 		})
 	}
 
-	// The KV key order is not guaranteed stable across calls, so a token
-	// addressing an offset into an unsorted set could skip or repeat a group.
-	slices.SortFunc(groups, func(a, b *ec2.SecurityGroupForVpc) int {
-		return strings.Compare(aws.StringValue(a.GroupId), aws.StringValue(b.GroupId))
-	})
-
-	page, nextToken, err := pageSecurityGroupsForVpc(groups, input.NextToken, input.MaxResults)
-	if err != nil {
-		slog.WarnContext(ctx, "GetSecurityGroupsForVpc: invalid NextToken", "next_token", aws.StringValue(input.NextToken))
-		return nil, err
-	}
+	page, nextToken := paging.EC2Page(groups, func(g *ec2.SecurityGroupForVpc) string { return *g.GroupId }, pageReq)
 
 	slog.InfoContext(ctx, "GetSecurityGroupsForVpc completed",
 		"count", len(page), "total", len(groups), "vpcId", vpcID, "accountID", accountID)
@@ -801,33 +793,6 @@ func sgForVpcMatchesFilters(record *SecurityGroupRecord, accountID string, filte
 	return awsfilters.MatchesTags(filters, record.Tags)
 }
 
-// pageSecurityGroupsForVpc slices one page out of a sorted group set, using an
-// opaque integer offset as the token. The last page carries a nil token.
-func pageSecurityGroupsForVpc(groups []*ec2.SecurityGroupForVpc, token *string, maxResults *int64) ([]*ec2.SecurityGroupForVpc, *string, error) {
-	start := 0
-	if aws.StringValue(token) != "" {
-		n, err := strconv.Atoi(*token)
-		if err != nil || n < 0 {
-			return nil, nil, errors.New(awserrors.ErrorInvalidParameterValue)
-		}
-		start = n
-	}
-	if start > len(groups) {
-		start = len(groups)
-	}
-
-	size := defaultSGForVpcPerPage
-	if maxResults != nil && *maxResults > 0 {
-		size = int(*maxResults)
-	}
-
-	end := start + size
-	if end >= len(groups) {
-		return groups[start:], nil, nil
-	}
-	return groups[start:end], aws.String(strconv.Itoa(end)), nil
-}
-
 // describeSecurityGroupRulesValidFilters defines the set of filter names accepted by DescribeSecurityGroupRules.
 var describeSecurityGroupRulesValidFilters = map[string]bool{
 	"group-id":               true,

spinifex/domains/ec2/vpc/security_group_modify_test.go

@@ -28,14 +28,6 @@ func requireAWSCode(t *testing.T, err error, code string) {
 	assert.Equal(t, code, got, "error: %v", err)
 }
 
-func requireAWSError(t *testing.T, err error, code, message string) {
-	t.Helper()
-	got, msg, ok := awserrors.ResolveErrorDetail(err)
-	require.True(t, ok, "error %v carries no AWS code", err)
-	assert.Equal(t, code, got)
-	assert.Equal(t, message, msg)
-}
-
 func modifyRules(svc *VPCServiceImpl, sgID string, updates ...*ec2.SecurityGroupRuleUpdate) error {
 	_, err := svc.ModifySecurityGroupRules(context.Background(), &ec2.ModifySecurityGroupRulesInput{
 		GroupId:            aws.String(sgID),

spinifex/domains/ec2/vpc/security_group_rule_validation_test.go

@@ -7,6 +7,7 @@ package vpc
 import (
 	"context"
 	"encoding/json"
+	testutil "github.com/mulgadc/spinifex/internal/testkit"
 	"testing"
 
 	"github.com/aws/aws-sdk-go/aws"
@@ -53,12 +54,12 @@ func TestAuthorizeSecurityGroup_RequiresProtocol(t *testing.T) {
 			IpProtocol: proto, FromPort: aws.Int64(80), ToPort: aws.Int64(80),
 			IpRanges: []*ec2.IpRange{{CidrIp: aws.String("10.0.0.0/24")}},
 		}
-		requireAWSError(t, authorizeIngress(svc, sgID, perm), awserrors.ErrorInvalidParameterValue, msg)
+		testutil.RequireAWSError(t, authorizeIngress(svc, sgID, perm), awserrors.ErrorInvalidParameterValue, msg)
 
 		_, err := svc.AuthorizeSecurityGroupEgress(context.Background(), &ec2.AuthorizeSecurityGroupEgressInput{
 			GroupId: aws.String(sgID), IpPermissions: []*ec2.IpPermission{perm},
 		}, testAccountID)
-		requireAWSError(t, err, awserrors.ErrorInvalidParameterValue, msg)
+		testutil.RequireAWSError(t, err, awserrors.ErrorInvalidParameterValue, msg)
 	}
 	assert.Equal(t, before, storedSGRecord(t, svc, sgID), "a rejected authorize must store nothing")
 }
@@ -94,7 +95,7 @@ func TestAuthorizeSecurityGroupIngress_TCPUDPRequireBothPorts(t *testing.T) {
 				IpProtocol: aws.String(proto), FromPort: ports[0], ToPort: ports[1],
 				IpRanges: []*ec2.IpRange{{CidrIp: aws.String("10.0.0.0/24")}},
 			})
-			requireAWSError(t, err, awserrors.ErrorInvalidParameterValue,
+			testutil.RequireAWSError(t, err, awserrors.ErrorInvalidParameterValue,
 				"Invalid value for portRange. Must specify both from and to ports with TCP/UDP.")
 		}
 	}

spinifex/domains/ec2/vpc/security_group_test.go

@@ -94,7 +94,7 @@ func TestCreateSecurityGroup_DuplicateName(t *testing.T) {
 		GroupName: aws.String("dup-sg"),
 		VpcId:     aws.String(vpcID),
 	}, testAccountID)
-	requireAWSError(t, err, awserrors.ErrorInvalidGroupDuplicate, "The security group 'dup-sg' already exists for VPC '"+vpcID+"'")
+	testutil.RequireAWSError(t, err, awserrors.ErrorInvalidGroupDuplicate, "The security group 'dup-sg' already exists for VPC '"+vpcID+"'")
 }
 
 // --- DeleteSecurityGroup ---
@@ -180,7 +180,7 @@ func TestDescribeSecurityGroups_NotFound(t *testing.T) {
 	_, err := svc.DescribeSecurityGroups(context.Background(), &ec2.DescribeSecurityGroupsInput{
 		GroupIds: []*string{aws.String("sg-0000000000000dead")},
 	}, testAccountID)
-	requireAWSError(t, err, awserrors.ErrorInvalidGroupNotFound, "The security group 'sg-0000000000000dead' does not exist")
+	testutil.RequireAWSError(t, err, awserrors.ErrorInvalidGroupNotFound, "The security group 'sg-0000000000000dead' does not exist")
 }
 
 // The shapes below are the answers AWS gave for each ID; a 17-digit ID is
@@ -196,9 +196,9 @@ func TestDescribeSecurityGroups_MalformedID(t *testing.T) {
 	}
 
 	for _, id := range []string{"sg-xyz", "sg-", "sg-0AAAAAAAAAAAAAAA1", "sg-000000001", "sg-0000000000000001", "sg-000000000000000001"} {
-		requireAWSError(t, describe(id), awserrors.ErrorInvalidGroupIdMalformed, fmt.Sprintf("Invalid id: %q", id))
+		testutil.RequireAWSError(t, describe(id), awserrors.ErrorInvalidGroupIdMalformed, fmt.Sprintf("Invalid id: %q", id))
 	}
-	requireAWSError(t, describe("foo"), awserrors.ErrorInvalidGroupIdMalformed, `Invalid id: "foo" (expecting "sg-...")`)
+	testutil.RequireAWSError(t, describe("foo"), awserrors.ErrorInvalidGroupIdMalformed, `Invalid id: "foo" (expecting "sg-...")`)
 	for _, id := range []string{"sg-a", "sg-0aaaaaaa", "sg-01234567890abcdef"} {
 		requireAWSCode(t, describe(id), awserrors.ErrorInvalidGroupNotFound)
 	}
@@ -647,7 +647,7 @@ func TestDeleteSecurityGroup_DependencyViolationForENI(t *testing.T) {
 	_, err := svc.DeleteSecurityGroup(context.Background(), &ec2.DeleteSecurityGroupInput{
 		GroupId: aws.String(sgID),
 	}, testAccountID)
-	requireAWSError(t, err, awserrors.ErrorDependencyViolation, "resource "+sgID+" has a dependent object")
+	testutil.RequireAWSError(t, err, awserrors.ErrorDependencyViolation, "resource "+sgID+" has a dependent object")
 }
 
 // TestDeleteSecurityGroup_DependencyViolationForReferencingSG pins the same
@@ -664,7 +664,7 @@ func TestDeleteSecurityGroup_DependencyViolationForReferencingSG(t *testing.T) {
 	_, err := svc.DeleteSecurityGroup(context.Background(), &ec2.DeleteSecurityGroupInput{
 		GroupId: aws.String(targetSG),
 	}, testAccountID)
-	requireAWSError(t, err, awserrors.ErrorDependencyViolation, "resource "+targetSG+" has a dependent object")
+	testutil.RequireAWSError(t, err, awserrors.ErrorDependencyViolation, "resource "+targetSG+" has a dependent object")
 }
 
 // TestRevokeSecurityGroupIngress_ByRuleID_BreaksMutualReferenceTeardown
@@ -1571,7 +1571,7 @@ func TestCreateSecurityGroup_RejectsReservedDefaultName(t *testing.T) {
 		Description: aws.String("user-supplied"),
 		VpcId:       aws.String(vpcID),
 	}, testAccountID)
-	requireAWSError(t, err, awserrors.ErrorInvalidParameterValue, "Cannot use reserved security group name: default")
+	testutil.RequireAWSError(t, err, awserrors.ErrorInvalidParameterValue, "Cannot use reserved security group name: default")
 }
 
 func TestCreateSecurityGroup_RejectsSgPrefixedName(t *testing.T) {
@@ -1607,12 +1607,12 @@ func TestDeleteSecurityGroup_RejectsDefault(t *testing.T) {
 	_, err := svc.DeleteSecurityGroup(context.Background(), &ec2.DeleteSecurityGroupInput{
 		GroupId: aws.String(sgID),
 	}, testAccountID)
-	requireAWSError(t, err, awserrors.ErrorCannotDelete, `the specified group: "`+sgID+`" name: "default" cannot be deleted by a user`)
+	testutil.RequireAWSError(t, err, awserrors.ErrorCannotDelete, `the specified group: "`+sgID+`" name: "default" cannot be deleted by a user`)
 
 	_, err = svc.DeleteSecurityGroup(context.Background(), &ec2.DeleteSecurityGroupInput{
 		GroupId: aws.String("sg-0000000000000dead"),
 	}, testAccountID)
-	requireAWSError(t, err, awserrors.ErrorInvalidGroupNotFound, "The security group 'sg-0000000000000dead' does not exist")
+	testutil.RequireAWSError(t, err, awserrors.ErrorInvalidGroupNotFound, "The security group 'sg-0000000000000dead' does not exist")
 }
 
 func TestDeleteSecurityGroup_RejectsAttachedToENI(t *testing.T) {
@@ -1666,7 +1666,7 @@ func TestDeleteVpc_BlocksOnNonDefaultSG(t *testing.T) {
 	createTestSG(t, svc, vpcID, "user-sg")
 
 	_, err := svc.DeleteVpc(context.Background(), &ec2.DeleteVpcInput{VpcId: aws.String(vpcID)}, testAccountID)
-	requireAWSError(t, err, awserrors.ErrorDependencyViolation, "The vpc '"+vpcID+"' has dependencies and cannot be deleted.")
+	testutil.RequireAWSError(t, err, awserrors.ErrorDependencyViolation, "The vpc '"+vpcID+"' has dependencies and cannot be deleted.")
 }
 
 func TestDeleteVpc_CascadesDefaultSG(t *testing.T) {
@@ -2394,7 +2394,7 @@ func TestAuthorizeSecurityGroupIngress_RejectsContentDuplicate(t *testing.T) {
 		GroupId:       aws.String(sgID),
 		IpPermissions: []*ec2.IpPermission{perm},
 	}, testAccountID)
-	requireAWSError(t, err, awserrors.ErrorInvalidPermissionDuplicate,
+	testutil.RequireAWSError(t, err, awserrors.ErrorInvalidPermissionDuplicate,
 		`the specified rule "peer: 10.0.0.0/24, TCP, from port: 22, to port: 22, ALLOW" already exists`)
 
 	// The default egress rule, re-added, is described without ports
@@ -2405,7 +2405,7 @@ func TestAuthorizeSecurityGroupIngress_RejectsContentDuplicate(t *testing.T) {
 			IpRanges:   []*ec2.IpRange{{CidrIp: aws.String("0.0.0.0/0")}},
 		}},
 	}, testAccountID)
-	requireAWSError(t, err, awserrors.ErrorInvalidPermissionDuplicate, `the specified rule "peer: 0.0.0.0/0, ALL, ALLOW" already exists`)
+	testutil.RequireAWSError(t, err, awserrors.ErrorInvalidPermissionDuplicate, `the specified rule "peer: 0.0.0.0/0, ALL, ALLOW" already exists`)
 
 	icmp := &ec2.IpPermission{
 		IpProtocol: aws.String("icmp"), FromPort: aws.Int64(8), ToPort: aws.Int64(-1),
@@ -2416,7 +2416,7 @@ func TestAuthorizeSecurityGroupIngress_RejectsContentDuplicate(t *testing.T) {
 			GroupId: aws.String(sgID), IpPermissions: []*ec2.IpPermission{icmp},
 		}, testAccountID)
 	}
-	requireAWSError(t, err, awserrors.ErrorInvalidPermissionDuplicate, `the specified rule "peer: 10.0.0.0/24, ICMP, type: 8, code: ALL, ALLOW" already exists`)
+	testutil.RequireAWSError(t, err, awserrors.ErrorInvalidPermissionDuplicate, `the specified rule "peer: 10.0.0.0/24, ICMP, type: 8, code: ALL, ALLOW" already exists`)
 
 	_, err = svc.AuthorizeSecurityGroupIngress(context.Background(), &ec2.AuthorizeSecurityGroupIngressInput{
 		GroupId: aws.String(sgID),
@@ -2425,12 +2425,12 @@ func TestAuthorizeSecurityGroupIngress_RejectsContentDuplicate(t *testing.T) {
 			IpRanges: []*ec2.IpRange{{CidrIp: aws.String("10.0.0.0/33")}},
 		}},
 	}, testAccountID)
-	requireAWSError(t, err, awserrors.ErrorInvalidParameterValue, "CIDR block 10.0.0.0/33 is malformed")
+	testutil.RequireAWSError(t, err, awserrors.ErrorInvalidParameterValue, "CIDR block 10.0.0.0/33 is malformed")
 
 	_, err = svc.DescribeSecurityGroupRules(context.Background(), &ec2.DescribeSecurityGroupRulesInput{
 		SecurityGroupRuleIds: []*string{aws.String("sgr-0000000000000dead")},
 	}, testAccountID)
-	requireAWSError(t, err, awserrors.ErrorInvalidSecurityGroupRuleIdNotFound, "The security group rule ID 'sgr-0000000000000dead' does not exist")
+	testutil.RequireAWSError(t, err, awserrors.ErrorInvalidSecurityGroupRuleIdNotFound, "The security group rule ID 'sgr-0000000000000dead' does not exist")
 }
 
 func TestDescribeSecurityGroupRules_InvalidFilter(t *testing.T) {

spinifex/domains/ec2/vpc/service_impl.go

@@ -631,12 +631,7 @@ func (s *VPCServiceImpl) DescribeVpcs(ctx context.Context, input *ec2.DescribeVp
 		return nil, err
 	}
 
-	vpcIDs := make(map[string]bool)
-	for _, id := range input.VpcIds {
-		if id != nil {
-			vpcIDs[*id] = true
-		}
-	}
+	vpcIDs := awsidentifiers.Set(input.VpcIds)
 
 	parsedFilters, err := awsfilters.ParseFilters(input.Filters, describeVpcsValidFilters)
 	if err != nil {
@@ -1010,12 +1005,7 @@ func (s *VPCServiceImpl) DescribeSubnets(ctx context.Context, input *ec2.Describ
 		return nil, err
 	}
 
-	subnetIDs := make(map[string]bool)
-	for _, id := range input.SubnetIds {
-		if id != nil {
-			subnetIDs[*id] = true
-		}
-	}
+	subnetIDs := awsidentifiers.Set(input.SubnetIds)
 
 	parsedFilters, err := awsfilters.ParseFilters(input.Filters, describeSubnetsValidFilters)
 	if err != nil {

spinifex/domains/ec2/vpc/service_impl_test.go

@@ -103,7 +103,7 @@ func TestCreateVpc_InvalidCidr(t *testing.T) {
 	_, err := svc.CreateVpc(context.Background(), &ec2.CreateVpcInput{
 		CidrBlock: aws.String("not-a-cidr"),
 	}, testAccountID)
-	requireAWSError(t, err, awserrors.ErrorInvalidParameterValue,
+	testutil.RequireAWSError(t, err, awserrors.ErrorInvalidParameterValue,
 		"Value (not-a-cidr) for parameter cidrBlock is invalid. This is not a valid CIDR block.")
 }
 
@@ -113,7 +113,7 @@ func TestCreateVpc_CidrTooLarge(t *testing.T) {
 	_, err := svc.CreateVpc(context.Background(), &ec2.CreateVpcInput{
 		CidrBlock: aws.String("10.0.0.0/8"),
 	}, testAccountID)
-	requireAWSError(t, err, "InvalidVpc.Range", "The CIDR '10.0.0.0/8' is invalid.")
+	testutil.RequireAWSError(t, err, "InvalidVpc.Range", "The CIDR '10.0.0.0/8' is invalid.")
 }
 
 func TestCreateVpc_CidrTooSmall(t *testing.T) {
@@ -122,7 +122,7 @@ func TestCreateVpc_CidrTooSmall(t *testing.T) {
 	_, err := svc.CreateVpc(context.Background(), &ec2.CreateVpcInput{
 		CidrBlock: aws.String("10.0.0.0/29"),
 	}, testAccountID)
-	requireAWSError(t, err, "InvalidVpc.Range", "The CIDR '10.0.0.0/29' is invalid.")
+	testutil.RequireAWSError(t, err, "InvalidVpc.Range", "The CIDR '10.0.0.0/29' is invalid.")
 }
 
 func TestCreateVpc_WithTags(t *testing.T) {
@@ -219,7 +219,7 @@ func TestDeleteVpc_WithSubnets(t *testing.T) {
 	_, err := svc.DeleteVpc(context.Background(), &ec2.DeleteVpcInput{
 		VpcId: aws.String(vpcID),
 	}, testAccountID)
-	requireAWSError(t, err, awserrors.ErrorDependencyViolation, "The vpc '"+vpcID+"' has dependencies and cannot be deleted.")
+	testutil.RequireAWSError(t, err, awserrors.ErrorDependencyViolation, "The vpc '"+vpcID+"' has dependencies and cannot be deleted.")
 }
 
 func TestDescribeVpcs_All(t *testing.T) {
@@ -261,10 +261,10 @@ func TestDescribeVpcs_NotFound(t *testing.T) {
 	_, err := svc.DescribeVpcs(context.Background(), &ec2.DescribeVpcsInput{
 		VpcIds: []*string{aws.String("vpc-nonexistent")},
 	}, testAccountID)
-	requireAWSError(t, err, awserrors.ErrorInvalidVpcIDNotFound, "The vpc ID 'vpc-nonexistent' does not exist")
+	testutil.RequireAWSError(t, err, awserrors.ErrorInvalidVpcIDNotFound, "The vpc ID 'vpc-nonexistent' does not exist")
 
 	_, err = svc.DeleteVpc(context.Background(), &ec2.DeleteVpcInput{VpcId: aws.String("vpc-nonexistent")}, testAccountID)
-	requireAWSError(t, err, awserrors.ErrorInvalidVpcIDNotFound, "The vpc ID 'vpc-nonexistent' does not exist")
+	testutil.RequireAWSError(t, err, awserrors.ErrorInvalidVpcIDNotFound, "The vpc ID 'vpc-nonexistent' does not exist")
 }
 
 // --- Subnet Tests ---
@@ -363,7 +363,7 @@ func TestCreateSubnet_InvalidCidr(t *testing.T) {
 		VpcId:     aws.String(vpcID),
 		CidrBlock: aws.String("not-a-cidr"),
 	}, testAccountID)
-	requireAWSError(t, err, awserrors.ErrorInvalidParameterValue,
+	testutil.RequireAWSError(t, err, awserrors.ErrorInvalidParameterValue,
 		"Value (not-a-cidr) for parameter cidrBlock is invalid. This is not a valid CIDR block.")
 
 	// AWS rejects the CIDR before it looks the VPC up.
@@ -382,7 +382,7 @@ func TestCreateSubnet_OutsideVpcCidr(t *testing.T) {
 		VpcId:     aws.String(vpcID),
 		CidrBlock: aws.String("192.168.1.0/24"),
 	}, testAccountID)
-	requireAWSError(t, err, awserrors.ErrorInvalidSubnetRange, "The CIDR '192.168.1.0/24' is invalid.")
+	testutil.RequireAWSError(t, err, awserrors.ErrorInvalidSubnetRange, "The CIDR '192.168.1.0/24' is invalid.")
 }
 
 func TestCreateSubnet_ConflictingCidr(t *testing.T) {
@@ -396,7 +396,7 @@ func TestCreateSubnet_ConflictingCidr(t *testing.T) {
 		VpcId:     aws.String(vpcID),
 		CidrBlock: aws.String("10.0.1.0/25"),
 	}, testAccountID)
-	requireAWSError(t, err, awserrors.ErrorInvalidSubnetConflict, "The CIDR '10.0.1.0/25' conflicts with another subnet")
+	testutil.RequireAWSError(t, err, awserrors.ErrorInvalidSubnetConflict, "The CIDR '10.0.1.0/25' conflicts with another subnet")
 }
 
 func TestCreateSubnet_WithTags(t *testing.T) {
@@ -857,7 +857,7 @@ func TestDeleteVpc_RejectsAttachedInternetGateway(t *testing.T) {
 
 	_, err = svc.DeleteVpc(context.Background(), &ec2.DeleteVpcInput{VpcId: aws.String(vpcID)}, testAccountID)
 	require.Error(t, err)
-	requireAWSError(t, err, awserrors.ErrorDependencyViolation, "The vpc '"+vpcID+"' has dependencies and cannot be deleted.")
+	testutil.RequireAWSError(t, err, awserrors.ErrorDependencyViolation, "The vpc '"+vpcID+"' has dependencies and cannot be deleted.")
 
 	desc, err := svc.DescribeVpcs(context.Background(), &ec2.DescribeVpcsInput{VpcIds: []*string{aws.String(vpcID)}}, testAccountID)
 	require.NoError(t, err)
@@ -877,7 +877,7 @@ func TestDeleteVpc_RejectsEgressOnlyInternetGateway(t *testing.T) {
 
 	_, err = svc.DeleteVpc(context.Background(), &ec2.DeleteVpcInput{VpcId: aws.String(vpcID)}, testAccountID)
 	require.Error(t, err)
-	requireAWSError(t, err, awserrors.ErrorDependencyViolation, "The vpc '"+vpcID+"' has dependencies and cannot be deleted.")
+	testutil.RequireAWSError(t, err, awserrors.ErrorDependencyViolation, "The vpc '"+vpcID+"' has dependencies and cannot be deleted.")
 
 	desc, err := svc.DescribeVpcs(context.Background(), &ec2.DescribeVpcsInput{VpcIds: []*string{aws.String(vpcID)}}, testAccountID)
 	require.NoError(t, err)
@@ -1138,7 +1138,7 @@ func TestCreateSubnet_CidrTooSmall(t *testing.T) {
 		VpcId:     aws.String(vpcID),
 		CidrBlock: aws.String("10.0.0.0/29"),
 	}, testAccountID)
-	requireAWSError(t, err, awserrors.ErrorInvalidSubnetRange, "The CIDR '10.0.0.0/29' is invalid.")
+	testutil.RequireAWSError(t, err, awserrors.ErrorInvalidSubnetRange, "The CIDR '10.0.0.0/29' is invalid.")
 }
 
 func TestVpcCidrBlockAssociation(t *testing.T) {

spinifex/domains/ecr/awsapi/handler.go

@@ -11,12 +11,9 @@ package awsapi
 import (
 	"context"
 	"errors"
-	"log/slog"
-	"net/http"
 	"reflect"
 	"sort"
 
-	"github.com/aws/aws-sdk-go/private/protocol/json/jsonutil"
 	"github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
 	"github.com/nats-io/nats.go"
 )
@@ -24,9 +21,6 @@ import (
 // TargetPrefix is the X-Amz-Target service prefix for ECR control-plane calls.
 const TargetPrefix = "AmazonEC2ContainerRegistry_V20150921"
 
-// JSONContentType is the AWS JSON 1.1 content type ECR clients expect.
-const JSONContentType = "application/x-amz-json-1.1"
-
 // Handler is the signature every ECR control-plane action implements. nc is the
 // gateway NATS connection (handlers relay onto ecr.* subjects); accountID is the
 // resolved caller account; body is the raw JSON 1.1 request payload.
@@ -129,19 +123,3 @@ func actionsWithHandler(target Handler) []string {
 	sort.Strings(actions)
 	return actions
 }
-
-// WriteJSONResponse serialises obj as a 200 AWS JSON 1.1 response using the
-// SDK's jsonutil marshaler (epoch-second time encoding), matching ACM/EKS.
-func WriteJSONResponse(w http.ResponseWriter, obj any) {
-	body, err := jsonutil.BuildJSON(obj)
-	if err != nil {
-		slog.Error("ECR: failed to marshal response JSON", "err", err)
-		http.Error(w, awserrors.ErrorInternalError, http.StatusInternalServerError)
-		return
-	}
-	w.Header().Set("Content-Type", JSONContentType)
-	w.WriteHeader(http.StatusOK)
-	if _, err := w.Write(body); err != nil {
-		slog.Error("ECR: failed to write response", "err", err)
-	}
-}

spinifex/domains/ecr/awsapi/handler_test.go

@@ -3,9 +3,6 @@ package awsapi
 import (
 	"context"
 
-	"encoding/json"
-	"net/http"
-	"net/http/httptest"
 	"testing"
 
 	"github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
@@ -61,19 +58,3 @@ func TestActions_ScanSurfaceUnsupported(t *testing.T) {
 			"action %q should resolve to the OperationNotSupported stub", action)
 	}
 }
-
-func TestWriteJSONResponse(t *testing.T) {
-	type repo struct {
-		RepositoryName *string `locationName:"repositoryName" type:"string"`
-	}
-	name := "team/app"
-	w := httptest.NewRecorder()
-	WriteJSONResponse(w, &repo{RepositoryName: &name})
-
-	assert.Equal(t, http.StatusOK, w.Code)
-	assert.Equal(t, JSONContentType, w.Header().Get("Content-Type"))
-
-	var decoded map[string]string
-	require.NoError(t, json.Unmarshal(w.Body.Bytes(), &decoded))
-	assert.Equal(t, "team/app", decoded["repositoryName"])
-}