Merge remote-tracking branch 'origin/feat/eks-node-lifecycle' into e2e/integration

WorkerNodeReady accepted any Ready Node, the tolerate-all DaemonSet check
dropped pods with no Ready condition, and nodegroup cleanup returned before
the nodegroup was gone. Scope the Ready check to the CPIsolation nodegroup,
count a pod with no Ready condition as not Ready, wait for each nodegroup
delete, and assert that deleted nodegroups leave no Node behind.

.github/e2e-cells.json

@@ -8,7 +8,8 @@
     "arch": "x86_64",
     "runner": "ci-single",
     "extra": "",
-    "suite": ""
+    "suite": "",
+    "timeout_minutes": 60
   },
   {
     "cell": 2,
@@ -19,7 +20,8 @@
     "arch": "x86_64",
     "runner": "ci-single",
     "extra": "",
-    "suite": ""
+    "suite": "",
+    "timeout_minutes": 60
   },
   {
     "cell": 3,
@@ -30,7 +32,8 @@
     "arch": "x86_64",
     "runner": "ci-single",
     "extra": "",
-    "suite": ""
+    "suite": "",
+    "timeout_minutes": 60
   },
   {
     "cell": 8,
@@ -200,7 +203,8 @@
     "arch": "x86_64",
     "runner": "ci-single",
     "extra": "",
-    "suite": ""
+    "suite": "",
+    "timeout_minutes": 60
   },
   {
     "cell": 35,

.github/scripts/hypervisor-disk-preflight.sh

@@ -0,0 +1,66 @@
+#!/usr/bin/env bash
+# Usage: hypervisor-disk-preflight.sh <env.tfvars>
+#
+# Fails fast when a hypervisor named in the tofu env is short of disk, before
+# `tofu apply` tries to define a domain there. libvirt writes each domain's XML
+# under /etc/libvirt/qemu on the root filesystem and its disks into the pool, and
+# a full one fails the apply minutes in as "Domain Creation Failed".
+#
+# Hosts are reached the way the libvirt provider reaches them (ssh <user>@<name>,
+# resolved through ~/.ssh/config). A host that cannot be reached is reported and
+# skipped: tofu's own error for it is clearer than anything said here.
+#
+# Env: HV_SSH_USER (tf-user), HV_MIN_ROOT_FREE_MB (1024), HV_MIN_POOL_FREE_GB (20).
+set -uo pipefail
+
+tfvars="${1:?usage: $0 <env.tfvars>}"
+user="${HV_SSH_USER:-tf-user}"
+min_root_kb=$(( ${HV_MIN_ROOT_FREE_MB:-1024} * 1024 ))
+min_pool_kb=$(( ${HV_MIN_POOL_FREE_GB:-20} * 1024 * 1024 ))
+
+if [ ! -r "$tfvars" ]; then
+  echo "::error::hypervisor disk preflight: cannot read $tfvars" >&2
+  exit 1
+fi
+
+# One node per line in every env file: { name = "...", ..., storage_pool = "..." }.
+pairs=$(grep -v '^[[:space:]]*#' "$tfvars" |
+  sed -nE 's/.*name[[:space:]]*=[[:space:]]*"([^"]+)".*storage_pool[[:space:]]*=[[:space:]]*"([^"]+)".*/\1 \2/p')
+if [ -z "$pairs" ]; then
+  echo "::warning::hypervisor disk preflight: no nodes with a storage_pool in $tfvars; nothing checked"
+  exit 0
+fi
+
+# Prints "<root-avail-kb> <pool-avail-kb> <pool-path>" for the pool named in $1.
+# shellcheck disable=SC2016 # expanded on the hypervisor, not here
+remote='set -u
+root=$(df -Pk /etc/libvirt/qemu | awk "NR==2{print \$4}")
+path=$(virsh -c qemu:///system pool-dumpxml "$1" 2>/dev/null | sed -n "s:.*<path>\(.*\)</path>.*:\1:p" | head -n1)
+pool=-
+[ -n "$path" ] && pool=$(df -Pk "$path" | awk "NR==2{print \$4}")
+echo "$root $pool ${path:--}"'
+
+fail=0
+while read -r host pool; do
+  [ -n "$host" ] || continue
+  if ! out=$(ssh -o BatchMode=yes -o ConnectTimeout=10 "${user}@${host}" bash -s -- "$pool" <<<"$remote" 2>&1); then
+    echo "::warning::hypervisor disk preflight: cannot reach $host ($out); its disk is unchecked"
+    continue
+  fi
+  read -r root_kb pool_kb pool_path <<<"$(tail -n1 <<<"$out")"
+  if ! [[ "$root_kb" =~ ^[0-9]+$ ]]; then
+    echo "::warning::hypervisor disk preflight: unexpected answer from $host: $out"
+    continue
+  fi
+  echo "$host: root fs $((root_kb / 1024)) MiB free; pool $pool ($pool_path) $([[ "$pool_kb" =~ ^[0-9]+$ ]] && echo "$((pool_kb / 1024 / 1024)) GiB free" || echo "not found")"
+  if [ "$root_kb" -lt "$min_root_kb" ]; then
+    echo "::error::hypervisor $host has $((root_kb / 1024)) MiB free on the filesystem holding /etc/libvirt/qemu (need ${HV_MIN_ROOT_FREE_MB:-1024} MiB): libvirt cannot write a domain definition there. Free space on $host before re-running." >&2
+    fail=1
+  fi
+  if [[ "$pool_kb" =~ ^[0-9]+$ ]] && [ "$pool_kb" -lt "$min_pool_kb" ]; then
+    echo "::error::hypervisor $host has $((pool_kb / 1024 / 1024)) GiB free in storage pool $pool at $pool_path (need ${HV_MIN_POOL_FREE_GB:-20} GiB): the node's disk cannot grow. Free space on $host before re-running." >&2
+    fail=1
+  fi
+done <<<"$pairs"
+
+exit "$fail"

.github/scripts/hypervisor-disk-preflight_test.sh

@@ -0,0 +1,87 @@
+#!/bin/sh
+# hypervisor-disk-preflight_test.sh — drives hypervisor-disk-preflight.sh against
+# a stub ssh that answers for each hypervisor from a canned file, so every
+# verdict is checked without a real host.
+set -eu
+
+SCRIPT_DIR=$(CDPATH='' cd -- "$(dirname -- "$0")" && pwd)
+PREFLIGHT="$SCRIPT_DIR/hypervisor-disk-preflight.sh"
+TMP=$(mktemp -d)
+trap 'rm -rf "$TMP"' EXIT
+
+fails=0
+pass() { echo "  ok   - $1"; }
+fail() { echo "  FAIL - $1"; fails=$((fails + 1)); }
+
+# The stub answers "<root-kb> <pool-kb> <path>" from $FAKE/<host>, or exits 255
+# when that host has no file, the way ssh does for a host it cannot reach.
+mkdir -p "$TMP/bin" "$TMP/hosts"
+cat > "$TMP/bin/ssh" <<'STUB'
+#!/bin/sh
+for a in "$@"; do case $a in *@*) host=${a#*@} ;; esac; done
+cat >/dev/null
+if [ -f "$FAKE/$host" ]; then cat "$FAKE/$host"; exit 0; fi
+echo "ssh: Could not resolve hostname $host" >&2
+exit 255
+STUB
+chmod +x "$TMP/bin/ssh"
+PATH="$TMP/bin:$PATH"
+FAKE="$TMP/hosts"
+export PATH FAKE
+
+cat > "$TMP/env.tfvars" <<'TFVARS'
+nodes = [
+  { name = "wattle", wan_ip = "192.0.2.1", storage_pool = "drive1-pool", base_image = "/x.qcow2" },
+  # { name = "retired", wan_ip = "192.0.2.9", storage_pool = "drive1-pool", base_image = "/x.qcow2" },
+  { name = "banksia", wan_ip = "192.0.2.2", storage_pool = "nvme1-pool", base_image = "/x.qcow2" },
+]
+TFVARS
+
+GiB=1048576
+plenty="$((50 * GiB)) $((200 * GiB)) /mnt/drive1/pool"
+
+# run <label> <want-rc> <want-text>
+run() {
+    rc=0
+    out=$(bash "$PREFLIGHT" "$TMP/env.tfvars" 2>&1) || rc=$?
+    if [ "$rc" -ne "$2" ]; then
+        fail "$1: exit $rc, want $2"
+        echo "$out" | sed 's/^/        /'
+        return
+    fi
+    case $out in
+        *"$3"*) pass "$1" ;;
+        *) fail "$1: output lacks '$3'"; echo "$out" | sed 's/^/        /' ;;
+    esac
+}
+
+echo "-- hypervisor-disk-preflight.sh"
+
+echo "$plenty" > "$FAKE/wattle"
+echo "$plenty" > "$FAKE/banksia"
+run "every host has room" 0 "banksia: root fs 51200 MiB free"
+
+echo "0 $((200 * GiB)) /mnt/drive1/pool" > "$FAKE/wattle"
+run "a full root filesystem fails fast" 1 "::error::hypervisor wattle has 0 MiB free on the filesystem holding /etc/libvirt/qemu"
+
+echo "$plenty" > "$FAKE/wattle"
+echo "$((50 * GiB)) $((3 * GiB)) /mnt/nvme1/pool" > "$FAKE/banksia"
+run "a nearly full pool fails fast" 1 "::error::hypervisor banksia has 3 GiB free in storage pool nvme1-pool"
+
+echo "$plenty" > "$FAKE/banksia"
+rm -f "$FAKE/wattle"
+run "an unreachable host is reported, not failed" 0 "::warning::hypervisor disk preflight: cannot reach wattle"
+
+echo "$plenty" > "$FAKE/wattle"
+echo "$((50 * GiB)) - -" > "$FAKE/banksia"
+run "a pool libvirt does not know is reported, not failed" 0 "pool nvme1-pool (-) not found"
+
+echo "0 0 /x" > "$FAKE/retired"
+echo "$plenty" > "$FAKE/banksia"
+run "a commented-out node is never checked" 0 "wattle: root fs"
+
+if [ "$fails" -ne 0 ]; then
+    echo "$fails failure(s)"
+    exit 1
+fi
+echo "  hypervisor-disk-preflight ok"

.github/scripts/suite-sets.sh

@@ -130,6 +130,10 @@ e2e_suite_timeout() {
     # budget on a node stopping and starting: seven units down, a raft election,
     # and the whole service stack back up with every stream caught up.
     kvquorum) echo "40m" ;;
+    # single is the broadest suite and finishes, it does not hang: passing runs
+    # took 24-29m, and an Oracle host ran out at 30m inside the last test's
+    # cleanup. The default left it no headroom at all.
+    single) echo "40m" ;;
     *) echo "30m" ;;
   esac
 }

.github/workflows/e2e-nightly.yml

@@ -184,6 +184,14 @@ jobs:
             echo "running qemu: $(pgrep -c qemu-system-x86 || echo 0)"
           } >> "${ARTIFACT_DIR}/hypervisor-load.txt"
 
+      # A hypervisor out of disk fails the apply minutes in, as "Domain Creation
+      # Failed" from libvirt writing the domain XML. Say so before starting.
+      - name: Preflight hypervisor disk
+        working-directory: mulga/scripts/tofu-cluster
+        run: |
+          "${GITHUB_WORKSPACE}/spinifex/.github/scripts/hypervisor-disk-preflight.sh" \
+            "envs/${SPINIFEX_CI_ENV}.tfvars"
+
       - name: Provision VMs
         working-directory: mulga/scripts/tofu-cluster
         env:
@@ -241,6 +249,11 @@ jobs:
           . "${GITHUB_WORKSPACE}/spinifex/.github/scripts/suite-sets.sh"
           e2e_suites_assert_subset "e2e-nightly single" "${E2E_SUITES_NIGHTLY_SINGLE}" "${E2E_SUITES_SINGLE}"
 
+          # The select job sized this cell's budget from these same timeouts, so
+          # resolve them here and carry the pairs into the remote loop.
+          SUITE_TMOS=""
+          for s in ${E2E_SUITES_NIGHTLY_SINGLE}; do SUITE_TMOS="${SUITE_TMOS}${s}:$(e2e_suite_timeout "${s}") "; done
+
           INVENTORY=$(tofu output -json inventory)
           WAN_IP=$(echo "$INVENTORY" | jq -r '.nodes[0].wan_ip')
           SSH_KEY=$(echo "$INVENTORY" | jq -r '.ssh_key_path')
@@ -250,9 +263,10 @@ jobs:
              for SUITE in ${E2E_SUITES_NIGHTLY_SINGLE}; do \
                echo \"::group::suite \$SUITE\"; \
                date -u +%Y-%m-%dT%H:%M:%SZ > '${ARTIFACT_DIR}/test-'\$SUITE'.start'; \
+               TMO=30m; for kv in ${SUITE_TMOS}; do case \$kv in \$SUITE:*) TMO=\${kv#*:} ;; esac; done; \
                GITHUB_ACTIONS=true FORCE_COLOR=1 SPINIFEX_E2E=1 SPINIFEX_MODE=single \
                  ARTIFACT_DIR='${ARTIFACT_DIR}' \
-                 ./\$SUITE.test -test.v -test.timeout=30m \
+                 ./\$SUITE.test -test.v -test.timeout=\$TMO \
                  2>&1 | tee '${ARTIFACT_DIR}/test-'\$SUITE'.log'; \
                rc=\${PIPESTATUS[0]}; echo \"::endgroup::\"; \
                if [ \$rc -ne 0 ]; then echo \"\$SUITE: FAILED\" >> '${ARTIFACT_DIR}/suite-status.txt'; FAIL=1; fi; \
@@ -324,16 +338,20 @@ jobs:
             NODES_CSV="${NODES_CSV:+$NODES_CSV,}$ip"
           done
 
+          SUITE_TMOS=""
+          for s in ${E2E_SUITES_NIGHTLY_MULTI}; do SUITE_TMOS="${SUITE_TMOS}${s}:$(e2e_suite_timeout "${s}") "; done
+
           ssh -tt -o StrictHostKeyChecking=no -i "$SSH_KEY" tf-user@${NODE1_IP} \
             "set -u; mkdir -p '${ARTIFACT_DIR}'; cd \$HOME/e2e-tests/_bin; FAIL=0; \
              for SUITE in ${E2E_SUITES_NIGHTLY_MULTI}; do \
                echo \"::group::suite \$SUITE\"; \
                date -u +%Y-%m-%dT%H:%M:%SZ > '${ARTIFACT_DIR}/test-'\$SUITE'.start'; \
+               TMO=30m; for kv in ${SUITE_TMOS}; do case \$kv in \$SUITE:*) TMO=\${kv#*:} ;; esac; done; \
                GITHUB_ACTIONS=true FORCE_COLOR=1 SPINIFEX_E2E=1 SPINIFEX_MODE=multinode \
                  SPINIFEX_NODES='${NODES_CSV}' SPINIFEX_NODE_IPS='${NODES_CSV}' \
                  SPINIFEX_SSH_USER=tf-user SPINIFEX_SSH_KEY=\$HOME/.ssh/tf-user-ap-southeast-2 \
                  ARTIFACT_DIR='${ARTIFACT_DIR}' \
-                 ./\$SUITE.test -test.v -test.timeout=30m \
+                 ./\$SUITE.test -test.v -test.timeout=\$TMO \
                  2>&1 | tee '${ARTIFACT_DIR}/test-'\$SUITE'.log'; \
                rc=\${PIPESTATUS[0]}; echo \"::endgroup::\"; \
                if [ \$rc -ne 0 ]; then echo \"\$SUITE: FAILED\" >> '${ARTIFACT_DIR}/suite-status.txt'; FAIL=1; fi; \

Makefile

@@ -247,6 +247,12 @@ test-build-scripts:
 		echo "-- $$t"; \
 		sh "$$t" || exit 1; \
 	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; \
+	done
+	shellcheck -S warning .github/scripts/hypervisor-disk-preflight.sh .github/scripts/hypervisor-disk-preflight_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)

architecture/layering/allowlist_test.go

@@ -104,7 +104,6 @@ var allowlist = []allowance{
 
 	{From: "domains/sts/awsapi", To: "domains/iam", Debt: "IAM implementation consumers in STS, Ochre and ELBv2", Reason: "GetCallerIdentity takes IAMService; needs an IAM capability"},
 
-	{From: "gateway", To: "domains/sts/awsapi", Debt: "gateway (service registration)", Reason: "IAM, STS, EKS and RDS dispatch stays in gateway until the ingress registration seam"},
 	{From: "domains/ecs", To: "domains/ec2/ebs/metadata", Debt: "close-out: handlers/ecs (ADR-0004 blocked-dependency list; inventory-ecs boundary imports)", Reason: "legacy ECS calls another domain's implementation as an internal API"},
 	{From: "domains/ecs", To: "domains/ec2/eip", Debt: "close-out: handlers/ecs (ADR-0004 blocked-dependency list; inventory-ecs boundary imports)", Reason: "legacy ECS calls another domain's implementation as an internal API"},
 	{From: "domains/ecs", To: "domains/ec2/instancetypes", Debt: "close-out: handlers/ecs (ADR-0004 blocked-dependency list; inventory-ecs boundary imports)", Reason: "legacy ECS calls another domain's implementation as an internal API"},

architecture/layering/layering_test.go

@@ -192,30 +192,29 @@ func TestADR0001_LayeringCheckerDetectsFaults(t *testing.T) {
 	}
 }
 
-// TestADR0001_InterimDaemonDirectionCoversRuntimeDaemon: the daemon left its
-// legacy root, so gateway importing runtime/daemon must still be rejected, and
-// the former handlers packages are held to it by R5 at their domain homes.
-func TestADR0001_InterimDaemonDirectionCoversRuntimeDaemon(t *testing.T) {
-	daemon, _ := classify("spinifex/runtime/daemon")
-	checks := []struct{ rule, from string }{
-		{"Interim_DaemonGatewayHandlersDirection", "spinifex/gateway"},
-		{"R5_DomainImportsNoRuntimeOrOperator", "spinifex/domains/ecs"},
+// TestADR0001_R2_IngressRejectsRuntimeOperatorAndDomains: the request pipeline
+// lives under ingress/aws, so it must not import the daemon, any runtime or
+// operator package, or a domain implementation.
+func TestADR0001_R2_IngressRejectsRuntimeOperatorAndDomains(t *testing.T) {
+	r := ruleByID("R2_IngressImportsNoImplementation")
+	from := "spinifex/ingress/aws/pipeline"
+	f, ok := classify(from)
+	if !ok || !r.subject(f, from) {
+		t.Fatalf("%s is not governed by R2", from)
 	}
-	for _, c := range checks {
-		r := ruleByID(c.rule)
-		f, _ := classify(c.from)
-		if !r.subject(f, c.from) || !r.forbids(f, daemon, c.from) {
-			t.Errorf("%s -> runtime/daemon is not rejected by %s", c.from, c.rule)
+	for _, to := range []string{
+		"spinifex/runtime/daemon", "spinifex/runtime/host/process", "spinifex/runtime/roles/awsgw",
+		"spinifex/operator/admin", "spinifex/domains/ecs", "spinifex/domains/iam/awsapi",
+	} {
+		target, _ := classify(to)
+		if !r.forbids(f, target, from) {
+			t.Errorf("%s -> %s is not rejected by R2", from, to)
 		}
 	}
-	other, _ := classify("spinifex/runtime/host/process")
-	if gw, _ := classify("spinifex/gateway"); ruleByID("Interim_DaemonGatewayHandlersDirection").forbids(gw, other, "spinifex/gateway") {
-		t.Error("gateway -> runtime/host/process is rejected, but it is not the daemon")
-	}
 }
 
 // TestADR0002_S7_RequestRecordRejectsOutsiders: the request record is readable
-// by the gateway and ingress/aws only.
+// by packages under ingress/aws only.
 func TestADR0002_S7_RequestRecordRejectsOutsiders(t *testing.T) {
 	record, ok := classify("spinifex/ingress/aws/reqctx")
 	if !ok || !record.requestRecord {
@@ -226,7 +225,6 @@ func TestADR0002_S7_RequestRecordRejectsOutsiders(t *testing.T) {
 		"domains/ecs/awsapi":   true,
 		"runtime/roles/awsgw":  true,
 		"operator/admin":       true,
-		"gateway":              false,
 		"ingress/aws/dispatch": false,
 	} {
 		from, ok := classify(fullPath(name))

architecture/layering/layers_test.go

@@ -29,12 +29,11 @@ const (
 type domainKind int
 
 const (
-	notDomain     domainKind = iota
-	domainImpl               // domains/<d>/... resource implementation
-	domainAWSAPI             // domains/<d>/awsapi/... AWS action adapters
-	domainProj               // domains/<d>/projection/... authorised view
-	legacyImpl               // handlers/<d>/... legacy domain implementation
-	legacyAdapter            // gateway/<d>/... legacy AWS action adapters
+	notDomain    domainKind = iota
+	domainImpl              // domains/<d>/... resource implementation
+	domainAWSAPI            // domains/<d>/awsapi/... AWS action adapters
+	domainProj              // domains/<d>/projection/... authorised view
+	legacyImpl              // handlers/<d>/... legacy domain implementation
 )
 
 type class struct {
@@ -60,17 +59,8 @@ func (c class) isLegacy(roots ...string) bool {
 	return slices.Contains(roots, c.legacy)
 }
 
-// legacyRoots are pre-ADR-0001 application roots still standing.
-var legacyRoots = map[string]bool{
-	"gateway": true,
-}
-
-// gatewayDomains names the domain each gateway/<service> adapter belongs to;
-// Bedrock is the AWS surface of the Ochre domain.
-var gatewayDomains = map[string]string{
-	"bedrock": "ochre", "ecs": "ecs", "eks": "eks", "elbv2": "elbv2", "iam": "iam",
-	"rds": "rds", "sts": "sts",
-}
+// legacyRoots are pre-ADR-0001 application roots still standing. None remain.
+var legacyRoots = map[string]bool{}
 
 // classify maps a module-relative package directory to its layer. ok is false
 // for a package that has no deliberate classification.
@@ -142,8 +132,6 @@ func classifyApp(app []string) (class, bool) {
 	switch {
 	case app[0] == "handlers" && len(app) >= 2:
 		c.domain, c.kind = app[1], legacyImpl
-	case app[0] == "gateway" && len(app) >= 2 && gatewayDomains[app[1]] != "":
-		c.domain, c.kind = gatewayDomains[app[1]], legacyAdapter
 	}
 	return c, true
 }

architecture/layering/legacy_test.go

@@ -9,13 +9,11 @@ import (
 // frozenLegacyPackages is every production package still under a legacy root.
 // It only shrinks: a new entry is a new legacy package, a missing one is a
 // completed move whose line must be deleted.
-var frozenLegacyPackages = []string{
-	"gateway",
-}
+var frozenLegacyPackages = []string{}
 
 // retiredLegacyRoots were legacy roots whose every package has moved to its
 // ADR-0001 target; recreating one would reopen a closed migration row.
-var retiredLegacyRoots = []string{"accountteardown", "admin", "daemon", "handlers", "lbagent", "services", "utils", "vpcd"}
+var retiredLegacyRoots = []string{"accountteardown", "admin", "daemon", "gateway", "handlers", "lbagent", "services", "utils", "vpcd"}
 
 // TestADR0001_LegacyRootsFrozen: the current-to-target map gives every legacy
 // root a destination, so none may gain a package and no retired root may return.
@@ -35,9 +33,6 @@ func TestADR0001_LegacyRootsFrozen(t *testing.T) {
 			present[name] = true
 		}
 	}
-	if len(present) == 0 {
-		t.Fatal("found no legacy packages; the freeze would pass vacuously")
-	}
 	frozen := map[string]bool{}
 	for _, p := range frozenLegacyPackages {
 		frozen[p] = true

architecture/layering/rules_test.go

@@ -126,10 +126,8 @@ var rules = []rule{
 			case t.layer == layerAgents:
 				return true
 			case t.layer == layerIngress:
-				return f.kind != domainAWSAPI && f.kind != legacyAdapter
-			case t.isLegacy("gateway") && !t.domainShaped():
-				return true
-			case t.isLegacy("handlers", "gateway"):
+				return f.kind != domainAWSAPI
+			case t.isLegacy("handlers"):
 				return f.layer == layerDomain
 			}
 			return false
@@ -156,29 +154,14 @@ var rules = []rule{
 	},
 	{
 		id:      "S7_RequestRecordPrivateToIngress",
-		clause:  "ADR-0002:S7: the request record is private to ingress, so no domain, runtime or operator package reads it; only the gateway and packages under ingress/aws may import ingress/aws/reqctx",
+		clause:  "ADR-0002:S7: the request record is private to ingress, so no domain, runtime or operator package reads it; only packages under ingress/aws may import ingress/aws/reqctx",
 		hint:    "receive what you need through dispatch.Invocation instead of the request record",
 		subject: func(_ class, _ string) bool { return true },
 		forbids: func(_, t class, fromName string) bool {
 			if !t.requestRecord {
 				return false
 			}
-			return fromName != "gateway" && !strings.HasPrefix(fromName, "ingress/aws/")
-		},
-	},
-	{
-		id:      "Interim_DaemonGatewayHandlersDirection",
-		clause:  "during migration the existing daemon -> gateway -> handlers rule remains the guardrail (Q-20, Q-29)",
-		hint:    "keep the dependency pointing daemon -> gateway -> handlers; invert it through a callback or capability",
-		subject: func(f class, _ string) bool { return f.isLegacy("handlers", "gateway") },
-		forbids: func(f, t class, _ string) bool {
-			switch {
-			case f.isLegacy("handlers"):
-				return t.isLegacy("gateway") || t.daemon
-			case f.isLegacy("gateway"):
-				return t.daemon
-			}
-			return false
+			return !strings.HasPrefix(fromName, "ingress/aws/")
 		},
 	},
 }

build/systemd/spinifex-firewall.service

@@ -1,11 +1,10 @@
-# spinifex-unit-version: 1
+# spinifex-unit-version: 2
 [Unit]
 Description=Spinifex host firewall (nftables input policy)
 Documentation=file:/etc/spinifex/firewall/spinifex.nft
 Wants=network-pre.target
 Before=network-pre.target shutdown.target
 DefaultDependencies=no
-ConditionPathExists=/etc/spinifex/firewall/peers.nft
 
 [Service]
 Type=oneshot
@@ -13,8 +12,11 @@ RemainAfterExit=yes
 StandardInput=null
 ProtectSystem=full
 ProtectHome=true
-ExecStart=/usr/local/lib/spinifex/spinifex-firewall-apply
-ExecReload=/usr/local/lib/spinifex/spinifex-firewall-apply
+# Unconditional: connection tracking loads on every node, so a flow that opens
+# while the policy is not armed survives the arm. The policy itself still loads
+# only where the daemon has written a peer file.
+ExecStart=/usr/local/lib/spinifex/spinifex-firewall-apply boot
+ExecReload=/usr/local/lib/spinifex/spinifex-firewall-apply boot
 TimeoutStopSec=10s
 
 # No ExecStop: flushing on stop would take vpcd's MASQUERADE and per-EIP FORWARD

cmd/aws-model-coverage/main.go

@@ -18,7 +18,6 @@ import (
 	rdsawsapi "github.com/mulgadc/spinifex/spinifex/domains/rds/awsapi"
 	route53awsapi "github.com/mulgadc/spinifex/spinifex/domains/route53/awsapi"
 	stsawsapi "github.com/mulgadc/spinifex/spinifex/domains/sts/awsapi"
-	"github.com/mulgadc/spinifex/spinifex/gateway"
 	"github.com/mulgadc/spinifex/spinifex/ingress/aws/dispatch"
 
 	_ "github.com/mulgadc/bluebottle/pkg/fipsboot"
@@ -57,7 +56,7 @@ func fail(err error) {
 func compareAll() ([]awsmodel.OperationCoverage, error) {
 	// Registered services contribute their declared inventory, which needs no
 	// live dependencies; nothing here serves a request.
-	inventories := gateway.AWSOperationInventory(map[string]dispatch.Inventory{
+	inventories := dispatch.AWSOperationInventory(map[string]dispatch.Inventory{
 		awsapi.ServiceName:        awsapi.OperationInventory(),
 		acmawsapi.ServiceName:     acmawsapi.OperationInventory(),
 		route53awsapi.ServiceName: route53awsapi.OperationInventory(),

cmd/eks-gateway-fetch/main.go

@@ -6,6 +6,7 @@
 //
 //	eks-gateway-fetch -resource addons   # GET /clusters/{cluster}/internal-addons/{accountId}
 //	eks-gateway-fetch -resource recovery -instance-id i-… # GET /clusters/{cluster}/internal-recovery/{accountId}/{instanceId}
+//	eks-gateway-fetch -resource workers  # GET /clusters/{cluster}/internal-workers/{accountId}
 //
 // Flags default to environment variables seeded by cloud-init:
 // EKS_GATEWAY_URL, EKS_GATEWAY_CA, EKS_ACCESS_KEY, EKS_SECRET_KEY, EKS_REGION,
@@ -33,7 +34,7 @@ func main() {
 	flag.StringVar(&cfg.AccountID, "account-id", os.Getenv("EKS_ACCOUNT_ID"), "Cluster account ID")
 	flag.StringVar(&cfg.ClusterName, "cluster", os.Getenv("EKS_CLUSTER_NAME"), "Cluster name")
 	flag.StringVar(&cfg.InstanceID, "instance-id", os.Getenv("EKS_INSTANCE_ID"), "This member's EC2 instance ID (recovery resource)")
-	flag.StringVar(&cfg.Resource, "resource", "addons", "Resource to fetch: addons | recovery")
+	flag.StringVar(&cfg.Resource, "resource", "addons", "Resource to fetch: addons | recovery | workers")
 	flag.Parse()
 
 	if err := gatewayfetch.Run(context.Background(), cfg); err != nil {

contracts/viperblockd/legacy/v1/README.md

@@ -60,6 +60,10 @@ moved to another node; the local daemon subscribes to stop the guest that
 was using it. Node-addressed, like mount and unmount: the guest is local, and
 no other node's daemon has anything to stop.
 
+The event is sent as a request and repeated until the daemon replies with `VolumeFencedAck`, because the fence often runs while the node's NATS server is restarting and a single publish can reach no subscriber.
+The daemon replies only when the event carries a reply subject, so a plain publish is still accepted.
+Repeats are harmless, since fencing a guest that is already fenced changes nothing.
+
 ## Abandon route
 
 `VolumeAbandonSubject(node)` is `ebs.<node>.abandon`. The daemon uses

contracts/viperblockd/legacy/v1/fence.go

@@ -19,3 +19,11 @@ type VolumeFencedEvent struct {
 	Winner string `json:"winner"`
 	Reason string `json:"reason"`
 }
+
+// VolumeFencedAck is the daemon's answer to a VolumeFencedEvent. Any answer
+// ends the sender's retries; Instance names the guest stopped, if one was, and
+// Error says why the daemon could not act on the event.
+type VolumeFencedAck struct {
+	Instance string `json:"instance,omitempty"`
+	Error    string `json:"error,omitempty"`
+}

contracts/viperblockd/legacy/v1/fence_test.go

@@ -37,3 +37,18 @@ func TestVolumeFencedEvent_JSON(t *testing.T) {
 	require.NoError(t, json.Unmarshal([]byte(golden), &decoded))
 	assert.Equal(t, event, decoded)
 }
+
+// TestVolumeFencedAck_JSON pins the daemon's answer, which is what ends
+// viperblockd's retries of a fence notice.
+func TestVolumeFencedAck_JSON(t *testing.T) {
+	golden := `{"instance":"i-1","error":"bad event"}`
+	ack := viperblocklegacyv1.VolumeFencedAck{Instance: "i-1", Error: "bad event"}
+
+	marshaled, err := json.Marshal(ack)
+	require.NoError(t, err)
+	assert.JSONEq(t, golden, string(marshaled))
+
+	empty, err := json.Marshal(viperblocklegacyv1.VolumeFencedAck{})
+	require.NoError(t, err)
+	assert.JSONEq(t, `{}`, string(empty), "an ack for a volume with no local guest carries nothing")
+}

docs/COMMANDS.md

@@ -26,7 +26,7 @@ Every service is `PartOf=spinifex.target` and ordered by its own `After=`/`Wants
 | `spinifex-qmp-collector.service` | `spx service qmp-collector start` | Reads the `contracts/telemetry/v1` per-VM sidecar, polls telemetry QMP sockets and tap counters, publishes to NATS `metrics.ec2.*` |
 | `spinifex-shutdown.service` | `/bin/true`; the work is its `ExecStop` | Runs `spx admin node drain --local --unless-restarting`. `After=` every service, so it stops first and drains guests while the stack is still up |
 | `spinifex-nats-watchdog.service` / `.timer` | `nats-js-watchdog.sh` | JetStream liveness probe every 30s. `try-restart`s NATS when JetStream stops accepting writes while the process stays alive — what `Restart=on-failure` cannot catch |
-| `spinifex-firewall.service` | `spinifex-firewall-apply` | Oneshot nftables input policy, `WantedBy=sysinit.target` rather than the target. No `ExecStop`: remove it with `nft delete table inet spinifex_filter` |
+| `spinifex-firewall.service` | `spinifex-firewall-apply boot` | Oneshot nftables input policy, `WantedBy=sysinit.target` rather than the target. Loads the `inet spinifex_conntrack` tracking table on every node, and the policy only once the daemon has written a peer file. No `ExecStop`: remove the policy with `spinifex-firewall-apply disable` |
 | `regenerate-ssh-host-keys.service` | `ssh-keygen -A` | Oneshot for AMI-built nodes, conditional on the host key being absent |
 
 #### Operations

docs/DESIGN.md

@@ -33,67 +33,32 @@ The gateway (`spinifex/runtime/roles/awsgw/awsgw.go`) is the entry point:
 // Connect to NATS (retries while the local broker comes up)
 natsConn, err := natsmsg.ConnectNATSWithRetry(...)
 
-// Load IAM (master key + JetStream KV) and create the gateway
-gw := gateway.GatewayConfig{
-    NATSConn:   natsConn,
-    Config:     nodeConfig.AWSGW.Config,
-    IAMService: iamService,
+// Load IAM (master key + JetStream KV), then compose the gateway: composeGateway
+// sets every production hook and wires the service registry from each domain's
+// awsapi registration.
+gw, err := composeGateway(gatewayInputs{
+    NATS:        natsConn,
+    IAM:         iamService,
+    Credentials: credentials,
+    Admission:   admission,
     // ...
-}
+}, ecrControlPlane, acmControlPlane, ec2ControlPlane /* ... */)
 
 // Serve over TLS
 server.ListenAndServeTLS("", "")
 ```
 
-Request routing (`spinifex/gateway/gateway.go`):
+Request routing (`spinifex/ingress/aws/pipeline/gateway.go`):
 
-1. **Authentication**: SigV4 middleware validates AWS credentials and resolves the account ID
-2. **Throttling**: Per-account+action token bucket rejects bursts post-auth
-3. **Service Detection**: Extracts service name from the Authorization header
-4. **Action Dispatch**: Routes to service-specific handler. A service registered through the ingress dispatch registry (`ingress/aws/dispatch`, wired by `runtime/roles/awsgw`) is dispatched by its registration; currently those are the ECR and ACM control planes, registered from `domains/ecr/awsapi` and `domains/acm/awsapi`. Every other service uses the legacy switch:
-
-```go
-switch svc {
-case "ec2":
-    err = gw.EC2_Request(w, r)
-case "iam":
-    err = gw.IAM_Request(w, r)
-case "sts":
-    err = gw.STS_Request(w, r)
-case "elasticloadbalancing":
-    err = gw.ELBv2_Request(w, r)
-case "eks":
-    err = gw.EKS_Request(w, r)
-case "bedrock":
-    err = gw.Bedrock_Request(w, r)
-// ... bedrock-runtime, bedrock-agent, bedrock-agent-runtime
-case "ecs":
-    err = gw.ECS_Request(w, r)
-case "rds":
-    err = gw.RDS_Request(w, r)
-case "tagging":
-    err = gw.Tagging_Request(w, r)
-case "spinifex":
-    err = gw.Spinifex_Request(w, r)
-}
-```
+1. **Public routes**: `/oidc/eks/...` and `/v2` (the ECR registry) are mounted ahead of SigV4 by `runtime/roles/awsgw`
+2. **Authentication**: SigV4 middleware validates AWS credentials and resolves the account ID
+3. **Throttling**: Per-account+action token bucket rejects bursts post-auth
+4. **Service Detection**: Extracts service name from the Authorization header
+5. **Action Dispatch**: Every service is a registration (`ingress/aws/dispatch`) built by its domain's `awsapi.NewRegistration` and wired by `runtime/roles/awsgw`. The pipeline dispatches by registered service name and holds no per-service code. The private `/admin/<Method>` surface dispatches to the registration `awsgw` names as `Admin.Service`.
 
 ### 3. EC2 Handler
 
-The EC2 handler (`spinifex/gateway/ec2.go`) parses the `Action` parameter and delegates to specific handlers via the generic `ec2Handler` and `ec2HandlerWithReq` wrappers. The account ID resolved by SigV4 auth is threaded into every handler so requests are scoped per IAM principal:
-
-```go
-"RunInstances": ec2HandlerWithReq(func(ctx context.Context, input *ec2.RunInstancesInput, gw *GatewayConfig, accountID string, r *http.Request) (any, error) {
-    // ... PassRole and launch-quota checks, then the vCPU quota charge
-    return ec2instanceapi.RunInstances(ctx, input, gw.NATSConn, gw.IAMService, accountID, passRoleCheck, launchQuotaCheck, gw.ExpectedNodes)
-}),
-"DescribeInstances": ec2Handler(func(ctx context.Context, input *ec2.DescribeInstancesInput, gw *GatewayConfig, accountID string) (any, error) {
-    return ec2instanceapi.DescribeInstancesChecked(ctx, input, gw.NATSConn, 0, gw.NodeIDs, accountID)
-}),
-// ... + volumes, snapshots, VPCs, subnets, route tables, IGWs, NAT gateways,
-// security groups, network interfaces, elastic IPs, placement groups, key pairs,
-// images, tags, capacity reservations, spot instances
-```
+The EC2 registration (`spinifex/domains/ec2/awsapi`, `NewRegistration`) parses the `Action` parameter and looks the handler up in its action table. The account ID resolved by SigV4 auth is threaded into every handler so requests are scoped per IAM principal. Each action's handler is built from the owning EC2 resource package (`domains/ec2/instance`, `volume`, `vpc` and so on), with PassRole, quota and client-token idempotency wrapped on at the action table rather than inside the resource.
 
 ### 4. NATS Messaging
 
@@ -238,12 +203,14 @@ Viperblock responds with an NBD URI that QEMU uses to access the block device.
 
 A direct-boot microVM keeps the launcher's saved `vm.Config` as a clean base, and `startQEMU` adds data volumes only to a launch-only copy of it (`runtime/compute/vm/lifecycle.go`).
 At each start the daemon takes one snapshot of the instance's already-mounted data requests, probes each NBD or host block export from that snapshot, and builds the command from the same snapshot, so the launched endpoint is the one that was checked and is never written back into the saved base.
-Each data volume becomes a `virtio-blk-device` on the MMIO bus with the same id, node name, I/O thread, serial and `werror=stop,rerror=stop` policy as the PCI device, and takes no PCIe hot-plug port.
+The microVM boots with `pcie=on`, so its NICs are `virtio-net-pci` devices with MSI-X interrupts instead of `virtio-net-device` MMIO devices.
+Each data volume becomes a `virtio-blk-pci` device on the default PCIe bus with the same id, node name, I/O thread, serial and `werror=stop,rerror=stop` policy as the q35 device, and takes no PCIe hot-plug port.
+A microVM started without `pcie=on` keeps the `virtio-*-device` MMIO transports.
 
 Boot and EFI requests are rejected on this path, and a direct-boot instance with no data requests launches its base unchanged.
 Disk-boot (q35) instances keep the existing PCI hot-plug path.
 
-Hot-attaching a volume to a running direct-boot microVM is not supported, because the MMIO bus has no PCIe root port to plug into.
+Hot-attaching a volume to a running direct-boot microVM is not supported, because the microVM has no PCIe hot-plug root port to plug into.
 `AttachVolume` refuses it before mounting the volume or changing any instance or volume state, and the EC2 API returns `UnsupportedOperation`.
 
 This covers only requests that are already mounted when the instance starts; nothing yet supplies data volumes to a system instance, and the path is verified with fake-QEMU unit tests, not with a real EBS volume in a guest.

docs/PACKAGE_BOUNDARY_MIGRATION.md

@@ -368,6 +368,8 @@ while the branch is in flight; they do not describe the umbrella repository's
 | `spinifex#1354` | `spinifex/gateway/ec2.go` (the EC2 action table, `ec2Handler`, `ec2Actions`, the dispatch pipeline), `ec2_idempotency.go`, `ec2_validation.go`, and the EC2 token-bucket fields of `GatewayConfig` | `spinifex/domains/ec2/awsapi` (`Handler`, `NewHandler`, `Deps`, `Quota`, `Dispatch`, `ActionNames`, `InputHasField`), composed by `spinifex/gateway/ec2.go` | Slice 9a of the EC2 move: the capabilities first, with no behaviour change. The EC2 action table, idempotency and validation move to `domains/ec2/awsapi` behind `Deps{NATS, IAM, Quota, ActiveNodes, ExpectedNodes, NodeIDs, AZ, Endpoint, InstanceStatus}`; the gateway keeps the composition and `EC2_Request`, still on the legacy switch until slice 9b registers EC2 through the ADR-0002 seam. `Quota` is declared in `awsapi` because `domains/admission/quota` imports the EC2 action packages and cannot be imported back, and the gateway wraps `quota.Service` to supply the NATS instance-type resolver its retype gate takes. The instance and spot-instance adapters take the two-method `InstanceProfiles` interface (`ResolveInstanceProfile`, `CanonicalResourceARN`) instead of `iamdomain.IAMService`, which pays off the `spotinstance` to `domains/admission/quota` and `domains/iam` allowlist rows. Allowlist changes are 20 deletions and no additions: the 18 `gateway` to `domains/ec2/awsapi/*` rows and those 2. The pipeline characterisation tests committed first (error envelopes, check order, gateway-local actions, idempotent replay, PassRole, ImportKeyPair padding, launch-template expansion before authorization, RunInstances instance-profile refusal order) pass unchanged. Package-local tests drive every routed action through `Dispatch` against a stub daemon. No change to `dispatch.Actor` or `dispatch.Invocation`. Live on env14: DescribeRegions, DescribeAvailabilityZones and DescribeAccountAttributes, an idempotent CreateVolume replay and its parameter mismatch, the DryRun and MaxResults errors, a no-policy user refused with UnauthorizedOperation (including on the gateway-local DescribeRegions), an assumed-role session allowed and refused per its policy, and one RunInstances to running then terminated. |
 | `spinifex#1355` | `spinifex/gateway/ec2.go` (`EC2_Request` in production, the legacy service switch, its `supportedServices` and JSON tables) and the `Quota` field of `GatewayConfig` | `spinifex/domains/ec2/awsapi` (`NewRegistration`, `OperationInventory`, `ServiceName`), composed by `spinifex/runtime/roles/awsgw` (`ec2.go`: `newEC2Registration`, `advertisedEndpoint`; `ec2quota`) | Slice 9b of the EC2 move: EC2 registers through the ADR-0002 seam as service `ec2` and the legacy gateway switch is empty. The EC2 query-protocol error envelope is kept through `ErrorEnvelopeXML`, and the policy denial match is now `*dispatch.PolicyDenialError`. The characterisation tests committed first also pin the Authorize service name and action, the trace span and audit action, the throttle key (account plus action) and the cluster-unavailable gate, which runs before policy and action lookup for gateway-local and NATS-backed actions alike. They and the 9a tests pass unchanged apart from setup: STS and EC2 test setups now register the services they sign as, and the two tests that claimed a legacy service name are removed because no legacy name remains. `EC2_Request` survives only as a test shim. Allowlist changes are two deletions (the `gateway` to `domains/ec2/awsapi` and `domains/ec2/awsapi/instance` rows) and one addition, the per-service model-coverage row `cmd/aws-model-coverage` to `domains/ec2/awsapi`. `startInstanceStatus` stays in `awsgw.go` because ADR-0007:S6 records the raw instance-state uses by file. No change to `dispatch.Actor` or `dispatch.Invocation`. Live on env14: DescribeRegions, DescribeAvailabilityZones and DescribeAccountAttributes, an idempotent CreateVolume replay and its parameter mismatch, the DryRun and InvalidVolume.NotFound errors, a no-policy user refused with UnauthorizedOperation (including on the gateway-local DescribeRegions), an assumed-role session allowed and refused per its policy, and one RunInstances to running then terminated. The gateway request logs carry `service=ec2` and the per-action label. Not exercised live: a throttled request (the default limits are not reachable from one client; the throttle key is pinned by the characterisation tests) and an agent-signed EC2 call. |
 | `spinifex#1363` | `spinifex/gateway` context keys (`ctxService`, `ctxRegion`, `ctxAction`, the principal and access key id keys, and the unread `ctxQueryArgs`), the duplicate `callerPrincipal` and `principalContext` types, `gateway/caller_arn.go` (`buildCallerARN`), and the direct `policyauth`, `accountadmission` and NATS-connection reads in the auth pipeline | `spinifex/ingress/aws/reqctx` (the request record), `credverify.CallerARN`, and the gateway-consumed capabilities `PolicyEvaluator`, `AccountAdmission` and `Readiness`, composed by `spinifex/runtime/roles/awsgw` (`clusterready`) | Slice 11a of the gateway removal: the request-context record and the ADR-0002:S8 capabilities, with no behaviour change. The record is a copy-on-write value under one context key; the principal is sealed once and a second `WithPrincipal` keeps the first caller. The layering rule `S7_RequestRecordPrivateToIngress` lets only `gateway` and `ingress/aws/*` import `reqctx`, with a fault-injection test, and slice 11c moves the package to `ingress/aws/internal/reqctx` and drops the rule. The caller ARN is built once in `credverify` (root cases kept) and filled on `Principal.CallerARN`. The policy evaluator, account admission (`credential.ErrAccountNotActive`, with `accountadmission.ErrNotActive` an alias) and cluster readiness (`Available() bool`) are interfaces the gateway consumes and fails closed on when nil. The step 1 characterisation tests (throttle against both readiness gates and the policy gate for ECS and IAM, `/admin` skipping `GetService` and the second gate, the nil-connection difference between the two gates, five error writers each minting a request id that differs from the audit record, and the credential-scope region reaching only the span) pass unchanged apart from setup. No allowlist rows change, and there is no change to `dispatch.Actor` or `dispatch.Invocation`. Debt for slice 11c: the test-only `defaultCapabilities` seam (a package variable set by a test `init` so about 60 gateway test files can build a config from `IAMService` alone), the nil-connection case of the early readiness gate, the remaining `NATSConn` readers (`DiscoverActiveNodes` and the field that `awsgw` and the integration harness set), and the `IAMService`, `STSService`, `Config`, `RootCAs` and `InternalSuffix` fields of `GatewayConfig`, which no production code reads. Live on env14: DescribeRegions, DescribeAvailabilityZones and DescribeAccountAttributes, an idempotent CreateVolume replay and its parameter mismatch, the DryRun and InvalidVolume.NotFound errors, a bad access key refused with InvalidClientTokenId on a Query service and InvalidClientTokenIdException on a JSON service, a no-policy user refused with UnauthorizedOperation, and an assumed-role session allowed and refused per its policy. Not exercised live: a suspended-account refusal, a throttled request and the NATS-down 503, which the step 1 tests pin. |
+| `spinifex#1378` | `spinifex/gateway` `oidc_discovery.go` and `registry.go` (the `/oidc/eks` routes and the `/v2` mount), the `DiscoverActiveNodes` method and its cache, and the `GatewayConfig` fields `Config`, `RootCAs`, `InternalSuffix`, `Version`, `Commit`, `IAMService`, `STSService`, `NATSConn` and `ExpectedNodes` | `spinifex/runtime/roles/awsgw` (`PublicRoutes`, `NodeDiscovery.ActiveNodes`, `composeGateway`); the gateway keeps one `PublicRoutes` mount hook and the `Admin` registration name supplied by the role |
+| `spinifex#1384` | `spinifex/gateway` (`GatewayConfig`, the SigV4 and policy middleware, `Request`, the admin dispatch, `AWSOperationInventory`) | `spinifex/ingress/aws/pipeline` (`Config`) and `spinifex/ingress/aws/dispatch` (`AWSOperationInventory`); the gateway layering rows are retired and R2 binds the pipeline |
 | `spinifex#1334` | `spinifex/gateway/oidc_discovery.go` (`eksAccountKV`, the discovery-document and JWKS read and construction) | `spinifex/domains/eks/cluster` (`OIDCPublisher`, `OIDCDiscoveryDocument`), composed by `spinifex/runtime/roles/awsgw` | The EKS domain now reads its own account bucket and builds the per-cluster discovery document and JWKS responses (ADR-0005:S4), and the gateway only maps the unauthenticated `/oidc/eks/...` routes onto an `OIDCPublisher` interface it is handed. Responses are byte-identical, pinned by `oidc_discovery_current_behaviour_test.go`, whose assertions are unchanged. The gateway no longer imports `domains/eks/cluster` in production code. `STS` keeps its own `ClusterJWKSResolver` (`OIDCJWKSReader`), untouched. Recorded debt, not repaired here: a stored JWKS is served verbatim with 200 even when it is not valid JSON. |
 The EC2 contract row is intentionally different from the directory moves: it creates
 a compatibility boundary. `ec2.cmd.*` retains its deployed one-token NATS
@@ -686,7 +688,7 @@ series above; "needs decision" is open.
 | `handlers/rds` | — | Done. The remaining RDS instance resource moved to `domains/rds/instance` (package `instance`) in `spinifex#1251` with no API, NATS subject, storage or behaviour change, fulfilling ADR-0001's organise-by-resource condition alongside the already-extracted subnet group, parameter group and engine catalogue. The ADR-0004 blocked-dependency imports (quota admission, the duplicated expected-node fan-out, the IAM ensurer and the Bluebottle master key) remain dependency-rule debt, with their allowlist entries now sourced from `domains/rds/instance`. |
 | `handlers/elbv2` | — | Done. The implementation moved to `domains/elbv2` (package `elbv2`) with no API, NATS subject, storage or behaviour change. Q-30's data-plane boundary remains deferred debt; the amended ADR-0001 row makes it not a relocation prerequisite. Its cross-domain, IAM and `agents/elbv2` imports remain dependency-rule debt, with their allowlist entries now sourced from `domains/elbv2`. |
 | `handlers/iam/mock` | — | Done. Moved with the implementation to `domains/iam/mock`. |
-| `gateway` (root) | Generic HTTP route assembly, SigV4/registry authentication, the policy gate, request-audit middleware, request-context keys, and the legacy per-service dispatch switch for every service not yet registered. | Deferred: the ADR-0002 phases above (auth/IAM and request-context boundary design) gate moving the audit, context-key and legacy-dispatch mechanics. |
+| `gateway` (root) | Generic HTTP route assembly, SigV4 authentication, the policy gate, request-audit middleware and request-context keys. | Done and deleted. The request pipeline moved into `ingress/aws/pipeline` (as `pipeline.Config`) and the operation inventory into `ingress/aws/dispatch` in `spinifex#1384`; the `/oidc`, `/v2` and node-discovery code had already moved to `runtime/roles/awsgw` in `spinifex#1378`. The layering test binds the pipeline with R2 and `gateway` is a retired legacy root. |
 | `gateway/bedrock` | — | Done. The whole package moved into `domains/ochre` in `spinifex#1266` with its package-private core and tests intact and no behaviour change. Separating the action adapters into `domains/ochre/awsapi` remains deferred, recorded as `domains/ochre` (action adapters) debt above. The action adapters then moved to `domains/ochre/awsapi` behind one ADR-0002 registration in `spinifex#1351`. |
 | `gateway/{iam,sts,eks,rds}` | — | Done. The action adapters moved to `domains/{iam,sts,eks,rds}/awsapi` in `spinifex#1265` with no API, NATS subject, storage or behaviour change. ADR-0001 does not make ADR-0002 registration a prerequisite for relocating a domain's adapters, as the EC2 adapter moves already showed; the root `gateway` dispatch that still imports them is recorded `gateway` (service registration) debt. |
 | `gateway/{ecs,elbv2}` | — | Done. The action adapters moved to `domains/{ecs,elbv2}/awsapi` in `spinifex#1273` with no API, NATS subject, storage or behaviour change; the root `gateway` dispatch that still imports them is recorded `gateway` (service registration) debt. |

docs/admin/host-firewall/README.md

@@ -155,6 +155,8 @@ Restarting the daemon is what re-arms it: the node rebuilds its peer list from t
 
 `disable` removes Spinifex's own table and the peer file and leaves every other nftables table alone. It is idempotent, and it is the recovery path — it runs even on a node with no policy installed, so it cannot leave you with a node you can neither arm nor disarm.
 
+`disable` leaves `table inet spinifex_conntrack` loaded, and the boot-time unit loads it on every node whether or not the policy is armed. It accepts everything and exists only to keep connection tracking running. Without it, a connection opened while the policy is off, such as an OVN RAFT link formed during that window, is first seen mid-stream when the policy comes back, and its replies are dropped as new connections. Do not delete it.
+
 > [!WARNING]
 > Do not flush the whole nftables ruleset to clear the policy. The VPC daemon writes SNAT and per-elastic-IP forward rules into the `ip nat` and `ip filter` tables and reinstalls them only when it starts, so `nft flush ruleset` breaks elastic IPs silently until the next `systemctl restart spinifex-vpcd`. Use `disable`, which only ever touches `inet spinifex_filter`.
 > On Oracle Linux with IPsec on, a flush also removes `inet spinifex_ipsec_guard`, so Geneve to a peer without an SA goes out in clear until the daemon puts the table back, within 60 seconds.
@@ -171,6 +173,7 @@ Everything lives under `/etc/spinifex/firewall/`. Which files are safe to edit f
 | `open-ports.nft` | The apply helper — the formation window only | No. Reset closed on install and upgrade |
 | `custom.nft` | **You** — `trusted_ssh_peers` and anything else you add | **Yes.** Created once, never rewritten |
 | `mode` | `setup.sh` — whether the install path armed the policy | No |
+| `conntrack.nft` | `setup.sh` — the tracking table, loaded armed or not | No. Edits are overwritten |
 
 The daemon cannot write `/etc` itself, and a blanket `nft` grant would be root-equivalent, so changes go through `/usr/local/lib/spinifex/spinifex-firewall-apply` — a root helper that accepts a fixed set of verbs and re-validates every address handed to it.
 
@@ -190,10 +193,10 @@ Dropped packets are logged, rate-limited to five a minute, so this tells you whe
 sudo journalctl -k | grep 'spinifex-fw drop'
 ```
 
-An empty log with a connection still failing means the policy is not the cause. A node with no table loaded at all is not armed:
+An empty log with a connection still failing means the policy is not the cause. A node with no peer file is not armed, and its boot-time unit loads only the tracking table:
 
 ```bash
-systemctl is-enabled spinifex-firewall.service
+sudo test -e /etc/spinifex/firewall/peers.nft && echo armed || echo not armed
 ```
 
 ## Troubleshooting

docs/install/install-oracle-linux/README.md

@@ -164,7 +164,7 @@ On Oracle Linux the installer makes these changes, which Debian hosts do not nee
 - **Repositories.** It enables `ol10_UEKR8` and `ol10_kvm_utils`, and adds `qemu-kvm*` and `qemu-img*` to the `excludepkgs` of `ol10_appstream`, keeping any excludes already set, so a later AppStream build can never replace the KVM Utilities QEMU.
 - **NFV SIG.** It writes `/etc/yum.repos.d/spinifex-nfv-sig.repo` with an `includepkgs=` list of the OVS and OVN packages, and imports the SIG key to `/etc/pki/rpm-gpg/RPM-GPG-KEY-CentOS-SIG-NFV` only when its fingerprint is `35154228174901BEFA8E69A621465E289D2A76A7`.
 - **OVS and OVN user.** Both run as root, as on Debian, through `OVS_USER_ID` and `OVN_USER_ID` in `/etc/sysconfig/openvswitch` and `/etc/sysconfig/ovn`.
-- **DNS.** Where systemd-resolved is installed but not running, it saves `/etc/resolv.conf` to `/var/lib/spinifex-reset/resolv.conf.orig`, starts resolved behind NetworkManager and links `/etc/resolv.conf` to the resolved stub. It then checks that names still resolve and stops, printing the commands that put DNS back, if they do not.
+- **DNS.** Where systemd-resolved is installed but not running, it saves `/etc/resolv.conf` to `/var/lib/spinifex-reset/resolv.conf.orig`, starts resolved behind NetworkManager, tells NetworkManager in `/etc/NetworkManager/conf.d/90-spinifex-dns.conf` to leave `/etc/resolv.conf` alone, and links `/etc/resolv.conf` to the resolved stub. It then checks that names still resolve and stops, printing the commands that put DNS back, if they do not.
 - **chrony.** It appends `confdir /etc/chrony.d` to `/etc/chrony.conf` once, and writes its step policy to `/etc/chrony.d/spinifex.conf`.
 - **`spx` under `sudo`.** Oracle Linux's `sudo` path leaves out `/usr/local/bin`, so the installer links `/usr/bin/spx` to `/usr/local/bin/spx`.
 

docs/package-boundary/eks-addon-lifecycle.md

@@ -15,7 +15,7 @@ There is no generation, so the record mixes desired fields with the latest obser
 
 | Item | Present behaviour | Evidence |
 |---|---|---|
-| Identity | Cluster name plus add-on name; one add-on per name per cluster. ARN `arn:aws:eks:{region}:{account}:addon/{cluster}/{addon}` (`foundation/aws/arn/eks.go` `FormatEKSAddon`), deterministic by name, with no per-incarnation suffix. The gateway authorizer derives the same ARN (`gateway/eks/authz.go` `addonARN`). | `handlers/eks/addon_lifecycle_test.go` `TestAddonPersistedLayout_RecordAndManifestKeysAndFieldNames`, `gateway/eks/authz_test.go` `TestResourceARNsFidelity` |
+| Identity | Cluster name plus add-on name; one add-on per name per cluster. ARN `arn:aws:eks:{region}:{account}:addon/{cluster}/{addon}` (`foundation/aws/arn/eks.go` `FormatEKSAddon`), deterministic by name, with no per-incarnation suffix. The gateway authorizer derives the same ARN (`domains/eks/awsapi/authz.go` `addonARN`). | `handlers/eks/addon_lifecycle_test.go` `TestAddonPersistedLayout_RecordAndManifestKeysAndFieldNames`, `domains/eks/awsapi/authz_test.go` `TestResourceARNsFidelity` |
 | Scope | Account and Region, within one cluster: keys live under `clusters/{cluster}/addons/` in the `eks-account-{account}` bucket. | `TestAddonPersistedLayout_RecordAndManifestKeysAndFieldNames`, `TestDeleteClusterPrefix_SweepsAddonRecordsAndManifestsScopedToCluster` |
 | Durable record | `clusters/{cluster}/addons/{addon}` holds `Record` (`domains/eks/addon/record.go`): `addonName`, `addonVersion`, `status`, `serviceAccountRoleArn`, `configurationValues`, `health`, `arn`, `tags`, `createdAt`, `modifiedAt` (the optional ones `omitempty`). `clusters/{cluster}/addons/{addon}/manifest` holds `Manifest` (`domains/eks/addon/record.go`): `addonName`, `addonVersion`, `serviceAccountRoleArn`, `configurationValues`. Listing skips the manifest sub-key by segment count. | `TestAddonPersistedLayout_RecordAndManifestKeysAndFieldNames`, `TestAddonRecord_DecodesHandWrittenJSON`, `domains/eks/addon/store_test.go` `TestRecord_RoundTrip`, `TestList_SkipsManifestSubKeys` |
 | Writers | Record: the owner in `domains/eks/addon/owner.go`, reached from the `CreateAddon`, `UpdateAddon` and `DeleteAddon` handlers, the cluster reconciler's report relay and the GPU node group launch: `Owner.Create` (unconditional put), `Owner.Update` and its failure marking (revision CAS), `Owner.Delete` (get then delete), `Owner.ApplyReport` (revision CAS), `Owner.EnsureGPUDevicePlugin` through `Owner.Create`; and the cluster prefix purge (`DeleteClusterPrefix`). Manifest: `StagingInstaller.Install` (unconditional put on create and update), `StagingInstaller.Uninstall` (`domains/eks/addon/delivery.go`), `Owner.Delete` and the prefix purge. | `TestAddonPersistedLayout_RecordAndManifestKeysAndFieldNames`, `TestUpdateAddon_ReportsSuccessfulWhileRecordIsUpdating`, `TestDeleteAddon_ReturnsDeletingAfterErasingRecordAndManifest`, `TestDeleteClusterPrefix_SweepsAddonRecordsAndManifestsScopedToCluster`, `domains/eks/addon/store_test.go` `TestCasUpdate`, `TestDeleteRecord_RemovesRecordAndManifest` |
@@ -29,16 +29,16 @@ There is no generation, so the record mixes desired fields with the latest obser
 | Readiness | `ready` maps any non-`ACTIVE` status to `ACTIVE` and clears `health`; `applied` and unknown phases change nothing. The agent reports `ready` when K3s records the add-on's `gvks` annotation, or after a 90 s grace once the bundle's workloads are rolled out; the check is per rendered file, not per version. | `domains/eks/addon/owner_test.go` `TestNextStatus`, `TestOwnerApplyReport`, `handlers/eks/addon_status_reconciler_test.go` `TestApplyAddonStatusReport`, `tests/e2e/eks/eks_test.go` `TestEKS/AddonDelivery` |
 | Failure | Staging failure on create or update marks the record `CREATE_FAILED` with the error as `health` and returns the raw error. A `failed` report maps `ACTIVE` to `DEGRADED`, keeps `DEGRADED` and `CREATE_FAILED`, and maps anything else (including `UPDATING`) to `CREATE_FAILED`. Neither failure state is terminal: a later `ready` report returns the add-on to `ACTIVE`. | `TestCreateAddon_InstallerFailureLeavesDurableCreateFailedRecord`, `TestUpdateAddon_InstallerFailureMarksCreateFailed`, `TestAddonStatusReport_MismatchedVersionIsAccepted`, `handlers/eks/addons_test.go` `TestCreateAddon_InstallerFailureMarksCreateFailed` |
 | Delete | Read the record, build a `DELETING` response from it, delete the manifest, then delete the record and manifest again (`Owner.Delete`). Both keys are gone when the call returns. The guest later notices the missing manifest, removes the rendered file and runs `kubectl delete --wait=false`; it sends no report, so nothing observes removal. | `TestDeleteAddon_ReturnsDeletingAfterErasingRecordAndManifest`, `TestListStagedAddonManifests`, `tests/e2e/eks/eks_test.go` `TestEKS/AddonDelivery` |
-| Guest fetch | `GET /clusters/{cluster}/internal-addons/{accountId}` (`gateway/eks.go`, `ListInternalAddons`) is admitted only for a control-plane agent serving that cluster (`AuthorizeInternal`), relays to `eks.ListStagedAddonManifests`, and wraps the result as `contracts/eks/v1` `InternalAddonsResponse`. The body is written by the SDK REST-JSON marshaller, which ignores the `json` tags: it is `{"Addons":[{"AddonName","AddonVersion","ServiceAccountRoleArn","ConfigurationValues"}]}` with empty strings present, sorted by add-on name. | `gateway/eks_addon_wire_test.go` `TestEKSRequest_InternalAddonsWireBody`, `contracts/eks/v1` `TestInternalAddonsResponse_RESTJSONBody`, `gateway/eks_authz_test.go` `TestEKSRequest_InternalRoutesAdmitTheCPAgentPrincipal`, `gateway/eks/internal_authz_test.go` `TestAuthorizeInternal_RejectsClusterTheCallerDoesNotServe` |
+| Guest fetch | `GET /clusters/{cluster}/internal-addons/{accountId}` (`domains/eks/awsapi/internal_addons.go`, `ListInternalAddons`) is admitted only for a control-plane agent serving that cluster (`AuthorizeInternal`), relays to `eks.ListStagedAddonManifests`, and wraps the result as `contracts/eks/v1` `InternalAddonsResponse`. The body is written by the SDK REST-JSON marshaller, which ignores the `json` tags: it is `{"Addons":[{"AddonName","AddonVersion","ServiceAccountRoleArn","ConfigurationValues"}]}` with empty strings present, sorted by add-on name. | `domains/eks/awsapi/gateway_wire_test.go` `TestEKSRequest_InternalAddonsWireBody`, `contracts/eks/v1` `TestInternalAddonsResponse_RESTJSONBody`, `domains/eks/awsapi/gateway_authz_test.go` `TestEKSRequest_InternalRoutesAdmitTheCPAgentPrincipal`, `domains/eks/awsapi/internal_authz_test.go` `TestAuthorizeInternal_RejectsClusterTheCallerDoesNotServe` |
 | TSV conversion | `eks-gateway-fetch -resource addons` (`agents/eks/gatewayfetch`) decodes the body into its own copy of the shape (matching the Go field names only because `encoding/json` is case-insensitive) and writes `name\tversion\troleArn\tbase64(config)` per add-on; an empty set writes nothing. It retries the GET up to 30 times at 5 s. | `agents/eks/gatewayfetch/addons_wire_test.go` `TestEmitAddonsTSV_GatewayWireBody`, `agents/eks/gatewayfetch/emit_test.go` `TestEmitAddonsTSV`, `TestEmitAddonsTSV_Empty`, `agents/eks/gatewayfetch/fetch_test.go` `TestRun_RendersAddons`, `TestRun_FailsAfterAttemptBudget` |
-| Guest render and report | Every 30 s (`ADDON_SYNC_INTERVAL`) the agent renders each staged add-on, substituting the role ARN into web-identity env when set, then publishes `{"addon","version","phase","message","ts"}` through `eks-gateway-publish -channel addon`. The gateway relays the raw payload to core NATS `eks.addon.{account}.{cluster}.status` (`gateway/eks/internal_publish.go`). A fetch failure keeps the current render. | `gateway/eks/internal_publish_test.go` `TestPublishInternal_RelaysToSubjects`, `agents/eks/gatewaypublish/publish_test.go` `TestRun_PublishesWrappedPayload`, `TestRun_FailsAfterAttemptBudget` |
+| Guest render and report | Every 30 s (`ADDON_SYNC_INTERVAL`) the agent renders each staged add-on, substituting the role ARN into web-identity env when set, then publishes `{"addon","version","phase","message","ts"}` through `eks-gateway-publish -channel addon`. The gateway relays the raw payload to core NATS `eks.addon.{account}.{cluster}.status` (`domains/eks/awsapi/internal_publish.go`). A fetch failure keeps the current render. | `domains/eks/awsapi/internal_publish_test.go` `TestPublishInternal_RelaysToSubjects`, `agents/eks/gatewaypublish/publish_test.go` `TestRun_PublishesWrappedPayload`, `TestRun_FailsAfterAttemptBudget` |
 | Retry meaning | Reports are level-triggered: each tick re-sends the current phase for every staged add-on, so a lost report is repeated on the next tick. `eks-gateway-publish` retries the POST up to 30 times at 5 s and the agent ignores its failure. Core NATS keeps nothing: a report published while no reconciler holds the cluster's lease, or after the reconciler stops on a terminal cluster status, is dropped. | none |
 | Report ordering | The reconciler subscribes plainly (no queue group) on the lease holder and applies reports in arrival order through a revision CAS. Nothing orders a report against a later update: a report produced for the previous version settles the new `UPDATING` record, and a stale `failed` turns it `CREATE_FAILED`. A report for a deleted add-on is a no-op. | `TestAddonStatusReport_MismatchedVersionIsAccepted`, `TestDeleteAddon_ReturnsDeletingAfterErasingRecordAndManifest` |
 | Cross-resource dependency | A GPU node group's launch calls `stageGPUDeviceAddon`, which checks the cluster exists and calls `Owner.EnsureGPUDevicePlugin` for `nvidia-device-plugin`; an existing record is a no-op and every other error is logged only. Deleting the node group never unstages it, and an existing record in any status, including `CREATE_FAILED`, blocks re-staging. | `handlers/eks/nodegroup_test.go` `TestCreateNodegroup_GPUNodegroupStagesDevicePluginAddon`, `domains/eks/addon/owner_test.go` `TestOwnerEnsureGPUDevicePlugin_StagesDefaultVersionOnceAndKeepsAnyExistingRecord`, `TestCreateNodegroup_NonGPUNodegroupDoesNotStageDevicePluginAddon`, `tests/e2e/gpu/eks_gpu_test.go` `TestEKSGPUPodExposure` |
 | Idempotency | `clientRequestToken` is ignored. A duplicate create, including an SDK retry with the same token, returns `ResourceInUseException`. A repeated update re-stages and returns another `Successful` with the same id. A repeated delete returns `ResourceNotFoundException`. | `TestCreateAddon_RepeatedClientRequestTokenIsResourceInUse`, `TestUpdateAddon_ReportsSuccessfulWhileRecordIsUpdating`, `TestDeleteAddon_ReturnsDeletingAfterErasingRecordAndManifest`, `handlers/eks/addons_test.go` `TestCreateAddon_CreatesStagingAndDefaultsVersion` |
 | Absent resource | An unknown cluster returns `ResourceNotFoundException` for every action, including the internal manifest listing. A missing add-on returns `ResourceNotFoundException` from describe, update and delete. | `handlers/eks/addons_test.go` `TestCreateAddon_UnknownClusterIsNotFound`, `TestDescribeDeleteAddon_MissingIsNotFound`, `TestUpdateAddon_MissingIsNotFound`, `TestListStagedAddonManifests` |
 | Dependencies | Only the cluster meta's existence is checked; cluster status is not read, so add-ons can be created on a `DELETING` cluster. | `TestAddonActionsIgnoreClusterLifecycleStatus` |
-| Authorization | Gateway policy uses the add-on ARN above. `serviceAccountRoleArn` requires `iam:PassRole` on `CreateAddon` and `UpdateAddon` (`gateway/eks/passrole.go`), the same check `CreateCluster`/`CreateNodegroup` apply, including that the role exists. Tag actions on an add-on ARN return `NotImplemented`. | `gateway/eks_addon_wire_test.go` `TestPassedRoleARNs_AddonServiceAccountRoleIsChecked`, `gateway/eks_passrole_test.go` `TestEKSCreateAddon_PassRole`, `TestEKSUpdateAddon_PassRole`, `gateway/eks/authz_test.go` `TestResourceARNsFidelity` |
+| Authorization | Gateway policy uses the add-on ARN above. `serviceAccountRoleArn` requires `iam:PassRole` on `CreateAddon` and `UpdateAddon` (`domains/eks/awsapi/passrole.go`), the same check `CreateCluster`/`CreateNodegroup` apply, including that the role exists. Tag actions on an add-on ARN return `NotImplemented`. | `domains/eks/awsapi/gateway_wire_test.go` `TestPassedRoleARNs_AddonServiceAccountRoleIsChecked`, `domains/eks/awsapi/gateway_passrole_test.go` `TestEKSCreateAddon_PassRole`, `TestEKSUpdateAddon_PassRole`, `domains/eks/awsapi/authz_test.go` `TestResourceARNsFidelity` |
 | Cluster deletion | Add-on records and manifests are erased only by the cluster prefix purge, after the cluster's infrastructure teardown succeeds; there is no per-add-on delete. Reclaiming a failed cluster keeps the meta and does not purge, so the recreated cluster inherits the add-ons. | `TestDeleteClusterPrefix_SweepsAddonRecordsAndManifestsScopedToCluster`, `handlers/eks/cluster_state_test.go` `TestDeleteClusterPrefix_SweepsEveryKey` |
 | Restart and interruption | Create interrupted between the record put and staging leaves a `CREATING` record with no manifest that nothing re-stages and that blocks a retry. Delete interrupted after unstaging leaves a record in its previous status while the guest removes the software. An update interrupted after the CAS leaves `UPDATING` with the previous manifest. A daemon restart loses nothing durable, but reports sent while no reconciler runs are lost until the next tick. | none |
 | End-to-end | `tests/e2e/eks/eks_test.go` `TestEKS/AddonDelivery` (create to `ACTIVE`, objects present, delete removes the namespace) and `tests/e2e/gpu/eks_gpu_test.go` `TestEKSGPUPodExposure`; `tests/integration/cross_tenant_test.go` `TestCrossTenant_EveryScopedActionIsAccountedFor`. No E2E for update, failure, configuration or the role. No Terraform apply/destroy evidence for `aws_eks_addon`. | |
@@ -52,7 +52,7 @@ There is no generation, so the record mixes desired fields with the latest obser
 5. `DELETING` exists only in the delete response; the record and manifest are erased before any guest removal, so `DescribeAddon` returns not found immediately.
 6. No observed deletion acknowledgement: the agent removes objects with `--wait=false` and reports nothing.
 7. `configurationValues` is stored, staged and transported but discarded by the agent; `resolveConflicts`, `preserve` and `podIdentityAssociations` are accepted and never used.
-8. Fixed: `serviceAccountRoleArn` now requires `iam:PassRole` on `CreateAddon` and `UpdateAddon`, matching `CreateCluster`/`CreateNodegroup` (`gateway/eks/passrole.go`).
+8. Fixed: `serviceAccountRoleArn` now requires `iam:PassRole` on `CreateAddon` and `UpdateAddon`, matching `CreateCluster`/`CreateNodegroup` (`domains/eks/awsapi/passrole.go`).
 9. AWS `UPDATE_FAILED` and `DELETE_FAILED` are not represented: a failed update becomes `CREATE_FAILED` and a failed delete leaves the prior status.
 10. The add-on ARN lacks the per-incarnation suffix AWS appends, so a recreated add-on reuses its predecessor's ARN. This is a compatibility gap; extraction must keep the current ARN.
 11. The guest/controller wire shape is versioned as `contracts/eks/v1` (subject, delivery phases, status report, staged manifest, internal-addons body), used directly by the handler, gateway and `agents/eks/gatewayfetch`. The add-on owner keeps its own `Manifest` and `Report`, and the shell agent still builds the report with `printf`. The HTTP body uses Go field names because the REST-JSON marshaller ignores the `json` tags, and the agent decodes it only through case-insensitive matching; the contract pins that form rather than changing it.

docs/package-boundary/rds-engine-catalogue-boundary.md

@@ -18,7 +18,7 @@ The characterization tests below were written in `handlers/rds` before the move;
 ## Discovered and corrected divergence
 
 `ValidateMasterUserPassword` built its empty-password error with a bare `errors.New`, which `awserrors` does not resolve, so `CreateDBInstance` without `MasterUserPassword` returned HTTP 500 `ServerInternal` where AWS returns a 400 client error.
-It now returns 400 `InvalidParameterValue`; `gateway/rds_test.go` `TestRDSRequest_CreateDBInstance_EmptyMasterUserPasswordReturns400` covers the gateway, NATS and RDS service path.
+It now returns 400 `InvalidParameterValue`; `ingress/aws/pipeline/rds_test.go` `TestRDSRequest_CreateDBInstance_EmptyMasterUserPasswordReturns400` covers the gateway, NATS and RDS service path.
 
 ## Boundaries kept by the move
 

docs/security/network-connections/README.md

@@ -237,6 +237,7 @@ Notes that make the difference between this working and locking you out:
 - **Use a dedicated table and never flush the others.** `vpcd` writes MASQUERADE and per-EIP FORWARD rules into the `ip filter` and `ip nat` tables, and only reinstalls them when the service starts. `iptables -F`, `nft flush ruleset`, `ufw enable` and `firewalld` all destroy them silently, and instance networking stays broken until the next restart.
 - **Do not add a `forward` hook.** nftables evaluates every table registered on a hook and any `drop` is final, so a default-deny forward chain here would override `vpcd`'s accepts in the other table and break every routed-NAT instance and every EIP. Filtering forwarded guest traffic is the security group's job, not the host firewall's.
 - **`vpcd` owns a second table named `spinifex_mss`, separate from `inet spinifex_filter`.** It is a `bridge` table on the veth uplink and an `inet` table on the routed uplink, it has a `forward` hook with policy `accept`, and it only rewrites the MSS option of TCP SYN packets on the host hop to the gateway. It never drops or accepts traffic and does not replace the forward-hook advice above. `vpcd` re-installs it every 30 seconds, so a flush is repaired within that interval, and there is none on a physical uplink.
+- **Keep connection tracking running while the policy is unloaded.** The kernel tracks connections only while some rule uses conntrack, so a node with no ruleset tracks nothing. A connection that opens then, such as an OVN RAFT link, is first seen mid-stream when the policy loads, and if the server's end speaks first its packet looks like a new connection to an ephemeral port and is dropped. Spinifex keeps `table inet spinifex_conntrack`, which accepts everything, loaded on every node for this reason.
 - **`output` is deliberately untouched.** The metadata service's reply path egresses through per-instance policy routing; filtering `output` breaks it in ways that are hard to attribute.
 - **On a single-NIC node the cluster and encap sets are the node's public addresses**, because the planes collapse. The rules are still correct, but they are no longer a boundary — an upstream ACL is the only real control there.
 

docs/testing/e2e-coverage.md

@@ -29,6 +29,7 @@ The largest suite, 27 top-level tests. Notable entry points:
 - `TestInstanceMetadata` — IMDS from inside a guest.
 - **`TestIMDSDatapathSurvivesVswitchdRestart`** restarts `ovs-vswitchd` under a guest with an Elastic IP, once through systemd and once with SIGKILL. After each, the guest must answer over its EIP and fetch its instance ID through IMDSv2 within 60 seconds, and `br-imds` must hold all 7 of the guest's own flows. It is sequential because it takes the whole node's datapath down.
 - `TestStopStart`, `TestEarlyRebootLiveness`, `TestGuestChurnDurability`, `TestSnapshotLifecycle`, `TestSnapshotBackedLaunch`, `TestCreateImage`, `TestVolumeLifecycle`, `TestSpotInstanceLifecycle`, `TestLaunchTemplateBoot`.
+- **`TestRebootLoop`** is opt-in and skipped in a whole-suite run. Name it in `-test.run`, or set `SPINIFEX_REBOOT_LOOP_ROUNDS` (default 10), and it reboots one fresh guest through the API that many times and then once from inside, requiring a new boot ID every time.
 - `TestNegativeErrorPaths` — the `8a`–`8t` AWS error-shape cases (malformed AMI ID, invalid instance type, volume in use, detach-root forbidden, unknown or malformed instance IDs on the lifecycle verbs, and so on).
 - **`TestKVReplication`** — the single-node half of the KV replication policy, and it is not the multi-node assertion with a smaller number in it. On one node one replica per bucket is the correct answer rather than the defect, so what it proves is the opposite: that nothing declared a replica count this deployment cannot place, that `spx admin kv replicas` exits 0 and reports `WANT 1`, and that `--repair` run twice against a cluster with nothing to repair changes no bucket's replica count and leaves every bucket serving. The multi-node suites cannot cover any of that, because on three nodes the unplaceable count is the right one. It lives here rather than in its own suite so it runs in every single-node nightly cell, across every host and network permutation.
 
@@ -122,6 +123,10 @@ Internal and internet-facing ALB and NLB, HTTPS listeners, listener rules, `Modi
 Unhealthy-target routing: an NLB with an HTTP health check on a TCP target group stops sending traffic to a target that fails it while still accepting TCP, sends traffic again once it recovers, and fails open when every target is unhealthy.
 An ALB whose targets all fail their health check keeps serving from them instead of answering 503.
 
+`TestInternalALBPrivateSubnet` launches an internal ALB in a private subnet of a non-default VPC with no internet gateway and no public route.
+The load balancer must reach active with a healthy target, which needs the owner stamp on its ENI to resolve, and a client inside the VPC must be served by it on every request.
+The client reports on its serial console because the runner has no route into the VPC.
+
 ### `eks`, `ecs`, `ecr` — container services
 
 `eks`: cluster create/delete, access entries, addon delivery, `GetToken`, kubeconfig artifacts, the EBS CSI volume path, an IRSA web-identity exchange under the stock AWS trust policy (`StringEquals` on the issuer host-path `:sub`/`:aud`, a mismatched `:sub` denied with `AccessDenied`, a `StringLike` `:sub` granted), and an IRSA pod wired with a projected SA token. Worker IMDS is covered too: a managed nodegroup worker reports IMDSv2 required with hop limit 2, a pod with no IRSA resolves to the node role through IMDS, and a nodegroup built from a launch template takes the template's metadata options. Control-plane isolation is covered as well: no control-plane host registers as a Node, a DaemonSet tolerating every taint runs on workers only, metrics-server is Ready, and logs work for every kube-system pod. `ecs`: clusters, container instances, the three network modes (`awsvpc`, `bridge`, `host`), task-role credentials fetched from the container credentials endpoint, a service behind an ELB, and ENI release on terminate. `ecr`: control-plane CRUD plus a real OCI data plane — docker login/build/push/pull round trips, multi-layer images, and raw blob HTTP.
@@ -135,7 +140,7 @@ The create/describe/connect path against a live cluster: a DB VM that actually b
 - **`storagefault`** — what a guest does when its storage backend stops answering, and what stop/start does when the seal failed. **The claim is about corruption, not availability.** SIGSTOPs a cluster-wide service, so it always runs last and in its own cell. Three-node, because the RS(2,1) shard floor and cross-node takeover after a failed seal do not exist on one node.
 - **`partialblock`** — concurrent sub-block writes to the same block surviving a cold reread through NBD from a real guest.
 - **`storagegrowth`** — volume growth and image pull. Requires `external_mode=pool`; skips on a nat env, which has no EIP and therefore no guest SSH.
-- **`reboot`** — a slow guest waited for and keeping what it wrote, and a wedged guest reset once its budget expires. Proves a real reboot by requiring a *new* boot ID.
+- **`reboot`** — a slow guest waited for and keeping what it wrote, and a wedged guest forced off and relaunched once its budget expires. Proves a real reboot by requiring a *new* boot ID.
 - **`diskperf`** — guest disk performance gate, pinned fio profile on a dedicated blank volume. Also requires `external_mode=pool` for guest SSH.
 
 ### Other

internal/testkit/gwfix/deps.go

@@ -0,0 +1,43 @@
+package gwfix
+
+import (
+	"sync"
+
+	iamdomain "github.com/mulgadc/spinifex/spinifex/domains/iam"
+	stsdomain "github.com/mulgadc/spinifex/spinifex/domains/sts"
+	"github.com/nats-io/nats.go"
+)
+
+// Deps are the collaborators a test composes a gateway from. The gateway
+// config no longer carries them, so test setup keeps them beside the config.
+type Deps struct {
+	IAM   iamdomain.IAMService
+	STS   stsdomain.STSService
+	NATS  *nats.Conn
+	Nodes int
+}
+
+var (
+	depsMu sync.Mutex
+	depsOf = map[any]*Deps{}
+)
+
+// WithDeps records d for gw and returns gw.
+func WithDeps[P comparable](gw P, d Deps) P {
+	depsMu.Lock()
+	defer depsMu.Unlock()
+	depsOf[gw] = &d
+	return gw
+}
+
+// DepsOf returns the deps recorded for gw, creating an empty set when none.
+func DepsOf[P comparable](gw P) *Deps {
+	depsMu.Lock()
+	defer depsMu.Unlock()
+	d, ok := depsOf[gw]
+	if !ok {
+		d = &Deps{}
+		depsOf[gw] = d
+	}
+	return d
+}

internal/testkit/kvfault.go

@@ -0,0 +1,108 @@
+package testutil
+
+import (
+	"context"
+	"sync"
+
+	"github.com/nats-io/nats.go/jetstream"
+)
+
+// StaleReadKV wraps a bucket so that the first Get of a key after a write to it
+// returns the entry as it stood before that write, as a lagging replica does
+// when it answers a direct get. Later Gets see the current entry.
+type StaleReadKV struct {
+	jetstream.KeyValue
+
+	mu    sync.Mutex
+	stale map[string]jetstream.KeyValueEntry
+}
+
+var _ jetstream.KeyValue = (*StaleReadKV)(nil)
+
+// NewStaleReadKV wraps kv.
+func NewStaleReadKV(kv jetstream.KeyValue) *StaleReadKV {
+	return &StaleReadKV{KeyValue: kv, stale: map[string]jetstream.KeyValueEntry{}}
+}
+
+// remember captures the entry about to be overwritten.
+func (k *StaleReadKV) remember(ctx context.Context, key string) jetstream.KeyValueEntry {
+	prev, err := k.KeyValue.Get(ctx, key)
+	if err != nil {
+		return nil
+	}
+	return prev
+}
+
+func (k *StaleReadKV) arm(key string, prev jetstream.KeyValueEntry) {
+	if prev == nil {
+		return
+	}
+	k.mu.Lock()
+	k.stale[key] = prev
+	k.mu.Unlock()
+}
+
+func (k *StaleReadKV) Get(ctx context.Context, key string) (jetstream.KeyValueEntry, error) {
+	k.mu.Lock()
+	prev, ok := k.stale[key]
+	delete(k.stale, key)
+	k.mu.Unlock()
+	if ok {
+		return prev, nil
+	}
+	return k.KeyValue.Get(ctx, key)
+}
+
+func (k *StaleReadKV) Update(ctx context.Context, key string, value []byte, last uint64) (uint64, error) {
+	prev := k.remember(ctx, key)
+	rev, err := k.KeyValue.Update(ctx, key, value, last)
+	if err == nil {
+		k.arm(key, prev)
+	}
+	return rev, err
+}
+
+func (k *StaleReadKV) Put(ctx context.Context, key string, value []byte) (uint64, error) {
+	prev := k.remember(ctx, key)
+	rev, err := k.KeyValue.Put(ctx, key, value)
+	if err == nil {
+		k.arm(key, prev)
+	}
+	return rev, err
+}
+
+// RejectingKV wraps a bucket so that writes for which reject returns an error
+// fail with it, and every other operation passes through.
+type RejectingKV struct {
+	jetstream.KeyValue
+
+	reject func(key string, value []byte) error
+}
+
+var _ jetstream.KeyValue = (*RejectingKV)(nil)
+
+// NewRejectingKV wraps kv. reject runs before each Put, Create and Update.
+func NewRejectingKV(kv jetstream.KeyValue, reject func(key string, value []byte) error) *RejectingKV {
+	return &RejectingKV{KeyValue: kv, reject: reject}
+}
+
+func (k *RejectingKV) Put(ctx context.Context, key string, value []byte) (uint64, error) {
+	if err := k.reject(key, value); err != nil {
+		return 0, err
+	}
+	return k.KeyValue.Put(ctx, key, value)
+}
+
+func (k *RejectingKV) Create(ctx context.Context, key string, value []byte, opts ...jetstream.KVCreateOpt) (uint64, error) {
+	if err := k.reject(key, value); err != nil {
+		return 0, err
+	}
+	return k.KeyValue.Create(ctx, key, value, opts...)
+}
+
+func (k *RejectingKV) Update(ctx context.Context, key string, value []byte, last uint64) (uint64, error) {
+	if err := k.reject(key, value); err != nil {
+		return 0, err
+	}
+	return k.KeyValue.Update(ctx, key, value, last)
+}

scripts/README.md

@@ -6,7 +6,7 @@ Node installation, dev-environment lifecycle, guest image builds, and verificati
 
 | Script | What it does |
 |---|---|
-| `setup.sh` | The production binary installer, served from `install.mulgadc.com`. Installs deps, users, sudoers, the host firewall policy, files, systemd units, logrotate and udev rules, and loads `openvswitch`, `geneve`, `vhost_net` and `tun`, listing them in `/etc/modules-load.d/spinifex.conf` for later boots. It masks firewalld wherever it is installed, since the Spinifex nftables policy is the only supported node firewall. It writes a needrestart drop-in so an apt upgrade never restarts the Spinifex, OVS or OVN units; restarting them after an upgrade is left to the operator. An unset `--firewall` keeps the mode recorded in `/etc/spinifex/firewall/mode`, and otherwise arms the policy when firewalld was active. Where systemd-resolved is installed but not running, resolvconf is absent and NetworkManager runs, it hands host DNS to resolved: it saves `/etc/resolv.conf` once to `/var/lib/spinifex-reset/resolv.conf.orig`, links `/etc/resolv.conf` to the resolved stub, and stops unless resolved has upstream servers and the package repository host still resolves. It links `/usr/bin/spx` to `/usr/local/bin/spx` where sudo's PATH leaves out `/usr/local/bin`, and takes the nbdkit plugin directory from `nbdkit --dump-config`, stopping when nbdkit cannot say. `SETUP_STAGES` runs a subset; `ISO_BUILD=1` runs it inside the ISO builder's chroot. |
+| `setup.sh` | The production binary installer, served from `install.mulgadc.com`. Installs deps, users, sudoers, the host firewall policy, files, systemd units, logrotate and udev rules, and loads `openvswitch`, `geneve`, `vhost_net` and `tun`, listing them in `/etc/modules-load.d/spinifex.conf` for later boots. It masks firewalld wherever it is installed, since the Spinifex nftables policy is the only supported node firewall. It writes a needrestart drop-in so an apt upgrade never restarts the Spinifex, OVS or OVN units; restarting them after an upgrade is left to the operator. An unset `--firewall` keeps the mode recorded in `/etc/spinifex/firewall/mode`, and otherwise arms the policy when firewalld was active. Where systemd-resolved is installed but not running, resolvconf is absent and NetworkManager runs, it hands host DNS to resolved: it saves `/etc/resolv.conf` once to `/var/lib/spinifex-reset/resolv.conf.orig`, writes `/etc/NetworkManager/conf.d/90-spinifex-dns.conf` (`dns=systemd-resolved`, `rc-manager=unmanaged`) so NetworkManager never rewrites `/etc/resolv.conf`, links `/etc/resolv.conf` to the resolved stub, and stops unless the link holds, resolved has upstream servers and the package repository host still resolves. A rerun after that handover adds a missing drop-in, reloads NetworkManager and relinks the stub. It links `/usr/bin/spx` to `/usr/local/bin/spx` where sudo's PATH leaves out `/usr/local/bin`, and takes the nbdkit plugin directory from `nbdkit --dump-config`, stopping when nbdkit cannot say. `SETUP_STAGES` runs a subset; `ISO_BUILD=1` runs it inside the ISO builder's chroot. |
 | `setup-ovn.sh` | Bootstraps a compute node for OVN VPC networking: installs OVN/OVS, creates `br-int`, configures the WAN bridge (bridged, DHCP, or routed-NAT uplink), sets the chassis identity, and applies overlay sysctls. It runs OVS and OVN as root, setting `OVS_USER_ID` and `OVN_USER_ID` where `/etc/sysconfig` holds them, and hooks the socket-permission helper onto `ovs-vswitchd`, `ovn-controller`, `openvswitch-ipsec` and, on EL, `ovn-northd`. `--management` also starts the OVN central services, writes their `ovn-ctl` flags to `/etc/default/ovn-central` (Debian) or `/etc/sysconfig/ovn-northd` (EL), and records the role in `/etc/spinifex/ovn-role`. With clustered databases it returns only once this node's NB and SB members are back in their RAFT clusters under a known leader, and fails after `OVN_RAFT_REJOIN_WAIT` seconds (default 60) otherwise, so reruns one node at a time never take two members of a cluster down. A later run without `--management` refuses while that file, or the `OVN_CTL_OPTS` line older nodes carry in `/etc/default/ovn-central`, records the role. It adds `maxsize 100M` to the OVS and OVN logrotate files, plus `rotate 5` where the file sets no count, and rotates as root, matching the daemons. On a host whose IPsec stack is libreswan it enables, stops and masks `openvswitch-ipsec.service` and `ipsec.service` for the daemon to unmask when IPsec is on, unless `openvswitch-ipsec.service` is already running; elsewhere it enables and starts `openvswitch-ipsec.service`, unless the daemon has masked it to keep IPsec off. The links it creates persist through systemd-networkd where that is enabled, and otherwise through `/etc/spinifex/hostnet.json`, which `spinifex-hostnet.service` (`spx admin hostnet apply`) replays at boot and on every OVS start, starting `ovn-controller` again when an OVS stop took it down; the first run records the choice in `/etc/spinifex/hostnet-backend`. Where NetworkManager runs, it leaves the links Spinifex owns unmanaged. |
 | `install-node.sh` | Forms a multi-node cluster across hosts that are already running Spinifex: rebuilds the OVN database in clustered form, runs `spx admin init` on the first host and `spx admin join` on the rest, then verifies. Before touching any host it refuses hosts of different OS families (`distro_os_family`) or `spx` versions, and `--ipsec on` when any host uses libreswan. The scripted form of [install-multi-node](../docs/install/install-multi-node/README.md). Destroys the joining nodes' logical network state and crypto — see its `--help`. |
 | `dev-install.sh` | Full local dev setup through the production installer: builds from source, packs a tarball, runs `setup.sh`, initialises the cluster, and starts services under systemd. |

scripts/appliance/firstboot.sh

@@ -90,6 +90,12 @@ else
     log "WARNING: clock not synced, continuing with $(date -uIs)"
 fi
 
+# ── Connection tracking ─────────────────────────────────────────────────────
+# Before OVN opens any link, so arming the firewall later cannot drop one the
+# kernel first saw mid-stream. Non-fatal; the unit reloads it on later boots.
+/usr/local/lib/spinifex/spinifex-firewall-apply track \
+    || log "WARNING: connection tracking not loaded; arming the firewall may drop OVN links"
+
 # ── OVS and OVN ─────────────────────────────────────────────────────────────
 # Start them before setup-ovn.sh so their databases exist when it runs: on
 # physical hardware first-boot DB initialisation outlasts its own timeout.

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

@@ -12,11 +12,12 @@ set -eu
 # once.
 #
 #   server      → eks-token-webhook, k3s (server), k3s-first-boot (bootstrap
-#                 publish), mulga-eks-state-report
+#                 publish), mulga-eks-state-report, mulga-eks-node-gc
 #   server-join → eks-token-webhook, k3s (server, joins the first server's etcd
-#                 quorum), mulga-eks-state-report. NO k3s-first-boot: the first
-#                 server already publishes the cluster-identical bootstrap
-#                 artifacts; a join re-publish only races the bootstrap bus.
+#                 quorum), mulga-eks-state-report, mulga-eks-node-gc. NO
+#                 k3s-first-boot: the first server already publishes the
+#                 cluster-identical bootstrap artifacts; a join re-publish only
+#                 races the bootstrap bus.
 #   agent       → k3s-agent
 #
 # Paths are overridable via env so the selection logic is unit-testable (bats)
@@ -140,6 +141,7 @@ case "${ROLE}" in
         # writer owns. HA multi-server addon delivery is tracked separately
         # Server-join nodes do not run this setup.
         svc_enable mulga-eks-addon-sync
+        svc_enable mulga-eks-node-gc
         svc_start eks-token-webhook
         svc_start mulga-eks-k3s-recovery
         svc_start k3s
@@ -147,6 +149,7 @@ case "${ROLE}" in
         svc_start k3s-first-boot
         svc_start mulga-eks-state-report
         svc_start mulga-eks-addon-sync
+        svc_start mulga-eks-node-gc
         ;;
     server-join)
         log "configuring server-join role"
@@ -158,11 +161,13 @@ case "${ROLE}" in
         svc_enable k3s
         svc_enable konnectivity-server
         svc_enable mulga-eks-state-report
+        svc_enable mulga-eks-node-gc
         svc_start eks-token-webhook
         svc_start mulga-eks-k3s-recovery
         svc_start k3s
         svc_start konnectivity-server
         svc_start mulga-eks-state-report
+        svc_start mulga-eks-node-gc
         ;;
     agent)
         log "configuring agent role"

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

@@ -141,6 +141,8 @@ for INIT in openrc systemd; do
             END { print (rn > 0 && kn > 0 && rn < kn) ? "yes" : "no" }
         ' "${CALLS}"
     )"
+    check "server(${INIT}): enables node-gc" "yes" "$(present "${CALLS}" "$(enable_cmd "${INIT}" mulga-eks-node-gc)")"
+    check "server(${INIT}): starts node-gc" "yes" "$(present "${CALLS}" "$(start_cmd "${INIT}" mulga-eks-node-gc)")"
     check "server(${INIT}): no k3s-agent" "no" "$(present "${CALLS}" "$(enable_cmd "${INIT}" k3s-agent)")"
     check "server(${INIT}): self-disables" "yes" "$(present "${CALLS}" "$(disable_cmd "${INIT}" eks-node-role)")"
 
@@ -153,6 +155,8 @@ for INIT in openrc systemd; do
     check "server-join(${INIT}): enables webhook" "yes" "$(present "${CALLS}" "$(enable_cmd "${INIT}" eks-token-webhook)")"
     check "server-join(${INIT}): enables k3s" "yes" "$(present "${CALLS}" "$(enable_cmd "${INIT}" k3s)")"
     check "server-join(${INIT}): enables state-report" "yes" "$(present "${CALLS}" "$(enable_cmd "${INIT}" mulga-eks-state-report)")"
+    check "server-join(${INIT}): enables node-gc" "yes" "$(present "${CALLS}" "$(enable_cmd "${INIT}" mulga-eks-node-gc)")"
+    check "server-join(${INIT}): starts node-gc" "yes" "$(present "${CALLS}" "$(start_cmd "${INIT}" mulga-eks-node-gc)")"
     check "server-join(${INIT}): enables konnectivity-server" "yes" "$(present "${CALLS}" "$(enable_cmd "${INIT}" konnectivity-server)")"
     check "server-join(${INIT}): enables k3s-recovery" "yes" "$(present "${CALLS}" "$(enable_cmd "${INIT}" mulga-eks-k3s-recovery)")"
     check "server-join(${INIT}): starts k3s-recovery" "yes" "$(present "${CALLS}" "$(start_cmd "${INIT}" mulga-eks-k3s-recovery)")"
@@ -170,7 +174,7 @@ for INIT in openrc systemd; do
     check "agent(${INIT}): exit 0" 0 "${rc}"
     check "agent(${INIT}): role file" "agent" "$(cat "${ROLE_FILE}")"
     check "agent(${INIT}): enables k3s-agent" "yes" "$(present "${CALLS}" "$(enable_cmd "${INIT}" k3s-agent)")"
-    check "agent(${INIT}): no server services" "yes" "$(absent_all "${CALLS}" "eks-token-webhook" "k3s-first-boot" "$(enable_cmd "${INIT}" k3s)")"
+    check "agent(${INIT}): no server services" "yes" "$(absent_all "${CALLS}" "eks-token-webhook" "k3s-first-boot" "mulga-eks-node-gc" "$(enable_cmd "${INIT}" k3s)")"
 
     # --- agent (inferred from agent.env, no explicit role) ---
     run_case "${INIT}" agent_inferred

scripts/images/eks-node/manifest.conf

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

scripts/images/eks-node/mulga-eks-node-gc.initd

@@ -0,0 +1,27 @@
+#!/sbin/openrc-run
+# mulga-eks-node-gc — server-role service that deletes the Node of every
+# nodegroup worker the nodegroup has terminated, every NODE_GC_INTERVAL
+# seconds. Enabled on every server role by eks-node-role.sh.
+
+description="Mulga EKS terminated-worker Node cleanup"
+command="/usr/local/sbin/mulga-eks-node-gc"
+command_background="yes"
+pidfile="/run/mulga-eks-node-gc.pid"
+output_log="/var/log/mulga-eks-node-gc.log"
+error_log="/var/log/mulga-eks-node-gc.log"
+
+# A Node goes NotReady about 40s after its worker stops, so 30s deletes it
+# within a couple of minutes of the terminate.
+export NODE_GC_INTERVAL="${NODE_GC_INTERVAL:-30}"
+
+depend() {
+    need net
+    after k3s
+}
+
+start_pre() {
+    if [ ! -f /etc/spinifex-eks/first-boot.env ]; then
+        eerror "/etc/spinifex-eks/first-boot.env not present — cloud-init did not seed gateway creds"
+        return 1
+    fi
+}

scripts/images/eks-node/mulga-eks-node-gc.sh

@@ -0,0 +1,89 @@
+#!/bin/sh
+set -eu
+
+# mulga-eks-node-gc — deletes the Node of a nodegroup worker whose instance the
+# nodegroup has terminated (DeleteNodegroup, scale-down, reclaim), as the AWS
+# cloud controller deletes a Node whose instance no longer exists. K3s runs no
+# AWS cloud provider: its own reports every instance as existing, so without
+# this a terminated worker stays a NotReady Node forever, its pods stay bound to
+# it, and once most Nodes are NotReady the node lifecycle controller stops
+# evicting at all.
+#
+# Pull (host→VM): GET /clusters/{cluster}/internal-workers/{accountId}, the
+# instance ID of every worker the cluster's nodegroups still own.
+#
+# A Node is deleted only when all of these hold, so a live worker is never
+# touched: it carries the eks.amazonaws.com/nodegroup label, its providerID is
+# aws:///<az>/<instance-id>, it is not Ready, and its instance is not in the
+# fetched set. A failed fetch deletes nothing.
+#
+# Runs on every server; deletes are idempotent. Loops every NODE_GC_INTERVAL
+# seconds in service mode, one-shot otherwise.
+
+ENVFILE=${ENVFILE:-/etc/spinifex-eks/first-boot.env}
+[ -f "${ENVFILE}" ] || { logger -t mulga-eks-node-gc "${ENVFILE} missing — exiting"; exit 0; }
+# set -a so the sourced creds are exported to the eks-gateway-fetch child.
+set -a
+# shellcheck disable=SC1090
+. "${ENVFILE}"
+set +a
+
+: "${EKS_GATEWAY_URL:?}"
+: "${EKS_ACCOUNT_ID:?}"
+: "${EKS_CLUSTER_NAME:?}"
+
+KUBECONFIG=${KUBECONFIG:-/etc/rancher/k3s/k3s.yaml}
+export KUBECONFIG
+
+# One "name|providerID|nodegroup|Ready" row per Node.
+NODE_ROWS='jsonpath={range .items[*]}{.metadata.name}{"|"}{.spec.providerID}{"|"}{.metadata.labels.eks\.amazonaws\.com/nodegroup}{"|"}{.status.conditions[?(@.type=="Ready")].status}{"\n"}{end}'
+
+log() {
+    logger -t mulga-eks-node-gc "$*"
+    echo "[mulga-eks-node-gc] $*" > "${CONSOLE:-/dev/console}" 2>/dev/null || true
+}
+
+# stale_nodes LIVE_FILE — reads Node rows on stdin and prints the name of each
+# Node to delete: a nodegroup worker, not Ready, whose instance is not listed
+# in LIVE_FILE.
+stale_nodes() {
+    awk -F'|' -v live="$1" '
+        BEGIN { while ((getline id < live) > 0) keep[id] = 1 }
+        $3 == "" || $4 == "True" { next }
+        $2 !~ /^aws:\/\/\/[^\/]+\/i-[0-9a-f]+$/ { next }
+        { n = split($2, p, "/"); if (!(p[n] in keep)) print $1 }
+    '
+}
+
+sync_once() {
+    # Nodes are listed before the workers are fetched. A worker is recorded on
+    # its nodegroup before it boots, so every Node listed here is already in a
+    # set fetched after.
+    if ! rows=$(kubectl get nodes -o "${NODE_ROWS}" 2>/dev/null); then
+        log "kubectl get nodes failed; skipping this pass"
+        return 0
+    fi
+    live=$(mktemp)
+    if ! eks-gateway-fetch -resource workers > "${live}" 2>/dev/null; then
+        log "fetch of the cluster's nodegroup workers failed; deleting nothing"
+        rm -f "${live}"
+        return 0
+    fi
+    printf '%s\n' "${rows}" | stale_nodes "${live}" | while read -r node; do
+        if kubectl delete node "${node}" --ignore-not-found --wait=false >/dev/null 2>&1; then
+            log "deleted Node ${node}: its instance is no longer a nodegroup worker"
+        else
+            log "delete of Node ${node} failed; retrying next pass"
+        fi
+    done
+    rm -f "${live}"
+}
+
+if [ -n "${NODE_GC_INTERVAL:-}" ] && [ "${NODE_GC_INTERVAL}" -gt 0 ] 2>/dev/null; then
+    while true; do
+        sync_once
+        sleep "${NODE_GC_INTERVAL}"
+    done
+else
+    sync_once
+fi

scripts/images/eks-node/mulga-eks-node-gc_test.sh

@@ -0,0 +1,107 @@
+#!/bin/sh
+# Self-contained POSIX test for mulga-eks-node-gc.sh. kubectl, eks-gateway-fetch
+# and logger are stubbed on PATH: kubectl serves canned Node rows and records
+# every delete, and eks-gateway-fetch serves a canned worker set or fails.
+#
+# Run: sh scripts/images/eks-node/mulga-eks-node-gc_test.sh
+set -eu
+
+SCRIPT_DIR=$(cd "$(dirname "$0")" && pwd)
+AGENT="${SCRIPT_DIR}/mulga-eks-node-gc.sh"
+WORK=$(mktemp -d)
+trap 'rm -rf "${WORK}"' EXIT
+
+STUBBIN="${WORK}/bin"
+mkdir -p "${STUBBIN}"
+
+cat > "${STUBBIN}/kubectl" <<'EOF'
+#!/bin/sh
+case "$1 $2" in
+    "get nodes")
+        [ "${NODES_RC:-0}" -eq 0 ] || exit "${NODES_RC}"
+        printf '%s\n' "${NODES_BODY:-}" ;;
+    "delete node") echo "$3" >> "${DELETED}" ;;
+esac
+EOF
+
+cat > "${STUBBIN}/eks-gateway-fetch" <<'EOF'
+#!/bin/sh
+echo "$*" >> "${FETCHED}"
+[ "${FETCH_RC:-0}" -eq 0 ] || exit "${FETCH_RC}"
+printf '%s' "${LIVE_BODY:-}"
+EOF
+
+cat > "${STUBBIN}/logger" <<'EOF'
+#!/bin/sh
+:
+EOF
+
+chmod +x "${STUBBIN}"/*
+PATH="${STUBBIN}:${PATH}"
+export PATH
+
+ENVFILE="${WORK}/first-boot.env"
+cat > "${ENVFILE}" <<'EOF'
+EKS_GATEWAY_URL=https://gw.invalid:9999
+EKS_ACCOUNT_ID=000000000001
+EKS_CLUSTER_NAME=demo3
+EOF
+CONSOLE="${WORK}/console"
+DELETED="${WORK}/deleted"
+FETCHED="${WORK}/fetched"
+export ENVFILE CONSOLE DELETED FETCHED
+
+PASS=0
+FAIL=0
+check() {
+    if [ "$2" = "$3" ]; then
+        PASS=$((PASS + 1))
+    else
+        FAIL=$((FAIL + 1))
+        printf 'FAIL: %s\n  want: %s\n  got:  %s\n' "$1" "$2" "$3"
+    fi
+}
+
+run_agent() {
+    : > "${DELETED}"
+    : > "${FETCHED}"
+    sh "${AGENT}"
+}
+
+deleted() { tr '\n' ' ' < "${DELETED}" | sed 's/ $//'; }
+
+# Every way a Node can look: only a NotReady nodegroup worker whose instance the
+# cluster's nodegroups no longer own is deleted.
+NODES_BODY='live|aws:///z1/i-0aaa|ng1|True
+gone|aws:///z1/i-0bbb|ng1|Unknown
+gone-never-reported|aws:///z1/i-0ddd|ng2|
+owned-notready|aws:///z1/i-0ccc|ng1|False
+gone-but-ready|aws:///z1/i-0eee|ng1|True
+self-managed|aws:///z1/i-0fff||Unknown
+k3s-provider|k3s://k3s-node|ng1|Unknown'
+LIVE_BODY='i-0aaa
+i-0ccc
+'
+export NODES_BODY LIVE_BODY
+
+run_agent
+check "deletes only NotReady nodegroup Nodes whose instance is gone" "gone gone-never-reported" "$(deleted)"
+check "fetches the workers resource" "-resource workers" "$(cat "${FETCHED}")"
+
+# A cluster with no nodegroups left: every NotReady nodegroup Node goes.
+LIVE_BODY='' run_agent
+check "empty worker set deletes every NotReady nodegroup Node" "gone gone-never-reported owned-notready" "$(deleted)"
+
+# A failed fetch is not an empty set: nothing is deleted.
+FETCH_RC=1 run_agent
+check "failed fetch deletes nothing" "" "$(deleted)"
+
+# Without a Node listing the workers are not fetched, so a set fetched before a
+# Node registered can never be compared against it.
+NODES_RC=1 run_agent
+check "failed node listing deletes nothing" "" "$(deleted)"
+check "failed node listing does not fetch" "" "$(cat "${FETCHED}")"
+
+echo "---"
+echo "PASS=${PASS} FAIL=${FAIL}"
+[ "${FAIL}" -eq 0 ]

scripts/images/eks-node/setup.sh

@@ -44,9 +44,10 @@ ln -sf /usr/local/bin/k3s /usr/local/bin/ctr
 # role's services are baked; the selector enables the right ones at first boot.
 chmod 0755 /etc/init.d/eks-node-role /etc/init.d/k3s /etc/init.d/k3s-agent \
     /etc/init.d/eks-token-webhook /etc/init.d/k3s-first-boot /etc/init.d/mulga-eks-state-report \
-    /etc/init.d/mulga-eks-addon-sync /etc/init.d/konnectivity-server
+    /etc/init.d/mulga-eks-addon-sync /etc/init.d/mulga-eks-node-gc /etc/init.d/konnectivity-server
 chmod 0755 /usr/local/sbin/eks-node-role /usr/local/sbin/k3s-first-boot \
-    /usr/local/sbin/mulga-eks-state-report /usr/local/sbin/mulga-eks-addon-sync
+    /usr/local/sbin/mulga-eks-state-report /usr/local/sbin/mulga-eks-addon-sync \
+    /usr/local/sbin/mulga-eks-node-gc
 # start_pre helpers shared with the systemd wrappers that will eventually
 # replace these runscripts.
 chmod 0755 /usr/local/sbin/k3s-prestart /usr/local/sbin/k3s-agent-prestart \

scripts/setup.sh

@@ -39,6 +39,7 @@ FIREWALL_PEERS="${FIREWALL_DIR}/peers.nft"
 FIREWALL_LOCAL="${FIREWALL_DIR}/local.nft"
 FIREWALL_OPEN="${FIREWALL_DIR}/open-ports.nft"
 FIREWALL_CUSTOM="${FIREWALL_DIR}/custom.nft"
+FIREWALL_TRACK="${FIREWALL_DIR}/conntrack.nft"
 FIREWALL_MODE_FILE="${SPINIFEX_FIREWALL_MODE_FILE:-${FIREWALL_DIR}/mode}"
 FIREWALL_APPLY="/usr/local/lib/spinifex/spinifex-firewall-apply"
 
@@ -749,6 +750,28 @@ RULES
     $SUDO chmod 0644 "$FIREWALL_RULES"
     info "  $FIREWALL_RULES"
 
+    # The kernel only tracks connections while some rule uses conntrack. Without
+    # this table a disarmed node tracks nothing, and a flow opened then (OVN's
+    # RAFT links) is first seen mid-stream at arm time, from whichever end speaks
+    # first. Seen from the server, that is a new flow to an ephemeral port, and
+    # the policy drops it. Kept loaded whether or not the policy is armed.
+    $SUDO tee "$FIREWALL_TRACK" > /dev/null << 'TRACK'
+# Spinifex connection tracking. Managed by scripts/setup.sh — edits are overwritten.
+# Accepts everything and filters nothing; it exists so conntrack keeps running
+# while inet spinifex_filter is not loaded. Never delete it on a running node.
+table inet spinifex_conntrack
+delete table inet spinifex_conntrack
+
+table inet spinifex_conntrack {
+    chain input {
+        type filter hook input priority filter; policy accept;
+        ct state invalid counter
+    }
+}
+TRACK
+    $SUDO chmod 0644 "$FIREWALL_TRACK"
+    info "  $FIREWALL_TRACK"
+
     # Host-local facts the policy needs that the daemon has no view of. Separate
     # from peers.nft because the daemon rewrites that file and would drop this.
     # Refreshed on every setup.sh run, so an sshd port change is picked up by an
@@ -816,6 +839,7 @@ set -eu
 RULES="/etc/spinifex/firewall/spinifex.nft"
 PEERS="/etc/spinifex/firewall/peers.nft"
 OPEN="/etc/spinifex/firewall/open-ports.nft"
+TRACK="/etc/spinifex/firewall/conntrack.nft"
 
 # 0 is a sentinel that keeps the set non-empty, because nft rejects an empty
 # one. No packet can match it: the kernel has no destination port 0.
@@ -837,13 +861,33 @@ write_open_ports() {
 # Deliberately ahead of the policy-file check below: turning the firewall off is
 # the recovery path, and refusing to run it because the policy is missing leaves
 # an operator with a node they can neither arm nor disarm.
+#
+# The tracking table goes in in the same transaction as the delete, so conntrack
+# never stops: a flow that opens while disarmed is then tracked from its SYN,
+# and re-arming does not drop it. The boot unit stays enabled to reload it.
 if [ "${1:-}" = "disable" ]; then
-    nft delete table inet spinifex_filter 2>/dev/null || true
+    if [ -r "$TRACK" ]; then
+        printf 'include "%s"\ntable inet spinifex_filter\ndelete table inet spinifex_filter\n' "$TRACK" | nft -f - \
+            || nft delete table inet spinifex_filter 2>/dev/null || true
+    else
+        nft delete table inet spinifex_filter 2>/dev/null || true
+    fi
     rm -f "$PEERS"
-    systemctl disable --now spinifex-firewall.service >/dev/null 2>&1 || true
     exit 0
 fi
 
+# track loads only the tracking table. boot is the unit's start: tracking on
+# every node, and the policy on top only where the daemon has written peers.
+if [ "${1:-}" = "track" ] || [ "${1:-}" = "boot" ]; then
+    if [ -r "$TRACK" ]; then
+        nft -f "$TRACK"
+    elif [ "$1" = "track" ]; then
+        echo "no tracking table at $TRACK" >&2
+        exit 1
+    fi
+    if [ "$1" = "track" ] || [ ! -r "$PEERS" ]; then exit 0; fi
+fi
+
 # Every remaining verb writes the policy, so it has to exist. Failing here
 # changes nothing, which leaves a node whose daemon has not yet resolved cluster
 # membership reachable rather than cut off from a cluster it cannot name.
@@ -917,10 +961,9 @@ if [ "${1:-}" = "set-peers" ]; then
     fi
     rm -f "$tmp" "$bak"
 
-    # The policy is loaded, so make it survive a reboot too. `disable` turns this
-    # unit off, and forming a cluster means disabling the firewall on every node
-    # first; without this, re-arming would look complete while silently leaving
-    # the node bare after its next boot. Idempotent and quiet in the normal case.
+    # The policy is loaded, so make it survive a reboot too. An older helper's
+    # `disable` turned this unit off; without this, re-arming such a node would
+    # look complete while leaving it bare after its next boot. Idempotent.
     systemctl enable spinifex-firewall.service >/dev/null 2>&1 || true
     exit 0
 fi
@@ -937,6 +980,13 @@ APPLY
     $SUDO chmod 0755 "$FIREWALL_APPLY"
     info "  $FIREWALL_APPLY"
 
+    # Now, not at the next boot: a fresh install forms its cluster before then,
+    # and an upgrade must start tracking before the daemon next arms. A chroot
+    # would load it into the build host's kernel.
+    if [ "${ISO_BUILD:-0}" != "1" ]; then
+        $SUDO "$FIREWALL_APPLY" track || warn "  Could not load $FIREWALL_TRACK: arming later may drop live cluster connections"
+    fi
+
     # Say which state ssh is in either way. A node whose allowlist was never
     # narrowed reads as configured otherwise, and that is the whole risk here.
     _ssh_srcs="$(grep -o '{[^}]*}' "$FIREWALL_CUSTOM" 2>/dev/null | tr -d '{}' | tr -s ' ')"
@@ -964,6 +1014,9 @@ RESOLV_CONF="${SPINIFEX_RESOLV_CONF:-/etc/resolv.conf}"
 RESOLVED_STUB=/run/systemd/resolve/stub-resolv.conf
 # node-reset.sh keeps /var/lib/spinifex-reset, so the copy survives until uninstall.
 RESOLV_CONF_SAVED="${SPINIFEX_RESOLV_CONF_SAVED:-/var/lib/spinifex-reset/resolv.conf.orig}"
+# NetworkManager started while resolv.conf is a plain file rewrites it on its
+# next DNS update, replacing the stub link; this drop-in stops that.
+NM_DNS_DROPIN="${SPINIFEX_NM_DNS_DROPIN:-/etc/NetworkManager/conf.d/90-spinifex-dns.conf}"
 
 # systemd-resolved ships resolvconf as a link to resolvectl, which only works
 # while resolved runs, so that one is not a resolvconf backend.
@@ -1016,10 +1069,52 @@ dns_handover_failed() {
     fatal "Host DNS through systemd-resolved does not work: $1.
     To put host DNS back as it was before this installer:
       sudo systemctl disable --now systemd-resolved
+      sudo rm -f $NM_DNS_DROPIN
       sudo cp -a --remove-destination $RESOLV_CONF_SAVED $RESOLV_CONF
       sudo systemctl restart NetworkManager"
 }
 
+# Prints the rc-manager NetworkManager would use if it is not unmanaged, which
+# means a file read after the Spinifex drop-in overrides it.
+nm_rc_manager_override() {
+    local config
+    config=$($SUDO NetworkManager --print-config) || return 1
+    printf '%s\n' "$config" | awk -F= '
+        /^\[/ { main = ($0 == "[main]"); next }
+        main && $1 == "rc-manager" { rc = $2 }
+        END { if (rc != "unmanaged") print "rc-manager=" (rc == "" ? "symlink" : rc) }'
+}
+
+# Writes the NetworkManager drop-in and checks nothing overrides it. Returns 1
+# when it was already in place.
+install_nm_dns_dropin() {
+    local want override changed=1
+    want=$(printf '[main]\ndns=systemd-resolved\nrc-manager=unmanaged')
+    if [ "$(cat "$NM_DNS_DROPIN" 2>/dev/null)" != "$want" ]; then
+        $SUDO mkdir -p "$(dirname "$NM_DNS_DROPIN")"
+        printf '%s\n' "$want" | $SUDO tee "$NM_DNS_DROPIN" > /dev/null || dns_handover_failed "cannot write $NM_DNS_DROPIN"
+        info "  $NM_DNS_DROPIN: NetworkManager leaves $RESOLV_CONF alone"
+        changed=0
+    fi
+    override=$(nm_rc_manager_override) || dns_handover_failed "cannot read NetworkManager's configuration with 'NetworkManager --print-config'"
+    if [ -n "$override" ]; then
+        dns_handover_failed "NetworkManager reads $override from a file after $NM_DNS_DROPIN, so it would replace the $RESOLV_CONF link. Remove that setting ('NetworkManager --print-config' shows the files it reads) and re-run this installer"
+    fi
+    return "$changed"
+}
+
+# Links the stub and checks the link is still there a second later, relinking a
+# bounded number of times in case a write NetworkManager had queued lands first.
+link_resolved_stub() {
+    local i
+    for i in 1 2 3 4 5; do
+        $SUDO ln -sfn "$RESOLVED_STUB" "$RESOLV_CONF"
+        sleep 1
+        [ "$(readlink "$RESOLV_CONF")" = "$RESOLVED_STUB" ] && return 0
+    done
+    dns_handover_failed "$RESOLV_CONF is not a link to $RESOLVED_STUB 5 seconds after linking it, so something keeps rewriting it"
+}
+
 # Checks resolved has upstream servers and still resolves the package
 # repository host, the one name this host is known to have resolved.
 check_resolved_dns() {
@@ -1053,8 +1148,13 @@ hand_dns_to_resolved() {
     [ "${ISO_BUILD:-0}" = "1" ] && return 0
 
     if systemctl is-active --quiet systemd-resolved.service; then
-        # A handover made by an earlier run is checked again on every run.
+        # A handover made by an earlier run is checked again, and repaired, on
+        # every run: installs before the drop-in could lose the link.
         if [ -e "$RESOLV_CONF_SAVED" ] || [ -L "$RESOLV_CONF_SAVED" ]; then
+            if install_nm_dns_dropin && systemctl is-active --quiet NetworkManager.service; then
+                $SUDO systemctl reload NetworkManager.service || dns_handover_failed "NetworkManager did not reload"
+            fi
+            [ "$(readlink "$RESOLV_CONF")" = "$RESOLVED_STUB" ] || link_resolved_stub
             check_resolved_dns
         fi
         return 0
@@ -1081,10 +1181,15 @@ hand_dns_to_resolved() {
         $SUDO cp -a "$RESOLV_CONF" "$RESOLV_CONF_SAVED" || fatal "Cannot save $RESOLV_CONF to $RESOLV_CONF_SAVED"
         info "  saved $RESOLV_CONF to $RESOLV_CONF_SAVED"
     fi
+    install_nm_dns_dropin || true
     $SUDO systemctl enable --now systemd-resolved.service > /dev/null 2>&1 \
         || dns_handover_failed "systemd-resolved did not start"
     $SUDO systemctl restart NetworkManager.service || dns_handover_failed "NetworkManager did not restart"
-    $SUDO ln -sfn "$RESOLVED_STUB" "$RESOLV_CONF"
+    # Best effort: a link with no DHCP server can hold startup past the timeout.
+    if command -v nm-online > /dev/null; then
+        nm-online -s -q --timeout=15 || true
+    fi
+    link_resolved_stub
     check_resolved_dns
     info "  $RESOLV_CONF -> $RESOLVED_STUB"
 }
@@ -1926,9 +2031,9 @@ EOF
     # takes effect once enabled) — without this the JetStream ENOSPC-latch
     # watchdog is inert and a full disk requires a manual restart forever.
     $SUDO systemctl enable --now spinifex-nats-watchdog.timer
-    # Enabled, not started: its ConditionPathExists holds it inert until the
-    # daemon writes a peer file, so this only decides that a node which has one
-    # gets the policy back at boot, before any service opens a socket.
+    # Enabled on every node: at boot it loads connection tracking everywhere,
+    # and the policy only on a node whose daemon has written a peer file, so it
+    # is back before any service opens a socket.
     $SUDO systemctl enable spinifex-firewall.service
     info "Systemd units installed and enabled (per-service users)"
 }

scripts/setup_firewall_conntrack_test.sh

@@ -0,0 +1,214 @@
+#!/bin/sh
+# setup_firewall_conntrack_test.sh — arms the real host firewall under a TCP
+# flow that opened while it was disarmed, in two network namespaces.
+#
+# The policy, the tracking table and spinifex-firewall-apply are cut out of
+# setup.sh with their paths moved under a temp dir, and the helper's verbs are
+# run inside the client namespace. Needs root or passwordless sudo, ip, nft and
+# python3; skips without them. A stub systemctl keeps the helper off the host.
+set -eu
+
+SCRIPT_DIR=$(CDPATH='' cd -- "$(dirname -- "$0")" && pwd)
+SETUP="$SCRIPT_DIR/setup.sh"
+
+echo "-- setup.sh firewall: flows that outlive a disarm"
+
+if [ "$(id -u)" = 0 ]; then SUDO=""
+elif sudo -n true 2>/dev/null; then SUDO="sudo -n"
+else echo "  skip - needs root or passwordless sudo"; exit 0
+fi
+for bin in ip nft python3; do
+    if ! PATH="$PATH:/usr/sbin:/sbin" command -v "$bin" > /dev/null 2>&1; then
+        echo "  skip - $bin not installed"; exit 0
+    fi
+done
+
+TMP=$(mktemp -d)
+S="fwct-s-$$"
+C="fwct-c-$$"
+cleanup() {
+    $SUDO ip netns del "$S" 2> /dev/null || true
+    $SUDO ip netns del "$C" 2> /dev/null || true
+    rm -rf "$TMP"
+}
+trap cleanup EXIT
+
+fails=0
+pass() { echo "  ok   - $1"; }
+fail() { echo "  FAIL - $1"; fails=$((fails + 1)); }
+
+# heredoc <tag> — the body of setup.sh's quoted heredoc named <tag>, with the
+# firewall directory moved under $TMP.
+heredoc() {
+    sed -n "/<< '$1'\$/,/^$1\$/p" "$SETUP" | sed '1d;$d' | sed "s#/etc/spinifex/firewall#$TMP/fw#g"
+}
+
+mkdir -p "$TMP/fw" "$TMP/bin"
+heredoc RULES > "$TMP/fw/spinifex.nft"
+heredoc APPLY > "$TMP/apply"
+heredoc TRACK > "$TMP/fw/conntrack.nft"
+printf 'define spinifex_ssh_ports = { 22 }\n' > "$TMP/fw/local.nft"
+printf 'define spinifex_open_ports = { 0 }\n' > "$TMP/fw/open-ports.nft"
+printf 'define trusted_ssh_peers = { 0.0.0.0/0 }\n' > "$TMP/fw/custom.nft"
+printf '#!/bin/sh\nexit 0\n' > "$TMP/bin/systemctl"
+chmod +x "$TMP/bin/systemctl"
+if [ ! -s "$TMP/fw/spinifex.nft" ] || [ ! -s "$TMP/apply" ]; then
+    echo "  FAIL - could not cut the policy or the helper out of setup.sh"; exit 1
+fi
+
+# One process owns both ends, so the order packets are sent in is exact. The
+# server is 10.99.0.1 (a peer) and also holds 10.99.0.9 (not a peer).
+cat > "$TMP/flow.py" << 'PY'
+import os, socket, subprocess, sys
+
+srv_ns, cli_ns, script = sys.argv[1], sys.argv[2], sys.argv[3]
+
+def enter(ns):
+    fd = os.open("/run/netns/" + ns, os.O_RDONLY)
+    os.setns(fd, os.CLONE_NEWNET)
+    os.close(fd)
+
+enter(srv_ns)
+listener = socket.socket()
+listener.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
+listener.bind(("10.99.0.1", 6644))
+listener.listen(1)
+enter(cli_ns)
+
+cli = srv = None
+for step in script.split("@@"):
+    if step == "connect":
+        cli = socket.create_connection(("10.99.0.1", 6644), timeout=3)
+        srv, _ = listener.accept()
+        cli.settimeout(2)
+        srv.settimeout(2)
+    elif step in ("s2c", "c2s"):
+        tx, rx = (srv, cli) if step == "s2c" else (cli, srv)
+        tx.sendall(b"x" * 300)
+        got = b""
+        try:
+            while len(got) < 300:
+                got += rx.recv(300 - len(got))
+        except socket.timeout:
+            print("lost " + step)
+            sys.exit(1)
+    else:
+        subprocess.run(step, shell=True, check=True)
+PY
+
+# dial <client source> <server addr> <port> — a new connection from the server
+# namespace to the client, which listens on the port. 0 when it connects.
+cat > "$TMP/dial.py" << 'PY'
+import os, socket, sys
+
+srv_ns, cli_ns, src, port = sys.argv[1], sys.argv[2], sys.argv[3], int(sys.argv[4])
+fd = os.open("/run/netns/" + cli_ns, os.O_RDONLY)
+os.setns(fd, os.CLONE_NEWNET)
+listener = socket.socket()
+listener.bind(("10.99.0.3", port))
+listener.listen(1)
+fd = os.open("/run/netns/" + srv_ns, os.O_RDONLY)
+os.setns(fd, os.CLONE_NEWNET)
+dialer = socket.socket()
+dialer.bind((src, 0))
+dialer.settimeout(2)
+try:
+    dialer.connect(("10.99.0.3", port))
+except OSError:
+    sys.exit(1)
+PY
+
+# A new pair of namespaces, and no peer file: every case starts disarmed.
+fresh() {
+    $SUDO rm -f "$TMP/fw/peers.nft"
+    $SUDO ip netns del "$S" 2> /dev/null || true
+    $SUDO ip netns del "$C" 2> /dev/null || true
+    $SUDO ip netns add "$S"
+    $SUDO ip netns add "$C"
+    $SUDO ip link add vs netns "$S" type veth peer name vc netns "$C"
+    $SUDO ip -n "$S" addr add 10.99.0.1/24 dev vs
+    $SUDO ip -n "$S" addr add 10.99.0.9/24 dev vs
+    $SUDO ip -n "$C" addr add 10.99.0.3/24 dev vc
+    for l in "$S vs" "$C vc"; do
+        # shellcheck disable=SC2086  # "<netns> <link>"
+        set -- $l
+        $SUDO ip -n "$1" link set lo up
+        $SUDO ip -n "$1" link set "$2" up
+    done
+}
+
+# Helper verbs and raw nft, run as root inside the client namespace.
+in_c="$SUDO env PATH=$TMP/bin:$PATH:/usr/sbin:/sbin ip netns exec $C"
+verb() { echo "$in_c sh $TMP/apply $*"; }
+arm="$(verb set-peers) < $TMP/peers"
+printf '10.99.0.1\n10.99.0.1\n' > "$TMP/peers"
+
+# flow <desc> <steps> — passes when every transfer is delivered.
+flow() {
+    if out=$($SUDO python3 "$TMP/flow.py" "$S" "$C" "$2" 2>&1); then pass "$1"; else fail "$1: $out"; fi
+}
+
+# Control first: with nothing tracking, arming under an open flow must drop the
+# server's next segment, or this kernel cannot show the failure and the cases
+# below prove nothing.
+fresh
+if $SUDO python3 "$TMP/flow.py" "$S" "$C" "connect@@s2c@@c2s@@$arm@@s2c" > /dev/null 2>&1; then
+    fail "control: a flow opened with no conntrack user survived the arm, so this kernel cannot reproduce the drop"
+else
+    pass "control: a flow opened with no conntrack user loses the server's next segment once armed"
+fi
+
+fresh
+flow "a flow opened after disable survives set-peers, server speaking first" \
+    "$(verb disable)@@connect@@s2c@@c2s@@$arm@@s2c@@c2s"
+
+fresh
+flow "a flow opened after boot with no peer file survives set-peers, server speaking first" \
+    "$(verb boot)@@connect@@s2c@@c2s@@$arm@@s2c@@c2s"
+
+fresh
+flow "a flow opened while armed survives disable and a re-arm, server speaking first" \
+    "$arm@@connect@@s2c@@c2s@@$(verb disable)@@s2c@@c2s@@s2c@@$arm@@s2c@@c2s"
+
+fresh
+flow "a flow opened before track survives set-peers once it has carried traffic" \
+    "connect@@s2c@@c2s@@$(verb track)@@s2c@@c2s@@$arm@@s2c@@c2s"
+
+# The tracking table must not open anything the policy drops.
+fresh
+sh -c "$(verb boot) && $arm"
+if $SUDO python3 "$TMP/dial.py" "$S" "$C" 10.99.0.1 6644; then
+    pass "armed: a peer still reaches a cluster port"
+else
+    fail "armed: a peer cannot reach a cluster port"
+fi
+if $SUDO python3 "$TMP/dial.py" "$S" "$C" 10.99.0.9 6644; then
+    fail "armed: a non-peer reaches a cluster port"
+else
+    pass "armed: a non-peer is dropped on a cluster port"
+fi
+if $SUDO python3 "$TMP/dial.py" "$S" "$C" 10.99.0.1 7777; then
+    fail "armed: a peer reaches a port the policy does not open"
+else
+    pass "armed: a peer is dropped on a port the policy does not open"
+fi
+
+# disable leaves only the tracking table, which accepts everything.
+sh -c "$(verb disable)"
+tables=$($in_c nft list tables | tr '\n' ' ')
+case "$tables" in
+    *spinifex_filter*) fail "disable left inet spinifex_filter loaded: $tables" ;;
+    *spinifex_conntrack*) pass "disable removes the policy and keeps the tracking table" ;;
+    *) fail "disable removed the tracking table: '$tables'" ;;
+esac
+if $SUDO python3 "$TMP/dial.py" "$S" "$C" 10.99.0.9 7777; then
+    pass "disarmed: the tracking table drops nothing"
+else
+    fail "disarmed: the tracking table dropped a connection"
+fi
+
+if [ "$fails" -gt 0 ]; then
+    echo "  $fails check(s) failed"
+    exit 1
+fi
+echo "  setup_firewall_conntrack ok"

scripts/setup_host_test.sh

@@ -20,9 +20,10 @@ fail() { echo "  FAIL - $1"; fails=$((fails + 1)); }
 echo "-- setup.sh host DNS, chrony, needrestart and nbdkit"
 
 mkdir -p "$TMP/sys"
-for tool in bash awk sed grep readlink basename dirname cp ln mkdir cat head tail wc tee rm chmod install; do
+for tool in bash awk sed grep readlink basename dirname cp ln mkdir cat head tail wc tee rm chmod install mv; do
     ln -s "$(command -v "$tool")" "$TMP/sys/$tool"
 done
+ln -s "$(command -v sleep)" "$TMP/sys/realsleep"
 
 # stub <name> <body> — a stub command; $S is the fixture's state directory.
 stub() {
@@ -30,10 +31,17 @@ stub() {
     chmod +x "$TMP/bin/$1"
 }
 
+# nm_settle — waits for the fixture's NetworkManager to finish writing.
+nm_settle() {
+    n=0
+    while [ -e "$TMP/state/nm-pending" ] && [ $n -lt 500 ]; do "$TMP/sys/realsleep" 0.01; n=$((n + 1)); done
+}
+
 # host — a fresh fixture: resolved installed and inactive, resolvconf the
 # resolvectl link, NetworkManager running with no DNS settings, a dnf host
 # whose first enabled repository is yum.oracle.com.
 host() {
+    nm_settle
     rm -rf "${TMP:?}/state" "${TMP:?}/root" "${TMP:?}/bin"
     mkdir -p "$TMP/state/active" "$TMP/root/etc" "$TMP/root/usr/lib/systemd/system" "$TMP/bin"
     : > "$TMP/root/usr/lib/systemd/system/systemd-resolved.service"
@@ -42,11 +50,28 @@ host() {
     printf '[main]\n# dns=default\nconfigure-and-quit=no\n' > "$TMP/state/nm-config"
     printf 'Global:\nLink 2 (br-wan): 192.168.1.1\n' > "$TMP/state/resolvectl-dns"
     printf '[ol10_UEKR8]\nbaseurl = https://mirror.example/uek/\nenabled = 0\n[ol10_baseos_latest]\nbaseurl = https://yum.oracle.com/repo/OracleLinux/OL10/baseos/latest/x86_64/\nenabled = 1\n' > "$TMP/state/dnf-dump"
+    # A (re)started NetworkManager whose rc-manager is not unmanaged writes
+    # resolv.conf as a plain file on its first DNS update, which can land just
+    # after the installer links the stub.
     stub systemctl '[ "$1" = is-active ] && { [ -e "$S/active/$3" ]; exit; }
 echo "$*" >> "$S/calls"
 [ "$1 $2" = "enable --now" ] && : > "$S/active/$3"
+case "$1 $2" in
+    "restart NetworkManager.service" | "reload NetworkManager.service")
+        NetworkManager --print-config | grep -qx rc-manager=unmanaged && exit 0
+        : > "$S/nm-pending"
+        (
+            n=0
+            while [ ! -L "$S/../root/etc/resolv.conf" ] && [ $n -lt 300 ]; do realsleep 0.01; n=$((n + 1)); done
+            printf "nameserver 192.168.1.1\n" > "$S/nm-resolv.conf"
+            mv -f "$S/nm-resolv.conf" "$S/../root/etc/resolv.conf"
+            rm -f "$S/nm-pending"
+        ) > /dev/null 2>&1 &
+        ;;
+esac
 exit 0'
-    stub NetworkManager 'cat "$S/nm-config"'
+    stub NetworkManager 'cat "$S/nm-config"; cat "$S/../root/nm/90-spinifex-dns.conf" "$S/nm-config-late" 2>/dev/null; :'
+    stub nm-online ':'
     stub resolvectl '[ "$1" = dns ] && cat "$S/resolvectl-dns"'
     stub resolvectl-real ':'
     ln -s resolvectl "$TMP/bin/resolvconf"
@@ -60,10 +85,17 @@ run() {
     RC=0
     OUT=$(PATH="$TMP/bin:$TMP/sys" DISTRO_ROOT="$TMP/root" \
         SPINIFEX_RESOLV_CONF="$TMP/root/etc/resolv.conf" SPINIFEX_RESOLV_CONF_SAVED="$TMP/root/saved/resolv.conf.orig" \
+        SPINIFEX_NM_DNS_DROPIN="$TMP/root/nm/90-spinifex-dns.conf" \
         INSTALL_SPINIFEX_LIB_ONLY=1 bash -c '. "$1"; shift; for f in "$@"; do $f; done' _ "$SETUP" "$@" 2>&1) || RC=$?
 }
 
 calls() { cat "$TMP/state/calls" 2>/dev/null || true; }
+# linked — whether resolv.conf is still the stub link once NetworkManager settles.
+linked() {
+    nm_settle
+    [ "$(readlink "$TMP/root/etc/resolv.conf")" = /run/systemd/resolve/stub-resolv.conf ]
+}
+dropin() { cat "$TMP/root/nm/90-spinifex-dns.conf" 2>/dev/null || echo "<none>"; }
 saved() { cat "$TMP/root/saved/resolv.conf.orig" 2>/dev/null || echo "<none>"; }
 expect_untouched() {
     if [ "$RC" = 0 ] && [ -z "$(calls)" ] && [ ! -L "$TMP/root/etc/resolv.conf" ] && [ "$(saved)" = "<none>" ]; then
@@ -86,7 +118,7 @@ expect_fatal() { # expect_fatal <desc> <text>...
 # --------------------------------------------------------------------------
 host
 run hand_dns_to_resolved
-if [ "$RC" = 0 ] && [ "$(readlink "$TMP/root/etc/resolv.conf")" = /run/systemd/resolve/stub-resolv.conf ] \
+if [ "$RC" = 0 ] && linked \
     && [ "$(saved)" = "$(printf 'nameserver 192.168.1.1\nsearch example.lan')" ] \
     && [ "$(calls)" = "$(printf 'enable --now systemd-resolved.service\nrestart NetworkManager.service')" ] \
     && [ "$(cat "$TMP/state/looked-up")" = yum.oracle.com ]; then
@@ -98,12 +130,44 @@ fi
 : > "$TMP/state/calls"
 rm -f "$TMP/state/looked-up"
 run hand_dns_to_resolved
-if [ "$RC" = 0 ] && [ -z "$(calls)" ] && [ "$(cat "$TMP/state/looked-up")" = yum.oracle.com ]; then
+if [ "$RC" = 0 ] && [ -z "$(calls)" ] && linked && [ "$(cat "$TMP/state/looked-up")" = yum.oracle.com ]; then
     pass "a rerun after a handover checks DNS again and changes nothing"
 else
     fail "rerun after handover: rc=$RC calls='$(calls)': $OUT"
 fi
 
+if [ "$(dropin)" = "$(printf '[main]\ndns=systemd-resolved\nrc-manager=unmanaged')" ]; then
+    pass "the handover tells NetworkManager to send DNS to resolved and leave resolv.conf alone"
+else
+    fail "NetworkManager drop-in: '$(dropin)'"
+fi
+
+# An install made before the drop-in, whose NetworkManager replaced the link.
+host
+mkdir -p "$TMP/root/saved"
+echo "the original" > "$TMP/root/saved/resolv.conf.orig"
+: > "$TMP/state/active/systemd-resolved.service"
+run hand_dns_to_resolved
+if [ "$RC" = 0 ] && linked && [ "$(calls)" = "reload NetworkManager.service" ] \
+    && [ "$(dropin)" = "$(printf '[main]\ndns=systemd-resolved\nrc-manager=unmanaged')" ]; then
+    pass "a rerun after a handover whose link NetworkManager replaced adds the drop-in, reloads NM and relinks the stub"
+else
+    fail "rerun repair: rc=$RC calls='$(calls)' link='$(readlink "$TMP/root/etc/resolv.conf")' dropin='$(dropin)': $OUT"
+fi
+
+host
+printf '[main]\nrc-manager=symlink\n' > "$TMP/state/nm-config-late"
+run hand_dns_to_resolved
+expect_fatal "a NetworkManager file read after the drop-in that sets rc-manager stops the install, naming it" \
+    "rc-manager=symlink" "rm -f $TMP/root/nm/90-spinifex-dns.conf"
+case "$(calls)" in *"restart NetworkManager"*) fail "rc-manager override: restarted NetworkManager first: $(calls)" ;; esac
+
+host
+stub ln ':'
+run hand_dns_to_resolved
+expect_fatal "a resolv.conf that will not stay linked to the stub stops the install" \
+    "is not a link to /run/systemd/resolve/stub-resolv.conf"
+
 host
 mkdir -p "$TMP/root/saved"
 echo "the original" > "$TMP/root/saved/resolv.conf.orig"
@@ -155,7 +219,8 @@ host
 printf 'Global:\nLink 2 (br-wan):\n' > "$TMP/state/resolvectl-dns"
 run hand_dns_to_resolved
 expect_fatal "no upstream servers stops the install, naming the check and printing the restore" \
-    "'resolvectl dns' lists no DNS servers" "systemctl disable --now systemd-resolved" "cp -a --remove-destination $TMP/root/saved/resolv.conf.orig"
+    "'resolvectl dns' lists no DNS servers" "systemctl disable --now systemd-resolved" "cp -a --remove-destination $TMP/root/saved/resolv.conf.orig" \
+    "rm -f $TMP/root/nm/90-spinifex-dns.conf"
 
 host
 : > "$TMP/state/getent-fail"

scripts/uninstall-spx.sh

@@ -293,18 +293,20 @@ fi
 # file behind it is unreachable after the next reboot and unexplainable to
 # whoever inherits it.
 #
-# Only our own table and our own marked rules, never a global flush: setup.sh
-# creates exactly `table inet spinifex_filter`, and vpcd writes iptables rules
-# carrying the comments below. Anything else on this host belongs to someone
-# else.
+# Only our own tables and our own marked rules, never a global flush: setup.sh
+# creates exactly `table inet spinifex_filter` and `table inet
+# spinifex_conntrack`, and vpcd writes iptables rules carrying the comments
+# below. Anything else on this host belongs to someone else.
 # ---------------------------------------------------------------------------
 log "unloading the host firewall policy"
 if command -v nft >/dev/null 2>&1; then
-    if sudo nft list table inet spinifex_filter >/dev/null 2>&1; then
-        run sudo nft delete table inet spinifex_filter
-    else
-        log "  no inet spinifex_filter table loaded"
-    fi
+    for nft_table in spinifex_filter spinifex_conntrack; do
+        if sudo nft list table inet "$nft_table" >/dev/null 2>&1; then
+            run sudo nft delete table inet "$nft_table"
+        else
+            log "  no inet $nft_table table loaded"
+        fi
+    done
 fi
 if command -v iptables-save >/dev/null 2>&1 && ! $DRY_RUN; then
     for marker in spinifex-nat-egress spinifex-eip-ingress spinifex-imds spinifex-imds-remap; do
@@ -641,9 +643,11 @@ if ! $DRY_RUN; then
     for svc in nats gw daemon storage northstar viperblock vpcd ui; do
         id -u "spinifex-${svc}" >/dev/null 2>&1 && LEFT+=("user spinifex-${svc}")
     done
-    if command -v nft >/dev/null 2>&1 && sudo nft list table inet spinifex_filter >/dev/null 2>&1; then
-        LEFT+=("nft table inet spinifex_filter")
-    fi
+    for nft_table in spinifex_filter spinifex_conntrack; do
+        if command -v nft >/dev/null 2>&1 && sudo nft list table inet "$nft_table" >/dev/null 2>&1; then
+            LEFT+=("nft table inet $nft_table")
+        fi
+    done
     $PURGE_DATA && [ -e "$DATA_DIR" ] && ! mountpoint -q "$DATA_DIR" 2>/dev/null && LEFT+=("$DATA_DIR")
 
     if [ "${#LEFT[@]}" -gt 0 ]; then

scripts/uninstall-spx_test.sh

@@ -158,9 +158,10 @@ fi
 
 # The nft teardown must name our own table. A global flush would take the
 # host's other rules with it.
-grep -q 'nft delete table inet spinifex_filter' "$UNINSTALL" \
-    && pass "nft teardown deletes only inet spinifex_filter" \
-    || fail "nft teardown does not delete inet spinifex_filter by name"
+grep -q 'nft delete table inet "$nft_table"' "$UNINSTALL" \
+    && grep -q 'for nft_table in spinifex_filter spinifex_conntrack; do' "$UNINSTALL" \
+    && pass "nft teardown deletes only inet spinifex_filter and spinifex_conntrack" \
+    || fail "nft teardown does not delete inet spinifex_filter and spinifex_conntrack by name"
 grep -qE 'nft[[:space:]]+(-[a-z][[:space:]]+)*flush ruleset' "$UNINSTALL" \
     && fail "nft teardown flushes the whole ruleset" \
     || pass "nft teardown never flushes the whole ruleset"

spinifex/agents/eks/gatewayfetch/fetch.go

@@ -16,6 +16,9 @@
 //
 // The recovery resource returns a single epoch\taction\tsnapshot line the on-VM
 // k3s-recovery agent applies before k3s starts (etcd cluster-reset / wipe-rejoin).
+//
+// The workers resource returns one instance ID per line: every worker the
+// cluster's nodegroups still own, which the on-VM node-gc agent keeps Nodes for.
 package gatewayfetch
 
 import (
@@ -46,7 +49,7 @@ type Config struct {
 	Region      string
 	AccountID   string
 	ClusterName string
-	// Resource is "addons" or "recovery".
+	// Resource is "addons", "recovery" or "workers".
 	Resource   string
 	InstanceID string
 	// Out receives the rendered TSV.
@@ -61,8 +64,8 @@ func Run(ctx context.Context, cfg Config) error {
 		return errors.New("--account-id is required (or set EKS_ACCOUNT_ID)")
 	case cfg.ClusterName == "":
 		return errors.New("--cluster is required (or set EKS_CLUSTER_NAME)")
-	case cfg.Resource != "addons" && cfg.Resource != "recovery":
-		return errors.New("--resource must be addons or recovery")
+	case cfg.Resource != "addons" && cfg.Resource != "recovery" && cfg.Resource != "workers":
+		return errors.New("--resource must be addons, recovery or workers")
 	case cfg.Resource == "recovery" && cfg.InstanceID == "":
 		return errors.New("--instance-id is required for --resource recovery (or set EKS_INSTANCE_ID)")
 	}
@@ -80,6 +83,9 @@ func Run(ctx context.Context, cfg Config) error {
 	case "recovery":
 		path = "/clusters/" + cfg.ClusterName + "/internal-recovery/" + cfg.AccountID + "/" + cfg.InstanceID
 		emit = emitRecoveryTSV
+	case "workers":
+		path = "/clusters/" + cfg.ClusterName + "/internal-workers/" + cfg.AccountID
+		emit = emitWorkers
 	default:
 		path = "/clusters/" + cfg.ClusterName + "/internal-addons/" + cfg.AccountID
 		emit = emitAddonsTSV
@@ -163,3 +169,24 @@ func emitRecoveryTSV(out io.Writer, body []byte) error {
 	}
 	return nil
 }
+
+type internalWorkersResponse struct {
+	InstanceIDs []string `json:"instanceIds"`
+}
+
+// emitWorkers writes one managed-worker instance ID per line. An empty set
+// writes nothing, which the node-gc agent tells apart from a failed fetch by
+// this command's exit status, never by the output being empty.
+func emitWorkers(out io.Writer, body []byte) error {
+	var resp internalWorkersResponse
+	if err := json.Unmarshal(body, &resp); err != nil {
+		return fmt.Errorf("unmarshal internal-workers response: %w", err)
+	}
+	w := bufio.NewWriter(out)
+	for _, id := range resp.InstanceIDs {
+		if _, err := fmt.Fprintln(w, id); err != nil {
+			return err
+		}
+	}
+	return w.Flush()
+}

spinifex/agents/eks/gatewayfetch/fetch_test.go

@@ -49,7 +49,7 @@ func TestRun_RejectsBeforeContactingGateway(t *testing.T) {
 	}{
 		{"account", func(c *Config) { c.AccountID = "" }, "--account-id is required (or set EKS_ACCOUNT_ID)"},
 		{"cluster", func(c *Config) { c.ClusterName = "" }, "--cluster is required (or set EKS_CLUSTER_NAME)"},
-		{"resource", func(c *Config) { c.Resource = "secrets" }, "--resource must be addons or recovery"},
+		{"resource", func(c *Config) { c.Resource = "secrets" }, "--resource must be addons, recovery or workers"},
 		{"recovery instance", func(c *Config) { c.Resource = "recovery" }, "--instance-id is required for --resource recovery (or set EKS_INSTANCE_ID)"},
 		{"gateway client", func(c *Config) { c.GatewayURL = "" }, "build gateway client: eksgw: baseURL is required"},
 	}
@@ -92,6 +92,26 @@ func TestRun_RendersRecoveryDirective(t *testing.T) {
 	}
 }
 
+// The gateway's restjson marshaler names the field InstanceIDs; both that and
+// the json-tag spelling decode, so the agent reads either.
+func TestRun_RendersManagedWorkers(t *testing.T) {
+	for _, body := range []string{
+		`{"InstanceIDs":["i-0aaa","i-0bbb"]}`,
+		`{"instanceIds":["i-0aaa","i-0bbb"]}`,
+	} {
+		var hits atomic.Int32
+		cfg, out := gateway(t, "/clusters/demo/internal-workers/123456789012", http.StatusOK, body, &hits)
+		cfg.Resource = "workers"
+
+		if err := Run(context.Background(), cfg); err != nil {
+			t.Fatalf("Run: %v", err)
+		}
+		if got, want := out.String(), "i-0aaa\ni-0bbb\n"; got != want {
+			t.Fatalf("stdout = %q, want %q", got, want)
+		}
+	}
+}
+
 func TestRun_RenderFailureIsNotRetried(t *testing.T) {
 	var hits atomic.Int32
 	cfg, _ := gateway(t, "/clusters/demo/internal-addons/123456789012", http.StatusOK, `{not-json`, &hits)

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

@@ -34,7 +34,8 @@ func ValidateRebootInstancesInput(input *ec2.RebootInstancesInput) error {
 }
 
 // RebootInstances sends reboot commands to specified instances via NATS.
-// Unlike stop+start, reboot keeps the instance running and sends a QMP system_reset.
+// Unlike stop+start, reboot keeps the instance running, on its node and volumes,
+// while the guest is shut down and started on a fresh QEMU.
 // Returns an empty response on success (AWS returns no state-change data).
 func RebootInstances(ctx context.Context, input *ec2.RebootInstancesInput, natsConn *nats.Conn, accountID string) (*ec2.RebootInstancesOutput, error) {
 	if err := ValidateRebootInstancesInput(input); err != nil {

spinifex/domains/ec2/awsapi/registration.go

@@ -7,14 +7,14 @@ import (
 )
 
 // NewRegistration returns the EC2 AWS Query control-plane registration. Errors
-// render in the EC2 <Response><Errors> body, which the gateway keys on the
-// service name.
+// render in the EC2 <Response><Errors> body.
 func NewRegistration(deps Deps) dispatch.Registration {
 	return dispatch.Registration{
 		Service:       ServiceName,
 		ResolveAction: resolveAction,
 		Dispatch:      NewHandler(deps).Dispatch,
 		Errors:        dispatch.ErrorEnvelopeXML,
+		Traits:        dispatch.ServiceTraits{XMLErrorShape: dispatch.XMLErrorShapeEC2},
 		Inventory:     OperationInventory(),
 	}
 }

spinifex/domains/ec2/vpc/eni.go

@@ -694,43 +694,39 @@ func eniMatchesFilters(record *ENIRecord, filters map[string][]string) bool {
 // AttachENI marks an ENI as attached to an instance (internal use by RunInstances).
 // accountID scopes the lookup to the correct KV key.
 func (s *VPCServiceImpl) AttachENI(accountID, eniId, instanceId string, deviceIndex int64) (string, error) {
-	return s.attachENI(context.Background(), accountID, eniId, instanceId, deviceIndex)
+	return s.attachENI(context.Background(), accountID, eniId, instanceId, deviceIndex, "")
 }
 
-func (s *VPCServiceImpl) attachENI(ctx context.Context, accountID, eniId, instanceId string, deviceIndex int64) (string, error) {
-	key := kvutil.AccountKey(accountID, eniId)
-	entry, err := s.eniKV.Get(ctx, key)
-	if err != nil {
-		return "", eniNotFoundError(eniId)
-	}
-
-	var record ENIRecord
-	if err := json.Unmarshal(entry.Value(), &record); err != nil {
-		return "", errors.New(awserrors.ErrorServerInternal)
-	}
-
-	if record.Status == "in-use" {
-		return "", errors.New(awserrors.ErrorInvalidNetworkInterfaceInUse)
-	}
+// AttachENIWithOwner attaches the ENI and records instanceOwnerID as the owner
+// of the attached instance in the same write, so no later write can lose the
+// stamp. An empty owner leaves the record's owner untouched.
+func (s *VPCServiceImpl) AttachENIWithOwner(accountID, eniId, instanceId string, deviceIndex int64, instanceOwnerID string) (string, error) {
+	return s.attachENI(context.Background(), accountID, eniId, instanceId, deviceIndex, instanceOwnerID)
+}
 
+func (s *VPCServiceImpl) attachENI(ctx context.Context, accountID, eniId, instanceId string, deviceIndex int64, instanceOwnerID string) (string, error) {
 	attachmentId := awsidentifiers.GenerateResourceID("eni-attach")
-	record.Status = "in-use"
-	record.AttachmentId = attachmentId
-	record.InstanceId = instanceId
-	record.DeviceIndex = deviceIndex
-	// Default DeleteOnTermination to true on first attach, matching what
-	// DescribeNetworkInterfaces already advertises; a record carrying an
-	// explicit value from an earlier attach keeps it across re-attach.
-	if record.DeleteOnTermination == nil {
-		record.DeleteOnTermination = aws.Bool(true)
-	}
-
-	data, err := json.Marshal(record)
+	err := s.casENI(ctx, kvutil.AccountKey(accountID, eniId), eniId, func(record *ENIRecord) error {
+		if record.Status == "in-use" {
+			return errors.New(awserrors.ErrorInvalidNetworkInterfaceInUse)
+		}
+		record.Status = "in-use"
+		record.AttachmentId = attachmentId
+		record.InstanceId = instanceId
+		record.DeviceIndex = deviceIndex
+		if instanceOwnerID != "" {
+			record.InstanceOwnerId = instanceOwnerID
+		}
+		// Default DeleteOnTermination to true on first attach, matching what
+		// DescribeNetworkInterfaces already advertises; a record carrying an
+		// explicit value from an earlier attach keeps it across re-attach.
+		if record.DeleteOnTermination == nil {
+			record.DeleteOnTermination = aws.Bool(true)
+		}
+		return nil
+	})
 	if err != nil {
-		return "", fmt.Errorf("failed to marshal ENI record: %w", err)
-	}
-	if _, err := s.eniKV.Update(ctx, key, data, entry.Revision()); err != nil {
-		return "", errors.New(awserrors.ErrorServerInternal)
+		return "", err
 	}
 
 	slog.InfoContext(ctx, "ENI attached", "eniId", eniId, "instanceId", instanceId, "attachmentId", attachmentId)
@@ -796,24 +792,69 @@ func (s *VPCServiceImpl) UpdateENI(accountID, eniId string, fn func(*ENIRecord))
 }
 
 func (s *VPCServiceImpl) updateENI(ctx context.Context, accountID, eniId string, fn func(*ENIRecord)) error {
-	key := kvutil.AccountKey(accountID, eniId)
-	entry, err := s.eniKV.Get(ctx, key)
-	if err != nil {
-		return eniNotFoundError(eniId)
+	return s.casENI(ctx, kvutil.AccountKey(accountID, eniId), eniId, func(r *ENIRecord) error {
+		fn(r)
+		return nil
+	})
+}
+
+const (
+	eniCASAttempts = 3
+	eniCASBackoff  = 10 * time.Millisecond
+)
+
+// storeError reports ServerInternal to API callers while keeping the store
+// failure reachable through errors.Is for the caller's log.
+type storeError struct{ cause error }
+
+func (e *storeError) Error() string { return awserrors.ErrorServerInternal }
+func (e *storeError) Unwrap() error { return e.cause }
+
+// casENI reads the ENI record, applies mutate and writes it back under the
+// revision it read. A direct get can be answered by a replica that has not yet
+// applied the caller's own earlier write, so a revision conflict re-reads and
+// re-applies mutate, bounded. mutate may run more than once and must only
+// depend on the record it is given.
+func (s *VPCServiceImpl) casENI(ctx context.Context, key, eniId string, mutate func(*ENIRecord) error) error {
+	var lastErr error
+	attempts := 0
+	for attempt := range eniCASAttempts {
+		attempts = attempt + 1
+		if attempt > 0 {
+			select {
+			case <-ctx.Done():
+				return &storeError{ctx.Err()}
+			case <-time.After(time.Duration(attempt) * eniCASBackoff):
+			}
+		}
+		entry, err := s.eniKV.Get(ctx, key)
+		if err != nil {
+			return eniNotFoundError(eniId)
+		}
+		var record ENIRecord
+		if err := json.Unmarshal(entry.Value(), &record); err != nil {
+			return errors.New(awserrors.ErrorServerInternal)
+		}
+		if err := mutate(&record); err != nil {
+			return err
+		}
+		data, err := json.Marshal(record)
+		if err != nil {
+			return fmt.Errorf("marshal ENI record: %w", err)
+		}
+		if _, err = s.eniKV.Update(ctx, key, data, entry.Revision()); err == nil {
+			if attempt > 0 {
+				slog.InfoContext(ctx, "ENI write succeeded after revision conflict", "eniId", eniId, "attempt", attempt+1)
+			}
+			return nil
+		}
+		lastErr = err
+		if !errors.Is(err, jetstream.ErrKeyExists) {
+			break
+		}
 	}
-	var record ENIRecord
-	if err := json.Unmarshal(entry.Value(), &record); err != nil {
-		return errors.New(awserrors.ErrorServerInternal)
-	}
-	fn(&record)
-	data, err := json.Marshal(record)
-	if err != nil {
-		return fmt.Errorf("marshal ENI record: %w", err)
-	}
-	if _, err := s.eniKV.Update(ctx, key, data, entry.Revision()); err != nil {
-		return errors.New(awserrors.ErrorServerInternal)
-	}
-	return nil
+	slog.ErrorContext(ctx, "ENI write failed", "eniId", eniId, "attempts", attempts, "err", lastErr)
+	return &storeError{lastErr}
 }
 
 // ListInstanceENIs returns every ENIRecord in accountID currently attached to

spinifex/domains/ec2/vpc/eni_kv_seam.go

@@ -0,0 +1,9 @@
+package vpc
+
+import "github.com/nats-io/nats.go/jetstream"
+
+// WrapENIKVForTest replaces the ENI bucket handle with wrap(current), so a test
+// can inject the read and write faults a replicated bucket produces.
+func (s *VPCServiceImpl) WrapENIKVForTest(wrap func(jetstream.KeyValue) jetstream.KeyValue) {
+	s.eniKV = wrap(s.eniKV)
+}

spinifex/domains/ec2/vpc/eni_stale_read_test.go

@@ -0,0 +1,125 @@
+package vpc
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"testing"
+
+	testutil "github.com/mulgadc/spinifex/internal/testkit"
+	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
+	"github.com/nats-io/nats.go/jetstream"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+// A replicated bucket can answer a direct get from a replica that has not yet
+// applied the caller's own previous write. The next compare-and-swap then
+// carries a revision that is already superseded.
+func newStaleReadENI(t *testing.T) (*VPCServiceImpl, string, func() ENIRecord) {
+	t.Helper()
+	svc := setupTestVPCService(t)
+	vpcID := createTestVPC(t, svc, "10.0.0.0/16")
+	subnetID := createTestSubnet(t, svc, vpcID, "10.0.1.0/24")
+	eniID := createTestENI(t, svc, subnetID)
+	var inner jetstream.KeyValue
+	svc.WrapENIKVForTest(func(kv jetstream.KeyValue) jetstream.KeyValue {
+		inner = kv
+		return testutil.NewStaleReadKV(kv)
+	})
+	// Reads the stored record directly: the wrapped bucket would serve the
+	// check itself from a lagging replica.
+	stored := func() ENIRecord {
+		entry, err := inner.Get(context.Background(), kvutil.AccountKey(testAccountID, eniID))
+		require.NoError(t, err)
+		var rec ENIRecord
+		require.NoError(t, json.Unmarshal(entry.Value(), &rec))
+		return rec
+	}
+	return svc, eniID, stored
+}
+
+func TestUpdateENIAfterOwnWriteSurvivesStaleReplicaRead(t *testing.T) {
+	t.Parallel()
+	svc, eniID, stored := newStaleReadENI(t)
+
+	_, err := svc.AttachENI(testAccountID, eniID, "i-test123", 0)
+	require.NoError(t, err)
+
+	err = svc.UpdateENI(testAccountID, eniID, func(r *ENIRecord) { r.InstanceOwnerId = "000000000000" })
+	require.NoError(t, err, "a read served from a lagging replica must not lose the update")
+
+	rec := stored()
+	assert.Equal(t, "000000000000", rec.InstanceOwnerId)
+	assert.Equal(t, "i-test123", rec.InstanceId, "the attach must survive the update")
+}
+
+func TestAttachENIAfterOwnWriteSurvivesStaleReplicaRead(t *testing.T) {
+	t.Parallel()
+	svc, eniID, stored := newStaleReadENI(t)
+
+	require.NoError(t, svc.UpdateENI(testAccountID, eniID, func(r *ENIRecord) { r.Description = "managed" }))
+
+	_, err := svc.AttachENI(testAccountID, eniID, "i-test123", 0)
+	require.NoError(t, err, "a read served from a lagging replica must not fail the attach")
+
+	rec := stored()
+	assert.Equal(t, "in-use", rec.Status)
+	assert.Equal(t, "managed", rec.Description)
+}
+
+func TestUpdateENIReportsTheStoreError(t *testing.T) {
+	t.Parallel()
+	svc := setupTestVPCService(t)
+	vpcID := createTestVPC(t, svc, "10.0.0.0/16")
+	subnetID := createTestSubnet(t, svc, vpcID, "10.0.1.0/24")
+	eniID := createTestENI(t, svc, subnetID)
+
+	cause := errors.New("stream unavailable")
+	svc.WrapENIKVForTest(func(kv jetstream.KeyValue) jetstream.KeyValue {
+		return testutil.NewRejectingKV(kv, func(string, []byte) error { return cause })
+	})
+
+	err := svc.UpdateENI(testAccountID, eniID, func(r *ENIRecord) { r.Description = "x" })
+	require.Error(t, err)
+	assert.ErrorIs(t, err, cause, "the caller logs this error, so it must name the store failure")
+}
+
+func TestAttachENIWithOwnerStampsTheOwnerInTheAttachWrite(t *testing.T) {
+	t.Parallel()
+	svc := setupTestVPCService(t)
+	vpcID := createTestVPC(t, svc, "10.0.0.0/16")
+	subnetID := createTestSubnet(t, svc, vpcID, "10.0.1.0/24")
+	eniID := createTestENI(t, svc, subnetID)
+
+	var writes []ENIRecord
+	svc.WrapENIKVForTest(func(kv jetstream.KeyValue) jetstream.KeyValue {
+		return testutil.NewRejectingKV(kv, func(_ string, value []byte) error {
+			var rec ENIRecord
+			require.NoError(t, json.Unmarshal(value, &rec))
+			writes = append(writes, rec)
+			return nil
+		})
+	})
+
+	_, err := svc.AttachENIWithOwner(testAccountID, eniID, "i-sys", 0, "000000000000")
+	require.NoError(t, err)
+
+	require.Len(t, writes, 1, "attach and owner stamp are one write")
+	assert.Equal(t, "in-use", writes[0].Status)
+	assert.Equal(t, "000000000000", writes[0].InstanceOwnerId)
+}
+
+func TestCASENIGivesUpAfterBoundedConflicts(t *testing.T) {
+	t.Parallel()
+	svc, eniID, _ := newStaleReadENI(t)
+	conflicts := 0
+	err := svc.casENI(t.Context(), kvutil.AccountKey(testAccountID, eniID), eniID, func(*ENIRecord) error {
+		conflicts++
+		// Move the record under the next compare-and-swap, as a competing writer does.
+		return svc.UpdateENI(testAccountID, eniID, func(r *ENIRecord) { r.Description = "other writer" })
+	})
+	require.Error(t, err)
+	assert.ErrorIs(t, err, jetstream.ErrKeyExists, "the conflict is reported, not hidden")
+	assert.Equal(t, eniCASAttempts, conflicts)
+}

spinifex/domains/eks/awsapi/authz.go

@@ -50,10 +50,11 @@ var eksScopes = map[string][]resourceSource{
 	"ListClusters":    {sourceAny},
 
 	// Internal control-plane routes. The cluster's owning account is a path
-	// segment on these two, and the ARN names that account rather than the
+	// segment on these three, and the ARN names that account rather than the
 	// system caller's, because that is the account the handler reads.
 	"ListInternalAddons":   {sourceInternalCluster},
 	"GetRecoveryDirective": {sourceInternalCluster},
+	"ListInternalWorkers":  {sourceInternalCluster},
 	// The owning account arrives in the body on these two, and the ARN names
 	// that account for the same reason: it is the account the handler acts in.
 	"PublishInternal":    {sourceInternalBodyCluster},

spinifex/domains/eks/awsapi/authz_test.go

@@ -179,6 +179,10 @@ func TestResourceARNsInternalRoutesUsePathAccount(t *testing.T) {
 	assert.Equal(t,
 		[]string{"arn:aws:eks:ap-southeast-2:999999999999:cluster/prod"},
 		resolveARNs(t, "GetRecoveryDirective", []string{"prod", "999999999999", "i-abc"}, ""))
+
+	assert.Equal(t,
+		[]string{"arn:aws:eks:ap-southeast-2:999999999999:cluster/prod"},
+		resolveARNs(t, "ListInternalWorkers", []string{"prod", "999999999999"}, ""))
 }
 
 // An identifier the gate cannot read authorizes account-wide, leaving the

spinifex/domains/eks/awsapi/gateway_wire_test.go

@@ -93,6 +93,46 @@ func TestEKSRequest_InternalAddonsWireBody(t *testing.T) {
 		rec.Body.String())
 }
 
+// The node-gc agent reads the same REST-JSON shape: the Go field name, sorted
+// IDs drawn from every nodegroup record of the cluster.
+func TestEKSRequest_InternalWorkersWireBody(t *testing.T) {
+	const cpInstanceID = "i-cp0000000000001"
+	_, nc, js := testutil.StartTestJetStream(t)
+	kv, err := ekscluster.GetOrCreateAccountBucket(t.Context(), js, authzAccountID)
+	require.NoError(t, err)
+	require.NoError(t, ekscluster.PutClusterMeta(t.Context(), kv, &ekscluster.ClusterMeta{
+		Name:              "prod",
+		Status:            ekscluster.ClusterStatusActive,
+		ControlPlaneNodes: []ekscluster.ControlPlaneNode{{InstanceID: cpInstanceID}},
+	}))
+	for _, rec := range []*ekscluster.NodegroupRecord{
+		{ClusterName: "prod", Name: "ng-b", InstanceIDs: []string{"i-0ccc", "i-0aaa"}},
+		{ClusterName: "prod", Name: "ng-a", InstanceIDs: []string{"i-0bbb"}},
+	} {
+		require.NoError(t, ekscluster.PutNodegroupRecord(t.Context(), kv, rec))
+	}
+
+	svc, err := ekscluster.NewEKSServiceImpl(ekscluster.EKSServiceDeps{NATSConn: nc, Region: authzRegion})
+	require.NoError(t, err)
+	sub, err := nc.Subscribe("eks.ListManagedWorkers", func(m *nats.Msg) {
+		acct := natsmsg.AccountIDFromMsg(m)
+		natsmsg.ServeNATSRequestCtx(m, func(ctx context.Context, in *ekscluster.ListManagedWorkersInput) (*ekscluster.ListManagedWorkersOutput, error) {
+			return svc.ListManagedWorkers(ctx, in, acct)
+		})
+	})
+	require.NoError(t, err)
+	t.Cleanup(func() { _ = sub.Unsubscribe() })
+
+	gw := scopedEKS(gwtest.Statement("Allow", "eks:*", "*"))
+	gw.nc = nc
+	rec := httptest.NewRecorder()
+	req := httptest.NewRequest(http.MethodGet, "/clusters/prod/internal-workers/"+authzAccountID, nil)
+	require.NoError(t, gw.as(cpAgentContext(cpInstanceID)).serve(rec, req))
+
+	assert.Equal(t, http.StatusOK, rec.Code)
+	assert.Equal(t, `{"InstanceIDs":["i-0aaa","i-0bbb","i-0ccc"]}`, rec.Body.String())
+}
+
 // CreateAddon and UpdateAddon hand their service-account role to iam:PassRole,
 // same as CreateCluster's roleArn and CreateNodegroup's nodeRole; an absent
 // role passes nothing.

spinifex/domains/eks/awsapi/internal_authz_test.go

@@ -97,6 +97,8 @@ func TestAuthorizeInternal_RejectsNonCPPrincipals(t *testing.T) {
 		t.Run(tc.name, func(t *testing.T) {
 			assertDenied(t, eksawsapi.AuthorizeInternal(t.Context(), nil, "ListInternalAddons",
 				tc.caller, []string{"alpha", tenantAccount}, nil))
+			assertDenied(t, eksawsapi.AuthorizeInternal(t.Context(), nil, "ListInternalWorkers",
+				tc.caller, []string{"alpha", tenantAccount}, nil))
 			assertDenied(t, eksawsapi.AuthorizeInternal(t.Context(), nil, "GetRecoveryDirective",
 				tc.caller, []string{"alpha", tenantAccount, cpInstanceID}, nil))
 			assertDenied(t, eksawsapi.AuthorizeInternal(t.Context(), nil, "PublishInternal",
@@ -124,6 +126,8 @@ func TestAuthorizeInternal_AllowsClusterMember(t *testing.T) {
 
 	require.NoError(t, eksawsapi.AuthorizeInternal(t.Context(), nc, "ListInternalAddons",
 		cpAgent(cpInstanceID), []string{"alpha", tenantAccount}, nil))
+	require.NoError(t, eksawsapi.AuthorizeInternal(t.Context(), nc, "ListInternalWorkers",
+		cpAgent(cpInstanceID), []string{"alpha", tenantAccount}, nil))
 	require.NoError(t, eksawsapi.AuthorizeInternal(t.Context(), nc, "GetRecoveryDirective",
 		cpAgent(cpInstanceID), []string{"alpha", tenantAccount, cpInstanceID}, nil))
 	require.NoError(t, eksawsapi.AuthorizeInternal(t.Context(), nc, "PublishInternal",

spinifex/domains/eks/awsapi/internal_workers.go

@@ -0,0 +1,35 @@
+package awsapi
+
+import (
+	"context"
+	"errors"
+
+	ekscluster "github.com/mulgadc/spinifex/spinifex/domains/eks/cluster"
+	"github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
+	"github.com/nats-io/nats.go"
+)
+
+// internalWorkersOutput is the body returned to the on-VM node-gc agent: the
+// instance ID of every worker the cluster's nodegroups still own.
+type internalWorkersOutput struct {
+	InstanceIDs []string `json:"instanceIds"`
+}
+
+// ListInternalWorkers — GET /clusters/{name}/internal-workers/{acct}.
+// Internal control-plane VM route (not an AWS-SDK action): same carve-out as
+// ListInternalAddons, gated by AuthorizeInternal, which admits only a CP agent
+// serving that account's cluster.
+func ListInternalWorkers(ctx context.Context, natsConn *nats.Conn, clusterName, accountID string) (*internalWorkersOutput, error) {
+	if natsConn == nil {
+		return nil, errors.New(awserrors.ErrorServerInternal)
+	}
+	if clusterName == "" || accountID == "" {
+		return nil, errors.New(awserrors.ErrorInvalidParameterValue)
+	}
+	out, err := ekscluster.NewNATSEKSService(natsConn).ListManagedWorkers(ctx,
+		&ekscluster.ListManagedWorkersInput{ClusterName: clusterName}, accountID)
+	if err != nil {
+		return nil, err
+	}
+	return &internalWorkersOutput{InstanceIDs: out.InstanceIDs}, nil
+}

spinifex/domains/eks/awsapi/routes.go

@@ -60,6 +60,14 @@ var routes = []route{
 			return GetRecoveryDirective(ctx, nc, p[0], p[1], p[2])
 		}},
 
+	// Internal control-plane route: the on-VM node-gc agent pulls the cluster's
+	// live nodegroup workers to delete the Nodes of terminated ones. Same
+	// system-cred carve-out as internal-addons.
+	{Method: "GET", Pattern: "/clusters/{clusterName}/internal-workers/{accountId}", Action: "ListInternalWorkers",
+		Handler: func(ctx context.Context, nc *nats.Conn, acct, callerARN string, p []string, b []byte) (any, error) {
+			return ListInternalWorkers(ctx, nc, p[0], p[1])
+		}},
+
 	// Nodegroup
 	{Method: "POST", Pattern: "/clusters/{clusterName}/node-groups", Action: "CreateNodegroup",
 		Handler: func(ctx context.Context, nc *nats.Conn, acct, callerARN string, p []string, b []byte) (any, error) {

spinifex/domains/eks/awsapi/routes_test.go

@@ -108,6 +108,7 @@ func TestLookupEKSAction_CoversAllActions(t *testing.T) {
 		"WebhookTokenReview":           false,
 		"ListInternalAddons":           false,
 		"GetRecoveryDirective":         false,
+		"ListInternalWorkers":          false,
 		"DescribeCluster":              false,
 		"ListClusters":                 false,
 		"DeleteCluster":                false,

spinifex/domains/eks/cluster/k3s_server_role.go

@@ -22,9 +22,10 @@ const (
 	// WebhookTokenReview (the eks-token-webhook relays `aws eks get-token` bearer
 	// tokens to the token-review broker for host-side STS verification), and
 	// GetRecoveryDirective (the k3s-recovery agent pulls its per-member etcd
-	// recovery directive at boot). The gateway evaluates these per request against
-	// the role's policies.
-	eksServerInlinePolicy = `{"Version":"2012-10-17","Statement":[{"Effect":"Allow","Action":["eks:PublishInternal","eks:ListInternalAddons","eks:WebhookTokenReview","eks:GetRecoveryDirective"],"Resource":"*"}]}`
+	// recovery directive at boot), and ListInternalWorkers (the node-gc agent
+	// pulls the live nodegroup workers). The gateway evaluates these per request
+	// against the role's policies.
+	eksServerInlinePolicy = `{"Version":"2012-10-17","Statement":[{"Effect":"Allow","Action":["eks:PublishInternal","eks:ListInternalAddons","eks:WebhookTokenReview","eks:GetRecoveryDirective","eks:ListInternalWorkers"],"Resource":"*"}]}`
 )
 
 // ensureCPInstanceProfile returns the CP instance-profile ARN, or "" to signal
@@ -35,7 +36,7 @@ func (s *EKSServiceImpl) ensureCPInstanceProfile(accountID string) string {
 	iam := s.iamEnsurer()
 	if iam == nil {
 		slog.Warn("EKS: IAM service unwired; CP VM falls back to baked static gateway creds and "+
-			"will be denied eks:PublishInternal, eks:WebhookTokenReview, eks:ListInternalAddons and eks:GetRecoveryDirective (no instance identity to bind)",
+			"will be denied eks:PublishInternal, eks:WebhookTokenReview, eks:ListInternalAddons, eks:GetRecoveryDirective and eks:ListInternalWorkers (no instance identity to bind)",
 			"accountID", accountID)
 		return ""
 	}
@@ -43,7 +44,7 @@ func (s *EKSServiceImpl) ensureCPInstanceProfile(accountID string) string {
 		CPInstanceRoleName, eksServerInlinePolicyName, eksServerInlinePolicy)
 	if err != nil {
 		slog.Error("EKS: ensure CP instance profile failed; CP VM falls back to baked static gateway creds and "+
-			"will be denied eks:PublishInternal, eks:WebhookTokenReview, eks:ListInternalAddons and eks:GetRecoveryDirective (no instance identity to bind)",
+			"will be denied eks:PublishInternal, eks:WebhookTokenReview, eks:ListInternalAddons, eks:GetRecoveryDirective and eks:ListInternalWorkers (no instance identity to bind)",
 			"accountID", accountID, "role", CPInstanceRoleName, "err", err)
 		return ""
 	}

spinifex/domains/eks/cluster/k3s_server_role_test.go

@@ -38,6 +38,9 @@ func TestEnsureK3sServerInstanceProfile_CreatesRolePolicyProfile(t *testing.T) {
 	// these IMDS creds; without this action the gateway 403s the boot-time fetch
 	// and a wedged control plane can never be auto-reset.
 	assert.Contains(t, policy, "eks:GetRecoveryDirective")
+	// The node-gc agent pulls the live nodegroup workers with these IMDS creds;
+	// without this action a terminated worker's Node is never deleted.
+	assert.Contains(t, policy, "eks:ListInternalWorkers")
 
 	require.NotNil(t, f.Profiles[CPInstanceRoleName])
 	assert.Len(t, f.Profiles[CPInstanceRoleName].Roles, 1)

spinifex/domains/eks/cluster/service.go

@@ -44,6 +44,10 @@ type EKSService interface {
 	// AWS-SDK action): the on-VM addon-sync agent fetches staged manifests for a
 	// cluster via the internal-addons gateway route.
 	ListStagedAddonManifests(ctx context.Context, input *ListStagedAddonManifestsInput, accountID string) (*ListStagedAddonManifestsOutput, error)
+	// ListManagedWorkers is an internal control-plane method (not an AWS-SDK
+	// action): the on-VM node-gc agent fetches the cluster's live nodegroup
+	// workers via the internal-workers gateway route.
+	ListManagedWorkers(ctx context.Context, input *ListManagedWorkersInput, accountID string) (*ListManagedWorkersOutput, error)
 
 	// GetRecoveryDirective / SetRecoveryDirective are internal control-plane
 	// methods (not AWS-SDK actions): the on-VM k3s-recovery agent fetches its

spinifex/domains/eks/cluster/service_nats.go

@@ -129,6 +129,10 @@ func (s *NATSEKSService) ListStagedAddonManifests(ctx context.Context, input *Li
 	return natsmsg.NATSRequest[ListStagedAddonManifestsOutput](ctx, s.natsConn, "eks.ListStagedAddonManifests", input, defaultTimeout, accountID)
 }
 
+func (s *NATSEKSService) ListManagedWorkers(ctx context.Context, input *ListManagedWorkersInput, accountID string) (*ListManagedWorkersOutput, error) {
+	return natsmsg.NATSRequest[ListManagedWorkersOutput](ctx, s.natsConn, "eks.ListManagedWorkers", input, defaultTimeout, accountID)
+}
+
 func (s *NATSEKSService) UpdateAddon(ctx context.Context, input *eks.UpdateAddonInput, accountID string) (*eks.UpdateAddonOutput, error) {
 	return natsmsg.NATSRequest[eks.UpdateAddonOutput](ctx, s.natsConn, "eks.UpdateAddon", input, defaultTimeout, accountID)
 }

spinifex/domains/eks/cluster/worker_membership.go

@@ -0,0 +1,46 @@
+package cluster
+
+import (
+	"context"
+	"errors"
+	"slices"
+
+	"github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
+)
+
+// ListManagedWorkersInput names the cluster whose managed-nodegroup workers to
+// list. It is an internal control-plane request (not an AWS-SDK shape), served
+// over NATS for the guest node-gc agent via the internal-workers route.
+type ListManagedWorkersInput struct {
+	ClusterName string `json:"clusterName"`
+}
+
+// ListManagedWorkersOutput carries the instance ID of every worker a managed
+// nodegroup of the cluster still owns, sorted.
+type ListManagedWorkersOutput struct {
+	InstanceIDs []string `json:"instanceIds"`
+}
+
+// ListManagedWorkers returns every worker instance ID recorded on the cluster's
+// nodegroups. A nodegroup drops an ID only after terminating it (DeleteNodegroup,
+// scale-down, reclaim), so the guest deletes a nodegroup Node whose instance is
+// absent here, as the AWS cloud controller deletes a Node whose instance is gone.
+func (s *EKSServiceImpl) ListManagedWorkers(ctx context.Context, input *ListManagedWorkersInput, accountID string) (*ListManagedWorkersOutput, error) {
+	if input == nil || input.ClusterName == "" {
+		return nil, errors.New(awserrors.ErrorInvalidParameterValue)
+	}
+	acctKV, err := s.acctKVForCluster(ctx, accountID, input.ClusterName)
+	if err != nil {
+		return nil, err
+	}
+	recs, err := ListNodegroupRecords(ctx, acctKV, input.ClusterName)
+	if err != nil {
+		return nil, err
+	}
+	ids := make([]string, 0)
+	for _, rec := range recs {
+		ids = append(ids, rec.InstanceIDs...)
+	}
+	slices.Sort(ids)
+	return &ListManagedWorkersOutput{InstanceIDs: slices.Compact(ids)}, nil
+}

spinifex/domains/eks/cluster/worker_membership_test.go

@@ -0,0 +1,74 @@
+package cluster
+
+import (
+	"context"
+	"testing"
+
+	"github.com/aws/aws-sdk-go/aws"
+	"github.com/aws/aws-sdk-go/service/eks"
+	"github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func managedWorkers(t *testing.T, f *eksServiceFixture, cluster string) []string {
+	t.Helper()
+	out, err := f.svc.ListManagedWorkers(context.Background(), &ListManagedWorkersInput{ClusterName: cluster}, testAccountID)
+	require.NoError(t, err)
+	return out.InstanceIDs
+}
+
+// A worker leaves the managed set exactly when its nodegroup terminates it, by
+// scale-down or DeleteNodegroup, so the guest node-gc deletes its Node then and
+// leaves every worker still owned by a nodegroup alone.
+func TestListManagedWorkers_DropsWorkersTheNodegroupTerminated(t *testing.T) {
+	t.Parallel()
+	f := newEKSServiceFixture(t)
+	seedActiveClusterWithToken(t, f, "c1")
+	for _, ng := range []string{"ng1", "ng2"} {
+		_, err := f.svc.CreateNodegroup(context.Background(), createNGInput("c1", ng, 2), testAccountID)
+		require.NoError(t, err)
+		markWorkersReady(t, f, "c1", ng, 2)
+		f.svc.WaitLaunches()
+	}
+	ng1, err := GetNodegroupRecord(t.Context(), f.kv, "c1", "ng1")
+	require.NoError(t, err)
+	ng2, err := GetNodegroupRecord(t.Context(), f.kv, "c1", "ng2")
+	require.NoError(t, err)
+	require.Len(t, ng1.InstanceIDs, 2)
+	assert.ElementsMatch(t, append(append([]string{}, ng1.InstanceIDs...), ng2.InstanceIDs...), managedWorkers(t, f, "c1"))
+
+	_, err = f.svc.UpdateNodegroupConfig(context.Background(), &eks.UpdateNodegroupConfigInput{
+		ClusterName:   aws.String("c1"),
+		NodegroupName: aws.String("ng1"),
+		ScalingConfig: &eks.NodegroupScalingConfig{DesiredSize: aws.Int64(1)},
+	}, testAccountID)
+	require.NoError(t, err)
+	require.Len(t, f.worker.terminateCalls, 1)
+	scaledIn := f.worker.terminateCalls[0]
+	got := managedWorkers(t, f, "c1")
+	assert.Len(t, got, 3)
+	for _, id := range scaledIn {
+		assert.NotContains(t, got, id, "a worker terminated by scale-down must leave the managed set")
+	}
+
+	_, err = f.svc.DeleteNodegroup(context.Background(), &eks.DeleteNodegroupInput{
+		ClusterName: aws.String("c1"), NodegroupName: aws.String("ng2"),
+	}, testAccountID)
+	require.NoError(t, err)
+	got = managedWorkers(t, f, "c1")
+	assert.Len(t, got, 1)
+	for _, id := range ng2.InstanceIDs {
+		assert.NotContains(t, got, id, "a deleted nodegroup's workers must leave the managed set")
+	}
+}
+
+func TestListManagedWorkers_UnknownClusterIsNotFound(t *testing.T) {
+	t.Parallel()
+	f := newEKSServiceFixture(t)
+
+	_, err := f.svc.ListManagedWorkers(context.Background(), &ListManagedWorkersInput{ClusterName: "ghost"}, testAccountID)
+	code, ok := awserrors.ResolveErrorCode(err)
+	require.True(t, ok)
+	assert.Equal(t, awserrors.ErrorEKSResourceNotFound, code)
+}

spinifex/domains/elbv2/reaper_test.go

@@ -2,9 +2,12 @@ package elbv2
 
 import (
 	"context"
+	"errors"
 	"testing"
 	"time"
 
+	"github.com/aws/aws-sdk-go/aws"
+	"github.com/aws/aws-sdk-go/service/ec2"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 )
@@ -80,3 +83,80 @@ func TestStartLifecycleReaper(t *testing.T) {
 
 	cancel()
 }
+
+// A load balancer the reaper gives up on must not leave its managed VM running:
+// the VM holds capacity and no customer data, and nothing else would stop it.
+func TestReapStuckProvisioningLBsTerminatesOnlyTheFailedLBsVM(t *testing.T) {
+	t.Parallel()
+	svc := setupTestService(t)
+	launcher := &mockSystemInstanceLauncher{}
+	svc.InstanceLauncher = launcher
+	now := time.Now().UTC()
+
+	stuck := testLBRecord("stuck0000000000", StateProvisioning, now.Add(-10*time.Minute), time.Time{})
+	stuck.InstanceID = "i-stuck"
+	noVM := testLBRecord("novm0000000000", StateProvisioning, now.Add(-10*time.Minute), time.Time{})
+	fresh := testLBRecord("fresh0000000000", StateProvisioning, now, time.Time{})
+	fresh.InstanceID = "i-fresh"
+	active := testLBRecord("active000000000", StateActive, now.Add(-10*time.Minute), time.Time{})
+	active.InstanceID = "i-active"
+	heartbeating := testLBRecord("heartbeat0000000", StateProvisioning, now.Add(-10*time.Minute), now.Add(-time.Second))
+	heartbeating.InstanceID = "i-heartbeating"
+	for _, lb := range []*LoadBalancerRecord{stuck, noVM, fresh, active, heartbeating} {
+		require.NoError(t, svc.store.PutLoadBalancer(t.Context(), lb))
+	}
+
+	svc.reapStuckProvisioningLBs(t.Context(), now)
+
+	got, err := svc.store.GetLoadBalancer(t.Context(), stuck.LoadBalancerID)
+	require.NoError(t, err)
+	assert.Equal(t, StateFailed, got.State)
+	assert.NotEmpty(t, got.StateReason, "the failure reason stays on the LB")
+
+	terminated := func() []string {
+		launcher.mu.Lock()
+		defer launcher.mu.Unlock()
+		return append([]string(nil), launcher.terminateCalls...)
+	}
+	require.Eventually(t, func() bool { return len(terminated()) > 0 }, 2*time.Second, 10*time.Millisecond,
+		"the reaper must terminate the failed LB's VM")
+	assert.Equal(t, []string{"i-stuck"}, terminated(),
+		"only the failed LB's own VM is terminated, never another LB's")
+}
+
+func TestReapStuckProvisioningLBsDetachesTheFailedLBsENIs(t *testing.T) {
+	t.Parallel()
+	for _, tc := range []struct {
+		name         string
+		terminateErr error
+		wantStatus   string
+	}{
+		{"terminated VM releases its ENI", nil, "available"},
+		{"a VM that cannot be stopped keeps its ENI attached", errors.New("nats timeout"), "in-use"},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			t.Parallel()
+			svc, vpcSvc, subnetID := setupMicrovmTestService(t)
+			svc.InstanceLauncher = &mockSystemInstanceLauncher{terminateErr: tc.terminateErr}
+
+			eniOut, err := vpcSvc.CreateNetworkInterface(t.Context(), &ec2.CreateNetworkInterfaceInput{SubnetId: aws.String(subnetID)}, testAccountID)
+			require.NoError(t, err)
+			eniID := aws.StringValue(eniOut.NetworkInterface.NetworkInterfaceId)
+			_, err = vpcSvc.AttachENI(testAccountID, eniID, "i-stuck", 0)
+			require.NoError(t, err)
+
+			now := time.Now().UTC()
+			lb := testLBRecord("enirel0000000000", StateProvisioning, now.Add(-10*time.Minute), time.Time{})
+			lb.AccountID = testAccountID
+			lb.InstanceID = "i-stuck"
+			lb.ENIs = []string{eniID}
+			require.NoError(t, svc.store.PutLoadBalancer(t.Context(), lb))
+
+			svc.reapStuckProvisioningLBs(t.Context(), now)
+
+			rec, err := vpcSvc.GetENIRecord(testAccountID, eniID)
+			require.NoError(t, err, "the ENI stays for DeleteLoadBalancer to remove")
+			assert.Equal(t, tc.wantStatus, rec.Status)
+		})
+	}
+}

spinifex/domains/elbv2/service_impl.go

@@ -1772,6 +1772,31 @@ func (s *ELBv2ServiceImpl) reapStuckProvisioningLBs(ctx context.Context, now tim
 			"lbArn", lb.LoadBalancerArn, "lbId", lb.LoadBalancerID,
 			"instanceId", lb.InstanceID, "createdAt", lb.CreatedAt)
 		s.markLBFailed(ctx, lb.LoadBalancerArn, reason)
+		s.releaseFailedLBVM(ctx, lb)
+	}
+}
+
+// releaseFailedLBVM stops the failed LB's own managed VM and detaches its ENIs,
+// so a failed LB holds no running capacity. The ENIs stay on the record for
+// DeleteLoadBalancer to remove. A VM that cannot be stopped keeps its ENIs
+// attached, since detaching under a running guest would only hide the leak.
+func (s *ELBv2ServiceImpl) releaseFailedLBVM(ctx context.Context, lb *LoadBalancerRecord) {
+	if lb.InstanceID == "" || s.InstanceLauncher == nil {
+		return
+	}
+	if err := s.InstanceLauncher.TerminateSystemInstance(lb.InstanceID); err != nil {
+		slog.Error("ELBv2 lifecycle reaper: failed to terminate failed LB VM",
+			"lbId", lb.LoadBalancerID, "instanceId", lb.InstanceID, "err", err)
+		return
+	}
+	if s.VPCService == nil {
+		return
+	}
+	for _, eniID := range lb.ENIs {
+		if err := s.VPCService.DetachENI(ctx, lb.AccountID, eniID); err != nil && !awserrors.IsNotFound(err) {
+			slog.Warn("ELBv2 lifecycle reaper: failed to detach failed LB ENI",
+				"lbId", lb.LoadBalancerID, "eniId", eniID, "err", err)
+		}
 	}
 }
 

spinifex/domains/iam/awsapi/registration.go

@@ -376,6 +376,7 @@ func NewRegistration(deps Deps) dispatch.Registration {
 		ResolveAction: resolveAction,
 		Dispatch:      deps.dispatch,
 		Errors:        dispatch.ErrorEnvelopeXML,
+		Traits:        dispatch.ServiceTraits{XMLThrottleCode: awserrors.ErrorThrottling},
 		Inventory:     OperationInventory(),
 	}
 }

spinifex/domains/network/external/igw.go

@@ -56,10 +56,9 @@ type IGWManager interface {
 	RebindGatewayIP(ctx context.Context, vpcID, newIP string, prefixLen int) error
 }
 
-// RoutedIngressHooks deliver host-side ingress plumbing in routed-NAT mode:
-// Ensure installs the host route sending VPC-CIDR traffic into OVN via the
-// gateway LRP transit IP; Remove tears it down on detach. Both idempotent.
-// Nil hooks (all other modes) are never called.
+// RoutedIngressHooks report routed-NAT ingress changes to the host: Ensure when
+// a VPC's gateway transit IP is set or re-asserted, Remove on detach. Both must
+// be idempotent. Nil hooks (all other modes) are never called.
 type RoutedIngressHooks struct {
 	Ensure func(ctx context.Context, vpcID, vpcCIDR, gwLrpIP string) error
 	Remove func(ctx context.Context, vpcID, vpcCIDR, gwLrpIP string) error
@@ -295,7 +294,7 @@ func (m *igwManager) AttachIGW(ctx context.Context, spec IGWSpec) error {
 // ensureRoutedLeg installs the routed-mode datapath legs for a VPC: the OVN
 // SNAT (VPC CIDR → gateway LRP transit IP) and the host ingress route into
 // it. No-op outside routed mode or without a gateway IP. Idempotent — the
-// reconcile loop replays AttachIGW, re-ensuring host state lost on reboot.
+// reconcile loop replays AttachIGW to re-ensure the SNAT.
 func (m *igwManager) ensureRoutedLeg(ctx context.Context, vpcID, gwLrpIP string) error {
 	if m.natMode != policy.NATModeRouted || gwLrpIP == "" {
 		return nil

spinifex/domains/network/host/routed_ingress.go

@@ -4,9 +4,15 @@ import (
 	"context"
 	"fmt"
 	"log/slog"
+	"net/netip"
 	"strings"
 )
 
+// VPCIngressRouteProto tags every VPC ingress route this package installs, so
+// the routes it may prune are told apart from the transit /24 and the EIP /32s
+// that share the spx-nat veth. Unassigned in iproute2's rt_protos.
+const VPCIngressRouteProto = "201"
+
 // EnsureVPCIngressRoute installs the host route that makes a VPC's private IPs
 // reachable in routed-NAT mode: traffic to the VPC CIDR enters OVN via the
 // gateway LRP's transit IP on the spx-nat veth. Idempotent (route replace).
@@ -18,44 +24,72 @@ func EnsureVPCIngressRoute(ctx context.Context, r Runner, vpcCIDR, gwLrpIP strin
 		return fmt.Errorf("EnsureVPCIngressRoute: gwLrpIP required")
 	}
 	if out, err := r.Run(ctx, "ip", "route", "replace", vpcCIDR,
-		"via", gwLrpIP, "dev", NATTransitHostEnd); err != nil {
+		"via", gwLrpIP, "dev", NATTransitHostEnd, "proto", VPCIngressRouteProto); err != nil {
 		return fmt.Errorf("install VPC ingress route %s via %s: %s: %w", vpcCIDR, gwLrpIP, string(out), err)
 	}
 	slog.Info("VPC ingress route installed", "vpc_cidr", vpcCIDR, "gw_lrp_ip", gwLrpIP)
 	return nil
 }
 
-// VPCIngressRouteVia returns the gateway IP of the existing transit-veth host
-// route for a VPC CIDR, or "" when none exists. Lets callers detect a
-// duplicate-CIDR collision (two VPCs sharing e.g. 172.31.0.0/16) before
-// replacing or deleting a route another VPC installed.
-func VPCIngressRouteVia(ctx context.Context, r Runner, vpcCIDR string) (string, error) {
-	if vpcCIDR == "" {
-		return "", fmt.Errorf("VPCIngressRouteVia: vpcCIDR required")
-	}
-	out, err := r.Run(ctx, "ip", "route", "show", vpcCIDR, "dev", NATTransitHostEnd)
+// ListVPCIngressRoutes returns the VPC ingress routes on this host, VPC CIDR to
+// gateway LRP transit IP. Only routes carrying VPCIngressRouteProto are listed.
+func ListVPCIngressRoutes(ctx context.Context, r Runner) (map[string]string, error) {
+	out, err := r.Run(ctx, "ip", "-o", "route", "show", "dev", NATTransitHostEnd, "proto", VPCIngressRouteProto)
 	if err != nil {
-		return "", fmt.Errorf("show VPC ingress route %s: %s: %w", vpcCIDR, string(out), err)
+		return nil, fmt.Errorf("list VPC ingress routes: %s: %w", string(out), err)
 	}
-	fields := strings.Fields(string(out))
-	for i, f := range fields {
-		if f == "via" && i+1 < len(fields) {
-			return fields[i+1], nil
+	routes := map[string]string{}
+	for line := range strings.SplitSeq(string(out), "\n") {
+		fields := strings.Fields(line)
+		if len(fields) == 0 {
+			continue
 		}
+		dst, err := netip.ParsePrefix(fields[0])
+		if err != nil {
+			slog.Warn("VPC ingress route list: unparseable destination", "line", strings.TrimSpace(line))
+			continue
+		}
+		via := ""
+		for i, f := range fields {
+			if f == "via" && i+1 < len(fields) {
+				via = fields[i+1]
+			}
+		}
+		routes[dst.Masked().String()] = via
 	}
-	return "", nil
+	return routes, nil
 }
 
-// RemoveVPCIngressRoute deletes the host route for a VPC CIDR on IGW detach.
-// A missing route is not an error (idempotent teardown).
+// RemoveVPCIngressRoute deletes the host route for a VPC CIDR. A missing route
+// is not an error (idempotent teardown).
 func RemoveVPCIngressRoute(ctx context.Context, r Runner, vpcCIDR string) error {
 	if vpcCIDR == "" {
 		return fmt.Errorf("RemoveVPCIngressRoute: vpcCIDR required")
 	}
-	if _, err := r.Run(ctx, "ip", "route", "del", vpcCIDR, "dev", NATTransitHostEnd); err != nil {
+	if _, err := r.Run(ctx, "ip", "route", "del", vpcCIDR, "dev", NATTransitHostEnd, "proto", VPCIngressRouteProto); err != nil {
 		slog.Debug("VPC ingress route already absent", "vpc_cidr", vpcCIDR, "err", err)
 		return nil
 	}
 	slog.Info("VPC ingress route removed", "vpc_cidr", vpcCIDR)
 	return nil
 }
+
+// VPCIngressRoutes is this host's VPC ingress route table behind a Runner.
+type VPCIngressRoutes struct {
+	Runner Runner
+}
+
+// List is ListVPCIngressRoutes.
+func (v VPCIngressRoutes) List(ctx context.Context) (map[string]string, error) {
+	return ListVPCIngressRoutes(ctx, v.Runner)
+}
+
+// Ensure is EnsureVPCIngressRoute.
+func (v VPCIngressRoutes) Ensure(ctx context.Context, vpcCIDR, gwLrpIP string) error {
+	return EnsureVPCIngressRoute(ctx, v.Runner, vpcCIDR, gwLrpIP)
+}
+
+// Remove is RemoveVPCIngressRoute.
+func (v VPCIngressRoutes) Remove(ctx context.Context, vpcCIDR string) error {
+	return RemoveVPCIngressRoute(ctx, v.Runner, vpcCIDR)
+}

spinifex/domains/network/host/routed_ingress_test.go

@@ -14,7 +14,7 @@ func TestEnsureVPCIngressRoute_InstallsRoute(t *testing.T) {
 	if err := EnsureVPCIngressRoute(context.Background(), r, "10.0.0.0/16", "100.127.0.10"); err != nil {
 		t.Fatalf("EnsureVPCIngressRoute: %v", err)
 	}
-	want := "ip route replace 10.0.0.0/16 via 100.127.0.10 dev " + NATTransitHostEnd
+	want := "ip route replace 10.0.0.0/16 via 100.127.0.10 dev " + NATTransitHostEnd + " proto " + VPCIngressRouteProto
 	if !r.called(want) {
 		t.Errorf("missing route call:\n  want %q\n  got  %v", want, r.calls)
 	}
@@ -43,43 +43,52 @@ func TestEnsureVPCIngressRoute_RouteFailure(t *testing.T) {
 	}
 }
 
-func TestVPCIngressRouteVia_ParsesHolder(t *testing.T) {
+// The list is what the per-node pass prunes from, so it must carry only the
+// tagged routes and key them exactly as intent renders a VPC CIDR.
+func TestListVPCIngressRoutes_ParsesTaggedRoutes(t *testing.T) {
 	r := newStubRunner()
-	r.expect("ip route show", []byte("172.31.0.0/16 via 100.127.0.239 dev spx-nat-host \n"), nil)
+	r.expect("ip -o route show", []byte("172.31.0.0/16 via 100.127.0.16 \n10.1.0.0/16 via 100.127.0.17 \n"), nil)
 
-	via, err := VPCIngressRouteVia(context.Background(), r, "172.31.0.0/16")
+	routes, err := ListVPCIngressRoutes(context.Background(), r)
 	if err != nil {
-		t.Fatalf("VPCIngressRouteVia: %v", err)
+		t.Fatalf("ListVPCIngressRoutes: %v", err)
 	}
-	if via != "100.127.0.239" {
-		t.Errorf("via = %q, want 100.127.0.239", via)
+	want := map[string]string{"172.31.0.0/16": "100.127.0.16", "10.1.0.0/16": "100.127.0.17"}
+	if len(routes) != len(want) {
+		t.Fatalf("routes = %v, want %v", routes, want)
 	}
-	want := "ip route show 172.31.0.0/16 dev " + NATTransitHostEnd
-	if !r.called(want) {
-		t.Errorf("missing route show call:\n  want %q\n  got  %v", want, r.calls)
+	for cidr, via := range want {
+		if routes[cidr] != via {
+			t.Errorf("routes[%s] = %q, want %q", cidr, routes[cidr], via)
+		}
+	}
+	call := "ip -o route show dev " + NATTransitHostEnd + " proto " + VPCIngressRouteProto
+	if !r.called(call) {
+		t.Errorf("missing list call:\n  want %q\n  got  %v", call, r.calls)
 	}
 }
 
-func TestVPCIngressRouteVia_NoRoute(t *testing.T) {
+func TestListVPCIngressRoutes_Empty(t *testing.T) {
 	r := newStubRunner()
-	r.expect("ip route show", nil, nil)
+	r.expect("ip -o route show", nil, nil)
 
-	via, err := VPCIngressRouteVia(context.Background(), r, "172.31.0.0/16")
+	routes, err := ListVPCIngressRoutes(context.Background(), r)
 	if err != nil {
-		t.Fatalf("VPCIngressRouteVia: %v", err)
+		t.Fatalf("ListVPCIngressRoutes: %v", err)
 	}
-	if via != "" {
-		t.Errorf("via = %q, want empty for absent route", via)
+	if len(routes) != 0 {
+		t.Errorf("routes = %v, want none", routes)
 	}
 }
 
-func TestVPCIngressRouteVia_ValidatesArgs(t *testing.T) {
+// A failed list must surface: read as empty it would prune nothing, but read as
+// a partial table the caller could not tell what it was missing.
+func TestListVPCIngressRoutes_Failure(t *testing.T) {
 	r := newStubRunner()
-	if _, err := VPCIngressRouteVia(context.Background(), r, ""); err == nil {
-		t.Error("empty vpcCIDR: expected error")
-	}
-	if len(r.calls) != 0 {
-		t.Errorf("validation failures must not run commands: %v", r.calls)
+	r.expect("ip -o route show", []byte("Cannot find device"), fmt.Errorf("exit 1"))
+
+	if _, err := ListVPCIngressRoutes(context.Background(), r); err == nil {
+		t.Fatal("expected list failure to be returned")
 	}
 }
 
@@ -90,7 +99,7 @@ func TestRemoveVPCIngressRoute_DeletesRoute(t *testing.T) {
 	if err := RemoveVPCIngressRoute(context.Background(), r, "10.0.0.0/16"); err != nil {
 		t.Fatalf("RemoveVPCIngressRoute: %v", err)
 	}
-	want := "ip route del 10.0.0.0/16 dev " + NATTransitHostEnd
+	want := "ip route del 10.0.0.0/16 dev " + NATTransitHostEnd + " proto " + VPCIngressRouteProto
 	if !r.called(want) {
 		t.Errorf("missing route del call:\n  want %q\n  got  %v", want, r.calls)
 	}

spinifex/domains/network/reconcile/host_ingress.go

@@ -0,0 +1,168 @@
+package reconcile
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"log/slog"
+	"time"
+
+	"github.com/mulgadc/spinifex/spinifex/domains/network/ovn"
+	"github.com/mulgadc/spinifex/spinifex/domains/network/policy"
+	"github.com/mulgadc/spinifex/spinifex/domains/network/topology"
+	loop "github.com/mulgadc/spinifex/spinifex/foundation/lifecycle/reconciler"
+	"github.com/nats-io/nats.go/jetstream"
+)
+
+// Host ingress cadence. The resync is short because a gateway IP rebind after a
+// DHCP lease move writes NB, not KV, so no watch announces it. The retry covers
+// an IGW whose gateway LRP the attaching node has not created yet.
+const (
+	defaultHostIngressResync = 30 * time.Second
+	hostIngressPendingRetry  = 5 * time.Second
+)
+
+// HostIngressRoutes is one host's table of routed-NAT VPC ingress routes, VPC
+// CIDR to the gateway LRP transit IP the traffic enters OVN through.
+type HostIngressRoutes interface {
+	// List returns only the routes Ensure installed, keyed by CIDR.
+	List(ctx context.Context) (map[string]string, error)
+	Ensure(ctx context.Context, cidr, via string) error
+	Remove(ctx context.Context, cidr string) error
+}
+
+// HostIngress converges this host's VPC ingress routes from intent. Every node
+// runs one, without the drift lease: the routes are the host's own, and the
+// attach event that used to install them reaches one vpcd of the queue group.
+type HostIngress struct {
+	// OVN is read only, for each attached VPC's gateway LRP transit IP.
+	OVN    ovn.Client
+	Routes HostIngressRoutes
+	// LoadIntent reads intent fresh each pass. An error abandons the pass, so
+	// an unreadable store never prunes a route.
+	LoadIntent func(ctx context.Context) (IntentState, error)
+	// PreferredVPC wins a CIDR shared by several attached VPCs; see ingressOutranks.
+	PreferredVPC string
+	// Resync is the backstop between passes. Zero means 30s.
+	Resync time.Duration
+}
+
+// hostIngressPlan is the route table a pass converges to. held names CIDRs whose
+// winning VPC has no resolvable gateway IP yet: whatever route they have is kept
+// as it is, since the winner is known but where it enters OVN is not.
+type hostIngressPlan struct {
+	want    map[string]string
+	held    map[string]struct{}
+	pending bool
+}
+
+// ingressOutranks reports whether VPC a takes a shared CIDR from VPC b: the
+// preferred VPC first, then the lower VPC ID. Every node ranks the same intent
+// the same way, so the holder never depends on which node saw which event.
+func ingressOutranks(a, b, preferred string) bool {
+	if a == preferred || b == preferred {
+		return a == preferred && b != preferred
+	}
+	return a < b
+}
+
+// planHostIngress picks one gateway per CIDR across the attached VPCs in intent.
+// gatewayIP is asked only about each CIDR's winner; "" marks it pending.
+func planHostIngress(intent IntentState, preferred string, gatewayIP func(vpcID string) string) hostIngressPlan {
+	winners := map[string]string{}
+	for vpcID := range intent.IGWs {
+		vpc, ok := intent.VPCs[vpcID]
+		if !ok || !vpc.CIDR.IsValid() {
+			continue
+		}
+		cidr := vpc.CIDR.Masked().String()
+		if cur, ok := winners[cidr]; !ok || ingressOutranks(vpcID, cur, preferred) {
+			winners[cidr] = vpcID
+		}
+	}
+	plan := hostIngressPlan{want: map[string]string{}, held: map[string]struct{}{}}
+	for cidr, vpcID := range winners {
+		if gw := gatewayIP(vpcID); gw != "" {
+			plan.want[cidr] = gw
+			continue
+		}
+		plan.held[cidr] = struct{}{}
+		plan.pending = true
+	}
+	return plan
+}
+
+// gatewayTransitIP is the stamped gateway IP of vpcID's gateway LRP, falling
+// back to its network address. "" when the port is absent or unreadable.
+func (h *HostIngress) gatewayTransitIP(ctx context.Context, vpcID string) string {
+	lrp, err := h.OVN.GetLogicalRouterPort(ctx, topology.GatewayRouterPort(vpcID))
+	if err != nil || lrp == nil {
+		return ""
+	}
+	if ip := lrp.ExternalIDs[policy.GatewayIPExtIDKey]; ip != "" {
+		return ip
+	}
+	return gatewayLRPIP(lrp)
+}
+
+// Pass converges the routes once. It asks to be revisited while a winning VPC's
+// gateway is still being created; the loop's resync covers everything else.
+func (h *HostIngress) Pass(ctx context.Context) (time.Duration, error) {
+	intent, err := h.LoadIntent(ctx)
+	if err != nil {
+		return 0, fmt.Errorf("load intent for host ingress routes: %w", err)
+	}
+	have, err := h.Routes.List(ctx)
+	if err != nil {
+		return 0, err
+	}
+	plan := planHostIngress(intent, h.PreferredVPC, func(vpcID string) string {
+		return h.gatewayTransitIP(ctx, vpcID)
+	})
+
+	var errs []error
+	for cidr, via := range plan.want {
+		if have[cidr] == via {
+			continue
+		}
+		if err := h.Routes.Ensure(ctx, cidr, via); err != nil {
+			errs = append(errs, err)
+		}
+	}
+	for cidr, via := range have {
+		_, wanted := plan.want[cidr]
+		_, held := plan.held[cidr]
+		if wanted || held {
+			continue
+		}
+		slog.InfoContext(ctx, "reconcile/host-ingress: pruning route no attached VPC holds",
+			"vpc_cidr", cidr, "gw_lrp_ip", via)
+		if err := h.Routes.Remove(ctx, cidr); err != nil {
+			errs = append(errs, err)
+		}
+	}
+
+	var revisit time.Duration
+	if plan.pending {
+		revisit = hostIngressPendingRetry
+	}
+	return revisit, errors.Join(errs...)
+}
+
+// RunHostIngress runs h on every intent change, every nudge and every
+// resync until ctx ends. nudge carries the attach and detach events
+// this node handled, which can land before the gateway they announce.
+func RunHostIngress(ctx context.Context, js jetstream.JetStream, h *HostIngress, nudge <-chan struct{}) {
+	resync := h.Resync
+	if resync <= 0 {
+		resync = defaultHostIngressResync
+	}
+	loop.Run(ctx, loop.Config{
+		Name:      "vpcd/host-ingress",
+		Sources:   []loop.Source{intentSource(js)},
+		Reconcile: h.Pass,
+		Trigger:   nudge,
+		Resync:    resync,
+		Debounce:  driftDebounce,
+	})
+}

spinifex/domains/network/reconcile/host_ingress_test.go

@@ -0,0 +1,439 @@
+package reconcile_test
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"maps"
+	"net/netip"
+	"sync"
+	"testing"
+	"time"
+
+	testutil "github.com/mulgadc/spinifex/internal/testkit"
+	ec2igw "github.com/mulgadc/spinifex/spinifex/domains/ec2/igw"
+	ec2vpc "github.com/mulgadc/spinifex/spinifex/domains/ec2/vpc"
+	"github.com/mulgadc/spinifex/spinifex/domains/network/external"
+	"github.com/mulgadc/spinifex/spinifex/domains/network/ovn/mock"
+	"github.com/mulgadc/spinifex/spinifex/domains/network/ovn/nbdb"
+	"github.com/mulgadc/spinifex/spinifex/domains/network/policy"
+	"github.com/mulgadc/spinifex/spinifex/domains/network/reconcile"
+	"github.com/mulgadc/spinifex/spinifex/domains/network/topology"
+	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
+	"github.com/nats-io/nats.go/jetstream"
+)
+
+// hostRoutes is one node's host route table.
+type hostRoutes struct {
+	mu     sync.Mutex
+	routes map[string]string
+	lists  int
+}
+
+func newHostRoutes(seed map[string]string) *hostRoutes {
+	r := &hostRoutes{routes: map[string]string{}}
+	maps.Copy(r.routes, seed)
+	return r
+}
+
+func (h *hostRoutes) List(context.Context) (map[string]string, error) {
+	h.mu.Lock()
+	defer h.mu.Unlock()
+	h.lists++
+	return maps.Clone(h.routes), nil
+}
+
+func (h *hostRoutes) Ensure(_ context.Context, cidr, via string) error {
+	h.mu.Lock()
+	defer h.mu.Unlock()
+	h.routes[cidr] = via
+	return nil
+}
+
+func (h *hostRoutes) Remove(_ context.Context, cidr string) error {
+	h.mu.Lock()
+	defer h.mu.Unlock()
+	delete(h.routes, cidr)
+	return nil
+}
+
+func (h *hostRoutes) listCount() int {
+	h.mu.Lock()
+	defer h.mu.Unlock()
+	return h.lists
+}
+
+func (h *hostRoutes) snapshot() map[string]string {
+	h.mu.Lock()
+	defer h.mu.Unlock()
+	return maps.Clone(h.routes)
+}
+
+// clusterIntent is the control-plane store every node reads.
+type clusterIntent struct {
+	mu   sync.Mutex
+	vpcs map[string]topology.VPCSpec
+	igws map[string]external.IGWSpec
+	err  error
+}
+
+func newClusterIntent() *clusterIntent {
+	return &clusterIntent{vpcs: map[string]topology.VPCSpec{}, igws: map[string]external.IGWSpec{}}
+}
+
+func (c *clusterIntent) addVPC(id, cidr string) {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	c.vpcs[id] = topology.VPCSpec{VPCID: id, CIDR: netip.MustParsePrefix(cidr)}
+}
+
+func (c *clusterIntent) attach(vpcID, igwID string) {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	c.igws[vpcID] = external.IGWSpec{VPCID: vpcID, InternetGatewayID: igwID}
+}
+
+func (c *clusterIntent) detach(vpcID string) {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	delete(c.igws, vpcID)
+}
+
+func (c *clusterIntent) load(context.Context) (reconcile.IntentState, error) {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	if c.err != nil {
+		return reconcile.IntentState{}, c.err
+	}
+	return reconcile.IntentState{VPCs: maps.Clone(c.vpcs), IGWs: maps.Clone(c.igws)}, nil
+}
+
+// node is one hypervisor's vpcd as far as host ingress goes.
+type node struct {
+	routes *hostRoutes
+	ingest *reconcile.HostIngress
+}
+
+func newNode(m *mock.Client, intent *clusterIntent, preferred string, seed map[string]string) *node {
+	routes := newHostRoutes(seed)
+	return &node{routes: routes, ingest: &reconcile.HostIngress{
+		OVN: m, Routes: routes, LoadIntent: intent.load, PreferredVPC: preferred,
+	}}
+}
+
+func (n *node) pass(t *testing.T) time.Duration {
+	t.Helper()
+	revisit, err := n.ingest.Pass(t.Context())
+	if err != nil {
+		t.Fatalf("host ingress pass: %v", err)
+	}
+	return revisit
+}
+
+func seedRouter(t *testing.T, m *mock.Client, vpcID, cidr string) {
+	t.Helper()
+	if err := m.CreateLogicalRouter(t.Context(), &nbdb.LogicalRouter{
+		Name:        topology.VPCRouter(vpcID),
+		ExternalIDs: map[string]string{"spinifex:vpc_id": vpcID, "spinifex:cidr": cidr},
+	}); err != nil {
+		t.Fatalf("seed router %s: %v", vpcID, err)
+	}
+}
+
+// seedGateway stands in for the gateway LRP an attach creates in NB.
+func seedGateway(t *testing.T, m *mock.Client, vpcID, gwIP string) {
+	t.Helper()
+	if err := m.CreateLogicalRouterPort(t.Context(), topology.VPCRouter(vpcID), &nbdb.LogicalRouterPort{
+		Name:        topology.GatewayRouterPort(vpcID),
+		Networks:    []string{gwIP + "/24"},
+		ExternalIDs: map[string]string{policy.GatewayIPExtIDKey: gwIP},
+	}); err != nil {
+		t.Fatalf("seed gateway %s: %v", vpcID, err)
+	}
+}
+
+// routedIGW is the IGW manager on the node the queue group hands an event to.
+// Its hooks only nudge, as in vpcd.
+func routedIGW(t *testing.T, m *mock.Client, nudges *int) external.IGWManager {
+	t.Helper()
+	nat, err := policy.NewNATManager(m, policy.NATModeRouted,
+		policy.WithSNATExemptSet(policy.NATExemptSetName, []string{"100.127.0.0/24"}))
+	if err != nil {
+		t.Fatalf("NewNATManager: %v", err)
+	}
+	nudge := func(context.Context, string, string, string) error { *nudges++; return nil }
+	mgr, err := external.NewIGWManager(external.IGWManagerConfig{
+		OVN: m, Routes: policy.NewRouteManager(m), NAT: nat,
+		Pool: &external.ExternalPoolConfig{
+			Name: "nat-transit", Gateway: "100.127.0.1", PrefixLen: 24,
+			GwLrpRangeStart: "100.127.0.240", GwLrpRangeEnd: "100.127.0.243",
+		},
+		Allocator: external.NewStaticRangeAllocator(m),
+		Chassis:   []string{"chassis-a"}, NATMode: policy.NATModeRouted,
+		RoutedIngress: &external.RoutedIngressHooks{Ensure: nudge, Remove: nudge},
+	})
+	if err != nil {
+		t.Fatalf("NewIGWManager: %v", err)
+	}
+	return mgr
+}
+
+func assertRoutes(t *testing.T, name string, got, want map[string]string) {
+	t.Helper()
+	if !maps.Equal(got, want) {
+		t.Errorf("%s routes = %v, want %v", name, got, want)
+	}
+}
+
+// The attach event reaches one vpcd of the queue group. Every other node must
+// still get the route from intent, or a host that neither handled the event
+// nor won a drift lease has no path into the VPC until a full resync.
+func TestHostIngress_AttachHandledOnOneNodeConvergesEveryNode(t *testing.T) {
+	m := mock.New()
+	intent := newClusterIntent()
+	seedRouter(t, m, "vpc-1", "10.1.0.0/16")
+	intent.addVPC("vpc-1", "10.1.0.0/16")
+	a := newNode(m, intent, "", nil)
+	b := newNode(m, intent, "", nil)
+
+	intent.attach("vpc-1", "igw-1")
+	nudges := 0
+	if err := routedIGW(t, m, &nudges).AttachIGW(t.Context(), external.IGWSpec{VPCID: "vpc-1", InternetGatewayID: "igw-1"}); err != nil {
+		t.Fatalf("AttachIGW on node A: %v", err)
+	}
+	if nudges == 0 {
+		t.Fatal("attach on node A never nudged its host ingress loop")
+	}
+
+	a.pass(t)
+	// Node B saw only the intent change, never the event.
+	b.pass(t)
+	want := map[string]string{"10.1.0.0/16": "100.127.0.240"}
+	assertRoutes(t, "node A", a.routes.snapshot(), want)
+	assertRoutes(t, "node B", b.routes.snapshot(), want)
+}
+
+// Detach has the same reach as attach, so the route must leave every node, not
+// only the one that ran DetachIGW.
+func TestHostIngress_DetachPrunesOnEveryNode(t *testing.T) {
+	m := mock.New()
+	intent := newClusterIntent()
+	seedRouter(t, m, "vpc-1", "10.1.0.0/16")
+	intent.addVPC("vpc-1", "10.1.0.0/16")
+	intent.attach("vpc-1", "igw-1")
+	nudges := 0
+	mgr := routedIGW(t, m, &nudges)
+	if err := mgr.AttachIGW(t.Context(), external.IGWSpec{VPCID: "vpc-1", InternetGatewayID: "igw-1"}); err != nil {
+		t.Fatalf("AttachIGW: %v", err)
+	}
+	nodes := []*node{newNode(m, intent, "", nil), newNode(m, intent, "", nil), newNode(m, intent, "", nil)}
+	for _, n := range nodes {
+		n.pass(t)
+		assertRoutes(t, "attached", n.routes.snapshot(), map[string]string{"10.1.0.0/16": "100.127.0.240"})
+	}
+
+	intent.detach("vpc-1")
+	if err := mgr.DetachIGW(t.Context(), "vpc-1"); err != nil {
+		t.Fatalf("DetachIGW on node A: %v", err)
+	}
+	for i, n := range nodes {
+		n.pass(t)
+		if got := n.routes.snapshot(); len(got) != 0 {
+			t.Errorf("node %d kept %v after detach, want no ingress route", i, got)
+		}
+	}
+}
+
+// Every account's default VPC is 172.31.0.0/16, so two attached VPCs contend
+// for one host route. Each node must pick the same holder whatever route it
+// had before and whichever order it met the VPCs in.
+func TestHostIngress_SharedCIDRPicksTheSameHolderOnEveryNode(t *testing.T) {
+	cases := []struct {
+		name      string
+		preferred string
+		wantVia   string
+	}{
+		{"preferred VPC wins over a lower ID", "vpc-b", "100.127.0.17"},
+		{"lower VPC ID wins without a preferred VPC", "", "100.127.0.16"},
+		{"a preferred VPC outside the contest changes nothing", "vpc-other", "100.127.0.16"},
+	}
+	for _, tc := range cases {
+		t.Run(tc.name, func(t *testing.T) {
+			m := mock.New()
+			intent := newClusterIntent()
+			for vpcID, gw := range map[string]string{"vpc-a": "100.127.0.16", "vpc-b": "100.127.0.17"} {
+				seedRouter(t, m, vpcID, "172.31.0.0/16")
+				seedGateway(t, m, vpcID, gw)
+				intent.addVPC(vpcID, "172.31.0.0/16")
+				intent.attach(vpcID, "igw-"+vpcID)
+			}
+			want := map[string]string{"172.31.0.0/16": tc.wantVia}
+			// Seeded with each holder, as if the event for either VPC had
+			// landed on that node first; repeated so map order cannot hide a
+			// choice that depends on it.
+			for _, seed := range []string{"", "100.127.0.16", "100.127.0.17"} {
+				for range 20 {
+					var initial map[string]string
+					if seed != "" {
+						initial = map[string]string{"172.31.0.0/16": seed}
+					}
+					n := newNode(m, intent, tc.preferred, initial)
+					n.pass(t)
+					assertRoutes(t, "seed "+seed, n.routes.snapshot(), want)
+				}
+			}
+		})
+	}
+}
+
+// Until the winning VPC's gateway exists the route is left alone: switching it
+// to the loser and back would cut the CIDR over twice for nothing.
+func TestHostIngress_PendingWinnerHoldsTheRouteAndRevisits(t *testing.T) {
+	m := mock.New()
+	intent := newClusterIntent()
+	seedRouter(t, m, "vpc-a", "172.31.0.0/16")
+	seedRouter(t, m, "vpc-b", "172.31.0.0/16")
+	seedGateway(t, m, "vpc-b", "100.127.0.17")
+	intent.addVPC("vpc-a", "172.31.0.0/16")
+	intent.addVPC("vpc-b", "172.31.0.0/16")
+	intent.attach("vpc-a", "igw-a")
+	intent.attach("vpc-b", "igw-b")
+	n := newNode(m, intent, "vpc-a", map[string]string{"172.31.0.0/16": "100.127.0.9"})
+
+	if revisit := n.pass(t); revisit <= 0 {
+		t.Errorf("revisit = %v, want a short retry while vpc-a's gateway is missing", revisit)
+	}
+	assertRoutes(t, "pending", n.routes.snapshot(), map[string]string{"172.31.0.0/16": "100.127.0.9"})
+
+	seedGateway(t, m, "vpc-a", "100.127.0.16")
+	if revisit := n.pass(t); revisit != 0 {
+		t.Errorf("revisit = %v after convergence, want 0", revisit)
+	}
+	assertRoutes(t, "converged", n.routes.snapshot(), map[string]string{"172.31.0.0/16": "100.127.0.16"})
+}
+
+// A restarted vpcd has no memory of the events it handled. Its first pass must
+// rebuild the table from intent: repoint a route left on a stale gateway, add
+// one it never had, and prune one whose VPC detached while it was down.
+func TestHostIngress_RestartConvergesFromIntent(t *testing.T) {
+	m := mock.New()
+	intent := newClusterIntent()
+	for vpcID, v := range map[string][2]string{
+		"vpc-1": {"10.1.0.0/16", "100.127.0.20"},
+		"vpc-2": {"10.2.0.0/16", "100.127.0.21"},
+	} {
+		seedRouter(t, m, vpcID, v[0])
+		seedGateway(t, m, vpcID, v[1])
+		intent.addVPC(vpcID, v[0])
+		intent.attach(vpcID, "igw-"+vpcID)
+	}
+	n := newNode(m, intent, "", map[string]string{
+		"10.1.0.0/16": "100.127.0.99",
+		"10.9.0.0/16": "100.127.0.30",
+	})
+
+	n.pass(t)
+	assertRoutes(t, "after restart", n.routes.snapshot(), map[string]string{
+		"10.1.0.0/16": "100.127.0.20",
+		"10.2.0.0/16": "100.127.0.21",
+	})
+}
+
+// An unreadable store is not an empty one: reading it as empty would strip
+// every VPC's ingress route from every node at once.
+func TestHostIngress_IntentReadFailureNeverPrunes(t *testing.T) {
+	m := mock.New()
+	intent := newClusterIntent()
+	intent.err = errors.New("kv unavailable")
+	seed := map[string]string{"10.1.0.0/16": "100.127.0.20"}
+	n := newNode(m, intent, "", seed)
+
+	if _, err := n.ingest.Pass(t.Context()); err == nil {
+		t.Fatal("expected the intent read failure to be returned")
+	}
+	assertRoutes(t, "after failed read", n.routes.snapshot(), seed)
+}
+
+// The loop has to wake on an intent write by itself, on every node at once,
+// while another node holds the drift lease and no event arrives, well inside
+// its resync.
+func TestRunHostIngress_IntentWriteConvergesEveryNodeWithoutLeaseOrEvent(t *testing.T) {
+	_, nc, js := testutil.StartTestJetStream(t)
+	ctx := t.Context()
+	vpcs, err := kvutil.GetOrCreateBucket(ctx, js, ec2vpc.KVBucketVPCs, 1)
+	if err != nil {
+		t.Fatalf("create VPC bucket: %v", err)
+	}
+	igws, err := kvutil.GetOrCreateBucket(ctx, js, ec2igw.KVBucketIGW, 1)
+	if err != nil {
+		t.Fatalf("create IGW bucket: %v", err)
+	}
+	put(t, vpcs, "acct/vpc-1", ec2vpc.VPCRecord{VpcId: "vpc-1", CidrBlock: "10.1.0.0/16"})
+	release, elected := reconcile.AcquireLeader(ctx, nc, reconcile.KVBucketVPCDReconcile, "node-elsewhere")
+	if !elected {
+		t.Fatal("could not take the drift lease for another node")
+	}
+	t.Cleanup(release)
+
+	m := mock.New()
+	seedRouter(t, m, "vpc-1", "10.1.0.0/16")
+	seedGateway(t, m, "vpc-1", "100.127.0.20")
+	loopCtx, cancel := context.WithCancel(ctx)
+	var wg sync.WaitGroup
+	t.Cleanup(func() {
+		cancel()
+		wg.Wait()
+	})
+	nodes := make([]*hostRoutes, 3)
+	for i := range nodes {
+		nodes[i] = newHostRoutes(nil)
+		h := &reconcile.HostIngress{
+			OVN: m, Routes: nodes[i], Resync: time.Minute,
+			LoadIntent: func(ctx context.Context) (reconcile.IntentState, error) {
+				return reconcile.LoadIntentFromKV(ctx, js, "us-east-1a")
+			},
+		}
+		wg.Go(func() { reconcile.RunHostIngress(loopCtx, js, h, nil) })
+	}
+	// The startup pass has to be over before the attach is written, or it could
+	// be what converged a node rather than the watch.
+	for _, n := range nodes {
+		for n.listCount() == 0 {
+			time.Sleep(10 * time.Millisecond)
+		}
+	}
+
+	want := map[string]string{"10.1.0.0/16": "100.127.0.20"}
+	converged := func() bool {
+		for _, n := range nodes {
+			if !maps.Equal(n.snapshot(), want) {
+				return false
+			}
+		}
+		return true
+	}
+	// Watchers attach asynchronously and see updates only, so the write is
+	// repeated until one lands rather than racing the loop's startup.
+	deadline := time.Now().Add(10 * time.Second)
+	for !converged() {
+		if time.Now().After(deadline) {
+			for i, n := range nodes {
+				t.Errorf("node %d routes = %v 10s after the attach was written, want %v", i, n.snapshot(), want)
+			}
+			t.FailNow()
+		}
+		put(t, igws, "acct/igw-1", ec2igw.IGWRecord{InternetGatewayId: "igw-1", VpcId: "vpc-1", State: "available"})
+		time.Sleep(500 * time.Millisecond)
+	}
+}
+
+func put(t *testing.T, kv jetstream.KeyValue, key string, rec any) {
+	t.Helper()
+	b, err := json.Marshal(rec)
+	if err != nil {
+		t.Fatalf("marshal %s: %v", key, err)
+	}
+	if _, err := kv.Put(t.Context(), key, b); err != nil {
+		t.Fatalf("put %s: %v", key, err)
+	}
+}

spinifex/domains/ochre/access.go

@@ -216,7 +216,7 @@ func (denyAllAccessResolver) Granted(_ context.Context, _, _ string) (bool, erro
 }
 
 // DenyAllAccessResolver grants no model to any account. It is the fallback for
-// a nil GatewayConfig.BedrockAccess: the insecure direction of this default is
+// a nil pipeline.Config.BedrockAccess: the insecure direction of this default is
 // "nothing works", which surfaces immediately, rather than "everything works",
 // which surfaces as a breach.
 var DenyAllAccessResolver AccessResolver = denyAllAccessResolver{}

spinifex/domains/ochre/awsapi/bedrock_request_test.go

@@ -71,14 +71,14 @@ type grantAllModels struct{}
 
 func (grantAllModels) Granted(_ context.Context, _, _ string) (bool, error) { return true, nil }
 
-// newBedrockRequestGateway builds a GatewayConfig with a real NATS connection
+// newBedrockRequestGateway builds a pipeline.Config with a real NATS connection
 // (satisfying Bedrock_Request/BedrockRuntime_Request's nil-check only — no
 // NATS subject handling is required for these routes), an allow-everything
 // IAMService so the policy gate passes, BedrockEndpoints pinned at the given
 // vLLM stub, and every model granted — access is deny-by-default, so without a
 // resolver these routes would all return before reaching the code under test.
 // ListFoundationModels/GetFoundationModel read the weights resolver through a
-// package-level setter rather than a GatewayConfig field, so it is installed
+// package-level setter rather than a pipeline.Config field, so it is installed
 // here and reset on cleanup.
 func newBedrockRequestGateway(t *testing.T, vllmURL string) *testGateway {
 	t.Helper()

spinifex/domains/ochre/awsapi/bedrockagent_request_test.go

@@ -22,7 +22,7 @@ import (
 	"github.com/stretchr/testify/require"
 )
 
-// newBedrockAgentRequestGateway builds a GatewayConfig with fresh
+// newBedrockAgentRequestGateway builds a pipeline.Config with fresh
 // gateway-owned KB/DataSource stores over a real (embedded) JetStream and an
 // allow-everything IAMService, so the policy gate passes on a real evaluation
 // rather than a bypass, mirroring newBedrockRequestGateway.

spinifex/domains/ochre/awsapi/bedrockagentruntime_request_test.go

@@ -23,7 +23,7 @@ import (
 	"github.com/stretchr/testify/require"
 )
 
-// newBedrockAgentRuntimeRequestGateway builds a GatewayConfig wired the same
+// newBedrockAgentRuntimeRequestGateway builds a pipeline.Config wired the same
 // way as newBedrockAgentRequestGateway (real JetStream-backed KB/DataSource
 // stores, allow-everything IAMService) plus newBedrockRequestGateway's vLLM
 // routing (BedrockEndpoints/BedrockAccess), since RetrieveAndGenerate's

spinifex/domains/ochre/credentials.go

@@ -106,5 +106,5 @@ func (noopCredentialResolver) Resolve(_ context.Context, _, _ string) (string, b
 }
 
 // NoopCredentialResolver resolves no vendors. Used as the fallback wherever
-// no CredentialStore is configured (nil GatewayConfig.BedrockCredentials).
+// no CredentialStore is configured (nil pipeline.Config.BedrockCredentials).
 var NoopCredentialResolver CredentialResolver = noopCredentialResolver{}

spinifex/domains/route53/awsapi/registration.go

@@ -89,6 +89,7 @@ func NewRegistration(deps Deps) dispatch.Registration {
 		ResolveAction: resolveAction,
 		Dispatch:      deps.dispatch,
 		Errors:        dispatch.ErrorEnvelopeXML,
+		Traits:        dispatch.ServiceTraits{XMLThrottleCode: awserrors.ErrorThrottling, GlobalSigningRegion: "us-east-1"},
 		Inventory:     OperationInventory(),
 	}
 }

spinifex/domains/sts/awsapi/preauth_test.go

@@ -0,0 +1,32 @@
+package awsapi
+
+import (
+	"io"
+	"net/http"
+	"net/http/httptest"
+	"strings"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+// TestAnonymousArgs_Classification covers the body-peek helper for anonymous action detection.
+func TestAnonymousArgs_Classification(t *testing.T) {
+	t.Run("anonymous action recognised, body restored", func(t *testing.T) {
+		const body = "Action=AssumeRoleWithWebIdentity&WebIdentityToken=tok"
+		req := httptest.NewRequest(http.MethodPost, "/", strings.NewReader(body))
+		args, ok := anonymousArgs(req)
+		require.True(t, ok)
+		assert.Equal(t, "AssumeRoleWithWebIdentity", args["Action"])
+
+		restored, _ := io.ReadAll(req.Body)
+		assert.Equal(t, body, string(restored), "body must be restored for downstream read")
+	})
+
+	t.Run("non-anonymous action rejected", func(t *testing.T) {
+		req := httptest.NewRequest(http.MethodPost, "/", strings.NewReader("Action=GetCallerIdentity"))
+		_, ok := anonymousArgs(req)
+		assert.False(t, ok)
+	})
+}

spinifex/domains/sts/awsapi/registration.go

@@ -107,6 +107,7 @@ func NewRegistration(deps Deps) dispatch.Registration {
 		ResolveAction: resolveAction,
 		Dispatch:      deps.dispatch,
 		Errors:        dispatch.ErrorEnvelopeXML,
+		Traits:        dispatch.ServiceTraits{XMLThrottleCode: awserrors.ErrorThrottling},
 		Inventory:     OperationInventory(),
 	}
 }
@@ -138,6 +139,63 @@ func NewAnonymousAssumeRoleWithWebIdentity(deps Deps) AnonymousFunc {
 	}
 }
 
+// anonymousActions lists the STS actions that carry no SigV4; a web-identity
+// token in the body authenticates them.
+var anonymousActions = map[string]bool{
+	"AssumeRoleWithWebIdentity": true,
+}
+
+// NewPreAuth serves the unsigned web-identity exchange ahead of SigV4. A
+// request with an Authorization header, a body over the payload cap or any
+// other action is left to the signed chain.
+func NewPreAuth(deps Deps) dispatch.PreAuth {
+	run := NewAnonymousAssumeRoleWithWebIdentity(deps)
+	return dispatch.PreAuth{
+		Service: ServiceName,
+		Serve: func(w http.ResponseWriter, r *http.Request) (bool, error) {
+			if r.Method != http.MethodPost || r.Header.Get("Authorization") != "" {
+				return false, nil
+			}
+			args, ok := anonymousArgs(r)
+			if !ok {
+				return false, nil
+			}
+			xmlOutput, err := run(args)
+			if err != nil {
+				return true, err
+			}
+			w.Header().Set("Content-Type", "text/xml")
+			w.WriteHeader(http.StatusOK)
+			if _, err := w.Write(xmlOutput); err != nil {
+				slog.Error("Failed to write STS response", "err", err)
+			}
+			return true, nil
+		},
+	}
+}
+
+// anonymousArgs peeks the request body and reports whether its Action is one
+// served without SigV4. This runs ahead of authentication, so an unauthenticated
+// client controls the allocation: one byte past the cap detects an oversize body,
+// which is handed back unread rather than dispatched.
+func anonymousArgs(r *http.Request) (map[string]string, bool) {
+	body, err := io.ReadAll(io.LimitReader(r.Body, sigv4.MaxPayloadLen+1))
+	if err != nil {
+		return nil, false
+	}
+	if int64(len(body)) > sigv4.MaxPayloadLen {
+		r.Body = io.NopCloser(io.MultiReader(bytes.NewReader(body), r.Body))
+		return nil, false
+	}
+	r.Body = io.NopCloser(bytes.NewReader(body))
+
+	args, err := query.ParseArgs(string(body))
+	if err != nil || !anonymousActions[args["Action"]] {
+		return nil, false
+	}
+	return args, true
+}
+
 // resolveAction is the one Query Action parser telemetry and dispatch share.
 // It leaves the body readable for the dispatcher and answers "" on any failure.
 func resolveAction(r *http.Request) string {

spinifex/gateway/discover_nodes_test.go

@@ -1,117 +0,0 @@
-package gateway_test
-
-import (
-	"context"
-	"encoding/json"
-	"sync/atomic"
-	"testing"
-	"time"
-
-	clusterv1 "github.com/mulgadc/spinifex/contracts/cluster/v1"
-	"github.com/mulgadc/spinifex/internal/testkit"
-	"github.com/mulgadc/spinifex/spinifex/gateway"
-	"github.com/nats-io/nats.go"
-)
-
-// respondAsNodes answers the discovery subject as n distinct nodes and reports
-// how many discovery rounds were served.
-func respondAsNodes(t *testing.T, nc *nats.Conn, n int) *atomic.Int64 {
-	t.Helper()
-
-	var rounds atomic.Int64
-	for i := range n {
-		node := string(rune('a' + i))
-		sub, err := nc.Subscribe(clusterv1.NodesDiscoverSubject, func(msg *nats.Msg) {
-			if node == "a" {
-				rounds.Add(1)
-			}
-			payload, err := json.Marshal(clusterv1.NodeDiscoverResponse{Node: node})
-			if err != nil {
-				return
-			}
-			_ = msg.Respond(payload)
-		})
-		if err != nil {
-			t.Fatalf("subscribe: %v", err)
-		}
-		t.Cleanup(func() { _ = sub.Unsubscribe() })
-	}
-	if err := nc.Flush(); err != nil {
-		t.Fatalf("flush: %v", err)
-	}
-	return &rounds
-}
-
-// Discovery fronts eight EC2 handlers, so whatever it costs is added to every
-// one of their calls. The gather counts whoever answers and therefore cannot
-// know when to stop, so it always sits out its whole timeout — that cost
-// has to be paid once in a while, not once per request.
-func TestDiscoverActiveNodesKeepsItsCostOffTheRequestPath(t *testing.T) {
-	const timeout = 50 * time.Millisecond
-	gateway.SetDiscoverActiveNodesTimeoutForTest(t, timeout)
-	_, nc := testutil.StartTestNATS(t)
-	respondAsNodes(t, nc, 4)
-
-	gw := &gateway.GatewayConfig{NATSConn: nc, ExpectedNodes: 4}
-
-	if count := gw.DiscoverActiveNodes(context.Background()); count != 4 {
-		t.Fatalf("DiscoverActiveNodes = %d, want 4", count)
-	}
-
-	start := time.Now()
-	for range 5 {
-		gw.DiscoverActiveNodes(context.Background())
-	}
-	elapsed := time.Since(start)
-
-	// Five uncached calls would cost five timeouts, so two bounds the cached path well clear of that.
-	if elapsed > 2*timeout {
-		t.Errorf("five further discoveries took %v; each was paying the fan-out timeout again", elapsed)
-	}
-}
-
-// Membership does not change per request. Re-deriving it on every call put a
-// fan-out in front of every API call that needs it, which is most of them.
-func TestDiscoverActiveNodesReusesARecentAnswer(t *testing.T) {
-	gateway.SetDiscoverActiveNodesTimeoutForTest(t, 50*time.Millisecond)
-	_, nc := testutil.StartTestNATS(t)
-	rounds := respondAsNodes(t, nc, 3)
-
-	gw := &gateway.GatewayConfig{NATSConn: nc, ExpectedNodes: 3}
-
-	if count := gw.DiscoverActiveNodes(context.Background()); count != 3 {
-		t.Fatalf("first discovery = %d, want 3", count)
-	}
-	first := rounds.Load()
-
-	for range 5 {
-		if count := gw.DiscoverActiveNodes(context.Background()); count != 3 {
-			t.Fatalf("cached discovery = %d, want 3", count)
-		}
-	}
-
-	if got := rounds.Load(); got != first {
-		t.Errorf("discovery ran %d more time(s) within the TTL, want the cached answer reused", got-first)
-	}
-}
-
-// A fallback is not evidence of anything: caching one would make a momentary
-// NATS problem outlive itself for the whole TTL.
-func TestDiscoverActiveNodesDoesNotCacheAFallback(t *testing.T) {
-	gateway.SetDiscoverActiveNodesTimeoutForTest(t, 50*time.Millisecond)
-	_, nc := testutil.StartTestNATS(t)
-
-	gw := &gateway.GatewayConfig{NATSConn: nc, ExpectedNodes: 2}
-
-	if count := gw.DiscoverActiveNodes(context.Background()); count != 2 {
-		t.Fatalf("fallback = %d, want the configured node count", count)
-	}
-
-	rounds := respondAsNodes(t, nc, 2)
-	if count := gw.DiscoverActiveNodes(context.Background()); count != 2 {
-		t.Fatalf("second discovery = %d, want 2", count)
-	}
-	if rounds.Load() == 0 {
-		t.Error("no discovery was published after a fallback, so the fallback had been cached")
-	}
-}

spinifex/gateway/oidc_discovery.go

@@ -1,37 +0,0 @@
-package gateway
-
-import (
-	"log/slog"
-	"net/http"
-
-	"github.com/go-chi/chi/v5"
-)
-
-// OIDCPublisher serves the unauthenticated per-cluster OIDC discovery document
-// and JWKS (IRSA). The EKS domain owns the records they are built from; the
-// gateway only maps the route onto the account and cluster.
-type OIDCPublisher interface {
-	ServeDiscoveryDocument(w http.ResponseWriter, r *http.Request, accountID, clusterName string)
-	ServeJWKS(w http.ResponseWriter, r *http.Request, accountID, clusterName string)
-}
-
-// OIDCDiscoveryDocument serves the cluster's OpenID discovery document.
-func (gw *GatewayConfig) OIDCDiscoveryDocument(w http.ResponseWriter, r *http.Request) {
-	if gw.OIDC == nil {
-		slog.Error("OIDC discovery: publisher not configured")
-		http.Error(w, "internal error", http.StatusInternalServerError)
-		return
-	}
-	gw.OIDC.ServeDiscoveryDocument(w, r, chi.URLParam(r, "accountID"), chi.URLParam(r, "clusterName"))
-}
-
-// OIDCJWKS serves the cluster's JWKS document as referenced by the discovery
-// document's jwks_uri.
-func (gw *GatewayConfig) OIDCJWKS(w http.ResponseWriter, r *http.Request) {
-	if gw.OIDC == nil {
-		slog.Error("OIDC JWKS: publisher not configured")
-		http.Error(w, "internal error", http.StatusInternalServerError)
-		return
-	}
-	gw.OIDC.ServeJWKS(w, r, chi.URLParam(r, "accountID"), chi.URLParam(r, "clusterName"))
-}

spinifex/gateway/registry.go

@@ -1,24 +0,0 @@
-package gateway
-
-import (
-	"net/http"
-
-	"github.com/go-chi/chi/v5"
-)
-
-// mountRegistry mounts the registry handler at /v2, ahead of the SigV4 group. The
-// registry authenticates by its own token rather than SigV4, so it is not part of the
-// AWS seam. Only the exact /v2 subtree is the registry; /v20 and the like stay SigV4.
-func (gw *GatewayConfig) mountRegistry(r chi.Router) {
-	h := gw.RegistryHandler
-	if h == nil {
-		h = http.HandlerFunc(registryNotConfigured)
-	}
-	r.Mount("/v2", h)
-}
-
-func registryNotConfigured(w http.ResponseWriter, _ *http.Request) {
-	w.Header().Set("Content-Type", "application/json")
-	w.WriteHeader(http.StatusNotImplemented)
-	_, _ = w.Write([]byte(`{"errors":[{"code":"UNSUPPORTED","message":"registry not configured"}]}`))
-}

spinifex/gateway/registry_helper_test.go

@@ -1,3 +0,0 @@
-package gateway
-
-import ()

spinifex/gateway/sts_anonymous.go

@@ -1,80 +0,0 @@
-package gateway
-
-import (
-	"bytes"
-	"errors"
-	"github.com/mulgadc/spinifex/spinifex/ingress/aws/reqctx"
-	"io"
-	"log/slog"
-	"net/http"
-
-	"github.com/mulgadc/bluebottle/pkg/sigv4"
-	"github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
-)
-
-// anonymousSTSInterceptor routes unsigned IRSA bootstrap actions
-// (AssumeRoleWithWebIdentity) to the STS dispatcher before the SigV4 middleware.
-// Callers carry no AWS credentials — authentication is via the ServiceAccount JWT
-// in the body. Signed requests (Authorization header present) pass through unchanged.
-func (gw *GatewayConfig) anonymousSTSInterceptor(next http.Handler) http.Handler {
-	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
-		if r.Method == http.MethodPost && r.Header.Get("Authorization") == "" {
-			if args, ok := gw.anonymousSTSArgs(r); ok {
-				r = r.WithContext(reqctx.WithService(r.Context(), "sts"))
-				if err := gw.serveAnonymousSTS(w, args); err != nil {
-					gw.ErrorHandler(w, r, err)
-				}
-				return
-			}
-		}
-		next.ServeHTTP(w, r)
-	})
-}
-
-// anonymousSTSArgs peeks the request body, parses query args, and reports
-// whether the Action is permitted without SigV4. Non-anonymous requests pass through unchanged.
-func (gw *GatewayConfig) anonymousSTSArgs(r *http.Request) (map[string]string, bool) {
-	// This runs ahead of SigV4, so an unauthenticated client controls the
-	// allocation. Read one byte past the cap to detect an oversized body and
-	// hand it back to the normal chain unread rather than dispatching it.
-	body, err := io.ReadAll(io.LimitReader(r.Body, sigv4.MaxPayloadLen+1))
-	if err != nil {
-		return nil, false
-	}
-	if int64(len(body)) > sigv4.MaxPayloadLen {
-		r.Body = io.NopCloser(io.MultiReader(bytes.NewReader(body), r.Body))
-		return nil, false
-	}
-	r.Body = io.NopCloser(bytes.NewReader(body))
-
-	args, err := ParseAWSQueryArgs(string(body))
-	if err != nil || !anonymousSTSActions[args["Action"]] {
-		return nil, false
-	}
-	return args, true
-}
-
-// anonymousSTSActions lists STS actions that carry no SigV4; authenticated by
-// a web-identity JWT instead. anonymousSTSInterceptor routes these before auth.
-var anonymousSTSActions = map[string]bool{
-	"AssumeRoleWithWebIdentity": true,
-}
-
-// serveAnonymousSTS runs an unsigned STS action through the composed
-// capability and writes its XML response.
-func (gw *GatewayConfig) serveAnonymousSTS(w http.ResponseWriter, args map[string]string) error {
-	if gw.AnonymousSTS == nil {
-		slog.Error("STS: anonymous service not initialized")
-		return errors.New(awserrors.ErrorInternalError)
-	}
-	xmlOutput, err := gw.AnonymousSTS(args)
-	if err != nil {
-		return err
-	}
-	w.Header().Set("Content-Type", "text/xml")
-	w.WriteHeader(http.StatusOK)
-	if _, err := w.Write(xmlOutput); err != nil {
-		slog.Error("Failed to write STS response", "err", err)
-	}
-	return nil
-}

spinifex/gateway/sts_anonymous_test.go

@@ -1,59 +0,0 @@
-package gateway
-
-import (
-	"github.com/aws/aws-sdk-go/service/sts"
-	"github.com/stretchr/testify/assert"
-	"github.com/stretchr/testify/require"
-	"io"
-	"net/http"
-	"net/http/httptest"
-	"strings"
-	"testing"
-)
-
-// TestAnonymousSTSArgs_Classification covers the body-peek helper for anonymous action detection.
-func TestAnonymousSTSArgs_Classification(t *testing.T) {
-	gw := &GatewayConfig{DisableLogging: true}
-
-	t.Run("anonymous action recognised, body restored", func(t *testing.T) {
-		const body = "Action=AssumeRoleWithWebIdentity&WebIdentityToken=tok"
-		req := httptest.NewRequest(http.MethodPost, "/", strings.NewReader(body))
-		args, ok := gw.anonymousSTSArgs(req)
-		require.True(t, ok)
-		assert.Equal(t, "AssumeRoleWithWebIdentity", args["Action"])
-
-		restored, _ := io.ReadAll(req.Body)
-		assert.Equal(t, body, string(restored), "body must be restored for downstream read")
-	})
-
-	t.Run("non-anonymous action rejected", func(t *testing.T) {
-		req := httptest.NewRequest(http.MethodPost, "/", strings.NewReader("Action=GetCallerIdentity"))
-		_, ok := gw.anonymousSTSArgs(req)
-		assert.False(t, ok)
-	})
-}
-
-// TestAnonymousSTS_SignedRequestFallsThrough confirms a request with an Authorization
-// header is not intercepted — it passes through to the signed surface.
-func TestAnonymousSTS_SignedRequestFallsThrough(t *testing.T) {
-	gw := &GatewayConfig{
-		DisableLogging: true,
-		STSService: &flexMockSTSService{
-			assumeWebIdentityFn: func(*sts.AssumeRoleWithWebIdentityInput) (*sts.AssumeRoleWithWebIdentityOutput, error) {
-				t.Fatal("signed request must not reach the anonymous STS dispatcher")
-				return nil, nil
-			},
-		},
-	}
-
-	nextCalled := false
-	next := http.HandlerFunc(func(http.ResponseWriter, *http.Request) { nextCalled = true })
-
-	req := httptest.NewRequest(http.MethodPost, "/",
-		strings.NewReader("Action=AssumeRoleWithWebIdentity&WebIdentityToken=x"))
-	req.Header.Set("Authorization", "AWS4-HMAC-SHA256 Credential=...")
-
-	rec := httptest.NewRecorder()
-	gw.anonymousSTSInterceptor(next).ServeHTTP(rec, req)
-	assert.True(t, nextCalled, "signed request must pass through to next handler")
-}

spinifex/ingress/aws/dispatch/operations.go

@@ -1,10 +1,6 @@
-package gateway
+package dispatch
 
-import (
-	"slices"
-
-	"github.com/mulgadc/spinifex/spinifex/ingress/aws/dispatch"
-)
+import "slices"
 
 // ServiceOperationInventory describes the authoritative dispatch state for an
 // AWS-compatible service. Registered operations minus Stubbed and Unsupported
@@ -19,7 +15,7 @@ type ServiceOperationInventory struct {
 // declared inventory of each registered service. S3 is intentionally absent:
 // Spinifex does not dispatch S3 operations and delegates that REST surface to
 // Predastore.
-func AWSOperationInventory(registered map[string]dispatch.Inventory) map[string]ServiceOperationInventory {
+func AWSOperationInventory(registered map[string]Inventory) map[string]ServiceOperationInventory {
 	inventory := make(map[string]ServiceOperationInventory, len(registered))
 	for name, inv := range registered {
 		inventory[name] = ServiceOperationInventory{

spinifex/ingress/aws/dispatch/preauth.go

@@ -0,0 +1,13 @@
+package dispatch
+
+import "net/http"
+
+// PreAuth serves a request that carries no credentials ahead of
+// authentication and throttling, because its own payload authenticates it.
+// Serve reports handled=false, having written nothing and left the body
+// readable, for a request that is not its own. An error it returns for a
+// handled request renders under Service.
+type PreAuth struct {
+	Service string
+	Serve   func(w http.ResponseWriter, r *http.Request) (handled bool, err error)
+}

spinifex/ingress/aws/dispatch/registry.go

@@ -160,6 +160,8 @@ type Registration struct {
 	// Unset, a REST-JSON or JSON 1.0 registration gets the protocol's own
 	// throttle error and any other keeps the gateway default. JSON only.
 	Throttle *ErrorSpec
+	// Traits are the per-service behaviours the pipeline reads by scope.
+	Traits ServiceTraits
 	// QueryErrorCodes maps an error code to the legacy query code sent in
 	// x-amzn-query-error; a code absent from it is sent as itself. JSON 1.0
 	// only.
@@ -194,6 +196,7 @@ var serviceNamePattern = regexp.MustCompile(`^[a-z0-9][a-z0-9-]*$`)
 // concurrent use; Build returns the immutable registry the server reads.
 type Builder struct {
 	entries   map[string]*Entry
+	scopes    map[string]ServiceTraits
 	inventory map[string]Inventory
 	sealed    bool
 }
@@ -202,6 +205,7 @@ type Builder struct {
 func NewBuilder() *Builder {
 	return &Builder{
 		entries:   make(map[string]*Entry),
+		scopes:    make(map[string]ServiceTraits),
 		inventory: make(map[string]Inventory),
 	}
 }
@@ -235,6 +239,7 @@ func (b *Builder) Register(reg Registration) error {
 		handle:     reg.Dispatch,
 		errors:     reg.Errors,
 		throttle:   reg.Throttle,
+		traits:     reg.Traits,
 		queryCodes: maps.Clone(reg.QueryErrorCodes),
 		internal:   reg.InternalErrorCode,
 		namespace:  reg.ErrorNamespace,
@@ -247,11 +252,32 @@ func (b *Builder) Register(reg Registration) error {
 	return nil
 }
 
+// DeclareScope declares the traits of a credential scope this gateway knows
+// but does not serve, so a rejection for it still renders as its clients
+// expect. It fails on an invalid name or a scope a registration serves.
+func (b *Builder) DeclareScope(scope string, traits ServiceTraits) error {
+	if b.sealed {
+		return ErrSealed
+	}
+	if !serviceNamePattern.MatchString(scope) {
+		return fmt.Errorf("dispatch: invalid scope name %q", scope)
+	}
+	if _, ok := b.entries[scope]; ok {
+		return fmt.Errorf("dispatch: scope %q is served by a registration", scope)
+	}
+	if _, ok := b.scopes[scope]; ok {
+		return fmt.Errorf("dispatch: scope %q is already declared", scope)
+	}
+	b.scopes[scope] = traits
+	return nil
+}
+
 // Build seals the builder and returns the registry. Further Register calls fail.
 func (b *Builder) Build() *Registry {
 	b.sealed = true
 	return &Registry{
 		entries:   maps.Clone(b.entries),
+		scopes:    maps.Clone(b.scopes),
 		inventory: maps.Clone(b.inventory),
 	}
 }
@@ -289,6 +315,9 @@ func validate(reg Registration) ([]string, error) {
 }
 
 func validateErrorSpecs(reg Registration) error {
+	if reg.Traits.XMLThrottleCode != "" && reg.Errors != ErrorEnvelopeXML {
+		return errors.New("XML throttle code needs the XML error envelope")
+	}
 	if t := reg.Throttle; t != nil {
 		if !reg.Errors.IsJSON() {
 			return errors.New("throttle error needs a JSON error envelope")
@@ -363,6 +392,7 @@ func sortedClone(s []string) []string {
 // for concurrent use because nothing mutates it after Build.
 type Registry struct {
 	entries   map[string]*Entry
+	scopes    map[string]ServiceTraits
 	inventory map[string]Inventory
 }
 
@@ -372,6 +402,16 @@ func (r *Registry) Lookup(service string) (*Entry, bool) {
 	return e, ok
 }
 
+// Traits returns the traits of a scope: a registration's own, else those
+// declared for a scope the gateway does not serve.
+func (r *Registry) Traits(scope string) (ServiceTraits, bool) {
+	if e, ok := r.entries[scope]; ok {
+		return e.traits, true
+	}
+	t, ok := r.scopes[scope]
+	return t, ok
+}
+
 // Inventory returns a fresh copy of every registration's operation inventory,
 // keyed by inventory name with each list sorted.
 func (r *Registry) Inventory() map[string]Inventory {
@@ -394,12 +434,16 @@ type Entry struct {
 	errors  ErrorEnvelope
 
 	throttle   *ErrorSpec
+	traits     ServiceTraits
 	queryCodes map[string]string
 	internal   string
 	namespace  string
 	namespaces map[string]string
 }
 
+// Traits returns the registration's declared service traits.
+func (e *Entry) Traits() ServiceTraits { return e.traits }
+
 // Service returns the registration's primary credential-scope name.
 func (e *Entry) Service() string { return e.service }
 

spinifex/ingress/aws/dispatch/registry_test.go

@@ -364,3 +364,38 @@ func TestEntryErrorNamespacePerCodeOverride(t *testing.T) {
 	assert.Equal(t, "com.amazonaws.sqs", e.ErrorNamespace("QueueDoesNotExist"))
 	assert.Equal(t, "com.amazon.coral.service", e.ErrorNamespace("SerializationException"))
 }
+
+func TestRegisterXMLThrottleCodeNeedsXMLEnvelope(t *testing.T) {
+	r := validReg("svc")
+	r.Errors = ErrorEnvelopeXML
+	r.Traits = ServiceTraits{XMLThrottleCode: "Throttling"}
+	b := NewBuilder()
+	require.NoError(t, b.Register(r))
+	e, _ := b.Build().Lookup("svc")
+	assert.Equal(t, "Throttling", e.Traits().XMLThrottleCode)
+
+	r = validReg("other")
+	r.Errors = ErrorEnvelopeJSON
+	r.Traits = ServiceTraits{XMLThrottleCode: "Throttling"}
+	assert.Error(t, NewBuilder().Register(r))
+}
+
+func TestDeclareScope(t *testing.T) {
+	b := NewBuilder()
+	require.NoError(t, b.Register(validReg("svc")))
+	require.NoError(t, b.DeclareScope("other", ServiceTraits{XMLErrorShape: XMLErrorShapeS3}))
+
+	assert.Error(t, b.DeclareScope("svc", ServiceTraits{}), "a served service cannot be declared")
+	assert.Error(t, b.DeclareScope("other", ServiceTraits{}), "a scope is declared once")
+	assert.Error(t, b.DeclareScope("Bad Name", ServiceTraits{}))
+
+	reg := b.Build()
+	traits, ok := reg.Traits("other")
+	require.True(t, ok)
+	assert.Equal(t, XMLErrorShapeS3, traits.XMLErrorShape)
+	_, served := reg.Lookup("other")
+	assert.False(t, served)
+	_, ok = reg.Traits("unknown")
+	assert.False(t, ok)
+	assert.ErrorIs(t, b.DeclareScope("late", ServiceTraits{}), ErrSealed)
+}

spinifex/ingress/aws/dispatch/selector.go

@@ -44,6 +44,14 @@ func (s *Selector) Lookup(service string) (*Entry, bool) {
 	return s.registry.Lookup(service)
 }
 
+// Traits returns the traits declared for a credential scope, served or not.
+func (s *Selector) Traits(service string) (ServiceTraits, bool) {
+	if s.registry == nil {
+		return ServiceTraits{}, false
+	}
+	return s.registry.Traits(service)
+}
+
 // Served reports whether either dispatch path serves service.
 func (s *Selector) Served(service string) bool {
 	if _, ok := s.Lookup(service); ok {

spinifex/ingress/aws/dispatch/traits.go

@@ -0,0 +1,32 @@
+package dispatch
+
+// XMLErrorShape selects the XML body an XML error renders in. The zero value
+// is the generic REST-XML ErrorResponse.
+type XMLErrorShape int
+
+const (
+	// XMLErrorShapeGeneric is the <ErrorResponse> envelope.
+	XMLErrorShapeGeneric XMLErrorShape = iota
+	// XMLErrorShapeEC2 is the EC2 <Response><Errors> envelope.
+	XMLErrorShapeEC2
+	// XMLErrorShapeS3 is the S3 <Error> envelope.
+	XMLErrorShapeS3
+)
+
+// ServiceTraits are the behaviours the pipeline applies by credential scope
+// without naming a service: the owning registration declares them, and the
+// composition declares them for a scope it knows but does not serve.
+type ServiceTraits struct {
+	// XMLThrottleCode is the code of the XML throttle rejection. Unset, the
+	// gateway default RequestLimitExceeded applies. XML envelopes only.
+	XMLThrottleCode string
+	// XMLErrorShape is the XML body the scope's errors render in.
+	XMLErrorShape XMLErrorShape
+	// GlobalSigningRegion is the credential-scope region clients of a global
+	// service sign with on its default endpoint. Unset, only the gateway's own
+	// region is accepted.
+	GlobalSigningRegion string
+	// PayloadBody marks a scope whose request body is object data, never
+	// query arguments, so nothing reads an Action out of it.
+	PayloadBody bool
+}

spinifex/ingress/aws/pipeline/acm_characterization_test.go

@@ -1,7 +1,8 @@
-package gateway
+package pipeline
 
 import (
 	"github.com/go-chi/chi/v5"
+	"github.com/mulgadc/spinifex/internal/testkit/gwfix"
 	acmawsapi "github.com/mulgadc/spinifex/spinifex/domains/acm/awsapi"
 	"github.com/mulgadc/spinifex/spinifex/foundation/telemetry"
 	"github.com/stretchr/testify/assert"
@@ -29,7 +30,7 @@ func TestACMActionFromTarget_FormsAndTelemetryAgree(t *testing.T) {
 		{"bare action", "ListCertificates", "ListCertificates"},
 		{"trailing dot", "CertificateManager.", ""},
 	}
-	acmEntry, ok := withACM(&GatewayConfig{}, acmawsapi.Deps{}).registered(acmawsapi.ServiceName)
+	acmEntry, ok := withACM(&Config{}, acmawsapi.Deps{}).registered(acmawsapi.ServiceName)
 	require.True(t, ok)
 	for _, tc := range cases {
 		t.Run(tc.name, func(t *testing.T) {
@@ -53,7 +54,7 @@ func TestACMPipeline_TraceActionEnricher_SpanNamedServiceAction(t *testing.T) {
 	defer otel.SetTracerProvider(prev)
 
 	gw, _ := acmPipelineApp(t, allowAllACMDocs())
-	subscribeACMCounter(t, gw.NATSConn, "acm.ListCertificates", []byte(`{}`))
+	subscribeACMCounter(t, gwfix.DepsOf(gw).NATS, "acm.ListCertificates", []byte(`{}`))
 
 	r := chi.NewRouter()
 	r.Use(otelsetup.HTTPMiddleware("awsgw"))

spinifex/ingress/aws/pipeline/actor_projection_characterization_test.go

@@ -1,4 +1,4 @@
-package gateway
+package pipeline
 
 //test:in-package — this characterization reads unexported request-context keys.
 
@@ -95,16 +95,12 @@ func actorProbeGateway(t *testing.T, e *cbEnv) (http.Handler, *[]actorObservatio
 			return nil
 		},
 	}))
-	gw := &GatewayConfig{
+	gw := gwfix.WithDeps(&Config{
 		DisableLogging: true,
-		NATSConn:       e.nc,
 		Region:         gwfix.Region,
-		ExpectedNodes:  1,
-		IAMService:     e.counter,
-		STSService:     e.sts,
-	}
+	}, gwfix.Deps{NATS: e.nc, Nodes: 1, IAM: e.counter, STS: e.sts})
 	withRealAuth(gw)
-	gw.AnonymousSTS = stsawsapi.NewAnonymousAssumeRoleWithWebIdentity(stsawsapi.Deps{STS: gw.STSService, IAM: gw.IAMService})
+	gw.PreAuth = append(gw.PreAuth, stsawsapi.NewPreAuth(stsawsapi.Deps{STS: gwfix.DepsOf(gw).STS, IAM: gwfix.DepsOf(gw).IAM}))
 	require.NoError(t, b.Register(stsRegistration(gw)))
 	gw.Services = b.Build()
 	return gw.SetupRoutes(), &seen
@@ -362,3 +358,9 @@ func TestActorProjection_AgentRolesNotAssumableOverHTTPS(t *testing.T) {
 	assert.Equal(t, e.accountID, o.RDSAccountID)
 	assert.NotEqual(t, global, o.RDSAccountID)
 }
+
+const (
+	testDiscAccount = "000000000001"
+	testDiscCluster = "h1"
+	testDiscIssuer  = "https://10.0.0.1:9999/oidc/eks/ap-southeast-2/000000000001/h1"
+)

spinifex/ingress/aws/pipeline/admin.go

@@ -1,6 +1,7 @@
-package gateway
+package pipeline
 
 import (
+	"context"
 	"encoding/json"
 	"github.com/mulgadc/spinifex/spinifex/ingress/aws/reqctx"
 	"log/slog"
@@ -8,19 +9,28 @@ import (
 	"uuid"
 
 	"github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
-	operatoradmin "github.com/mulgadc/spinifex/spinifex/operator/admin"
 )
 
+// AdminSurface names the registration the private /admin surface dispatches
+// to and how the request ID reaches it. The composing role supplies both, so
+// the gateway names no service.
+type AdminSurface struct {
+	Service       string
+	WithRequestID func(ctx context.Context, requestID string) context.Context
+}
+
 // Admin_Request serves the private /admin/<Method> surface. The route stays
 // here so it keeps its place ahead of the service selector; the work is the
 // spinifex registration's, dispatched exactly as a Query request is. Only the
 // error rendering differs: this surface answers in JSON with a request ID.
-func (gw *GatewayConfig) Admin_Request(w http.ResponseWriter, r *http.Request) {
+func (gw *Config) Admin_Request(w http.ResponseWriter, r *http.Request) {
 	requestID := uuid.NewV4().String()
 	w.Header().Set("X-Amzn-Requestid", requestID)
-	r = r.WithContext(operatoradmin.WithRequestID(r.Context(), requestID))
+	if gw.Admin.WithRequestID != nil {
+		r = r.WithContext(gw.Admin.WithRequestID(r.Context(), requestID))
+	}
 
-	entry, ok := gw.registered(operatoradmin.ServiceName)
+	entry, ok := gw.registered(gw.Admin.Service)
 	if !ok {
 		slog.Error("Admin: spinifex service is not registered", "requestId", requestID)
 		gw.writeAdminError(w, requestID, awserrors.ErrorInternalError, "")
@@ -29,12 +39,12 @@ func (gw *GatewayConfig) Admin_Request(w http.ResponseWriter, r *http.Request) {
 
 	action := reqctx.Action(r.Context())
 	if err := gw.dispatchRegistered(entry, w, r); err != nil {
-		slog.Log(r.Context(), serviceErrorLevel(operatoradmin.ServiceName, err), "Service request error",
-			"service", operatoradmin.ServiceName, "action", action, "requestId", requestID, "error", err)
+		slog.Log(r.Context(), serviceErrorLevel(gw.Admin.Service, err), "Service request error",
+			"service", gw.Admin.Service, "action", action, "requestId", requestID, "error", err)
 		gw.writeAdminErrorFor(w, requestID, err)
 		return
 	}
-	slog.Info("Service request completed", "service", operatoradmin.ServiceName, "action", action, "requestId", requestID)
+	slog.Info("Service request completed", "service", gw.Admin.Service, "action", action, "requestId", requestID)
 }
 
 // adminErrorBody is the JSON error envelope for /admin. This surface is not an
@@ -52,7 +62,7 @@ type adminErrorDetail struct {
 // writeAdminErrorFor resolves an error to its code and any message its call
 // site attached, so a wrapped error renders as its own code instead of falling
 // through to InternalError.
-func (gw *GatewayConfig) writeAdminErrorFor(w http.ResponseWriter, requestID string, err error) {
+func (gw *Config) writeAdminErrorFor(w http.ResponseWriter, requestID string, err error) {
 	code, message, ok := awserrors.ResolveErrorDetail(err)
 	if !ok {
 		// The cause reaches neither the wire nor the envelope's own unmapped-code
@@ -66,7 +76,7 @@ func (gw *GatewayConfig) writeAdminErrorFor(w http.ResponseWriter, requestID str
 // writeAdminError writes the JSON error envelope. message overrides the
 // registry text when a specific cause is safe to disclose; a policy denial
 // names principal, action and resource, all of them caller-derivable.
-func (gw *GatewayConfig) writeAdminError(w http.ResponseWriter, requestID, code, message string) {
+func (gw *Config) writeAdminError(w http.ResponseWriter, requestID, code, message string) {
 	detail, ok := awserrors.ErrorLookup[code]
 	if !ok {
 		slog.Error("Admin: unmapped error code", "code", code)

spinifex/ingress/aws/pipeline/admin_outcome_test.go

@@ -1,8 +1,9 @@
-package gateway
+package pipeline
 
 import (
 	"bytes"
 	"encoding/json"
+	"github.com/mulgadc/spinifex/internal/testkit/gwfix"
 	"log/slog"
 	"net/http"
 	"strings"
@@ -48,13 +49,13 @@ func TestAdminOutcome_LogFollowsTheRealResult(t *testing.T) {
 	cases := []struct {
 		name       string
 		accountID  string
-		mutate     func(*GatewayConfig)
+		mutate     func(*Config)
 		wantStatus int
 		wantLevel  string
 	}{
 		{"success", awsidentifiers.DefaultAccountID(), nil, http.StatusOK, ""},
 		{"denied caller", authzAccountID, nil, http.StatusForbidden, "INFO"},
-		{"cluster fault", awsidentifiers.DefaultAccountID(), func(gw *GatewayConfig) { gw.NATSConn = nil }, http.StatusServiceUnavailable, "ERROR"},
+		{"cluster fault", awsidentifiers.DefaultAccountID(), func(gw *Config) { gwfix.DepsOf(gw).NATS = nil }, http.StatusServiceUnavailable, "ERROR"},
 	}
 	for _, tc := range cases {
 		t.Run(tc.name, func(t *testing.T) {

spinifex/ingress/aws/pipeline/admin_test.go

@@ -1,4 +1,4 @@
-package gateway
+package pipeline
 
 //test:in-package — the authorization gates read unexported context keys and
 // principal-type constants, and asserting on them through the router alone
@@ -41,7 +41,7 @@ func createAccountPolicy() *gwfix.PolicyIAM {
 }
 
 // adminRequest drives Admin_Request with the chi route param the router sets.
-func adminRequest(gw *GatewayConfig, method string, body string) *httptest.ResponseRecorder {
+func adminRequest(gw *Config, method string, body string) *httptest.ResponseRecorder {
 	req := httptest.NewRequest(http.MethodPost, "/admin/"+method, strings.NewReader(body))
 	rctx := chi.NewRouteContext()
 	rctx.URLParams.Add("method", method)
@@ -62,7 +62,7 @@ func decodeAdminError(t *testing.T, rec *httptest.ResponseRecorder) adminErrorBo
 // An unknown method is rejected before authorization, so it never discloses
 // whether the caller would have been permitted.
 func TestAdminRequestUnknownMethod(t *testing.T) {
-	gw := withOperator(withRealAuth(&GatewayConfig{DisableLogging: true, IAMService: allowAllIAMService()}))
+	gw := withOperator(withRealAuth(gwfix.WithDeps(&Config{DisableLogging: true}, gwfix.Deps{IAM: allowAllIAMService()})))
 
 	rec := adminRequest(gw, "PurgeEverything", `{}`)
 
@@ -71,7 +71,7 @@ func TestAdminRequestUnknownMethod(t *testing.T) {
 }
 
 func TestAdminRequestDeniesUnauthorizedCaller(t *testing.T) {
-	gw := withOperator(withRealAuth(&GatewayConfig{DisableLogging: true, IAMService: allowAllIAMService()}))
+	gw := withOperator(withRealAuth(gwfix.WithDeps(&Config{DisableLogging: true}, gwfix.Deps{IAM: allowAllIAMService()})))
 
 	rec := adminRequest(gw, "CreateAccount", `{}`)
 
@@ -83,7 +83,7 @@ func TestAdminRequestDeniesUnauthorizedCaller(t *testing.T) {
 // The request ID ties a client failure to the gateway log line, so it must be
 // present on the header and in the body even for a denial.
 func TestAdminRequestAlwaysCarriesRequestID(t *testing.T) {
-	gw := withOperator(withRealAuth(&GatewayConfig{DisableLogging: true, IAMService: allowAllIAMService()}))
+	gw := withOperator(withRealAuth(gwfix.WithDeps(&Config{DisableLogging: true}, gwfix.Deps{IAM: allowAllIAMService()})))
 
 	rec := adminRequest(gw, "CreateAccount", `{}`)
 
@@ -94,7 +94,7 @@ func TestAdminRequestAlwaysCarriesRequestID(t *testing.T) {
 
 // An unmapped code must not escape as a 200 or an empty body.
 func TestWriteAdminErrorFallsBackOnUnknownCode(t *testing.T) {
-	gw := &GatewayConfig{DisableLogging: true}
+	gw := &Config{DisableLogging: true}
 	rec := httptest.NewRecorder()
 
 	gw.writeAdminError(rec, "req-1", "NotARegisteredCode", "")
@@ -106,7 +106,7 @@ func TestWriteAdminErrorFallsBackOnUnknownCode(t *testing.T) {
 // The router serves every verb from one pattern, so the handler is what
 // refuses a GET — a second route would shadow the POST one.
 func TestAdminRequestRejectsNonPost(t *testing.T) {
-	gw := withOperator(withRealAuth(&GatewayConfig{DisableLogging: true, IAMService: allowAllIAMService()}))
+	gw := withOperator(withRealAuth(gwfix.WithDeps(&Config{DisableLogging: true}, gwfix.Deps{IAM: allowAllIAMService()})))
 	req := httptest.NewRequest(http.MethodGet, "/admin/CreateAccount", nil)
 	rctx := chi.NewRouteContext()
 	rctx.URLParams.Add("method", "CreateAccount")
@@ -121,13 +121,13 @@ func TestAdminRequestRejectsNonPost(t *testing.T) {
 
 // authorizedAdminRequest builds a request that clears every authorization gate,
 // so the guards behind them can be tested on their own.
-func authorizedAdminRequest(gw *GatewayConfig, body string) *httptest.ResponseRecorder {
+func authorizedAdminRequest(gw *Config, body string) *httptest.ResponseRecorder {
 	return authorizedAdminRequestForMethod(gw, "CreateAccount", body)
 }
 
 // authorizedAdminRequestForMethod is the same for a named method, so a grant
 // that covers one method can be shown not to cover another.
-func authorizedAdminRequestForMethod(gw *GatewayConfig, method, body string) *httptest.ResponseRecorder {
+func authorizedAdminRequestForMethod(gw *Config, method, body string) *httptest.ResponseRecorder {
 	req := httptest.NewRequest(http.MethodPost, "/admin/"+method, strings.NewReader(body))
 	rctx := chi.NewRouteContext()
 	rctx.URLParams.Add("method", method)
@@ -146,7 +146,7 @@ func authorizedAdminRequestForMethod(gw *GatewayConfig, method, body string) *ht
 // An authorized caller must still get a clear answer when the cluster is not
 // reachable, rather than a panic or a silent success.
 func TestAdminRequestReportsClusterUnavailable(t *testing.T) {
-	gw := withOperator(withRealAuth(&GatewayConfig{DisableLogging: true, IAMService: createAccountPolicy()}))
+	gw := withOperator(withRealAuth(gwfix.WithDeps(&Config{DisableLogging: true}, gwfix.Deps{IAM: createAccountPolicy()})))
 
 	rec := authorizedAdminRequest(gw, `{}`)
 
@@ -156,7 +156,7 @@ func TestAdminRequestReportsClusterUnavailable(t *testing.T) {
 // Each admin method is a separate grant. The signup Worker holds
 // spinifex:CreateAccount and must not reach deletion with it.
 func TestCreateAccountGrantDoesNotAllowDeleteAccount(t *testing.T) {
-	gw := withOperator(withRealAuth(&GatewayConfig{DisableLogging: true, IAMService: createAccountPolicy()}))
+	gw := withOperator(withRealAuth(gwfix.WithDeps(&Config{DisableLogging: true}, gwfix.Deps{IAM: createAccountPolicy()})))
 
 	for _, method := range []string{"DeleteAccount", "DescribeAccountDeletion", "ListAccounts"} {
 		t.Run(method, func(t *testing.T) {

spinifex/ingress/aws/pipeline/auth.go

@@ -1,4 +1,4 @@
-package gateway
+package pipeline
 
 import (
 	"bytes"
@@ -31,7 +31,7 @@ type authSource struct {
 }
 
 // SigV4AuthMiddleware returns stdlib middleware that validates AWS Signature V4 authentication.
-func (gw *GatewayConfig) SigV4AuthMiddleware() func(http.Handler) http.Handler {
+func (gw *Config) SigV4AuthMiddleware() func(http.Handler) http.Handler {
 	if gw.RateLimiter == nil {
 		gw.RateLimiter = authlimit.NewAuthRateLimiter()
 	}
@@ -148,7 +148,7 @@ func (gw *GatewayConfig) SigV4AuthMiddleware() func(http.Handler) http.Handler {
 					slog.Warn("Auth failure: verification failed",
 						"accessKeyID", sig.Credential.AccessKeyID, "sourceIP", src.logIP,
 						"service", sig.Credential.Service,
-						"action", mismatchAction(r, sig.Credential.Service),
+						"action", gw.mismatchAction(r, sig.Credential.Service),
 						"canonicalRequest", sig.RedactedCanonicalRequest(),
 						"err", sigErr)
 					// Fingerprinted by the signature as well as the key id: each guess at a
@@ -248,9 +248,9 @@ func signingTime(r *http.Request) string {
 
 // mismatchAction recovers the Action for a rejected query-protocol request so the log names
 // the operation. Auth fails before the dispatcher parses it, which otherwise leaves the
-// trace as a bare "POST /". S3 is skipped: its body is object data, not query args.
-func mismatchAction(r *http.Request, service string) string {
-	if service == "s3" {
+// trace as a bare "POST /". A scope whose body is object data is skipped.
+func (gw *Config) mismatchAction(r *http.Request, service string) string {
+	if traits, ok := gw.selector().Traits(service); ok && traits.PayloadBody {
 		return ""
 	}
 
@@ -274,7 +274,7 @@ func mismatchAction(r *http.Request, service string) string {
 }
 
 // recordCredentialFailure counts a verifier rejection toward the source's lockout.
-func (gw *GatewayConfig) recordCredentialFailure(src authSource, f *credential.Failure) {
+func (gw *Config) recordCredentialFailure(src authSource, f *credential.Failure) {
 	if f.FingerprintKind == "" {
 		return
 	}
@@ -287,7 +287,7 @@ func (gw *GatewayConfig) recordCredentialFailure(src authSource, f *credential.F
 // request. message overrides the registry text from ErrorLookup when the
 // call site knows a more truthful reason to give; pass "" to use the
 // registry text unchanged.
-func (gw *GatewayConfig) writeSigV4Error(w http.ResponseWriter, r *http.Request, errorCode, message string) {
+func (gw *Config) writeSigV4Error(w http.ResponseWriter, r *http.Request, errorCode, message string) {
 	requestID := uuid.NewV4().String()
 	audit.From(r.Context()).SetAuthError(errorCode)
 
@@ -328,7 +328,7 @@ func (gw *GatewayConfig) writeSigV4Error(w http.ResponseWriter, r *http.Request,
 		return
 	}
 
-	xmlError := xmlErrorBody(svc, errorCode, errorMsg.Message, requestID, r.URL.Path)
+	xmlError := gw.xmlErrorBody(svc, errorCode, errorMsg.Message, requestID, r.URL.Path)
 
 	w.Header().Set("Content-Type", envelope.XMLContentType)
 	w.WriteHeader(errorMsg.HTTPCode)
@@ -357,19 +357,13 @@ func signedService(r *http.Request) string {
 	return reqctx.Service(r.Context())
 }
 
-// globalSigningRegions maps a global AWS service to the region its clients sign
-// with when they use the default endpoint. A client given an endpoint override
-// signs with its own configured region instead, so both are accepted.
-var globalSigningRegions = map[string]string{
-	"route53": "us-east-1",
-}
-
 // expectedSigningRegion returns the credential-scope region a request for
 // service must carry: the global service's signing region when the client
-// claimed it, else the gateway's own region.
-func (gw *GatewayConfig) expectedSigningRegion(service, claimed string) string {
-	if global, ok := globalSigningRegions[service]; ok && claimed == global {
-		return global
+// claimed it, else the gateway's own region. A client given an endpoint
+// override signs with its own configured region, so both are accepted.
+func (gw *Config) expectedSigningRegion(service, claimed string) string {
+	if traits, ok := gw.selector().Traits(service); ok && traits.GlobalSigningRegion != "" && claimed == traits.GlobalSigningRegion {
+		return traits.GlobalSigningRegion
 	}
 	return gw.Region
 }

spinifex/ingress/aws/pipeline/auth_accountstatus.go

@@ -1,4 +1,4 @@
-package gateway
+package pipeline
 
 import (
 	"errors"
@@ -12,7 +12,7 @@ import (
 //
 // This is the only enforcement point for account status on the AWS surface:
 // until it existed, suspending an account blocked ECR and nothing else.
-func (gw *GatewayConfig) checkAccountActive(accountID string, src authSource) string {
+func (gw *Config) checkAccountActive(accountID string, src authSource) string {
 	if gw.Admission == nil {
 		return awserrors.ErrorInternalError
 	}

spinifex/ingress/aws/pipeline/auth_accountstatus_test.go

@@ -1,4 +1,4 @@
-package gateway
+package pipeline
 
 //test:in-package — the status gate is an unexported step of the auth
 // middleware, and its cache is deliberately not part of the package's API.
@@ -19,7 +19,7 @@ import (
 func statusGateRouter(t *testing.T, svc iamdomain.IAMService) (chi.Router, *http.Request) {
 	t.Helper()
 
-	gw := withEC2(withRealAuth(&GatewayConfig{DisableLogging: true, Region: gwfix.Region, IAMService: svc}))
+	gw := withEC2(withRealAuth(gwfix.WithDeps(&Config{DisableLogging: true, Region: gwfix.Region}, gwfix.Deps{IAM: svc})))
 	r := chi.NewRouter()
 	r.Use(gw.SigV4AuthMiddleware())
 	r.HandleFunc("/*", func(w http.ResponseWriter, _ *http.Request) { w.WriteHeader(http.StatusOK) })

spinifex/ingress/aws/pipeline/auth_current_behaviour_test.go

@@ -1,4 +1,4 @@
-package gateway
+package pipeline
 
 //test:in-package — this characterization drives unexported gateway wiring.
 
@@ -125,14 +125,10 @@ func newCBEnv(t *testing.T, opts cbEnvOpts) *cbEnv {
 
 // gateway builds a router over the env's services with its own (cold) caches.
 func (e *cbEnv) gateway() http.Handler {
-	gw := &GatewayConfig{
+	gw := gwfix.WithDeps(&Config{
 		DisableLogging: true,
-		NATSConn:       e.nc,
 		Region:         gwfix.Region,
-		ExpectedNodes:  1,
-		IAMService:     e.counter,
-		STSService:     e.sts,
-	}
+	}, gwfix.Deps{NATS: e.nc, Nodes: 1, IAM: e.counter, STS: e.sts})
 	withIAMAndSTS(withRealAuth(gw))
 	return gw.SetupRoutes()
 }

spinifex/ingress/aws/pipeline/auth_silentmismatch_test.go

@@ -1,4 +1,4 @@
-package gateway
+package pipeline
 
 import (
 	"bytes"
@@ -99,14 +99,12 @@ func TestSigV4Auth_FailureLogsLoopbackGatedClientIP(t *testing.T) {
 		t.Run(tc.name, func(t *testing.T) {
 			rl := authlimit.NewAuthRateLimiter()
 			defer rl.Stop()
-			gw := &GatewayConfig{
+			gw := gwfix.WithDeps(&Config{
 				DisableLogging: true,
 				Region:         gwfix.Region,
-				IAMService:     &gwfix.IAM{MasterKey: gwfix.MasterKey},
 				Credentials:    credverify.New(&gwfix.IAM{MasterKey: gwfix.MasterKey}, &gwfix.STS{}),
-				STSService:     &gwfix.STS{},
 				RateLimiter:    rl,
-			}
+			}, gwfix.Deps{IAM: &gwfix.IAM{MasterKey: gwfix.MasterKey}, STS: &gwfix.STS{}})
 			withEC2(gw)
 			handler := gw.SigV4AuthMiddleware()(http.HandlerFunc(func(http.ResponseWriter, *http.Request) {
 				t.Fatal("an unknown session credential must not reach the handler")