test: require a client failure after the fault before the recovery counts

The reservation callback now resets its admission state on every CAS retry,
leadership is checked before the reservation and again before dispatch, and a
suspension left by a failed resume converges through the heartbeat of the VM
the load balancer names, with a 30-minute backstop. An established failure
clears only on a clean probe, and a heartbeat never rebinds a bound record.

build/microvm/init.sh

@@ -32,6 +32,8 @@ load_mod qemu_fw_cfg || load_mod fw_cfg_sysfs || true
 load_mod failover || true
 load_mod net_failover || true
 load_mod virtio_net || true
+# lb-agent probes each data NIC's gateway on an AF_PACKET socket.
+load_mod af_packet || true
 
 NETCFG=/sys/firmware/qemu_fw_cfg/by_name/opt/spinifex/netcfg/raw
 i=0

docs/testing/e2e-coverage.md

@@ -108,6 +108,8 @@ Three entry points, in the order they run:
 
 **Its own suite and its own cell, for instancerecovery's reasons plus one.** It takes a node away, so anything sharing the environment reports that outage as its own failure; it needs its own VPC, public subnet and a backend placement no shared fixture describes; and a failure here has to read as "the load balancer did not survive its host" rather than as one more red subtest in a suite about load balancer features.
 
+**A dead data NIC is the other half, and there the state does change.** `LBSelfHealsADeadDataNIC` disables one data NIC of a running load balancer guest through a second QMP client and runs for an internet-facing NLB and an internal ALB. The backends and an unrelated customer instance in the same subnet are launched more than 21 minutes before the fault, so a repair that reached a customer instance would find one the platform has long stopped expecting to change. A state watch samples `DescribeLoadBalancers` every two seconds and requires the sequence `active`, `active_impaired`, `active` with nothing in between, while the internet-facing run also watches the name continuously and serves traffic through it within five minutes. Afterwards the guest, its interfaces and its address are unchanged but its process is not, the backends and the bystander keep their instance ID, launch time and process (ADR-0003:S7), and the daemon journals exactly one `ELBv2 data-path repair started` line carrying the fields an operator needs. The internal ALB has no client inside the VPC, so it asserts the state sequence and target health rather than traffic through its name.
+
 ### `iam` — everything IAM and STS
 
 Policy evaluation against live requests rather than unit-level: a deny applying to the next request without a new credential, resource-scoped grants naming one key pair, a user rename needing `iam:UpdateUser` on both the old and the new name, `aws:SourceIp` seeing the real client address, assumed-role sessions carrying role policies, and revocation stopping one key alone.
@@ -129,7 +131,7 @@ The client reports on its serial console because the runner has no route into th
 
 ### `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 whose spec names only a ServiceAccount with the role-arn annotation, where the pod-identity webhook injects the token volume, `AWS_*` environment, STS endpoint and CA bundle and the pod assumes the role, while a pod of an unannotated ServiceAccount is left untouched. The unmodified aws-load-balancer-controller chart 1.11.0 installs with the values an AWS user passes plus the gateway endpoint override, the Shield and WAF switches and the single-subnet feature gate, and reconciles an Ingress into an active ALB. 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.
+`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 whose spec names only a ServiceAccount with the role-arn annotation, where the pod-identity webhook injects the token volume, `AWS_*` environment, STS endpoint and CA bundle and the pod assumes the role, while a pod of an unannotated ServiceAccount is left untouched. The unmodified aws-load-balancer-controller chart 1.11.0 installs with the values an AWS user passes plus the gateway endpoint override, the Shield and WAF switches and the single-subnet feature gate, and reconciles an Ingress into an active ALB. 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. The private-endpoint NIC of the cluster's endpoint NLB is disabled through QMP and every worker must be Ready again within ten minutes while the cluster stays `ACTIVE` and the NLB keeps its guest instance. 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.
 
 ### `rds` — managed databases
 

scripts/build-microvm-image.sh

@@ -209,6 +209,7 @@ echo "[build-microvm-image] nginx: binary + stream module present"
 #   virtio*.ko*    — virtio bus, net, blk, rng, console, and helpers
 #   *fw_cfg*.ko*   — QEMU fw_cfg sysfs driver (delivers boot blobs)
 #   *failover*.ko* — net_failover + failover (virtio_net deps for live migration)
+#   af_packet.ko*  — AF_PACKET sockets (lb-agent's per-NIC gateway ARP probe)
 # All other loadable modules are removed. Drivers compiled into the kernel
 # (CONFIG_VIRTIO_MMIO=y etc.) are unaffected — they live in vmlinuz, not here.
 echo "[build-microvm-image] stripping kernel modules to virtio+fw_cfg only..."
@@ -222,6 +223,7 @@ for kver_dir in "$CHROOT_DIR/lib/modules"/*/; do
     find "$kernel_dir" -name "virtio*.ko.gz" -o -name "virtio*.ko" \
                        -o -name "*fw_cfg*.ko.gz" -o -name "*fw_cfg*.ko" \
                        -o -name "*failover*.ko.gz" -o -name "*failover*.ko" \
+                       -o -name "af_packet.ko.gz" -o -name "af_packet.ko" \
         2>/dev/null | while read -r mod; do
         rel="${mod#"$kernel_dir/"}"
         dest_dir="$tmp_keep/$(dirname "$rel")"
@@ -240,6 +242,15 @@ for kver_dir in "$CHROOT_DIR/lib/modules"/*/; do
     done)
     rm -rf "$tmp_keep"
 
+    # lb-agent's data-path probe needs AF_PACKET; without it every LB would
+    # report a dead NIC. Accept a kernel with CONFIG_PACKET built in.
+    if ! find "$kernel_dir" -name "af_packet.ko*" | grep -q .; then
+        if ! grep -q "CONFIG_PACKET=y" "$KERNEL_CONFIG" 2>/dev/null; then
+            echo "ERROR: af_packet.ko missing from the kept modules — lb-agent's data-path probe cannot open its socket" >&2
+            exit 1
+        fi
+    fi
+
     # Decompress kept modules to plain .ko. Alpine ships .ko.gz, but a host
     # kmod that lacks gzip support (e.g. Debian trixie's kmod 34) silently
     # produces an EMPTY modules.dep from compressed modules — the guest's

spinifex/agents/elbv2/agent.go

@@ -21,6 +21,7 @@ import (
 	"path/filepath"
 	"strconv"
 	"strings"
+	"sync"
 	"syscall"
 	"time"
 
@@ -94,6 +95,18 @@ type Agent struct {
 	statsFn func(socketPath string) ([]ServerStatus, error)
 	// For testing: override the active health prober (nginx/NLB).
 	probeFn func(targets []HealthTarget) []ServerStatus
+
+	// Data-path probe: netcfg and sysfs locations, the ARP exchange, the
+	// instance ID lookup and per-NIC miss counters (overridable in tests).
+	netcfgPath   string
+	sysNetDir    string
+	probeWait    time.Duration
+	arpProbeFn   arpProbeFn
+	instanceIDFn func() (string, error)
+	instanceID   string
+	idSkip       int
+	probeMu      sync.Mutex
+	probeStates  map[string]*nicProbeState
 }
 
 // New creates a new LB agent for the given load balancer.
@@ -150,6 +163,12 @@ func New(lbID, gatewayURL, accessKey, secretKey, region string) (*Agent, error)
 		reloadNginxFn: reloadNginx,
 		statsFn:       queryHAProxyStats,
 		probeFn:       probeHealthTargets,
+		netcfgPath:    netcfgPath,
+		sysNetDir:     sysClassNet,
+		probeWait:     probeWait,
+		arpProbeFn:    arpProbe,
+		instanceIDFn:  fetchInstanceID,
+		probeStates:   map[string]*nicProbeState{},
 	}, nil
 }
 
@@ -207,7 +226,7 @@ func (a *Agent) tick() {
 		}
 	}
 
-	resp, err := a.sendHeartbeat(servers)
+	resp, err := a.sendHeartbeat(servers, a.dataPathEvidence())
 	if err != nil {
 		// Include the dial target so the serial console disambiguates a
 		// transport failure (never reached AWSGW — "send request: dial ...")
@@ -288,7 +307,7 @@ func (a *Agent) enginePaths(engine string) (configPath, pidPath, certDir string,
 }
 
 // sendHeartbeat sends a heartbeat with health report to the gateway.
-func (a *Agent) sendHeartbeat(servers []ServerStatus) (*heartbeatResponse, error) {
+func (a *Agent) sendHeartbeat(servers []ServerStatus, ev dataPathEvidence) (*heartbeatResponse, error) {
 	params := url.Values{}
 	params.Set("Action", "LBAgentHeartbeat")
 	params.Set("Version", "2015-12-01")
@@ -300,6 +319,7 @@ func (a *Agent) sendHeartbeat(servers []ServerStatus) (*heartbeatResponse, error
 		params.Set("Servers.member."+idx+".Server", s.Server)
 		params.Set("Servers.member."+idx+".Status", s.Status)
 	}
+	ev.addTo(params)
 
 	body, err := a.signedPost(params)
 	if err != nil {

spinifex/agents/elbv2/arp_linux.go

@@ -0,0 +1,62 @@
+//go:build linux
+
+package elbv2
+
+import (
+	"errors"
+	"fmt"
+	"net"
+	"net/netip"
+	"syscall"
+	"time"
+)
+
+// arpProbe sends one ARP request for gateway out of iface on an AF_PACKET
+// socket bound to it and waits for the gateway's reply on the same socket.
+func arpProbe(iface string, mac net.HardwareAddr, src, gateway netip.Addr, wait time.Duration) error {
+	ifi, err := net.InterfaceByName(iface)
+	if err != nil {
+		return fmt.Errorf("interface %s: %w", iface, err)
+	}
+
+	proto := htons(syscall.ETH_P_ARP)
+	fd, err := syscall.Socket(syscall.AF_PACKET, syscall.SOCK_RAW, int(proto))
+	if err != nil {
+		return fmt.Errorf("packet socket: %w", err)
+	}
+	defer syscall.Close(fd)
+
+	addr := &syscall.SockaddrLinklayer{Protocol: proto, Ifindex: ifi.Index}
+	if err := syscall.Bind(fd, addr); err != nil {
+		return fmt.Errorf("bind %s: %w", iface, err)
+	}
+
+	deadline := time.Now().Add(wait)
+	if err := syscall.Sendto(fd, buildARPRequest(mac, src, gateway), 0, addr); err != nil {
+		return fmt.Errorf("send ARP request: %w", err)
+	}
+
+	buf := make([]byte, 128)
+	for {
+		left := time.Until(deadline)
+		if left <= 0 {
+			return errARPNoReply
+		}
+		tv := syscall.NsecToTimeval(int64(left))
+		if err := syscall.SetsockoptTimeval(fd, syscall.SOL_SOCKET, syscall.SO_RCVTIMEO, &tv); err != nil {
+			return fmt.Errorf("set receive deadline: %w", err)
+		}
+		n, _, err := syscall.Recvfrom(fd, buf, 0)
+		if err != nil {
+			if errors.Is(err, syscall.EAGAIN) || errors.Is(err, syscall.EINTR) {
+				continue
+			}
+			return fmt.Errorf("receive ARP reply: %w", err)
+		}
+		if isARPReplyFrom(buf[:n], mac, gateway) {
+			return nil
+		}
+	}
+}
+
+func htons(v uint16) uint16 { return v<<8 | v>>8 }

spinifex/agents/elbv2/arp_other.go

@@ -0,0 +1,15 @@
+//go:build !linux
+
+package elbv2
+
+import (
+	"errors"
+	"net"
+	"net/netip"
+	"time"
+)
+
+// arpProbe is only meaningful inside the Linux guest.
+func arpProbe(string, net.HardwareAddr, netip.Addr, netip.Addr, time.Duration) error {
+	return errors.New("ARP probe is only supported on linux")
+}

spinifex/agents/elbv2/datapath.go

@@ -0,0 +1,303 @@
+package elbv2
+
+import (
+	"bufio"
+	"bytes"
+	"encoding/binary"
+	"errors"
+	"fmt"
+	"net"
+	"net/netip"
+	"os"
+	"path/filepath"
+	"sort"
+	"strconv"
+	"strings"
+	"sync"
+	"time"
+)
+
+const (
+	// netcfgPath is the fw_cfg blob the microVM init script reads to configure NICs.
+	netcfgPath = "/sys/firmware/qemu_fw_cfg/by_name/opt/spinifex/netcfg/raw"
+
+	sysClassNet = "/sys/class/net"
+
+	// probeFailureThreshold is the number of consecutive missed probes after
+	// which a NIC is reported as failing (about 15 s at the poll interval).
+	probeFailureThreshold = 3
+
+	// arpExchangeWait bounds one ARP exchange. probeWait bounds how long a tick
+	// waits for all probes, so a stuck probe never delays the heartbeat.
+	arpExchangeWait = time.Second
+	probeWait       = 1500 * time.Millisecond
+)
+
+// nicConfig is one NIC of the netcfg blob that has a gateway to probe.
+type nicConfig struct {
+	MAC     string
+	Addr    netip.Addr
+	Gateway netip.Addr
+}
+
+// NICReport is the per-NIC data-path evidence carried in the heartbeat.
+type NICReport struct {
+	MAC                 string
+	ProbeOK             bool
+	ConsecutiveFailures int64
+	RxPackets           int64
+	TxPackets           int64
+	RxDropped           int64
+	TxDropped           int64
+}
+
+// arpProbeFn sends one ARP request for gateway out of iface and returns nil
+// when the gateway's reply arrives within timeout.
+type arpProbeFn func(iface string, mac net.HardwareAddr, src, gateway netip.Addr, timeout time.Duration) error
+
+// parseNetcfg reads the KEY=value netcfg blob and returns the NICs that have a
+// gateway, in NIC order. The mgmt NIC carries no gateway and is skipped, as
+// is any NIC without a usable MAC, address or gateway.
+func parseNetcfg(blob []byte) []nicConfig {
+	type raw struct{ mac, cidr, gw string }
+	byIndex := map[int]*raw{}
+
+	sc := bufio.NewScanner(bytes.NewReader(blob))
+	for sc.Scan() {
+		line := strings.TrimSpace(sc.Text())
+		key, val, ok := strings.Cut(line, "=")
+		if !ok || !strings.HasPrefix(key, "NIC") {
+			continue
+		}
+		idxStr, field, ok := strings.Cut(strings.TrimPrefix(key, "NIC"), "_")
+		if !ok {
+			continue
+		}
+		idx, err := strconv.Atoi(idxStr)
+		if err != nil {
+			continue
+		}
+		r := byIndex[idx]
+		if r == nil {
+			r = &raw{}
+			byIndex[idx] = r
+		}
+		switch field {
+		case "MAC":
+			r.mac = strings.ToLower(val)
+		case "CIDR":
+			r.cidr = val
+		case "GW":
+			r.gw = val
+		}
+	}
+
+	idxs := make([]int, 0, len(byIndex))
+	for i := range byIndex {
+		idxs = append(idxs, i)
+	}
+	sort.Ints(idxs)
+
+	var out []nicConfig
+	for _, i := range idxs {
+		r := byIndex[i]
+		if r.mac == "" || r.cidr == "" || r.gw == "" {
+			continue
+		}
+		prefix, err := netip.ParsePrefix(r.cidr)
+		if err != nil || !prefix.Addr().Is4() {
+			continue
+		}
+		gw, err := netip.ParseAddr(r.gw)
+		if err != nil || !gw.Is4() {
+			continue
+		}
+		out = append(out, nicConfig{MAC: r.mac, Addr: prefix.Addr(), Gateway: gw})
+	}
+	return out
+}
+
+// ifaceByMAC returns the interface under sysNetDir whose address is mac.
+func ifaceByMAC(sysNetDir, mac string) (string, error) {
+	entries, err := os.ReadDir(sysNetDir)
+	if err != nil {
+		return "", err
+	}
+	for _, e := range entries {
+		b, err := os.ReadFile(filepath.Join(sysNetDir, e.Name(), "address"))
+		if err != nil {
+			continue
+		}
+		if strings.EqualFold(strings.TrimSpace(string(b)), mac) {
+			return e.Name(), nil
+		}
+	}
+	return "", fmt.Errorf("no interface with MAC %s", mac)
+}
+
+// readCounters fills the sysfs statistics of iface into r.
+func readCounters(sysNetDir, iface string, r *NICReport) {
+	read := func(name string) int64 {
+		b, err := os.ReadFile(filepath.Join(sysNetDir, iface, "statistics", name))
+		if err != nil {
+			return 0
+		}
+		v, _ := strconv.ParseInt(strings.TrimSpace(string(b)), 10, 64)
+		return v
+	}
+	r.RxPackets = read("rx_packets")
+	r.TxPackets = read("tx_packets")
+	r.RxDropped = read("rx_dropped")
+	r.TxDropped = read("tx_dropped")
+}
+
+var errARPNoReply = errors.New("no ARP reply from the gateway")
+
+// buildARPRequest returns the Ethernet frame asking who has target.
+func buildARPRequest(mac net.HardwareAddr, src, target netip.Addr) []byte {
+	f := make([]byte, 0, 42)
+	f = append(f, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff)
+	f = append(f, mac...)
+	f = binary.BigEndian.AppendUint16(f, 0x0806)
+	f = binary.BigEndian.AppendUint16(f, 1)      // Ethernet
+	f = binary.BigEndian.AppendUint16(f, 0x0800) // IPv4
+	f = append(f, 6, 4)
+	f = binary.BigEndian.AppendUint16(f, 1) // request
+	f = append(f, mac...)
+	s := src.As4()
+	f = append(f, s[:]...)
+	f = append(f, 0, 0, 0, 0, 0, 0)
+	t := target.As4()
+	f = append(f, t[:]...)
+	return f
+}
+
+// isARPReplyFrom reports whether frame is an ARP reply sent by sender to mac.
+func isARPReplyFrom(frame []byte, mac net.HardwareAddr, sender netip.Addr) bool {
+	if len(frame) < 42 || binary.BigEndian.Uint16(frame[12:14]) != 0x0806 {
+		return false
+	}
+	if binary.BigEndian.Uint16(frame[20:22]) != 2 || !bytes.Equal(frame[32:38], mac) {
+		return false
+	}
+	s := sender.As4()
+	return bytes.Equal(frame[28:32], s[:])
+}
+
+// nicProbeState carries the miss counter of one NIC across ticks. A probe that
+// has not returned is still in flight, and counts as a miss without starting
+// another goroutine behind it.
+type nicProbeState struct {
+	mu       sync.Mutex
+	failures int64
+	inflight bool
+}
+
+type probeResult struct {
+	ok  bool
+	set bool
+}
+
+// probeNICs probes every gateway-bearing data NIC once, in parallel, and
+// returns within probeWait however long a probe blocks.
+func (a *Agent) probeNICs() []NICReport {
+	blob, err := os.ReadFile(a.netcfgPath)
+	if err != nil {
+		return nil
+	}
+	nics := parseNetcfg(blob)
+	if len(nics) == 0 {
+		return nil
+	}
+
+	reports := make([]NICReport, len(nics))
+	results := make([]chan probeResult, len(nics))
+	for i, n := range nics {
+		results[i] = make(chan probeResult, 1)
+		st := a.probeState(n.MAC)
+
+		st.mu.Lock()
+		busy := st.inflight
+		if !busy {
+			st.inflight = true
+		}
+		st.mu.Unlock()
+		if busy {
+			results[i] <- probeResult{set: true}
+			continue
+		}
+
+		go func(n nicConfig, st *nicProbeState, out chan<- probeResult) {
+			ok := a.probeOne(n)
+			st.mu.Lock()
+			st.inflight = false
+			st.mu.Unlock()
+			out <- probeResult{ok: ok, set: true}
+		}(n, st, results[i])
+	}
+
+	deadline := time.NewTimer(a.probeWait)
+	defer deadline.Stop()
+	expired := false
+	for i, n := range nics {
+		var res probeResult
+		if expired {
+			select {
+			case res = <-results[i]:
+			default:
+			}
+		} else {
+			select {
+			case res = <-results[i]:
+			case <-deadline.C:
+				// Out of time: an unfinished probe is a miss for this tick.
+				expired = true
+			}
+		}
+
+		st := a.probeState(n.MAC)
+		st.mu.Lock()
+		if res.ok {
+			st.failures = 0
+		} else {
+			st.failures++
+		}
+		failures := st.failures
+		st.mu.Unlock()
+
+		r := NICReport{
+			MAC:                 n.MAC,
+			ProbeOK:             failures < probeFailureThreshold,
+			ConsecutiveFailures: failures,
+		}
+		if iface, err := ifaceByMAC(a.sysNetDir, n.MAC); err == nil {
+			readCounters(a.sysNetDir, iface, &r)
+		}
+		reports[i] = r
+	}
+	return reports
+}
+
+// probeOne runs one ARP exchange for n, false on any failure.
+func (a *Agent) probeOne(n nicConfig) bool {
+	iface, err := ifaceByMAC(a.sysNetDir, n.MAC)
+	if err != nil {
+		return false
+	}
+	hw, err := net.ParseMAC(n.MAC)
+	if err != nil {
+		return false
+	}
+	return a.arpProbeFn(iface, hw, n.Addr, n.Gateway, arpExchangeWait) == nil
+}
+
+func (a *Agent) probeState(mac string) *nicProbeState {
+	a.probeMu.Lock()
+	defer a.probeMu.Unlock()
+	st := a.probeStates[mac]
+	if st == nil {
+		st = &nicProbeState{}
+		a.probeStates[mac] = st
+	}
+	return st
+}

spinifex/agents/elbv2/datapath_test.go

@@ -0,0 +1,484 @@
+package elbv2
+
+import (
+	"errors"
+	"fmt"
+	"net"
+	"net/http"
+	"net/http/httptest"
+	"net/netip"
+	"net/url"
+	"os"
+	"path/filepath"
+	"sync"
+	"sync/atomic"
+	"testing"
+	"time"
+)
+
+const (
+	testDataMAC = "02:00:00:aa:00:01"
+	testMgmtMAC = "02:00:00:aa:00:02"
+	testNetcfg  = "NIC0_MAC=" + testDataMAC + "\nNIC0_CIDR=10.0.1.5/24\nNIC0_GW=10.0.1.1\nNIC0_DEFAULT=1\n" +
+		"NIC1_MAC=" + testMgmtMAC + "\nNIC1_CIDR=10.255.0.9/24\nNIC1_DEFAULT=0\n"
+)
+
+// datapathAgent returns a test agent whose netcfg and sysfs live in a temp dir.
+func datapathAgent(t *testing.T, gwURL, netcfg string) *Agent {
+	t.Helper()
+	a := newTestAgent(t, gwURL)
+	dir := t.TempDir()
+
+	a.netcfgPath = filepath.Join(dir, "netcfg")
+	if err := os.WriteFile(a.netcfgPath, []byte(netcfg), 0o600); err != nil {
+		t.Fatal(err)
+	}
+
+	a.sysNetDir = filepath.Join(dir, "net")
+	for name, mac := range map[string]string{"ens3": testDataMAC, "ens4": testMgmtMAC} {
+		stats := filepath.Join(a.sysNetDir, name, "statistics")
+		if err := os.MkdirAll(stats, 0o755); err != nil {
+			t.Fatal(err)
+		}
+		if err := os.WriteFile(filepath.Join(a.sysNetDir, name, "address"), []byte(mac+"\n"), 0o600); err != nil {
+			t.Fatal(err)
+		}
+		for file, v := range map[string]string{"rx_packets": "100", "tx_packets": "200", "rx_dropped": "3", "tx_dropped": "4"} {
+			if err := os.WriteFile(filepath.Join(stats, file), []byte(v+"\n"), 0o600); err != nil {
+				t.Fatal(err)
+			}
+		}
+	}
+
+	a.instanceIDFn = func() (string, error) { return "i-0123456789abcdef0", nil }
+	return a
+}
+
+func TestParseNetcfg_SkipsMgmtNICAndMissingGateway(t *testing.T) {
+	blob := testNetcfg +
+		"NIC2_MAC=02:00:00:aa:00:03\nNIC2_CIDR=10.0.2.5/24\nNIC2_GW=10.0.2.1\nNIC2_DEFAULT=0\n" +
+		"NIC3_MAC=02:00:00:aa:00:04\nNIC3_DHCP=1\nNIC3_DEFAULT=0\n" +
+		"NIC4_CIDR=10.0.4.5/24\nNIC4_GW=10.0.4.1\n" +
+		"NIC5_MAC=02:00:00:aa:00:06\nNIC5_CIDR=not-a-cidr\nNIC5_GW=10.0.5.1\n"
+
+	got := parseNetcfg([]byte(blob))
+	if len(got) != 2 {
+		t.Fatalf("got %d NICs, want the 2 with a gateway: %+v", len(got), got)
+	}
+	if got[0].MAC != testDataMAC || got[0].Gateway != netip.MustParseAddr("10.0.1.1") || got[0].Addr != netip.MustParseAddr("10.0.1.5") {
+		t.Errorf("first NIC = %+v", got[0])
+	}
+	if got[1].MAC != "02:00:00:aa:00:03" || got[1].Gateway != netip.MustParseAddr("10.0.2.1") {
+		t.Errorf("second NIC = %+v", got[1])
+	}
+}
+
+func TestParseNetcfg_OrdersByNICIndexAndLowercasesMAC(t *testing.T) {
+	blob := "NIC10_MAC=02:00:00:AA:00:0A\nNIC10_CIDR=10.0.9.5/24\nNIC10_GW=10.0.9.1\n" +
+		"NIC2_MAC=02:00:00:AA:00:02\nNIC2_CIDR=10.0.2.5/24\nNIC2_GW=10.0.2.1\n"
+	got := parseNetcfg([]byte(blob))
+	if len(got) != 2 || got[0].MAC != "02:00:00:aa:00:02" || got[1].MAC != "02:00:00:aa:00:0a" {
+		t.Fatalf("order or case wrong: %+v", got)
+	}
+}
+
+func TestIfaceByMACAndCounters(t *testing.T) {
+	a := datapathAgent(t, "http://unused", testNetcfg)
+
+	iface, err := ifaceByMAC(a.sysNetDir, "02:00:00:AA:00:01")
+	if err != nil || iface != "ens3" {
+		t.Fatalf("iface = %q, err = %v", iface, err)
+	}
+	if _, err := ifaceByMAC(a.sysNetDir, "02:00:00:ff:ff:ff"); err == nil {
+		t.Error("an unknown MAC must not resolve to an interface")
+	}
+
+	var r NICReport
+	readCounters(a.sysNetDir, iface, &r)
+	if r.RxPackets != 100 || r.TxPackets != 200 || r.RxDropped != 3 || r.TxDropped != 4 {
+		t.Errorf("counters = %+v", r)
+	}
+}
+
+func TestARPFrames_RequestShapeAndReplyMatching(t *testing.T) {
+	mac, _ := net.ParseMAC(testDataMAC)
+	src, gw := netip.MustParseAddr("10.0.1.5"), netip.MustParseAddr("10.0.1.1")
+
+	req := buildARPRequest(mac, src, gw)
+	if len(req) != 42 || req[12] != 0x08 || req[13] != 0x06 || req[21] != 1 {
+		t.Fatalf("request frame malformed: % x", req)
+	}
+	if string(req[38:42]) != string(gw.AsSlice()) {
+		t.Errorf("target IP not the gateway: % x", req[38:42])
+	}
+
+	reply := make([]byte, 42)
+	copy(reply[0:6], mac)
+	reply[12], reply[13] = 0x08, 0x06
+	reply[21] = 2
+	copy(reply[28:32], gw.AsSlice())
+	copy(reply[32:38], mac)
+	if !isARPReplyFrom(reply, mac, gw) {
+		t.Error("a reply from the gateway to our MAC must match")
+	}
+	if isARPReplyFrom(reply, mac, netip.MustParseAddr("10.0.1.99")) {
+		t.Error("a reply from another host must not match")
+	}
+	reply[21] = 1
+	if isARPReplyFrom(reply, mac, gw) {
+		t.Error("a request must not count as a reply")
+	}
+	if isARPReplyFrom(reply[:20], mac, gw) {
+		t.Error("a short frame must not match")
+	}
+}
+
+func TestProbeNICs_ProbesOnlyTheGatewayNIC(t *testing.T) {
+	a := datapathAgent(t, "http://unused", testNetcfg)
+	var mu sync.Mutex
+	var probed []string
+	a.arpProbeFn = func(iface string, _ net.HardwareAddr, src, gw netip.Addr, _ time.Duration) error {
+		mu.Lock()
+		defer mu.Unlock()
+		probed = append(probed, fmt.Sprintf("%s %s>%s", iface, src, gw))
+		return nil
+	}
+
+	reports := a.probeNICs()
+	if len(reports) != 1 || reports[0].MAC != testDataMAC || !reports[0].ProbeOK || reports[0].ConsecutiveFailures != 0 {
+		t.Fatalf("reports = %+v", reports)
+	}
+	if len(probed) != 1 || probed[0] != "ens3 10.0.1.5>10.0.1.1" {
+		t.Errorf("probed = %v, want only the data NIC", probed)
+	}
+	if reports[0].RxPackets != 100 || reports[0].TxDropped != 4 {
+		t.Errorf("counters missing from report: %+v", reports[0])
+	}
+}
+
+func TestProbeNICs_TwoMissesThenSuccessNeverReportsFailing(t *testing.T) {
+	a := datapathAgent(t, "http://unused", testNetcfg)
+	results := []error{errARPNoReply, errARPNoReply, nil, errARPNoReply, errARPNoReply, nil}
+	var n atomic.Int32
+	a.arpProbeFn = func(string, net.HardwareAddr, netip.Addr, netip.Addr, time.Duration) error {
+		return results[n.Add(1)-1]
+	}
+
+	for i, want := range []int64{1, 2, 0, 1, 2, 0} {
+		r := a.probeNICs()
+		if len(r) != 1 || !r[0].ProbeOK || r[0].ConsecutiveFailures != want {
+			t.Fatalf("tick %d: report = %+v, want ok with %d failures", i, r, want)
+		}
+	}
+}
+
+func TestProbeNICs_ThreeMissesReportFailingAndRecoveryClears(t *testing.T) {
+	a := datapathAgent(t, "http://unused", testNetcfg)
+	var fail atomic.Bool
+	fail.Store(true)
+	a.arpProbeFn = func(string, net.HardwareAddr, netip.Addr, netip.Addr, time.Duration) error {
+		if fail.Load() {
+			return errARPNoReply
+		}
+		return nil
+	}
+
+	for i := 1; i <= 4; i++ {
+		r := a.probeNICs()
+		wantOK := i < 3 // failing from the third consecutive miss
+		if r[0].ProbeOK != wantOK || r[0].ConsecutiveFailures != int64(i) {
+			t.Fatalf("miss %d: report = %+v, want ProbeOK=%v", i, r[0], wantOK)
+		}
+	}
+
+	fail.Store(false)
+	if r := a.probeNICs(); !r[0].ProbeOK || r[0].ConsecutiveFailures != 0 {
+		t.Errorf("a successful probe must clear the failure: %+v", r[0])
+	}
+}
+
+func TestProbeNICs_BlockedProbeIsBoundedAndCountsAsMiss(t *testing.T) {
+	a := datapathAgent(t, "http://unused", testNetcfg)
+	a.probeWait = 50 * time.Millisecond
+	release := make(chan struct{})
+	defer close(release)
+	var calls atomic.Int32
+	a.arpProbeFn = func(string, net.HardwareAddr, netip.Addr, netip.Addr, time.Duration) error {
+		calls.Add(1)
+		<-release
+		return nil
+	}
+
+	for i := 1; i <= 3; i++ {
+		start := time.Now()
+		r := a.probeNICs()
+		if d := time.Since(start); d > time.Second {
+			t.Fatalf("tick %d waited %s on a stuck probe", i, d)
+		}
+		if len(r) != 1 || r[0].ConsecutiveFailures != int64(i) {
+			t.Fatalf("tick %d: report = %+v, want %d misses", i, r, i)
+		}
+	}
+	if got := calls.Load(); got != 1 {
+		t.Errorf("a stuck probe was started %d times, want 1 (no goroutine pile-up)", got)
+	}
+}
+
+func TestTick_BlockedProbeDoesNotDelayHeartbeatBeyondOneTick(t *testing.T) {
+	var got atomic.Pointer[url.Values]
+	gw := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		_ = r.ParseForm()
+		if r.FormValue("Action") == "LBAgentHeartbeat" {
+			v := r.Form
+			got.Store(&v)
+		}
+		fmt.Fprint(w, `<LBAgentHeartbeatResponse><LBAgentHeartbeatResult><Status>active</Status><ConfigHash>h1</ConfigHash></LBAgentHeartbeatResult></LBAgentHeartbeatResponse>`)
+	}))
+	defer gw.Close()
+
+	a := datapathAgent(t, gw.URL, testNetcfg)
+	a.localConfigHash = "h1"
+	a.probeWait = 50 * time.Millisecond
+	release := make(chan struct{})
+	defer close(release)
+	a.arpProbeFn = func(string, net.HardwareAddr, netip.Addr, netip.Addr, time.Duration) error {
+		<-release
+		return nil
+	}
+
+	start := time.Now()
+	a.tick()
+	if d := time.Since(start); d >= pollInterval {
+		t.Fatalf("tick took %s with a stuck probe, want under one %s tick", d, pollInterval)
+	}
+	if got.Load() == nil {
+		t.Fatal("the heartbeat was never sent")
+	}
+}
+
+func TestHeartbeat_EncodesEvidenceWhenPresent(t *testing.T) {
+	var got atomic.Pointer[url.Values]
+	gw := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		_ = r.ParseForm()
+		v := r.Form
+		got.Store(&v)
+		fmt.Fprint(w, `<LBAgentHeartbeatResponse><LBAgentHeartbeatResult><Status>active</Status><ConfigHash>h1</ConfigHash></LBAgentHeartbeatResult></LBAgentHeartbeatResponse>`)
+	}))
+	defer gw.Close()
+
+	a := datapathAgent(t, gw.URL, testNetcfg)
+	a.localConfigHash = "h1"
+	a.arpProbeFn = func(string, net.HardwareAddr, netip.Addr, netip.Addr, time.Duration) error { return errARPNoReply }
+
+	a.tick()
+	v := got.Load()
+	if v == nil {
+		t.Fatal("no heartbeat received")
+	}
+	want := map[string]string{
+		"InstanceID":                        "i-0123456789abcdef0",
+		"NICs.member.1.MAC":                 testDataMAC,
+		"NICs.member.1.ProbeOK":             "true",
+		"NICs.member.1.ConsecutiveFailures": "1",
+		"NICs.member.1.RxPackets":           "100",
+		"NICs.member.1.TxPackets":           "200",
+		"NICs.member.1.RxDropped":           "3",
+		"NICs.member.1.TxDropped":           "4",
+	}
+	for k, w := range want {
+		if g := v.Get(k); g != w {
+			t.Errorf("%s = %q, want %q", k, g, w)
+		}
+	}
+	if v.Get("NICs.member.2.MAC") != "" {
+		t.Error("the mgmt NIC must not be reported")
+	}
+}
+
+func TestHeartbeat_OmitsEvidenceWithoutNetcfg(t *testing.T) {
+	var got atomic.Pointer[url.Values]
+	gw := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		_ = r.ParseForm()
+		v := r.Form
+		got.Store(&v)
+		fmt.Fprint(w, `<LBAgentHeartbeatResponse><LBAgentHeartbeatResult><Status>active</Status><ConfigHash>h1</ConfigHash></LBAgentHeartbeatResult></LBAgentHeartbeatResponse>`)
+	}))
+	defer gw.Close()
+
+	a := datapathAgent(t, gw.URL, testNetcfg)
+	a.localConfigHash = "h1"
+	a.netcfgPath = filepath.Join(t.TempDir(), "absent")
+	a.instanceIDFn = func() (string, error) { return "", errors.New("must not be asked without NICs") }
+
+	a.tick()
+	v := got.Load()
+	if v == nil {
+		t.Fatal("no heartbeat received")
+	}
+	for k := range *v {
+		if k == "InstanceID" || len(k) >= 4 && k[:4] == "NICs" {
+			t.Errorf("unexpected evidence parameter %s", k)
+		}
+	}
+}
+
+func TestDataPathEvidence_InstanceIDRetriedUntilKnown(t *testing.T) {
+	a := datapathAgent(t, "http://unused", testNetcfg)
+	a.arpProbeFn = func(string, net.HardwareAddr, netip.Addr, netip.Addr, time.Duration) error { return nil }
+	var calls atomic.Int32
+	a.instanceIDFn = func() (string, error) {
+		if calls.Add(1) < 3 {
+			return "", errors.New("metadata not ready")
+		}
+		return "i-late", nil
+	}
+
+	// A failure backs the lookup off for imdsRetryTicks-1 ticks.
+	ticks := func(n int) string {
+		var id string
+		for range n {
+			id = a.dataPathEvidence().InstanceID
+		}
+		return id
+	}
+	if got := ticks(imdsRetryTicks); got != "" || calls.Load() != 1 {
+		t.Fatalf("after %d ticks: id %q, calls %d; want no id and one lookup", imdsRetryTicks, got, calls.Load())
+	}
+	if got := ticks(imdsRetryTicks); got != "" || calls.Load() != 2 {
+		t.Fatalf("second window: id %q, calls %d; want no id and two lookups", got, calls.Load())
+	}
+	if got := ticks(imdsRetryTicks); got != "i-late" || calls.Load() != 3 {
+		t.Fatalf("third window: id %q, calls %d; want i-late and three lookups", got, calls.Load())
+	}
+	ticks(imdsRetryTicks)
+	if calls.Load() != 3 {
+		t.Errorf("metadata asked %d times, want 3 (stop once known)", calls.Load())
+	}
+}
+
+func TestTick_BlockedInstanceIDLookupDoesNotDelayHeartbeat(t *testing.T) {
+	var got atomic.Pointer[url.Values]
+	gw := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		_ = r.ParseForm()
+		if r.FormValue("Action") == "LBAgentHeartbeat" {
+			v := r.Form
+			got.Store(&v)
+		}
+		fmt.Fprint(w, `<LBAgentHeartbeatResponse><LBAgentHeartbeatResult><Status>active</Status><ConfigHash>h1</ConfigHash></LBAgentHeartbeatResult></LBAgentHeartbeatResponse>`)
+	}))
+	defer gw.Close()
+
+	a := datapathAgent(t, gw.URL, testNetcfg)
+	a.localConfigHash = "h1"
+	a.arpProbeFn = func(string, net.HardwareAddr, netip.Addr, netip.Addr, time.Duration) error { return nil }
+	release := make(chan struct{})
+	defer close(release)
+	var calls atomic.Int32
+	a.instanceIDFn = func() (string, error) {
+		calls.Add(1)
+		<-release
+		return "", errors.New("unreachable")
+	}
+
+	for i := range 3 {
+		start := time.Now()
+		a.tick()
+		if d := time.Since(start); d >= pollInterval {
+			t.Fatalf("tick %d took %s with IMDS blocked, want under one %s tick", i, d, pollInterval)
+		}
+	}
+	v := got.Load()
+	if v == nil {
+		t.Fatal("the heartbeat was never sent")
+	}
+	if v.Get("InstanceID") != "" {
+		t.Error("InstanceID must be omitted while the lookup fails")
+	}
+	if v.Get("NICs.member.1.MAC") == "" {
+		t.Error("NIC evidence must still be sent")
+	}
+	if calls.Load() != 1 {
+		t.Errorf("blocked lookup started %d times in 3 ticks, want 1 (backoff)", calls.Load())
+	}
+}
+
+func TestFetchInstanceID_SharesOneDeadline(t *testing.T) {
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		time.Sleep(3 * imdsWait)
+	}))
+	defer srv.Close()
+	old := imdsBase
+	imdsBase = srv.URL + "/latest"
+	defer func() { imdsBase = old }()
+
+	start := time.Now()
+	if _, err := fetchInstanceID(); err == nil {
+		t.Fatal("a hung metadata service must fail the lookup")
+	}
+	if d := time.Since(start); d > imdsWait+imdsWait/2 {
+		t.Fatalf("lookup took %s, want about one %s deadline", d, imdsWait)
+	}
+}
+
+func TestARPProbe_NoResponderFailsClosed(t *testing.T) {
+	mac, _ := net.ParseMAC(testDataMAC)
+	// Nobody answers an ARP request sent on loopback, so a probe that returned
+	// nil here would mask a dead data path. It fails either on the socket (no
+	// CAP_NET_RAW) or on the missing reply.
+	err := arpProbe("lo", mac, netip.MustParseAddr("10.9.9.2"), netip.MustParseAddr("10.9.9.1"), 100*time.Millisecond)
+	if err == nil {
+		t.Fatal("a probe with no responder must fail")
+	}
+	if err := arpProbe("no-such-iface0", mac, netip.MustParseAddr("10.9.9.2"), netip.MustParseAddr("10.9.9.1"), 100*time.Millisecond); err == nil {
+		t.Fatal("a probe on a missing interface must fail")
+	}
+}
+
+func TestFetchInstanceID(t *testing.T) {
+	serve := func(tokenStatus, idStatus int, wantToken string) *httptest.Server {
+		return httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+			switch {
+			case r.Method == http.MethodPut && r.URL.Path == "/latest/api/token":
+				w.WriteHeader(tokenStatus)
+				fmt.Fprint(w, "tok123")
+			case r.URL.Path == "/latest/meta-data/instance-id":
+				if got := r.Header.Get("X-Aws-Ec2-Metadata-Token"); got != wantToken {
+					t.Errorf("token header = %q, want %q", got, wantToken)
+				}
+				w.WriteHeader(idStatus)
+				fmt.Fprint(w, " i-abc123 \n")
+			default:
+				http.NotFound(w, r)
+			}
+		}))
+	}
+	saved := imdsBase
+	defer func() { imdsBase = saved }()
+
+	for _, tc := range []struct {
+		name                  string
+		tokenStatus, idStatus int
+		wantToken             string
+		wantID                string
+		wantErr               bool
+	}{
+		{"token then id", http.StatusOK, http.StatusOK, "tok123", "i-abc123", false},
+		{"token refused falls back to v1", http.StatusNotFound, http.StatusOK, "", "i-abc123", false},
+		{"id endpoint error", http.StatusOK, http.StatusServiceUnavailable, "tok123", "", true},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			srv := serve(tc.tokenStatus, tc.idStatus, tc.wantToken)
+			defer srv.Close()
+			imdsBase = srv.URL + "/latest"
+			id, err := fetchInstanceID()
+			if (err != nil) != tc.wantErr || id != tc.wantID {
+				t.Fatalf("id = %q, err = %v", id, err)
+			}
+		})
+	}
+
+	imdsBase = "http://127.0.0.1:1/latest"
+	if _, err := fetchInstanceID(); err == nil {
+		t.Error("an unreachable metadata service must fail")
+	}
+}

spinifex/agents/elbv2/evidence.go

@@ -0,0 +1,133 @@
+package elbv2
+
+import (
+	"context"
+	"fmt"
+	"io"
+	"net/http"
+	"net/url"
+	"strconv"
+	"strings"
+	"time"
+)
+
+const (
+	imdsWait       = time.Second
+	imdsRetryTicks = 6
+)
+
+// imdsBase is the in-VM metadata endpoint (a variable so tests can point it at a fake).
+var imdsBase = "http://169.254.169.254/latest"
+
+// dataPathEvidence is the optional part of a heartbeat: which VM the agent runs
+// in and what each data NIC's gateway probe saw. An empty value adds nothing,
+// which is what an agent without a netcfg blob sends.
+type dataPathEvidence struct {
+	InstanceID string
+	NICs       []NICReport
+}
+
+// addTo encodes the evidence as additional heartbeat query parameters.
+func (e dataPathEvidence) addTo(params url.Values) {
+	if e.InstanceID != "" {
+		params.Set("InstanceID", e.InstanceID)
+	}
+	for i, n := range e.NICs {
+		p := "NICs.member." + strconv.Itoa(i+1) + "."
+		params.Set(p+"MAC", n.MAC)
+		params.Set(p+"ProbeOK", strconv.FormatBool(n.ProbeOK))
+		params.Set(p+"ConsecutiveFailures", strconv.FormatInt(n.ConsecutiveFailures, 10))
+		params.Set(p+"RxPackets", strconv.FormatInt(n.RxPackets, 10))
+		params.Set(p+"TxPackets", strconv.FormatInt(n.TxPackets, 10))
+		params.Set(p+"RxDropped", strconv.FormatInt(n.RxDropped, 10))
+		params.Set(p+"TxDropped", strconv.FormatInt(n.TxDropped, 10))
+	}
+}
+
+// dataPathEvidence gathers the probe results and instance ID for one heartbeat.
+// Neither step can fail the heartbeat: missing evidence is simply omitted.
+func (a *Agent) dataPathEvidence() dataPathEvidence {
+	nics := a.probeNICs()
+	if len(nics) == 0 {
+		return dataPathEvidence{}
+	}
+	if a.instanceID == "" {
+		a.lookupInstanceID()
+	}
+	return dataPathEvidence{InstanceID: a.instanceID, NICs: nics}
+}
+
+// lookupInstanceID asks for the instance ID without ever holding up a tick: the
+// call is bounded by imdsWait and, after a failure, retried only every
+// imdsRetryTicks ticks.
+func (a *Agent) lookupInstanceID() {
+	if a.idSkip > 0 {
+		a.idSkip--
+		return
+	}
+	type result struct {
+		id  string
+		err error
+	}
+	ch := make(chan result, 1)
+	go func() {
+		id, err := a.instanceIDFn()
+		ch <- result{id, err}
+	}()
+	timer := time.NewTimer(imdsWait)
+	defer timer.Stop()
+	select {
+	case r := <-ch:
+		if r.err == nil && r.id != "" {
+			a.instanceID = r.id
+			return
+		}
+	case <-timer.C:
+	}
+	a.idSkip = imdsRetryTicks - 1
+}
+
+// fetchInstanceID reads the instance ID from the in-VM metadata service. The
+// token request and the read share one deadline.
+func fetchInstanceID() (string, error) {
+	ctx, cancel := context.WithTimeout(context.Background(), imdsWait)
+	defer cancel()
+	c := &http.Client{}
+
+	var token string
+	if req, err := http.NewRequestWithContext(ctx, http.MethodPut, imdsBase+"/api/token", nil); err == nil {
+		req.Header.Set("X-Aws-Ec2-Metadata-Token-Ttl-Seconds", "60")
+		if resp, err := c.Do(req); err == nil {
+			b, _ := io.ReadAll(io.LimitReader(resp.Body, 256))
+			resp.Body.Close()
+			if resp.StatusCode == http.StatusOK {
+				token = strings.TrimSpace(string(b))
+			}
+		}
+	}
+
+	req, err := http.NewRequestWithContext(ctx, http.MethodGet, imdsBase+"/meta-data/instance-id", nil)
+	if err != nil {
+		return "", err
+	}
+	if token != "" {
+		req.Header.Set("X-Aws-Ec2-Metadata-Token", token)
+	}
+	resp, err := c.Do(req)
+	if err != nil {
+		return "", err
+	}
+	defer resp.Body.Close()
+	if resp.StatusCode != http.StatusOK {
+		return "", fmt.Errorf("instance-id: status %d", resp.StatusCode)
+	}
+	b, err := io.ReadAll(io.LimitReader(resp.Body, 256))
+	if err != nil {
+		return "", err
+	}
+	id := strings.TrimSpace(string(b))
+	if id == "" {
+		return "", fmt.Errorf("instance-id: empty")
+	}
+	return id, nil
+}

spinifex/domains/ec2/systeminstance/launcher.go

@@ -57,7 +57,7 @@ const (
 	BootAMI BootMode = "ami"
 )
 
-// SystemInstanceLauncher launches and terminates system-managed VMs.
+// SystemInstanceLauncher launches, reboots and terminates system-managed VMs.
 // Defined here to avoid a circular import between elbv2/eks and daemon.
 type SystemInstanceLauncher interface {
 	// LaunchSystemInstance creates and starts a system-managed VM, returning its ID and private IP.
@@ -65,6 +65,10 @@ type SystemInstanceLauncher interface {
 
 	// TerminateSystemInstance stops and cleans up a system-managed VM.
 	TerminateSystemInstance(instanceID string) error
+
+	// RebootSystemInstance restarts a system-managed load balancer VM into a
+	// fresh QEMU on its owning node, keeping its ID, ENIs and addresses.
+	RebootSystemInstance(instanceID string) error
 }
 
 // SystemInstanceInput describes a system VM to launch.

spinifex/domains/elbv2/agent_heartbeat_wire_test.go

@@ -0,0 +1,73 @@
+package elbv2
+
+import (
+	"testing"
+
+	"github.com/mulgadc/spinifex/spinifex/ingress/aws/query"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func TestLBAgentHeartbeatInput_OldAgentDecodesUnchanged(t *testing.T) {
+	t.Parallel()
+	// Exactly the parameters an agent without the data-path probe sends.
+	params := map[string]string{
+		"Action":                   "LBAgentHeartbeat",
+		"Version":                  "2015-12-01",
+		"LBID":                     "lb-abc123",
+		"Servers.member.1.Backend": "bk_tg1",
+		"Servers.member.1.Server":  "srv_i-web1",
+		"Servers.member.1.Status":  "UP",
+		"Servers.member.2.Backend": "bk_tg1",
+		"Servers.member.2.Server":  "srv_i-web2",
+		"Servers.member.2.Status":  "DOWN",
+	}
+
+	var in LBAgentHeartbeatInput
+	require.NoError(t, query.QueryParamsToStruct(params, &in))
+
+	require.NotNil(t, in.LBID)
+	assert.Equal(t, "lb-abc123", *in.LBID)
+	require.Len(t, in.Servers, 2)
+	assert.Equal(t, "DOWN", *in.Servers[1].Status)
+	assert.Nil(t, in.InstanceID, "an old agent reports no instance ID")
+	assert.Empty(t, in.NICs, "an old agent reports no NICs")
+	assert.Len(t, in.toHealthReport().Servers, 2)
+}
+
+func TestLBAgentHeartbeatInput_DecodesDataPathEvidence(t *testing.T) {
+	t.Parallel()
+	params := map[string]string{
+		"LBID":                              "lb-abc123",
+		"InstanceID":                        "i-0123456789abcdef0",
+		"NICs.member.1.MAC":                 "02:00:00:aa:00:01",
+		"NICs.member.1.ProbeOK":             "false",
+		"NICs.member.1.ConsecutiveFailures": "4",
+		"NICs.member.1.RxPackets":           "100",
+		"NICs.member.1.TxPackets":           "200",
+		"NICs.member.1.RxDropped":           "3",
+		"NICs.member.1.TxDropped":           "9",
+		"NICs.member.2.MAC":                 "02:00:00:aa:00:03",
+		"NICs.member.2.ProbeOK":             "true",
+		"NICs.member.2.ConsecutiveFailures": "0",
+	}
+
+	var in LBAgentHeartbeatInput
+	require.NoError(t, query.QueryParamsToStruct(params, &in))
+
+	require.NotNil(t, in.InstanceID)
+	assert.Equal(t, "i-0123456789abcdef0", *in.InstanceID)
+	require.Len(t, in.NICs, 2)
+
+	n := in.NICs[0]
+	assert.Equal(t, "02:00:00:aa:00:01", *n.MAC)
+	assert.False(t, *n.ProbeOK)
+	assert.EqualValues(t, 4, *n.ConsecutiveFailures)
+	assert.EqualValues(t, 100, *n.RxPackets)
+	assert.EqualValues(t, 200, *n.TxPackets)
+	assert.EqualValues(t, 3, *n.RxDropped)
+	assert.EqualValues(t, 9, *n.TxDropped)
+
+	assert.True(t, *in.NICs[1].ProbeOK)
+	assert.Nil(t, in.NICs[1].RxPackets, "an omitted counter stays unset")
+}

spinifex/domains/elbv2/agent_types.go

@@ -7,9 +7,25 @@ import (
 
 // LBAgentHeartbeatInput is sent by the LB agent on each heartbeat tick,
 // including its health report so the daemon can update target health.
+//
+// InstanceID and NICs are optional: an older agent sends neither.
 type LBAgentHeartbeatInput struct {
-	LBID    *string                `locationName:"LBID" type:"string"`
-	Servers []*LBAgentServerStatus `locationName:"Servers" type:"list"`
+	LBID       *string                `locationName:"LBID" type:"string"`
+	Servers    []*LBAgentServerStatus `locationName:"Servers" type:"list"`
+	InstanceID *string                `locationName:"InstanceID" type:"string"`
+	NICs       []*LBAgentNICStatus    `locationName:"NICs" type:"list"`
+}
+
+// LBAgentNICStatus is the gateway probe result and interface counters of one
+// data NIC, as reported by the LB agent.
+type LBAgentNICStatus struct {
+	MAC                 *string `locationName:"MAC" type:"string"`
+	ProbeOK             *bool   `locationName:"ProbeOK" type:"boolean"`
+	ConsecutiveFailures *int64  `locationName:"ConsecutiveFailures" type:"long"`
+	RxPackets           *int64  `locationName:"RxPackets" type:"long"`
+	TxPackets           *int64  `locationName:"TxPackets" type:"long"`
+	RxDropped           *int64  `locationName:"RxDropped" type:"long"`
+	TxDropped           *int64  `locationName:"TxDropped" type:"long"`
 }
 
 // LBAgentServerStatus represents a single backend server's health status as reported by the LB agent.

spinifex/domains/elbv2/launcher_test.go

@@ -14,6 +14,9 @@ type mockSystemInstanceLauncher struct {
 	launchErr      error
 	terminateErr   error
 	terminateDone  chan struct{} // closed after TerminateSystemInstance completes
+	onTerminate    func(instanceID string)
+	rebootCalls    []string
+	rebootErr      error
 }
 
 func (m *mockSystemInstanceLauncher) LaunchSystemInstance(input *SystemInstanceInput) (*SystemInstanceOutput, error) {
@@ -25,12 +28,22 @@ func (m *mockSystemInstanceLauncher) TerminateSystemInstance(instanceID string)
 	m.mu.Lock()
 	m.terminateCalls = append(m.terminateCalls, instanceID)
 	m.mu.Unlock()
+	if m.onTerminate != nil {
+		m.onTerminate(instanceID)
+	}
 	if m.terminateDone != nil {
 		close(m.terminateDone)
 	}
 	return m.terminateErr
 }
 
+func (m *mockSystemInstanceLauncher) RebootSystemInstance(instanceID string) error {
+	m.mu.Lock()
+	m.rebootCalls = append(m.rebootCalls, instanceID)
+	m.mu.Unlock()
+	return m.rebootErr
+}
+
 // waitTerminate blocks until TerminateSystemInstance has been called.
 func (m *mockSystemInstanceLauncher) waitTerminate() {
 	if m.terminateDone != nil {

spinifex/domains/elbv2/lb_converge_test.go

@@ -0,0 +1,192 @@
+package elbv2
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	"github.com/aws/aws-sdk-go/aws"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func TestLBRepair_AnotherReservationWinsTheFirstCAS(t *testing.T) {
+	t.Parallel()
+	e := newRepairEnv(t)
+	lb := e.seed("lbr20", true)
+	cands := e.candidates()
+	require.Len(t, cands, 1)
+	require.True(t, e.rec.eligible(context.Background(), cands[0], e.clock))
+
+	_, err := e.svc.store.UpdateLBHealth(context.Background(), lb.LoadBalancerID, func(rec *LBDataPathRecord) (bool, error) {
+		rec.Repairs = append(rec.Repairs, LBRepair{At: e.clock, InstanceID: lb.InstanceID, Outcome: RepairReserved})
+		return true, nil
+	})
+	require.NoError(t, err)
+
+	assert.False(t, e.rec.repair(context.Background(), cands[0]))
+	assert.Empty(t, e.reboots())
+	assert.Len(t, e.health(lb.LoadBalancerID).Repairs, 1, "the losing reservation must not add an entry")
+}
+
+func TestLBRepair_LeaseLostAfterReservationKeepsItCountedAndDoesNotDispatch(t *testing.T) {
+	t.Parallel()
+	e := newRepairEnv(t)
+	e.seed("lbr21", true)
+	cands := e.candidates()
+	require.Len(t, cands, 1)
+
+	calls := 0
+	e.rec.isLeader = func() bool {
+		calls++
+		return calls == 1
+	}
+	assert.True(t, e.rec.repair(context.Background(), cands[0]), "a written reservation is reported as spent")
+	assert.Empty(t, e.reboots())
+
+	repairs := e.health("lbr21").Repairs
+	require.Len(t, repairs, 1)
+	assert.Equal(t, RepairReserved, repairs[0].Outcome, "the entry stays counted with no outcome")
+}
+
+func TestLBRepair_NonLeaderNeverReserves(t *testing.T) {
+	t.Parallel()
+	e := newRepairEnv(t)
+	e.seed("lbr22", true)
+	cands := e.candidates()
+	require.Len(t, cands, 1)
+
+	e.rec.isLeader = func() bool { return false }
+	assert.False(t, e.rec.repair(context.Background(), cands[0]))
+	assert.Empty(t, e.health("lbr22").Repairs)
+}
+
+func TestLBHealth_FailedResumeThenNewVMHeartbeatsConverge(t *testing.T) {
+	t.Parallel()
+	svc := setupTestService(t)
+	lb := seedHealthLB(t, svc)
+	ctx := context.Background()
+
+	// The old VM's evidence was suspended by SetSubnets, whose resume write then failed.
+	_, err := svc.store.UpdateLBHealth(ctx, lb.LoadBalancerID, func(rec *LBDataPathRecord) (bool, error) {
+		rec.InstanceID = "i-oldvm"
+		rec.Suspended = true
+		rec.SuspendedAt = time.Now().UTC()
+		return true, nil
+	})
+	require.NoError(t, err)
+
+	healthBeat(t, svc, aws.String(healthInst), nicStatus(healthMACData, true, 0, 10))
+	rec, err := svc.store.GetLBHealth(ctx, lb.LoadBalancerID)
+	require.NoError(t, err)
+	assert.False(t, rec.Suspended, "a heartbeat from the VM the LB names lifts the stale suspension")
+	assert.Equal(t, healthInst, rec.InstanceID)
+
+	healthBeat(t, svc, aws.String(healthInst), nicStatus(healthMACData, false, 3, 11))
+	rec, err = svc.store.GetLBHealth(ctx, lb.LoadBalancerID)
+	require.NoError(t, err)
+	require.Contains(t, rec.NICs, healthMACData, "a later NIC failure is recorded and repairable")
+}
+
+func TestLBHealth_SuspensionIsNotLiftedByAHeartbeatFromAVMTheLBDoesNotName(t *testing.T) {
+	t.Parallel()
+	svc := setupTestService(t)
+	lb := seedHealthLB(t, svc)
+	ctx := context.Background()
+
+	_, err := svc.store.UpdateLBHealth(ctx, lb.LoadBalancerID, func(rec *LBDataPathRecord) (bool, error) {
+		rec.InstanceID = "i-oldvm"
+		rec.Suspended = true
+		rec.SuspendedAt = time.Now().UTC()
+		return true, nil
+	})
+	require.NoError(t, err)
+
+	healthBeat(t, svc, aws.String("i-oldvm"), nicStatus(healthMACData, false, 3, 1))
+	rec, err := svc.store.GetLBHealth(ctx, lb.LoadBalancerID)
+	require.NoError(t, err)
+	assert.True(t, rec.Suspended)
+	assert.Equal(t, "i-oldvm", rec.InstanceID)
+}
+
+func TestLBHealth_SuspensionBackstopLiftsOnlyAfterThirtyMinutes(t *testing.T) {
+	t.Parallel()
+	svc := setupTestService(t)
+	lb := seedHealthLB(t, svc)
+	ctx := context.Background()
+
+	suspend := func(age time.Duration) {
+		_, err := svc.store.UpdateLBHealth(ctx, lb.LoadBalancerID, func(rec *LBDataPathRecord) (bool, error) {
+			rec.InstanceID = healthInst
+			rec.Suspended = true
+			rec.SuspendedAt = time.Now().UTC().Add(-age)
+			return true, nil
+		})
+		require.NoError(t, err)
+	}
+
+	suspend(29 * time.Minute)
+	healthBeat(t, svc, aws.String(healthInst), nicStatus(healthMACData, false, 3, 1))
+	rec, err := svc.store.GetLBHealth(ctx, lb.LoadBalancerID)
+	require.NoError(t, err)
+	assert.True(t, rec.Suspended, "inside the backstop the suspension holds")
+
+	suspend(31 * time.Minute)
+	healthBeat(t, svc, aws.String(healthInst), nicStatus(healthMACData, false, 3, 2))
+	rec, err = svc.store.GetLBHealth(ctx, lb.LoadBalancerID)
+	require.NoError(t, err)
+	assert.False(t, rec.Suspended, "past the backstop the current VM's heartbeat lifts it")
+	assert.Contains(t, rec.NICs, healthMACData)
+}
+
+func TestLBHealth_EstablishedFailureSurvivesAnAgentRestart(t *testing.T) {
+	t.Parallel()
+	svc := setupTestService(t)
+	lb := seedHealthLB(t, svc)
+	ctx := context.Background()
+
+	healthBeat(t, svc, aws.String(healthInst), nicStatus(healthMACData, false, 3, 1))
+	first, err := svc.store.GetLBHealth(ctx, lb.LoadBalancerID)
+	require.NoError(t, err)
+	since := first.NICs[healthMACData].FailingSince
+
+	// A restarted agent reports healthy until it has missed three probes again.
+	healthBeat(t, svc, aws.String(healthInst), nicStatus(healthMACData, true, 1, 2))
+	healthBeat(t, svc, aws.String(healthInst), nicStatus(healthMACData, true, 2, 3))
+	rec, err := svc.store.GetLBHealth(ctx, lb.LoadBalancerID)
+	require.NoError(t, err)
+	require.Contains(t, rec.NICs, healthMACData)
+	assert.True(t, since.Equal(rec.NICs[healthMACData].FailingSince), "FailingSince is kept")
+
+	healthBeat(t, svc, aws.String(healthInst), nicStatus(healthMACData, true, 0, 4))
+	rec, err = svc.store.GetLBHealth(ctx, lb.LoadBalancerID)
+	require.NoError(t, err)
+	assert.Empty(t, rec.NICs, "only a clean probe clears the failure")
+}
+
+func TestLBHealth_OldVMHeartbeatPausedAcrossPublicationLeavesTheNewBinding(t *testing.T) {
+	t.Parallel()
+	svc := setupTestService(t)
+	lb := seedHealthLB(t, svc)
+	ctx := context.Background()
+
+	// SetSubnets has already published the new VM and rebound the evidence.
+	lb.InstanceID = "i-newvm"
+	require.NoError(t, svc.store.PutLoadBalancer(ctx, lb))
+	require.NoError(t, svc.store.ResumeLBHealth(ctx, lb.LoadBalancerID, "i-newvm"))
+
+	// The old VM's heartbeat read the LB before that and runs now.
+	stale := *lb
+	stale.InstanceID = healthInst
+	svc.recordDataPathEvidence(ctx, &stale, &LBAgentHeartbeatInput{
+		LBID:       aws.String(lb.LoadBalancerID),
+		InstanceID: aws.String(healthInst),
+		NICs:       []*LBAgentNICStatus{nicStatus(healthMACData, false, 5, 1)},
+	})
+
+	rec, err := svc.store.GetLBHealth(ctx, lb.LoadBalancerID)
+	require.NoError(t, err)
+	require.NotNil(t, rec)
+	assert.Equal(t, "i-newvm", rec.InstanceID)
+	assert.Empty(t, rec.NICs, "the old VM's failure must not land on the new binding")
+}

spinifex/domains/elbv2/lb_health.go

@@ -0,0 +1,333 @@
+package elbv2
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"log/slog"
+	"strings"
+	"time"
+
+	"github.com/aws/aws-sdk-go/aws"
+	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
+	"github.com/nats-io/nats.go/jetstream"
+)
+
+const (
+	// KeyPrefixLBHealth holds the data-path evidence of one load balancer. It is
+	// a separate key from the LB record so whole-record writes cannot drop it.
+	KeyPrefixLBHealth = "lbhealth."
+
+	// StateActiveImpaired is what DescribeLoadBalancers reports while a data NIC
+	// of the LB's current VM is failing its gateway probe.
+	StateActiveImpaired = "active_impaired"
+
+	// lbSuspendBackstop is how long a suspension may outlive its mutation before
+	// a heartbeat from the current VM lifts it.
+	lbSuspendBackstop = 30 * time.Minute
+)
+
+// Outcomes of a recorded repair. Reserved is written before the repair is
+// dispatched, so an entry whose outcome was never updated still counts.
+const (
+	RepairReserved    = "reserved"
+	RepairDispatched  = "dispatched"
+	RepairRefused     = "refused"
+	RepairUnreachable = "unreachable"
+)
+
+// LBNICFailure is one data NIC that was failing its gateway probe when the
+// transition was recorded.
+type LBNICFailure struct {
+	ENIID               string    `json:"eni_id,omitempty"`
+	FailingSince        time.Time `json:"failing_since"`
+	ConsecutiveFailures int64     `json:"consecutive_failures"`
+	RxPackets           int64     `json:"rx_packets"`
+	TxPackets           int64     `json:"tx_packets"`
+	RxDropped           int64     `json:"rx_dropped"`
+	TxDropped           int64     `json:"tx_dropped"`
+}
+
+// LBRepair is one repair of an LB's VM and how it ended.
+type LBRepair struct {
+	At         time.Time `json:"at"`
+	InstanceID string    `json:"instance_id"`
+	ENIID      string    `json:"eni_id,omitempty"`
+	Reason     string    `json:"reason,omitempty"`
+	Outcome    string    `json:"outcome"`
+}
+
+// LBDataPathRecord is the data-path evidence of one load balancer VM. NICs holds
+// only failing NICs, keyed by lower-case MAC; an absent NIC is healthy.
+type LBDataPathRecord struct {
+	InstanceID  string                   `json:"instance_id"`
+	Suspended   bool                     `json:"suspended,omitempty"`
+	SuspendedAt time.Time                `json:"suspended_at,omitzero"`
+	BoundAt     time.Time                `json:"bound_at,omitzero"`
+	NICs        map[string]*LBNICFailure `json:"nics,omitempty"`
+	Repairs     []LBRepair               `json:"repairs,omitempty"`
+}
+
+// failing reports whether any NIC of the record is failing.
+func (r *LBDataPathRecord) failing() bool {
+	return r != nil && len(r.NICs) > 0
+}
+
+// GetLBHealth returns the evidence record of a load balancer, or nil when none
+// has been written.
+func (s *Store) GetLBHealth(ctx context.Context, lbID string) (*LBDataPathRecord, error) {
+	entry, err := s.kv.Get(ctx, KeyPrefixLBHealth+lbID)
+	if err != nil {
+		if errors.Is(err, jetstream.ErrKeyNotFound) {
+			return nil, nil
+		}
+		return nil, err
+	}
+	var rec LBDataPathRecord
+	if err := json.Unmarshal(entry.Value(), &rec); err != nil {
+		return nil, fmt.Errorf("unmarshal lb health: %w", err)
+	}
+	return &rec, nil
+}
+
+// UpdateLBHealth applies mutate to the evidence record under optimistic
+// concurrency, creating it when absent. mutate returns false to write nothing.
+func (s *Store) UpdateLBHealth(ctx context.Context, lbID string, mutate func(*LBDataPathRecord) (bool, error)) (*LBDataPathRecord, error) {
+	return kvutil.Update(ctx, s.kv, KeyPrefixLBHealth+lbID, kvutil.CASConfig{
+		Attempts:       targetGroupCASAttempts,
+		CreateIfAbsent: true,
+		Exhausted: func(_ string, attempts int) error {
+			return fmt.Errorf("elbv2: lb health %s contended, CAS exhausted after %d attempts", lbID, attempts)
+		},
+	}, mutate)
+}
+
+// DeleteLBHealth removes the evidence record of a load balancer.
+func (s *Store) DeleteLBHealth(ctx context.Context, lbID string) error {
+	err := s.kv.Delete(ctx, KeyPrefixLBHealth+lbID)
+	if err != nil && !errors.Is(err, jetstream.ErrKeyNotFound) {
+		return err
+	}
+	return nil
+}
+
+// SuspendLBHealth marks the load balancer as being torn down or reshaped, so
+// no repair is admitted against its VM until the mutation finishes.
+func (s *Store) SuspendLBHealth(ctx context.Context, lbID string) error {
+	_, err := s.UpdateLBHealth(ctx, lbID, func(rec *LBDataPathRecord) (bool, error) {
+		if rec.Suspended {
+			return false, nil
+		}
+		rec.Suspended = true
+		rec.SuspendedAt = time.Now().UTC()
+		return true, nil
+	})
+	return err
+}
+
+// ResumeLBHealth lifts a suspension. A non-empty instanceID also binds the
+// record to that VM and drops the old VM's failing NICs.
+func (s *Store) ResumeLBHealth(ctx context.Context, lbID, instanceID string) error {
+	_, err := s.UpdateLBHealth(ctx, lbID, func(rec *LBDataPathRecord) (bool, error) {
+		changed := rec.Suspended
+		rec.Suspended = false
+		rec.SuspendedAt = time.Time{}
+		if instanceID != "" && rec.InstanceID != instanceID {
+			rec.InstanceID = instanceID
+			rec.BoundAt = time.Now().UTC()
+			rec.NICs = nil
+			changed = true
+		}
+		return changed, nil
+	})
+	return err
+}
+
+// failingNICMACs returns the lower-case MACs that are failing. A probe that
+// failed starts a failure. An established failure ends only when the latest
+// probe succeeded, because an agent restarted by a repair reports healthy for
+// its first misses.
+func failingNICMACs(nics []*LBAgentNICStatus, established map[string]*LBNICFailure) map[string]*LBAgentNICStatus {
+	out := make(map[string]*LBAgentNICStatus)
+	for _, n := range nics {
+		if n == nil || n.MAC == nil || n.ProbeOK == nil {
+			continue
+		}
+		mac := strings.ToLower(*n.MAC)
+		if !*n.ProbeOK || (established[mac] != nil && aws.Int64Value(n.ConsecutiveFailures) != 0) {
+			out[mac] = n
+		}
+	}
+	return out
+}
+
+// recordDataPathEvidence stores the data-path evidence of one heartbeat. It
+// writes only when the set of failing NICs changes, and ignores evidence that
+// does not come from the VM the LB record currently names.
+func (s *ELBv2ServiceImpl) recordDataPathEvidence(ctx context.Context, lb *LoadBalancerRecord, input *LBAgentHeartbeatInput) {
+	if lb.InstanceID == "" || input.InstanceID == nil || *input.InstanceID != lb.InstanceID || len(input.NICs) == 0 {
+		return
+	}
+
+	rec, err := s.store.GetLBHealth(ctx, lb.LoadBalancerID)
+	if err != nil {
+		slog.WarnContext(ctx, "LBAgentHeartbeat: failed to read data-path evidence", "lbId", lb.LoadBalancerID, "err", err)
+		return
+	}
+	converge := false
+	if rec != nil && rec.Suspended {
+		if !s.suspensionConverges(ctx, lb, rec, *input.InstanceID) {
+			return
+		}
+		converge = true
+	}
+	var known map[string]*LBNICFailure
+	if rec != nil && rec.InstanceID == lb.InstanceID {
+		known = rec.NICs
+	}
+	failing := failingNICMACs(input.NICs, known)
+	if !converge && sameMACs(known, failing) && (len(failing) == 0 || rec != nil) {
+		return
+	}
+
+	eniByMAC := s.resolveLBENIs(ctx, lb, failing)
+	now := time.Now().UTC()
+	_, err = s.store.UpdateLBHealth(ctx, lb.LoadBalancerID, func(cur *LBDataPathRecord) (bool, error) {
+		changed := false
+		if cur.Suspended {
+			if !converge || !suspensionLifts(cur, lb.InstanceID, now) {
+				return false, nil
+			}
+			cur.Suspended = false
+			cur.SuspendedAt = time.Time{}
+			changed = true
+			if cur.InstanceID != lb.InstanceID {
+				cur.InstanceID = lb.InstanceID
+				cur.BoundAt = now
+				cur.NICs = nil
+			}
+		}
+		if cur.InstanceID != "" && cur.InstanceID != lb.InstanceID {
+			return changed, nil
+		}
+		prev := cur.NICs
+		failing := failingNICMACs(input.NICs, prev)
+		if cur.InstanceID == lb.InstanceID && sameMACs(prev, failing) {
+			return changed, nil
+		}
+		next := make(map[string]*LBNICFailure, len(failing))
+		for mac, n := range failing {
+			f := &LBNICFailure{
+				ENIID:               eniByMAC[mac],
+				FailingSince:        now,
+				ConsecutiveFailures: aws.Int64Value(n.ConsecutiveFailures),
+				RxPackets:           aws.Int64Value(n.RxPackets),
+				TxPackets:           aws.Int64Value(n.TxPackets),
+				RxDropped:           aws.Int64Value(n.RxDropped),
+				TxDropped:           aws.Int64Value(n.TxDropped),
+			}
+			if old, ok := prev[mac]; ok {
+				f.FailingSince = old.FailingSince
+			}
+			next[mac] = f
+		}
+		cur.InstanceID = lb.InstanceID
+		cur.NICs = next
+		return true, nil
+	})
+	if err != nil {
+		slog.WarnContext(ctx, "LBAgentHeartbeat: failed to store data-path evidence", "lbId", lb.LoadBalancerID, "err", err)
+		return
+	}
+	for mac, n := range failing {
+		if _, ok := known[mac]; !ok {
+			slog.WarnContext(ctx, "ELBv2 data-path probe failing",
+				"lbArn", lb.LoadBalancerArn, "lbId", lb.LoadBalancerID, "instanceId", lb.InstanceID,
+				"eniId", eniByMAC[mac], "mac", mac, "consecutive_failures", aws.Int64Value(n.ConsecutiveFailures),
+				"rx_packets", aws.Int64Value(n.RxPackets), "tx_packets", aws.Int64Value(n.TxPackets),
+				"rx_dropped", aws.Int64Value(n.RxDropped), "tx_dropped", aws.Int64Value(n.TxDropped))
+		}
+	}
+	for mac := range known {
+		if _, ok := failing[mac]; !ok {
+			slog.InfoContext(ctx, "ELBv2 data-path probe recovered",
+				"lbArn", lb.LoadBalancerArn, "lbId", lb.LoadBalancerID, "instanceId", lb.InstanceID, "mac", mac)
+		}
+	}
+}
+
+// suspensionLifts reports whether a suspended record may be lifted by a
+// heartbeat from instanceID: it is bound to an older VM, or it has outlived the
+// backstop while bound to the current one.
+func suspensionLifts(rec *LBDataPathRecord, instanceID string, now time.Time) bool {
+	if rec.InstanceID != "" && rec.InstanceID != instanceID {
+		return true
+	}
+	return !rec.SuspendedAt.IsZero() && now.Sub(rec.SuspendedAt) >= lbSuspendBackstop
+}
+
+// suspensionConverges confirms against a fresh read of the LB record that the
+// VM the heartbeat reports is the one the LB currently names.
+func (s *ELBv2ServiceImpl) suspensionConverges(ctx context.Context, lb *LoadBalancerRecord, rec *LBDataPathRecord, heartbeatInstance string) bool {
+	if !suspensionLifts(rec, heartbeatInstance, time.Now().UTC()) {
+		return false
+	}
+	fresh, err := s.store.GetLoadBalancer(ctx, lb.LoadBalancerID)
+	return err == nil && fresh != nil && fresh.InstanceID == heartbeatInstance
+}
+
+// sameMACs reports whether two failing sets name the same NICs.
+func sameMACs[A, B any](a map[string]A, b map[string]B) bool {
+	if len(a) != len(b) {
+		return false
+	}
+	for mac := range a {
+		if _, ok := b[mac]; !ok {
+			return false
+		}
+	}
+	return true
+}
+
+// resolveLBENIs maps failing MACs to ENI IDs from the LB's cross-account ENIs
+// and one describe of its own ENIs. A MAC it cannot resolve maps to nothing.
+func (s *ELBv2ServiceImpl) resolveLBENIs(ctx context.Context, lb *LoadBalancerRecord, failing map[string]*LBAgentNICStatus) map[string]string {
+	out := make(map[string]string, len(failing))
+	for _, x := range lb.CrossAccountENIs {
+		mac := strings.ToLower(x.ENIMac)
+		if _, ok := failing[mac]; ok {
+			out[mac] = x.ENIID
+		}
+	}
+	if len(out) == len(failing) {
+		return out
+	}
+	for id, eni := range s.describeENIs(ctx, lb.ENIs, lb.AccountID) {
+		mac := strings.ToLower(aws.StringValue(eni.MacAddress))
+		if _, ok := failing[mac]; ok && out[mac] == "" {
+			out[mac] = id
+		}
+	}
+	return out
+}
+
+// impairedByEvidence reports whether the LB's current VM has a failing data NIC.
+func (s *ELBv2ServiceImpl) impairedByEvidence(ctx context.Context, r *LoadBalancerRecord) bool {
+	if r.State != StateActive || r.InstanceID == "" {
+		return false
+	}
+	rec, err := s.store.GetLBHealth(ctx, r.LoadBalancerID)
+	if err != nil {
+		slog.DebugContext(ctx, "lbRecordToSDK: failed to read data-path evidence", "lbId", r.LoadBalancerID, "err", err)
+		return false
+	}
+	return rec != nil && !rec.Suspended && rec.InstanceID == r.InstanceID && rec.failing()
+}
+
+// resumeLBHealth lifts a suspension after a failure that left the VM untouched.
+func (s *ELBv2ServiceImpl) resumeLBHealth(ctx context.Context, lbID string) {
+	if err := s.store.ResumeLBHealth(ctx, lbID, ""); err != nil {
+		slog.WarnContext(ctx, "failed to lift data-path suspension", "lbId", lbID, "err", err)
+	}
+}

spinifex/domains/elbv2/lb_health_test.go

@@ -0,0 +1,370 @@
+package elbv2
+
+import (
+	"context"
+	"errors"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"github.com/aws/aws-sdk-go/aws"
+	"github.com/aws/aws-sdk-go/service/elbv2"
+	"github.com/nats-io/nats.go/jetstream"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+const (
+	healthLBID    = "lbhealth01"
+	healthInst    = "i-lbvm01"
+	healthMACData = "02:00:00:aa:00:01"
+	healthMACX    = "02:00:00:aa:00:02"
+)
+
+func seedHealthLB(t *testing.T, svc *ELBv2ServiceImpl) *LoadBalancerRecord {
+	t.Helper()
+	lb := &LoadBalancerRecord{
+		LoadBalancerArn: "arn:aws:elasticloadbalancing:us-east-1:" + testAccountID + ":loadbalancer/app/health-lb/" + healthLBID,
+		LoadBalancerID:  healthLBID,
+		Name:            "health-lb",
+		State:           StateActive,
+		InstanceID:      healthInst,
+		AccountID:       testAccountID,
+		CrossAccountENIs: []ExtraENIInput{
+			{ENIID: "eni-cross01", ENIMac: healthMACX},
+		},
+		CreatedAt: time.Now().UTC(),
+	}
+	require.NoError(t, svc.store.PutLoadBalancer(context.Background(), lb))
+	return lb
+}
+
+func nicStatus(mac string, ok bool, failures, rx int64) *LBAgentNICStatus {
+	return &LBAgentNICStatus{
+		MAC:                 aws.String(mac),
+		ProbeOK:             aws.Bool(ok),
+		ConsecutiveFailures: aws.Int64(failures),
+		RxPackets:           aws.Int64(rx),
+		TxPackets:           aws.Int64(rx + 5),
+		RxDropped:           aws.Int64(0),
+		TxDropped:           aws.Int64(2),
+	}
+}
+
+func healthBeat(t *testing.T, svc *ELBv2ServiceImpl, instanceID *string, nics ...*LBAgentNICStatus) {
+	t.Helper()
+	_, err := svc.LBAgentHeartbeat(context.Background(), &LBAgentHeartbeatInput{
+		LBID:       aws.String(healthLBID),
+		InstanceID: instanceID,
+		NICs:       nics,
+	}, testAccountID)
+	require.NoError(t, err)
+}
+
+func healthRevision(t *testing.T, svc *ELBv2ServiceImpl) uint64 {
+	t.Helper()
+	entry, err := svc.store.kv.Get(context.Background(), KeyPrefixLBHealth+healthLBID)
+	require.NoError(t, err)
+	return entry.Revision()
+}
+
+func TestLBHealth_SteadyHeartbeatsWriteNothing(t *testing.T) {
+	t.Parallel()
+	svc := setupTestService(t)
+	seedHealthLB(t, svc)
+
+	for range 5 {
+		healthBeat(t, svc, aws.String(healthInst), nicStatus(healthMACData, true, 0, 100))
+	}
+	rec, err := svc.store.GetLBHealth(context.Background(), healthLBID)
+	require.NoError(t, err)
+	assert.Nil(t, rec, "healthy heartbeats must not create an evidence record")
+
+	healthBeat(t, svc, aws.String(healthInst), nicStatus(healthMACData, false, 3, 100))
+	rev := healthRevision(t, svc)
+	for i := range 5 {
+		healthBeat(t, svc, aws.String(healthInst), nicStatus(healthMACData, false, int64(4+i), int64(100+i)))
+	}
+	assert.Equal(t, rev, healthRevision(t, svc), "a NIC that stays failing must not rewrite the record")
+}
+
+func TestLBHealth_FailingNICWritesFailingSinceOnceAndRecoveryClears(t *testing.T) {
+	t.Parallel()
+	svc := setupTestService(t)
+	seedHealthLB(t, svc)
+
+	before := time.Now().UTC().Add(-time.Second)
+	healthBeat(t, svc, aws.String(healthInst), nicStatus(healthMACData, false, 3, 777), nicStatus(healthMACX, true, 0, 1))
+
+	rec, err := svc.store.GetLBHealth(context.Background(), healthLBID)
+	require.NoError(t, err)
+	require.NotNil(t, rec)
+	assert.Equal(t, healthInst, rec.InstanceID)
+	require.Len(t, rec.NICs, 1)
+	f := rec.NICs[healthMACData]
+	require.NotNil(t, f)
+	assert.True(t, f.FailingSince.After(before))
+	assert.EqualValues(t, 777, f.RxPackets)
+	assert.EqualValues(t, 782, f.TxPackets)
+	assert.EqualValues(t, 2, f.TxDropped)
+	assert.EqualValues(t, 3, f.ConsecutiveFailures)
+
+	healthBeat(t, svc, aws.String(healthInst), nicStatus(healthMACData, true, 0, 900))
+	rec, err = svc.store.GetLBHealth(context.Background(), healthLBID)
+	require.NoError(t, err)
+	require.NotNil(t, rec)
+	assert.Empty(t, rec.NICs, "recovery clears the failing NIC")
+}
+
+func TestLBHealth_SecondNICFailingKeepsFirstFailingSince(t *testing.T) {
+	t.Parallel()
+	svc := setupTestService(t)
+	seedHealthLB(t, svc)
+
+	healthBeat(t, svc, aws.String(healthInst), nicStatus(healthMACData, false, 3, 1))
+	first, err := svc.store.GetLBHealth(context.Background(), healthLBID)
+	require.NoError(t, err)
+	since := first.NICs[healthMACData].FailingSince
+
+	healthBeat(t, svc, aws.String(healthInst), nicStatus(healthMACData, false, 9, 2), nicStatus(healthMACX, false, 3, 2))
+	rec, err := svc.store.GetLBHealth(context.Background(), healthLBID)
+	require.NoError(t, err)
+	require.Len(t, rec.NICs, 2)
+	assert.True(t, since.Equal(rec.NICs[healthMACData].FailingSince))
+	assert.Equal(t, "eni-cross01", rec.NICs[healthMACX].ENIID, "a cross-account NIC resolves to its ENI without a describe")
+}
+
+func TestLBHealth_WrongOrMissingInstanceIDIsIgnored(t *testing.T) {
+	t.Parallel()
+	svc := setupTestService(t)
+	seedHealthLB(t, svc)
+
+	failing := nicStatus(healthMACData, false, 5, 1)
+	healthBeat(t, svc, aws.String("i-someoneelse"), failing)
+	healthBeat(t, svc, aws.String(""), failing)
+	healthBeat(t, svc, nil, failing)
+
+	rec, err := svc.store.GetLBHealth(context.Background(), healthLBID)
+	require.NoError(t, err)
+	assert.Nil(t, rec, "evidence from another VM, or with no instance ID, must be ignored")
+}
+
+func TestLBHealth_WholeRecordPutLeavesEvidenceIntact(t *testing.T) {
+	t.Parallel()
+	svc := setupTestService(t)
+	lb := seedHealthLB(t, svc)
+
+	healthBeat(t, svc, aws.String(healthInst), nicStatus(healthMACData, false, 3, 1))
+	want, err := svc.store.GetLBHealth(context.Background(), healthLBID)
+	require.NoError(t, err)
+	rev := healthRevision(t, svc)
+
+	lb.StateReason = "rewritten by an older daemon"
+	require.NoError(t, svc.store.PutLoadBalancer(context.Background(), lb))
+
+	got, err := svc.store.GetLBHealth(context.Background(), healthLBID)
+	require.NoError(t, err)
+	assert.Equal(t, want, got)
+	assert.Equal(t, rev, healthRevision(t, svc))
+}
+
+func TestLBHealth_DeleteLoadBalancerRemovesTheKey(t *testing.T) {
+	t.Parallel()
+	svc := setupTestService(t)
+	lb := seedHealthLB(t, svc)
+	healthBeat(t, svc, aws.String(healthInst), nicStatus(healthMACData, false, 3, 1))
+
+	_, err := svc.DeleteLoadBalancer(context.Background(), &elbv2.DeleteLoadBalancerInput{
+		LoadBalancerArn: aws.String(lb.LoadBalancerArn),
+	}, testAccountID)
+	require.NoError(t, err)
+
+	rec, err := svc.store.GetLBHealth(context.Background(), healthLBID)
+	require.NoError(t, err)
+	assert.Nil(t, rec)
+}
+
+func TestLBHealth_DeleteLoadBalancerSuspendsBeforeTerminatingTheVM(t *testing.T) {
+	t.Parallel()
+	svc := setupTestService(t)
+	lb := seedHealthLB(t, svc)
+	seen := make(chan bool, 1)
+	svc.InstanceLauncher = &mockSystemInstanceLauncher{
+		onTerminate: func(string) {
+			rec, err := svc.store.GetLBHealth(context.Background(), healthLBID)
+			seen <- err == nil && rec != nil && rec.Suspended
+		},
+	}
+
+	_, err := svc.DeleteLoadBalancer(context.Background(), &elbv2.DeleteLoadBalancerInput{
+		LoadBalancerArn: aws.String(lb.LoadBalancerArn),
+	}, testAccountID)
+	require.NoError(t, err)
+
+	select {
+	case suspended := <-seen:
+		assert.True(t, suspended, "repair must be excluded before the VM is terminated")
+	case <-time.After(5 * time.Second):
+		t.Fatal("the LB VM was never terminated")
+	}
+}
+
+func TestLBHealth_SetSubnetsSuspendsFirstAndResumesOnTheNewVM(t *testing.T) {
+	t.Parallel()
+	svc, vpcSvc, mock := setupSubnetTestService(t)
+	vid := vpcID(t, vpcSvc)
+	sub1 := getTestSubnetID(t, vpcSvc, vid, "10.0.40.0/24", "us-east-1a")
+	sub2 := getTestSubnetID(t, vpcSvc, vid, "10.0.41.0/24", "us-east-1b")
+
+	out, err := svc.CreateLoadBalancer(context.Background(), &elbv2.CreateLoadBalancerInput{
+		Name:    aws.String("health-sub-lb"),
+		Subnets: []*string{aws.String(sub1)},
+	}, testAccountID)
+	require.NoError(t, err)
+	svc.WaitLaunches()
+	lbID := aws.StringValue(out.LoadBalancers[0].LoadBalancerArn)
+	rec0, err := svc.store.GetLoadBalancerByArn(context.Background(), lbID)
+	require.NoError(t, err)
+	require.Equal(t, "i-lb", rec0.InstanceID)
+
+	_, err = svc.LBAgentHeartbeat(context.Background(), &LBAgentHeartbeatInput{
+		LBID:       aws.String(rec0.LoadBalancerID),
+		InstanceID: aws.String("i-lb"),
+		NICs:       []*LBAgentNICStatus{nicStatus(healthMACData, false, 3, 1)},
+	}, testAccountID)
+	require.NoError(t, err)
+
+	suspendedAtTerminate := false
+	mock.onTerminate = func(string) {
+		rec, gerr := svc.store.GetLBHealth(context.Background(), rec0.LoadBalancerID)
+		suspendedAtTerminate = gerr == nil && rec != nil && rec.Suspended
+	}
+	mock.launchResult = &SystemInstanceOutput{InstanceID: "i-new", PrivateIP: "10.0.1.6"}
+
+	_, err = svc.SetSubnets(context.Background(), &elbv2.SetSubnetsInput{
+		LoadBalancerArn: aws.String(lbID),
+		Subnets:         []*string{aws.String(sub1), aws.String(sub2)},
+	}, testAccountID)
+	require.NoError(t, err)
+	assert.True(t, suspendedAtTerminate, "repair must be excluded before the VM is terminated")
+
+	rec, err := svc.store.GetLBHealth(context.Background(), rec0.LoadBalancerID)
+	require.NoError(t, err)
+	require.NotNil(t, rec)
+	assert.False(t, rec.Suspended, "the suspension is lifted once the new VM is published")
+	assert.Equal(t, "i-new", rec.InstanceID)
+	assert.Empty(t, rec.NICs, "the old VM's failing NIC does not carry over")
+}
+
+func TestLBHealth_SetSubnetsFailureBeforeTheVMIsTouchedResumes(t *testing.T) {
+	t.Parallel()
+	svc, vpcSvc, mock := setupSubnetTestService(t)
+	vid := vpcID(t, vpcSvc)
+	sub1 := getTestSubnetID(t, vpcSvc, vid, "10.0.42.0/24", "us-east-1a")
+	sub2 := getTestSubnetID(t, vpcSvc, vid, "10.0.43.0/24", "us-east-1b")
+
+	out, err := svc.CreateLoadBalancer(context.Background(), &elbv2.CreateLoadBalancerInput{
+		Name:    aws.String("health-resume-lb"),
+		Subnets: []*string{aws.String(sub1)},
+	}, testAccountID)
+	require.NoError(t, err)
+	svc.WaitLaunches()
+	arn := aws.StringValue(out.LoadBalancers[0].LoadBalancerArn)
+	rec0, err := svc.store.GetLoadBalancerByArn(context.Background(), arn)
+	require.NoError(t, err)
+
+	mock.terminateErr = assert.AnError
+	_, err = svc.SetSubnets(context.Background(), &elbv2.SetSubnetsInput{
+		LoadBalancerArn: aws.String(arn),
+		Subnets:         []*string{aws.String(sub1), aws.String(sub2)},
+	}, testAccountID)
+	require.Error(t, err)
+
+	rec, err := svc.store.GetLBHealth(context.Background(), rec0.LoadBalancerID)
+	require.NoError(t, err)
+	require.NotNil(t, rec)
+	assert.False(t, rec.Suspended, "a failed reshape that left the VM running must not leave repair excluded")
+}
+
+func TestLBHealth_DescribeShowsImpairedOnlyForFailingNICOfCurrentVM(t *testing.T) {
+	t.Parallel()
+	svc := setupTestService(t)
+	lb := seedHealthLB(t, svc)
+	code := func() string {
+		out, err := svc.DescribeLoadBalancers(context.Background(), &elbv2.DescribeLoadBalancersInput{
+			LoadBalancerArns: []*string{aws.String(lb.LoadBalancerArn)},
+		}, testAccountID)
+		require.NoError(t, err)
+		require.Len(t, out.LoadBalancers, 1)
+		return aws.StringValue(out.LoadBalancers[0].State.Code)
+	}
+
+	assert.Equal(t, StateActive, code())
+
+	healthBeat(t, svc, aws.String(healthInst), nicStatus(healthMACData, false, 3, 1))
+	assert.Equal(t, StateActiveImpaired, code())
+
+	healthBeat(t, svc, aws.String(healthInst), nicStatus(healthMACData, true, 0, 2))
+	assert.Equal(t, StateActive, code(), "active again once the probe recovers")
+
+	// Evidence recorded against a VM the LB no longer names never impairs it.
+	_, err := svc.store.UpdateLBHealth(context.Background(), healthLBID, func(r *LBDataPathRecord) (bool, error) {
+		r.InstanceID = "i-oldvm"
+		r.NICs = map[string]*LBNICFailure{healthMACData: {FailingSince: time.Now().UTC()}}
+		return true, nil
+	})
+	require.NoError(t, err)
+	assert.Equal(t, StateActive, code())
+
+	// A LB that is not active is reported as it is.
+	_, err = svc.store.UpdateLBHealth(context.Background(), healthLBID, func(r *LBDataPathRecord) (bool, error) {
+		r.InstanceID = healthInst
+		return true, nil
+	})
+	require.NoError(t, err)
+	require.Equal(t, StateActiveImpaired, code())
+	_, err = svc.store.UpdateLoadBalancer(context.Background(), healthLBID, func(r *LoadBalancerRecord) (bool, error) {
+		r.State = StateFailed
+		return true, nil
+	})
+	require.NoError(t, err)
+	assert.Equal(t, StateFailed, code())
+}
+
+// failKeysKV fails key listing once armed, which is what the listener cascade
+// of DeleteLoadBalancer reads first.
+type failKeysKV struct {
+	jetstream.KeyValue
+
+	armed atomic.Bool
+}
+
+func (k *failKeysKV) Keys(ctx context.Context, opts ...jetstream.WatchOpt) ([]string, error) {
+	if k.armed.Load() {
+		return nil, errors.New("injected key listing failure")
+	}
+	return k.KeyValue.Keys(ctx, opts...)
+}
+
+func TestLBHealth_DeleteLoadBalancerListenerCascadeFailureResumes(t *testing.T) {
+	t.Parallel()
+	svc := setupTestService(t)
+	lb := seedHealthLB(t, svc)
+	kv := &failKeysKV{KeyValue: svc.store.kv}
+	svc.store.kv = kv
+	mock := &mockSystemInstanceLauncher{}
+	svc.InstanceLauncher = mock
+	kv.armed.Store(true)
+
+	_, err := svc.DeleteLoadBalancer(context.Background(), &elbv2.DeleteLoadBalancerInput{
+		LoadBalancerArn: aws.String(lb.LoadBalancerArn),
+	}, testAccountID)
+	require.Error(t, err)
+	kv.armed.Store(false)
+
+	rec, err := svc.store.GetLBHealth(context.Background(), healthLBID)
+	require.NoError(t, err)
+	require.NotNil(t, rec)
+	assert.False(t, rec.Suspended, "a delete that failed before touching the VM must not leave repair excluded")
+	assert.Empty(t, mock.terminateCalls)
+}

spinifex/domains/elbv2/lb_repair.go

@@ -0,0 +1,377 @@
+package elbv2
+
+import (
+	"context"
+	"errors"
+	"log/slog"
+	"sort"
+	"strings"
+	"sync"
+	"time"
+
+	"github.com/mulgadc/spinifex/spinifex/domains/ec2/systeminstance"
+	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvlease"
+	"github.com/mulgadc/spinifex/spinifex/foundation/telemetry"
+	"github.com/nats-io/nats.go"
+	"github.com/nats-io/nats.go/jetstream"
+)
+
+const (
+	// KVBucketELBv2Leader holds the lease key of the data-path reconciler.
+	KVBucketELBv2Leader        = "spinifex-elbv2-leader"
+	KVBucketELBv2LeaderVersion = 1
+	KVBucketELBv2LeaderTTL     = 60 * time.Second
+
+	lbRepairLeaderKey     = "datapath"
+	lbRepairLeaseRenew    = 20 * time.Second
+	lbRepairPassInterval  = 30 * time.Second
+	lbRepairMinFailing    = 60 * time.Second
+	lbRepairMinVMAge      = 5 * time.Minute
+	lbRepairCooldown      = 10 * time.Minute
+	lbRepairBudgetWindow  = 24 * time.Hour
+	lbRepairBudget        = 3
+	lbRepairFleetMinCount = 3
+
+	// A heartbeat that persists only every heartbeatPersistInterval is still
+	// fresh for twice that long.
+	lbRepairHeartbeatMaxAge = 2 * heartbeatPersistInterval
+)
+
+// SystemInstanceRebooter is the only capability the repair path has over a VM.
+type SystemInstanceRebooter interface {
+	RebootSystemInstance(instanceID string) error
+}
+
+var _ SystemInstanceRebooter = SystemInstanceLauncher(nil)
+
+// counts reports whether a repair entry uses up cooldown and budget. Only a
+// repair no node received is free.
+func (r LBRepair) counts() bool {
+	return r.Outcome != RepairUnreachable
+}
+
+// repairWindow counts the entries inside window ending at now, and returns the
+// newest of them.
+func repairWindow(repairs []LBRepair, now time.Time, window time.Duration) (n int, newest time.Time) {
+	for _, r := range repairs {
+		if !r.counts() || now.Sub(r.At) >= window {
+			continue
+		}
+		n++
+		if r.At.After(newest) {
+			newest = r.At
+		}
+	}
+	return n, newest
+}
+
+// lbRepairReconciler restarts the VM of a load balancer whose data NIC has been
+// failing its gateway probe. Only the lease holder acts.
+type lbRepairReconciler struct {
+	svc      *ELBv2ServiceImpl
+	rebooter SystemInstanceRebooter
+	isLeader func() bool
+	now      func() time.Time
+
+	mu             sync.Mutex
+	budgetReported map[string]bool
+}
+
+func newLBRepairReconciler(svc *ELBv2ServiceImpl, rebooter SystemInstanceRebooter, isLeader func() bool) *lbRepairReconciler {
+	return &lbRepairReconciler{
+		svc:            svc,
+		rebooter:       rebooter,
+		isLeader:       isLeader,
+		now:            func() time.Time { return time.Now().UTC() },
+		budgetReported: make(map[string]bool),
+	}
+}
+
+// StartDataPathReconciler elects a leader among the daemons and runs the repair
+// pass on it every lbRepairPassInterval until ctx or the service is cancelled.
+func (s *ELBv2ServiceImpl) StartDataPathReconciler(ctx context.Context, holder string) {
+	if s.InstanceLauncher == nil || s.store == nil || s.store.js == nil {
+		return
+	}
+	lease, err := kvlease.New(kvlease.Config{
+		Name:   "elbv2/datapath",
+		Bucket: s.leaderBucket,
+		Key:    lbRepairLeaderKey,
+		Holder: holder,
+		TTL:    KVBucketELBv2LeaderTTL,
+		Renew:  lbRepairLeaseRenew,
+		Retry:  lbRepairLeaseRenew,
+	})
+	if err != nil {
+		slog.ErrorContext(ctx, "ELBv2 data-path reconciler: lease config invalid", "err", err)
+		return
+	}
+	rec := newLBRepairReconciler(s, s.InstanceLauncher, lease.Held)
+	go lease.Run(ctx)
+	go func() {
+		ticker := time.NewTicker(lbRepairPassInterval)
+		defer ticker.Stop()
+		for {
+			select {
+			case <-s.ctx.Done():
+				return
+			case <-ctx.Done():
+				return
+			case <-ticker.C:
+				rec.pass(ctx)
+			}
+		}
+	}()
+}
+
+func (s *ELBv2ServiceImpl) leaderBucket(ctx context.Context) (jetstream.KeyValue, error) {
+	return kvlease.OpenBucket(ctx, s.store.js, kvlease.BucketConfig{
+		Name:    KVBucketELBv2Leader,
+		TTL:     KVBucketELBv2LeaderTTL,
+		Version: KVBucketELBv2LeaderVersion,
+	})
+}
+
+// candidate is a load balancer that is failing and may be repaired.
+type lbRepairCandidate struct {
+	lb    *LoadBalancerRecord
+	rec   *LBDataPathRecord
+	mac   string
+	nic   *LBNICFailure
+	since time.Time
+}
+
+// pass runs one repair pass. It starts at most one repair.
+func (r *lbRepairReconciler) pass(ctx context.Context) {
+	if !r.isLeader() {
+		return
+	}
+	lbs, err := r.svc.store.ListLoadBalancersStrict(ctx)
+	if err != nil {
+		slog.WarnContext(ctx, "ELBv2 data-path reconciler: failed to list load balancers", "err", err)
+		return
+	}
+	now := r.now()
+
+	var heartbeating, failing int
+	var cands []lbRepairCandidate
+	live := make(map[string]bool, len(lbs))
+	for _, lb := range lbs {
+		live[lb.LoadBalancerID] = true
+		if lb.State != StateActive || lb.InstanceID == "" || now.Sub(lb.LastHeartbeat) >= lbRepairHeartbeatMaxAge {
+			continue
+		}
+		heartbeating++
+		rec, err := r.svc.store.GetLBHealth(ctx, lb.LoadBalancerID)
+		if err != nil || rec == nil || rec.Suspended || rec.InstanceID != lb.InstanceID || !rec.failing() {
+			continue
+		}
+		failing++
+		c := lbRepairCandidate{lb: lb, rec: rec}
+		for mac, nic := range rec.NICs {
+			if c.nic == nil || nic.FailingSince.Before(c.since) {
+				c.mac, c.nic, c.since = mac, nic, nic.FailingSince
+			}
+		}
+		cands = append(cands, c)
+	}
+	r.forgetBudgetReports(live)
+
+	if failing >= lbRepairFleetMinCount && failing*2 > heartbeating {
+		slog.WarnContext(ctx, "ELBv2 data-path repair suppressed: fleet-wide probe failure",
+			"failing_lbs", failing, "heartbeating_lbs", heartbeating)
+		return
+	}
+
+	sort.Slice(cands, func(i, j int) bool { return cands[i].since.Before(cands[j].since) })
+	for _, c := range cands {
+		if r.eligible(ctx, c, now) && r.repair(ctx, c) {
+			return
+		}
+	}
+}
+
+// eligible applies the per-LB conditions that need no write.
+func (r *lbRepairReconciler) eligible(ctx context.Context, c lbRepairCandidate, now time.Time) bool {
+	if now.Sub(c.since) < lbRepairMinFailing {
+		return false
+	}
+	if now.Sub(vmRunningSince(c.lb, c.rec)) < lbRepairMinVMAge {
+		return false
+	}
+	n, newest := repairWindow(c.rec.Repairs, now, lbRepairBudgetWindow)
+	if n >= lbRepairBudget {
+		r.reportBudgetSpent(ctx, c, n)
+		return false
+	}
+	r.clearBudgetReport(c.lb.LoadBalancerID)
+	return now.Sub(newest) >= lbRepairCooldown || newest.IsZero()
+}
+
+// vmRunningSince is when the LB's current VM last came up as far as the control
+// plane can tell: its activation, a relaunch, or the last repair.
+func vmRunningSince(lb *LoadBalancerRecord, rec *LBDataPathRecord) time.Time {
+	since := lb.ActiveSince
+	if since.IsZero() {
+		since = lb.CreatedAt
+	}
+	if rec.BoundAt.After(since) {
+		since = rec.BoundAt
+	}
+	for _, e := range rec.Repairs {
+		if e.counts() && e.At.After(since) {
+			since = e.At
+		}
+	}
+	return since
+}
+
+// repair reserves the attempt durably, then dispatches it. It reports whether a
+// repair was started, which ends the pass.
+func (r *lbRepairReconciler) repair(ctx context.Context, c lbRepairCandidate) bool {
+	if !r.isLeader() {
+		return false
+	}
+	lb, err := r.svc.store.GetLoadBalancer(ctx, c.lb.LoadBalancerID)
+	if err != nil || lb == nil || lb.State != StateActive || lb.InstanceID != c.lb.InstanceID {
+		return false
+	}
+	now := r.now()
+	reason := "data NIC failing gateway probe"
+	var attempt int
+	reserved := false
+	_, err = r.svc.store.UpdateLBHealth(ctx, lb.LoadBalancerID, func(cur *LBDataPathRecord) (bool, error) {
+		// The callback reruns after a conflict, so admission is judged by the run
+		// that wrote the reservation and nothing earlier.
+		reserved, attempt = false, 0
+		nic := cur.NICs[c.mac]
+		if cur.Suspended || cur.InstanceID != lb.InstanceID || nic == nil || now.Sub(nic.FailingSince) < lbRepairMinFailing {
+			return false, nil
+		}
+		n, newest := repairWindow(cur.Repairs, now, lbRepairBudgetWindow)
+		if n >= lbRepairBudget || (!newest.IsZero() && now.Sub(newest) < lbRepairCooldown) {
+			return false, nil
+		}
+		cur.Repairs = append(pruneRepairs(cur.Repairs, now), LBRepair{
+			At: now, InstanceID: lb.InstanceID, ENIID: nic.ENIID, Reason: reason, Outcome: RepairReserved,
+		})
+		attempt = n + 1
+		reserved = true
+		return true, nil
+	})
+	if err != nil {
+		slog.WarnContext(ctx, "ELBv2 data-path repair: reservation failed", "lbId", lb.LoadBalancerID, "err", err)
+		return false
+	}
+	if !reserved {
+		return false
+	}
+
+	attrs := repairLogAttrs(lb, c, attempt, reason, now)
+	if !r.isLeader() {
+		slog.WarnContext(ctx, "ELBv2 data-path repair not dispatched: leadership lost after reservation", attrs...)
+		return true
+	}
+	slog.WarnContext(ctx, "ELBv2 data-path repair started", attrs...)
+	outcome := r.dispatch(lb.InstanceID)
+	if outcome != "" {
+		r.recordOutcome(ctx, lb.LoadBalancerID, now, lb.InstanceID, outcome)
+	}
+	switch outcome {
+	case RepairDispatched:
+		slog.InfoContext(ctx, "ELBv2 data-path repair dispatched", attrs...)
+	case RepairRefused:
+		slog.ErrorContext(ctx, "ELBv2 data-path repair refused by the owning node", attrs...)
+	case RepairUnreachable:
+		slog.WarnContext(ctx, "ELBv2 data-path repair: owning node unreachable, will retry", attrs...)
+	default:
+		slog.WarnContext(ctx, "ELBv2 data-path repair: outcome unknown, counted", attrs...)
+	}
+	return true
+}
+
+// dispatch asks the owning node to restart exactly the LB's own VM. It returns
+// an empty outcome when the reply was lost, which leaves the entry reserved.
+func (r *lbRepairReconciler) dispatch(instanceID string) string {
+	err := r.rebooter.RebootSystemInstance(instanceID)
+	switch {
+	case err == nil:
+		return RepairDispatched
+	case errors.Is(err, nats.ErrNoResponders), strings.Contains(err.Error(), "no responders"):
+		return RepairUnreachable
+	case errors.Is(err, systeminstance.ErrSystemInstanceNotFound),
+		errors.Is(err, systeminstance.ErrSystemInstanceNotSystem),
+		errors.Is(err, systeminstance.ErrSystemInstanceOwnershipUnknown):
+		return RepairRefused
+	default:
+		return ""
+	}
+}
+
+func (r *lbRepairReconciler) recordOutcome(ctx context.Context, lbID string, at time.Time, instanceID, outcome string) {
+	_, err := r.svc.store.UpdateLBHealth(ctx, lbID, func(cur *LBDataPathRecord) (bool, error) {
+		for i := range cur.Repairs {
+			e := &cur.Repairs[i]
+			if e.At.Equal(at) && e.InstanceID == instanceID && e.Outcome == RepairReserved {
+				e.Outcome = outcome
+				return true, nil
+			}
+		}
+		return false, nil
+	})
+	if err != nil {
+		slog.WarnContext(ctx, "ELBv2 data-path repair: failed to record outcome", "lbId", lbID, "outcome", outcome, "err", err)
+	}
+}
+
+// pruneRepairs drops entries older than the budget window.
+func pruneRepairs(repairs []LBRepair, now time.Time) []LBRepair {
+	kept := repairs[:0:0]
+	for _, e := range repairs {
+		if now.Sub(e.At) < lbRepairBudgetWindow {
+			kept = append(kept, e)
+		}
+	}
+	return kept
+}
+
+func repairLogAttrs(lb *LoadBalancerRecord, c lbRepairCandidate, attempt int, reason string, now time.Time) []any {
+	return []any{
+		"lbArn", lb.LoadBalancerArn, "lbId", lb.LoadBalancerID, "instanceId", lb.InstanceID, "nodeId", lb.NodeID,
+		"eniId", c.nic.ENIID, "mac", c.mac, "failing_since", c.nic.FailingSince,
+		"failing_ms", otelsetup.Millis(now.Sub(c.nic.FailingSince)),
+		"consecutive_failures", c.nic.ConsecutiveFailures,
+		"rx_packets", c.nic.RxPackets, "tx_packets", c.nic.TxPackets,
+		"rx_dropped", c.nic.RxDropped, "tx_dropped", c.nic.TxDropped,
+		"repair_attempt", attempt, "reason", reason,
+	}
+}
+
+func (r *lbRepairReconciler) reportBudgetSpent(ctx context.Context, c lbRepairCandidate, n int) {
+	r.mu.Lock()
+	already := r.budgetReported[c.lb.LoadBalancerID]
+	r.budgetReported[c.lb.LoadBalancerID] = true
+	r.mu.Unlock()
+	if already {
+		return
+	}
+	slog.ErrorContext(ctx, "ELBv2 data-path repair budget exhausted",
+		"lbArn", c.lb.LoadBalancerArn, "lbId", c.lb.LoadBalancerID, "instanceId", c.lb.InstanceID,
+		"eniId", c.nic.ENIID, "mac", c.mac, "repairs_24h", n)
+}
+
+func (r *lbRepairReconciler) clearBudgetReport(lbID string) {
+	r.mu.Lock()
+	delete(r.budgetReported, lbID)
+	r.mu.Unlock()
+}
+
+func (r *lbRepairReconciler) forgetBudgetReports(live map[string]bool) {
+	r.mu.Lock()
+	defer r.mu.Unlock()
+	for id := range r.budgetReported {
+		if !live[id] {
+			delete(r.budgetReported, id)
+		}
+	}
+}

spinifex/domains/elbv2/lb_repair_test.go

@@ -0,0 +1,418 @@
+package elbv2
+
+import (
+	"bytes"
+	"context"
+	"errors"
+	"fmt"
+	"log/slog"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/mulgadc/spinifex/spinifex/domains/ec2/systeminstance"
+	"github.com/nats-io/nats.go"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+const repairMAC = "02:00:00:bb:00:01"
+
+type repairEnv struct {
+	t        *testing.T
+	svc      *ELBv2ServiceImpl
+	rec      *lbRepairReconciler
+	launcher *mockSystemInstanceLauncher
+	clock    time.Time
+	leader   bool
+}
+
+func newRepairEnv(t *testing.T) *repairEnv {
+	t.Helper()
+	e := &repairEnv{
+		t:        t,
+		svc:      setupTestService(t),
+		launcher: &mockSystemInstanceLauncher{},
+		clock:    time.Date(2026, 11, 5, 12, 0, 0, 0, time.UTC),
+		leader:   true,
+	}
+	e.rec = newLBRepairReconciler(e.svc, e.launcher, func() bool { return e.leader })
+	e.rec.now = func() time.Time { return e.clock }
+	return e
+}
+
+// seed stores an active LB whose VM has run for ten minutes and whose data NIC
+// has been failing for two, with a heartbeat from this instant.
+func (e *repairEnv) seed(id string, failing bool) *LoadBalancerRecord {
+	e.t.Helper()
+	lb := &LoadBalancerRecord{
+		LoadBalancerArn: "arn:aws:elasticloadbalancing:us-east-1:" + testAccountID + ":loadbalancer/app/" + id + "/" + id,
+		LoadBalancerID:  id,
+		Name:            id,
+		State:           StateActive,
+		InstanceID:      "i-" + id,
+		AccountID:       testAccountID,
+		NodeID:          "node-1",
+		CreatedAt:       e.clock.Add(-time.Hour),
+		ActiveSince:     e.clock.Add(-10 * time.Minute),
+		LastHeartbeat:   e.clock,
+	}
+	require.NoError(e.t, e.svc.store.PutLoadBalancer(context.Background(), lb))
+	if failing {
+		e.fail(id, 2*time.Minute)
+	}
+	return lb
+}
+
+func (e *repairEnv) fail(id string, age time.Duration) {
+	e.t.Helper()
+	_, err := e.svc.store.UpdateLoadBalancer(context.Background(), id, func(lb *LoadBalancerRecord) (bool, error) {
+		lb.LastHeartbeat = e.clock
+		return true, nil
+	})
+	require.NoError(e.t, err)
+	_, err = e.svc.store.UpdateLBHealth(context.Background(), id, func(rec *LBDataPathRecord) (bool, error) {
+		rec.InstanceID = "i-" + id
+		rec.NICs = map[string]*LBNICFailure{
+			repairMAC: {ENIID: "eni-" + id, FailingSince: e.clock.Add(-age), ConsecutiveFailures: 9, RxPackets: 4, TxPackets: 8},
+		}
+		return true, nil
+	})
+	require.NoError(e.t, err)
+}
+
+func (e *repairEnv) health(id string) *LBDataPathRecord {
+	e.t.Helper()
+	rec, err := e.svc.store.GetLBHealth(context.Background(), id)
+	require.NoError(e.t, err)
+	require.NotNil(e.t, rec)
+	return rec
+}
+
+func (e *repairEnv) pass() { e.rec.pass(context.Background()) }
+
+func (e *repairEnv) reboots() []string {
+	e.launcher.mu.Lock()
+	defer e.launcher.mu.Unlock()
+	return append([]string(nil), e.launcher.rebootCalls...)
+}
+
+func captureLogs(t *testing.T) *bytes.Buffer {
+	t.Helper()
+	var buf bytes.Buffer
+	prev := slog.Default()
+	slog.SetDefault(slog.New(slog.NewTextHandler(&buf, &slog.HandlerOptions{Level: slog.LevelDebug})))
+	t.Cleanup(func() { slog.SetDefault(prev) })
+	return &buf
+}
+
+func TestLBRepair_RepairsOnlyAfterSixtySecondsOfFailure(t *testing.T) {
+	t.Parallel()
+	e := newRepairEnv(t)
+	e.seed("lbr1", false)
+	e.fail("lbr1", 59*time.Second)
+	e.pass()
+	assert.Empty(t, e.reboots())
+
+	e.clock = e.clock.Add(2 * time.Second)
+	e.pass()
+	assert.Equal(t, []string{"i-lbr1"}, e.reboots())
+	h := e.health("lbr1")
+	require.Len(t, h.Repairs, 1)
+	assert.Equal(t, RepairDispatched, h.Repairs[0].Outcome)
+	assert.Equal(t, "eni-lbr1", h.Repairs[0].ENIID)
+}
+
+func TestLBRepair_ContinuouslyHeartbeatingLBIsNeverStale(t *testing.T) {
+	t.Parallel()
+	// A healthy agent persists its heartbeat only every minute, so the stored
+	// timestamp can be any age below that plus one pass at the reconciler.
+	for age := time.Duration(0); age < lbRepairHeartbeatMaxAge; age += 15 * time.Second {
+		e := newRepairEnv(t)
+		lb := e.seed("lbr2", true)
+		lb.LastHeartbeat = e.clock.Add(-age)
+		require.NoError(t, e.svc.store.PutLoadBalancer(context.Background(), lb))
+		e.pass()
+		assert.Len(t, e.reboots(), 1, "heartbeat %s old", age)
+	}
+
+	e := newRepairEnv(t)
+	lb := e.seed("lbr2", true)
+	lb.LastHeartbeat = e.clock.Add(-lbRepairHeartbeatMaxAge)
+	require.NoError(t, e.svc.store.PutLoadBalancer(context.Background(), lb))
+	e.pass()
+	assert.Empty(t, e.reboots(), "a heartbeat older than the bound is stale")
+}
+
+func TestLBRepair_NeverWithoutCurrentEvidence(t *testing.T) {
+	t.Parallel()
+	mutate := map[string]func(*LoadBalancerRecord){
+		"non-active LB":      func(lb *LoadBalancerRecord) { lb.State = StateProvisioning },
+		"failed LB":          func(lb *LoadBalancerRecord) { lb.State = StateFailed },
+		"empty instance ID":  func(lb *LoadBalancerRecord) { lb.InstanceID = "" },
+		"changed instance":   func(lb *LoadBalancerRecord) { lb.InstanceID = "i-other" },
+		"stale heartbeat":    func(lb *LoadBalancerRecord) { lb.LastHeartbeat = lb.LastHeartbeat.Add(-time.Hour) },
+		"VM under 5 minutes": func(lb *LoadBalancerRecord) { lb.ActiveSince = lb.LastHeartbeat.Add(-4 * time.Minute) },
+	}
+	for name, fn := range mutate {
+		t.Run(name, func(t *testing.T) {
+			t.Parallel()
+			e := newRepairEnv(t)
+			lb := e.seed("lbr3", true)
+			fn(lb)
+			require.NoError(t, e.svc.store.PutLoadBalancer(context.Background(), lb))
+			e.pass()
+			assert.Empty(t, e.reboots())
+		})
+	}
+}
+
+func TestLBRepair_NonHolderNeverRepairs(t *testing.T) {
+	t.Parallel()
+	e := newRepairEnv(t)
+	e.seed("lbr4", true)
+	e.leader = false
+	e.pass()
+	assert.Empty(t, e.reboots())
+	e.leader = true
+	e.pass()
+	assert.Len(t, e.reboots(), 1)
+}
+
+func TestLBRepair_OneRepairPerPass(t *testing.T) {
+	t.Parallel()
+	e := newRepairEnv(t)
+	e.seed("lbr5a", true)
+	e.seed("lbr5b", true)
+	for i := range 3 {
+		e.seed(fmt.Sprintf("lbr5h%d", i), false)
+	}
+	e.fail("lbr5b", 3*time.Minute)
+
+	e.pass()
+	assert.Equal(t, []string{"i-lbr5b"}, e.reboots(), "the longest-failing LB goes first")
+	e.pass()
+	assert.Equal(t, []string{"i-lbr5b", "i-lbr5a"}, e.reboots())
+}
+
+func TestLBRepair_FleetWideFailureRepairsNothing(t *testing.T) {
+	e := newRepairEnv(t)
+	logs := captureLogs(t)
+	e.seed("lbr6a", true)
+	e.seed("lbr6b", true)
+	e.seed("lbr6c", true)
+	e.seed("lbr6d", false)
+	e.pass()
+	assert.Empty(t, e.reboots())
+	assert.Contains(t, logs.String(), "ELBv2 data-path repair suppressed: fleet-wide probe failure")
+
+	// Three failing out of seven is not more than half.
+	for i := range 3 {
+		e.seed(fmt.Sprintf("lbr6h%d", i), false)
+	}
+	e.pass()
+	assert.Len(t, e.reboots(), 1)
+}
+
+func TestLBRepair_CooldownAndDailyBudget(t *testing.T) {
+	e := newRepairEnv(t)
+	logs := captureLogs(t)
+	e.seed("lbr7", true)
+
+	e.pass()
+	require.Len(t, e.reboots(), 1)
+
+	e.clock = e.clock.Add(9 * time.Minute)
+	e.fail("lbr7", 2*time.Minute)
+	e.pass()
+	assert.Len(t, e.reboots(), 1, "inside the 10 minute cooldown")
+
+	for want := 2; want <= 3; want++ {
+		e.clock = e.clock.Add(11 * time.Minute)
+		e.fail("lbr7", 2*time.Minute)
+		e.pass()
+		assert.Len(t, e.reboots(), want)
+	}
+
+	e.clock = e.clock.Add(11 * time.Minute)
+	e.fail("lbr7", 2*time.Minute)
+	e.pass()
+	e.pass()
+	assert.Len(t, e.reboots(), 3, "the fourth repair in 24 hours is refused")
+	assert.Equal(t, 1, strings.Count(logs.String(), "ELBv2 data-path repair budget exhausted"))
+
+	e.clock = e.clock.Add(24 * time.Hour)
+	e.fail("lbr7", 2*time.Minute)
+	e.pass()
+	assert.Len(t, e.reboots(), 4, "the budget frees after 24 hours")
+}
+
+func TestLBRepair_UnreachableOwnerIsRetriedAndNotCounted(t *testing.T) {
+	t.Parallel()
+	e := newRepairEnv(t)
+	e.launcher.rebootErr = fmt.Errorf("nats request: %w", nats.ErrNoResponders)
+	e.seed("lbr8", true)
+	for range 4 {
+		e.pass()
+		e.clock = e.clock.Add(30 * time.Second)
+		e.fail("lbr8", 2*time.Minute)
+	}
+	assert.Len(t, e.reboots(), 4, "no node received it, so cooldown and budget do not apply")
+	for _, r := range e.health("lbr8").Repairs {
+		assert.Equal(t, RepairUnreachable, r.Outcome)
+	}
+
+	e.launcher.rebootErr = nil
+	e.pass()
+	assert.Len(t, e.reboots(), 5)
+}
+
+func TestLBRepair_RefusedAndAmbiguousRepairsStillCount(t *testing.T) {
+	t.Parallel()
+	for name, err := range map[string]error{
+		"refused": fmt.Errorf("%w: i-lbr9", systeminstance.ErrSystemInstanceNotSystem),
+		"lost":    errors.New("timeout waiting for reply"),
+	} {
+		t.Run(name, func(t *testing.T) {
+			t.Parallel()
+			e := newRepairEnv(t)
+			e.launcher.rebootErr = err
+			e.seed("lbr9", true)
+			e.pass()
+			e.clock = e.clock.Add(30 * time.Second)
+			e.fail("lbr9", 2*time.Minute)
+			e.pass()
+			assert.Len(t, e.reboots(), 1, "inside the cooldown")
+			want := RepairRefused
+			if name == "lost" {
+				want = RepairReserved
+			}
+			assert.Equal(t, want, e.health("lbr9").Repairs[0].Outcome)
+		})
+	}
+}
+
+func TestLBRepair_ReservationWithoutOutcomeCountsAfterCrashOrLostLease(t *testing.T) {
+	t.Parallel()
+	e := newRepairEnv(t)
+	e.seed("lbr10", true)
+	// A leader reserved and died before dispatching.
+	_, err := e.svc.store.UpdateLBHealth(context.Background(), "lbr10", func(rec *LBDataPathRecord) (bool, error) {
+		rec.Repairs = append(rec.Repairs, LBRepair{At: e.clock.Add(-time.Minute), InstanceID: "i-lbr10", Outcome: RepairReserved})
+		return true, nil
+	})
+	require.NoError(t, err)
+
+	e.pass()
+	assert.Empty(t, e.reboots(), "the new leader sees the counted reservation")
+
+	e.leader = false
+	e.pass()
+	e.leader = true
+	assert.Empty(t, e.reboots())
+}
+
+func TestLBRepair_PausedBetweenEligibilityAndReservationRepairsNothing(t *testing.T) {
+	t.Parallel()
+	cases := map[string]func(e *repairEnv, lb *LoadBalancerRecord){
+		"DeleteLoadBalancer starts": func(e *repairEnv, lb *LoadBalancerRecord) {
+			require.NoError(t, e.svc.store.SuspendLBHealth(context.Background(), lb.LoadBalancerID))
+		},
+		"SetSubnets relaunches the VM": func(e *repairEnv, lb *LoadBalancerRecord) {
+			lb.InstanceID = "i-new"
+			require.NoError(t, e.svc.store.PutLoadBalancer(context.Background(), lb))
+		},
+		"SetSubnets rebinds the evidence": func(e *repairEnv, lb *LoadBalancerRecord) {
+			require.NoError(t, e.svc.store.ResumeLBHealth(context.Background(), lb.LoadBalancerID, "i-new"))
+		},
+	}
+	for name, interleave := range cases {
+		t.Run(name, func(t *testing.T) {
+			t.Parallel()
+			e := newRepairEnv(t)
+			lb := e.seed("lbr11", true)
+			cands := e.candidates()
+			require.Len(t, cands, 1)
+			require.True(t, e.rec.eligible(context.Background(), cands[0], e.clock))
+
+			interleave(e, lb)
+
+			assert.False(t, e.rec.repair(context.Background(), cands[0]))
+			assert.Empty(t, e.reboots())
+		})
+	}
+}
+
+func (e *repairEnv) candidates() []lbRepairCandidate {
+	e.t.Helper()
+	var out []lbRepairCandidate
+	lbs, err := e.svc.store.ListLoadBalancersStrict(context.Background())
+	require.NoError(e.t, err)
+	for _, lb := range lbs {
+		rec, err := e.svc.store.GetLBHealth(context.Background(), lb.LoadBalancerID)
+		require.NoError(e.t, err)
+		if rec == nil {
+			continue
+		}
+		c := lbRepairCandidate{lb: lb, rec: rec}
+		for mac, nic := range rec.NICs {
+			c.mac, c.nic, c.since = mac, nic, nic.FailingSince
+		}
+		out = append(out, c)
+	}
+	return out
+}
+
+// Verifies: ADR-0003:S7.
+func TestS7_LBRepairActsOnlyOnTheLBsOwnVM(t *testing.T) {
+	t.Parallel()
+	e := newRepairEnv(t)
+	e.seed("lbr12", true)
+	e.seed("lbr12ok", false)
+	e.pass()
+	e.pass()
+	assert.Equal(t, []string{"i-lbr12"}, e.reboots(), "only the failing LB's own instance ID reaches the launcher")
+	e.launcher.mu.Lock()
+	defer e.launcher.mu.Unlock()
+	assert.Empty(t, e.launcher.terminateCalls)
+	assert.Empty(t, e.launcher.launchCalls)
+}
+
+func TestLBRepair_TeardownNeverTriggersRepair(t *testing.T) {
+	t.Parallel()
+	ctx := context.Background()
+	t.Run("DeleteLoadBalancer", func(t *testing.T) {
+		t.Parallel()
+		e := newRepairEnv(t)
+		lb := e.seed("lbr13a", true)
+		require.NoError(t, e.svc.store.SuspendLBHealth(ctx, lb.LoadBalancerID))
+		e.pass()
+		assert.Empty(t, e.reboots())
+	})
+	t.Run("SetSubnets window", func(t *testing.T) {
+		t.Parallel()
+		e := newRepairEnv(t)
+		lb := e.seed("lbr13b", true)
+		require.NoError(t, e.svc.store.SuspendLBHealth(ctx, lb.LoadBalancerID))
+		lb.InstanceID = "i-relaunched"
+		require.NoError(t, e.svc.store.PutLoadBalancer(ctx, lb))
+		e.pass()
+		assert.Empty(t, e.reboots(), "suspended, and the old VM's evidence names another instance")
+	})
+	t.Run("transient probe failure on the relaunched VM", func(t *testing.T) {
+		t.Parallel()
+		e := newRepairEnv(t)
+		lb := e.seed("lbr13c", false)
+		lb.InstanceID = "i-relaunched"
+		require.NoError(t, e.svc.store.PutLoadBalancer(ctx, lb))
+		require.NoError(t, e.svc.store.ResumeLBHealth(ctx, lb.LoadBalancerID, "i-relaunched"))
+		_, err := e.svc.store.UpdateLBHealth(ctx, lb.LoadBalancerID, func(rec *LBDataPathRecord) (bool, error) {
+			rec.BoundAt = e.clock.Add(-90 * time.Second)
+			rec.NICs = map[string]*LBNICFailure{repairMAC: {FailingSince: e.clock.Add(-80 * time.Second)}}
+			return true, nil
+		})
+		require.NoError(t, err)
+		e.pass()
+		assert.Empty(t, e.reboots(), "a freshly relaunched VM gets five minutes")
+	})
+}

spinifex/domains/elbv2/service_impl.go

@@ -1138,6 +1138,7 @@ func (s *ELBv2ServiceImpl) LBAgentHeartbeat(ctx context.Context, input *LBAgentH
 			if rec.State == StateProvisioning {
 				rec.State = StateActive
 				rec.StateReason = ""
+				rec.ActiveSince = now
 			}
 			rec.LastHeartbeat = now
 			return true, nil
@@ -1162,6 +1163,8 @@ func (s *ELBv2ServiceImpl) LBAgentHeartbeat(ctx context.Context, input *LBAgentH
 		}
 	}
 
+	s.recordDataPathEvidence(ctx, lb, input)
+
 	// Process health report directly — no JSON round-trip needed.
 	if len(input.Servers) > 0 {
 		s.hc.handleHealthReportDirect(ctx, input.toHealthReport())
@@ -1853,15 +1856,23 @@ func (s *ELBv2ServiceImpl) DeleteLoadBalancer(ctx context.Context, input *elbv2.
 		return &elbv2.DeleteLoadBalancerOutput{}, nil
 	}
 
+	// Exclude data-path repair before anything touches the VM or its ENIs.
+	if err := s.store.SuspendLBHealth(ctx, lb.LoadBalancerID); err != nil {
+		slog.ErrorContext(ctx, "DeleteLoadBalancer: failed to suspend data-path repair", "lbId", lb.LoadBalancerID, "err", err)
+		return nil, errors.New(awserrors.ErrorServerInternal)
+	}
+
 	// Cascade-delete all listeners and their rules so no orphan pins a TG as ResourceInUse.
 	listeners, err := s.store.ListListenersByLB(ctx, lb.LoadBalancerArn)
 	if err != nil {
 		slog.ErrorContext(ctx, "DeleteLoadBalancer: failed to list listeners for cascade", "lbArn", lb.LoadBalancerArn, "err", err)
+		s.resumeLBHealth(ctx, lb.LoadBalancerID)
 		return nil, errors.New(awserrors.ErrorServerInternal)
 	}
 	for _, l := range listeners {
 		if err := s.deleteListenerCascade(ctx, l); err != nil {
 			slog.ErrorContext(ctx, "DeleteLoadBalancer: failed to cascade listener delete", "listenerID", l.ListenerID, "err", err)
+			s.resumeLBHealth(ctx, lb.LoadBalancerID)
 			return nil, errors.New(awserrors.ErrorServerInternal)
 		}
 	}
@@ -1910,6 +1921,9 @@ func (s *ELBv2ServiceImpl) DeleteLoadBalancer(ctx context.Context, input *elbv2.
 		slog.ErrorContext(ctx, "DeleteLoadBalancer: failed to delete record", "lbId", lb.LoadBalancerID, "err", err)
 		return nil, errors.New(awserrors.ErrorServerInternal)
 	}
+	if err := s.store.DeleteLBHealth(ctx, lb.LoadBalancerID); err != nil {
+		slog.WarnContext(ctx, "DeleteLoadBalancer: failed to delete data-path evidence", "lbId", lb.LoadBalancerID, "err", err)
+	}
 
 	// Release the name claim so the name is reusable. Idempotent on a missing
 	// key, so a delete that races the record removal still converges.
@@ -3456,6 +3470,9 @@ func (s *ELBv2ServiceImpl) lbRecordToSDK(ctx context.Context, r *LoadBalancerRec
 	if r.StateReason != "" {
 		lb.State.Reason = aws.String(r.StateReason)
 	}
+	if s.impairedByEvidence(ctx, r) {
+		lb.State.Code = aws.String(StateActiveImpaired)
+	}
 
 	for _, sg := range r.SecurityGroups {
 		lb.SecurityGroups = append(lb.SecurityGroups, aws.String(sg))

spinifex/domains/elbv2/service_impl_lb_network.go

@@ -172,6 +172,18 @@ func (s *ELBv2ServiceImpl) SetSubnets(ctx context.Context, input *elbv2.SetSubne
 		return s.setSubnetsOutput(ctx, lb), nil
 	}
 
+	// Exclude data-path repair before any ENI or VM call. It stays excluded once the
+	// VM is terminated and is lifted when the new VM is published.
+	if err := s.store.SuspendLBHealth(ctx, lb.LoadBalancerID); err != nil {
+		slog.ErrorContext(ctx, "SetSubnets: failed to suspend data-path repair", "arn", *input.LoadBalancerArn, "err", err)
+		return nil, errors.New(awserrors.ErrorServerInternal)
+	}
+	resume := func() {
+		if err := s.store.ResumeLBHealth(ctx, lb.LoadBalancerID, ""); err != nil {
+			slog.WarnContext(ctx, "SetSubnets: failed to lift data-path suspension", "lbId", lb.LoadBalancerID, "err", err)
+		}
+	}
+
 	// Create ENIs for added subnets; roll back on failure to avoid leaks.
 	newENIBySubnet := make(map[string]string, len(toAdd))
 	newAZBySubnet := make(map[string]string, len(toAdd))
@@ -188,6 +200,7 @@ func (s *ELBv2ServiceImpl) SetSubnets(ctx context.Context, input *elbv2.SetSubne
 		eniID, az, eniErr := s.createLBENI(ctx, subnetID, lb, accountID)
 		if eniErr != nil {
 			rollbackNewENIs()
+			resume()
 			slog.ErrorContext(ctx, "SetSubnets: failed to create ENI", "subnet", subnetID, "err", eniErr)
 			return nil, errors.New(awserrors.ErrorELBv2SubnetNotFound)
 		}
@@ -209,6 +222,7 @@ func (s *ELBv2ServiceImpl) SetSubnets(ctx context.Context, input *elbv2.SetSubne
 	if lb.InstanceID != "" && s.InstanceLauncher != nil {
 		if err := s.InstanceLauncher.TerminateSystemInstance(lb.InstanceID); err != nil {
 			rollbackNewENIs()
+			resume()
 			slog.ErrorContext(ctx, "SetSubnets: failed to terminate LB VM for relaunch", "arn", *input.LoadBalancerArn, "instanceId", lb.InstanceID, "err", err)
 			return nil, errors.New(awserrors.ErrorServerInternal)
 		}
@@ -268,6 +282,9 @@ func (s *ELBv2ServiceImpl) SetSubnets(ctx context.Context, input *elbv2.SetSubne
 		slog.ErrorContext(ctx, "SetSubnets: failed to persist LB", "arn", *input.LoadBalancerArn, "err", err)
 		return nil, errors.New(awserrors.ErrorServerInternal)
 	}
+	if err := s.store.ResumeLBHealth(ctx, lb.LoadBalancerID, lb.InstanceID); err != nil {
+		slog.WarnContext(ctx, "SetSubnets: failed to lift data-path suspension", "lbId", lb.LoadBalancerID, "err", err)
+	}
 
 	slog.InfoContext(ctx, "SetSubnets completed", "arn", *input.LoadBalancerArn, "subnets", len(desired), "added", len(toAdd), "removed", len(toRemove), "state", lb.State)
 	return s.setSubnetsOutput(ctx, lb), nil

spinifex/domains/elbv2/service_impl_test.go

@@ -2108,6 +2108,8 @@ func (m *mockTerminateLauncher) TerminateSystemInstance(instanceID string) error
 	return nil
 }
 
+func (m *mockTerminateLauncher) RebootSystemInstance(_ string) error { return nil }
+
 // --- Scheme unit tests ---
 
 func TestCreateLoadBalancer_InternetFacingScheme_DNSName(t *testing.T) {

spinifex/domains/elbv2/system_instance_launcher_nats.go

@@ -96,4 +96,25 @@ func (l *natsSystemInstanceLauncher) TerminateSystemInstance(instanceID string)
 	return nil
 }
 
+func (l *natsSystemInstanceLauncher) RebootSystemInstance(instanceID string) error {
+	if instanceID == "" {
+		return fmt.Errorf("system instance reboot: empty instance ID")
+	}
+
+	subject := fmt.Sprintf("system.RebootInstance.%s", instanceID)
+	reply, err := l.nc.Request(subject, nil, l.timeout)
+	if err != nil {
+		return fmt.Errorf("nats request %s: %w", subject, err)
+	}
+
+	var env natsSystemInstanceTerminateEnvelope
+	if err := json.Unmarshal(reply.Data, &env); err != nil {
+		return fmt.Errorf("decode reboot reply: %w", err)
+	}
+	if env.Error != "" {
+		return fmt.Errorf("%s", env.Error)
+	}
+	return nil
+}
+
 var _ SystemInstanceLauncher = (*natsSystemInstanceLauncher)(nil)

spinifex/domains/elbv2/system_instance_launcher_nats_test.go

@@ -4,6 +4,7 @@ import (
 	"encoding/json"
 	"errors"
 	"sync"
+	"sync/atomic"
 	"testing"
 	"time"
 
@@ -311,3 +312,69 @@ func TestNATSSystemInstanceLauncher_RefusesNonLoadBalancerKindBeforeSending(t *t
 	defer mu.Unlock()
 	assert.Equal(t, 1, received, "only the load balancer launch may be sent")
 }
+
+func TestNATSSystemInstanceLauncher_RebootEmptyID(t *testing.T) {
+	t.Parallel()
+	_, nc, _ := testutil.StartTestJetStream(t)
+	defer nc.Close()
+
+	err := NewNATSSystemInstanceLauncher(nc, time.Second).RebootSystemInstance("")
+	require.Error(t, err)
+	assert.Contains(t, err.Error(), "empty instance ID")
+}
+
+func TestNATSSystemInstanceLauncher_RebootRoundTrip(t *testing.T) {
+	t.Parallel()
+	_, nc, _ := testutil.StartTestJetStream(t)
+	defer nc.Close()
+
+	var asked atomic.Int32
+	sub, err := nc.Subscribe("system.RebootInstance.i-lb", func(msg *nats.Msg) {
+		asked.Add(1)
+		_ = msg.Respond([]byte(`{}`))
+	})
+	require.NoError(t, err)
+	defer func() { _ = sub.Unsubscribe() }()
+
+	require.NoError(t, NewNATSSystemInstanceLauncher(nc, 2*time.Second).RebootSystemInstance("i-lb"))
+	assert.EqualValues(t, 1, asked.Load())
+}
+
+func TestNATSSystemInstanceLauncher_RebootNoResponderKeepsTransportError(t *testing.T) {
+	t.Parallel()
+	_, nc, _ := testutil.StartTestJetStream(t)
+	defer nc.Close()
+
+	err := NewNATSSystemInstanceLauncher(nc, 250*time.Millisecond).RebootSystemInstance("i-no-one-listens")
+	require.ErrorIs(t, err, nats.ErrNoResponders)
+}
+
+func TestNATSSystemInstanceLauncher_RebootDecodeFailureAndRemoteError(t *testing.T) {
+	t.Parallel()
+	_, nc, _ := testutil.StartTestJetStream(t)
+	defer nc.Close()
+
+	bad, err := nc.Subscribe("system.RebootInstance.i-decode", func(msg *nats.Msg) {
+		_ = msg.Respond([]byte("{not valid json"))
+	})
+	require.NoError(t, err)
+	defer func() { _ = bad.Unsubscribe() }()
+
+	refused, err := nc.Subscribe("system.RebootInstance.i-refused", func(msg *nats.Msg) {
+		payload, _ := json.Marshal(struct {
+			Error string `json:"error,omitempty"`
+		}{Error: "not a system instance"})
+		_ = msg.Respond(payload)
+	})
+	require.NoError(t, err)
+	defer func() { _ = refused.Unsubscribe() }()
+
+	l := NewNATSSystemInstanceLauncher(nc, 2*time.Second)
+	err = l.RebootSystemInstance("i-decode")
+	require.Error(t, err)
+	assert.Contains(t, err.Error(), "decode reboot reply")
+
+	err = l.RebootSystemInstance("i-refused")
+	require.Error(t, err)
+	assert.Contains(t, err.Error(), "not a system instance")
+}

spinifex/domains/elbv2/types.go

@@ -155,6 +155,7 @@ type LoadBalancerRecord struct {
 	Tags             map[string]string  `json:"tags,omitempty"`
 	AccountID        string             `json:"account_id"`
 	CreatedAt        time.Time          `json:"created_at"`
+	ActiveSince      time.Time          `json:"active_since,omitzero"` // First heartbeat of the current VM generation
 }
 
 // HealthTargetSpec is a backend the nginx agent actively health-checks, computed

spinifex/domains/ochre/fakes_test.go

@@ -360,6 +360,8 @@ func (f *fakeInstanceLauncher) TerminateSystemInstance(instanceID string) error
 	return nil
 }
 
+func (f *fakeInstanceLauncher) RebootSystemInstance(string) error { return nil }
+
 // fakeAMIResolver always resolves the same serving AMI.
 type fakeAMIResolver struct{}
 

spinifex/runtime/daemon/daemon.go

@@ -2331,6 +2331,7 @@ func (d *Daemon) startCluster() error {
 	}
 
 	d.elbv2Service.StartLifecycleReaper(context.Background())
+	d.elbv2Service.StartDataPathReconciler(d.ctx, d.node)
 
 	d.eksService, err = initServiceWithRetry("EKS service", func() (*ekscluster.EKSServiceImpl, error) {
 		return ekscluster.NewEKSServiceImpl(d.buildEKSServiceDeps())

spinifex/runtime/daemon/daemon_system_dispatch.go

@@ -97,6 +97,14 @@ func (d *Daemon) serveSystemLaunch(msg *nats.Msg, launch systemLaunchFunc) strin
 			"instanceId", output.InstanceID, "err", subErr)
 	}
 
+	d.mu.Lock()
+	rebootErr := d.subscribeSystemRebootLocked(output.InstanceID)
+	d.mu.Unlock()
+	if rebootErr != nil {
+		slog.Error("system.LaunchInstance: failed to subscribe reboot subject",
+			"instanceId", output.InstanceID, "err", rebootErr)
+	}
+
 	respondWithSystemLaunchOutput(msg, output)
 	return outcomeSuccess
 }
@@ -175,21 +183,121 @@ func (d *Daemon) terminateSystemInstanceRemote(instanceID string) error {
 		return fmt.Errorf("decode routed terminate reply %s: %w", instanceID, err)
 	}
 	if env.Error != "" {
-		if strings.Contains(env.Error, systeminstance.ErrSystemInstanceNotFound.Error()) {
-			return fmt.Errorf("%w: %s", systeminstance.ErrSystemInstanceNotFound, instanceID)
-		}
-		if strings.Contains(env.Error, systeminstance.ErrSystemInstanceNotSystem.Error()) {
-			return fmt.Errorf("%w: %s", systeminstance.ErrSystemInstanceNotSystem, instanceID)
-		}
-		if strings.Contains(env.Error, systeminstance.ErrSystemInstanceOwnershipUnknown.Error()) {
-			return fmt.Errorf("%w: %s", systeminstance.ErrSystemInstanceOwnershipUnknown, instanceID)
-		}
-		return fmt.Errorf("routed terminate %s: %s", instanceID, env.Error)
+		return routedSystemError("terminate", instanceID, env.Error)
 	}
 	slog.Info("terminateSystemInstanceRemote: owner terminated VM", "instanceId", instanceID)
 	return nil
 }
 
+// routedSystemError maps an owner's error text back to the sentinel it carried,
+// so callers keep telling gone, customer-owned and unreadable apart.
+func routedSystemError(op, instanceID, text string) error {
+	for _, sentinel := range []error{
+		systeminstance.ErrSystemInstanceNotFound,
+		systeminstance.ErrSystemInstanceNotSystem,
+		systeminstance.ErrSystemInstanceOwnershipUnknown,
+	} {
+		if strings.Contains(text, sentinel.Error()) {
+			return fmt.Errorf("%w: %s", sentinel, instanceID)
+		}
+	}
+	return fmt.Errorf("routed %s %s: %s", op, instanceID, text)
+}
+
+// rebootSystemInstanceRemote routes a reboot to the owning daemon over
+// system.RebootInstance.{id}. No responder means the owner is unreachable and
+// is reported as ErrSystemInstanceNotFound; the caller retries and must not
+// count it as a repair.
+func (d *Daemon) rebootSystemInstanceRemote(instanceID string) error {
+	if d.natsConn == nil {
+		return fmt.Errorf("%w: %s", systeminstance.ErrSystemInstanceNotFound, instanceID)
+	}
+	subject := systemRebootSubject(instanceID)
+	reply, err := d.natsConn.Request(subject, nil, systemRebootRemoteTimeout)
+	if err != nil {
+		if errors.Is(err, nats.ErrNoResponders) {
+			return fmt.Errorf("%w: %s", systeminstance.ErrSystemInstanceNotFound, instanceID)
+		}
+		return fmt.Errorf("route reboot %s: %w", instanceID, err)
+	}
+	var env systemInstanceTerminateEnvelope
+	if err := json.Unmarshal(reply.Data, &env); err != nil {
+		return fmt.Errorf("decode routed reboot reply %s: %w", instanceID, err)
+	}
+	if env.Error != "" {
+		return routedSystemError("reboot", instanceID, env.Error)
+	}
+	return nil
+}
+
+// systemRebootRemoteTimeout caps a routed reboot round trip. The owner admits
+// the reboot and returns; the guest restart runs in the background.
+const systemRebootRemoteTimeout = 15 * time.Second
+
+func systemRebootSubject(instanceID string) string {
+	return "system.RebootInstance." + instanceID
+}
+
+// subscribeSystemRebootLocked registers the owning-node subscription for
+// system.RebootInstance.{instanceID}. Idempotent; the caller holds d.mu.
+func (d *Daemon) subscribeSystemRebootLocked(instanceID string) error {
+	if d.natsConn == nil {
+		return nil
+	}
+	subject := systemRebootSubject(instanceID)
+	if _, exists := d.natsSubscriptions[subject]; exists {
+		return nil
+	}
+	sub, err := d.natsConn.Subscribe(subject, d.handleSystemRebootInstance)
+	if err != nil {
+		return fmt.Errorf("subscribe %s: %w", subject, err)
+	}
+	d.natsSubscriptions[subject] = sub
+	return nil
+}
+
+// subscribeSystemReboot is subscribeSystemRebootLocked for callers without d.mu.
+func (d *Daemon) subscribeSystemReboot(instanceID string) error {
+	d.mu.Lock()
+	defer d.mu.Unlock()
+	return d.subscribeSystemRebootLocked(instanceID)
+}
+
+// handleSystemRebootInstance is the NATS subscriber for
+// system.RebootInstance.{instanceID}. Only the owning daemon subscribes.
+func (d *Daemon) handleSystemRebootInstance(msg *nats.Msg) {
+	d.systemDispatchWg.Go(func() {
+		defer func() {
+			if r := recover(); r != nil {
+				slog.Error("system.RebootInstance: handler panic", "subject", msg.Subject, "panic", r)
+				respondWithSystemRebootReply(msg, awserrors.ErrorServerInternal)
+			}
+		}()
+		id, ok := strings.CutPrefix(msg.Subject, "system.RebootInstance.")
+		if !ok || id == "" {
+			respondWithSystemRebootReply(msg, awserrors.ErrorInvalidInstanceIDNotFound)
+			return
+		}
+		// The local body, not the routing wrapper: only the owner subscribes.
+		if err := d.rebootSystemInstanceLocal(id); err != nil {
+			respondWithSystemRebootReply(msg, err.Error())
+			return
+		}
+		respondWithSystemRebootReply(msg, "")
+	})
+}
+
+func respondWithSystemRebootReply(msg *nats.Msg, errMsg string) {
+	payload, err := json.Marshal(systemInstanceTerminateEnvelope{Error: errMsg})
+	if err != nil {
+		slog.Error("system.RebootInstance: marshal reply failed", "err", err)
+		return
+	}
+	if err := msg.Respond(payload); err != nil {
+		slog.Error("system.RebootInstance: respond failed", "err", err)
+	}
+}
+
 // subscribeSystemTerminate registers an owning-node subscription for
 // system.TerminateInstance.{instanceID}. Idempotent — calling for an
 // already-bound subscription is a no-op.
@@ -266,6 +374,10 @@ func (d *Daemon) serveSystemTerminateInstance(msg *nats.Msg) {
 		}
 		delete(d.natsSubscriptions, subject)
 	}
+	if sub, exists := d.natsSubscriptions[systemRebootSubject(parts)]; exists {
+		_ = sub.Unsubscribe()
+		delete(d.natsSubscriptions, systemRebootSubject(parts))
+	}
 	d.mu.Unlock()
 
 	respondWithSystemTerminateOK(msg)

spinifex/runtime/daemon/daemon_system_instance.go

@@ -682,13 +682,19 @@ func (d *Daemon) reclaimSystemInstanceEIP(instanceID string, cls systeminstance.
 // customer ID is refused as not a system instance, an unknown ID answered as
 // gone and an unreadable record as ownership unknown; nothing is touched.
 func (d *Daemon) requireSystemInstance(instanceID string) (systeminstance.Classification, error) {
+	return d.requireSystemInstanceFor("TerminateSystemInstance", instanceID)
+}
+
+// requireSystemInstanceFor is requireSystemInstance with the calling operation
+// named in the refusal log.
+func (d *Daemon) requireSystemInstanceFor(op, instanceID string) (systeminstance.Classification, error) {
 	cls, err := d.classifySystemInstance(instanceID)
 	if err != nil {
-		slog.Warn("TerminateSystemInstance: refused, ownership could not be read", "instanceId", instanceID, "err", err)
+		slog.Warn(op+": refused, ownership could not be read", "instanceId", instanceID, "err", err)
 		return cls, fmt.Errorf("%w: %s: %w", systeminstance.ErrSystemInstanceOwnershipUnknown, instanceID, err)
 	}
 	if cls.Ownership != systeminstance.OwnershipSystem {
-		slog.Warn("TerminateSystemInstance: refused, not a system instance", "instanceId", instanceID, "ownership", cls.Ownership)
+		slog.Warn(op+": refused, not a system instance", "instanceId", instanceID, "ownership", cls.Ownership)
 		if cls.Ownership == systeminstance.OwnershipCustomer {
 			return cls, fmt.Errorf("%w: %s", systeminstance.ErrSystemInstanceNotSystem, instanceID)
 		}
@@ -738,6 +744,43 @@ func (d *Daemon) terminateSystemInstanceLocal(instanceID string) error {
 	return nil
 }
 
+// RebootSystemInstance restarts a load balancer VM into a fresh QEMU, keeping
+// its instance ID, ENIs and addresses. When this node does not run the VM the
+// request is routed to the owning daemon; no responder means the owner is
+// unreachable, reported as ErrSystemInstanceNotFound for the caller to retry.
+func (d *Daemon) RebootSystemInstance(instanceID string) error {
+	if _, err := d.requireSystemInstanceFor("RebootSystemInstance", instanceID); err != nil {
+		return err
+	}
+	if _, exists := d.vmMgr.Get(instanceID); exists {
+		return d.rebootSystemInstanceLocal(instanceID)
+	}
+	return d.rebootSystemInstanceRemote(instanceID)
+}
+
+// rebootSystemInstanceLocal restarts a VM owned by this node. It refuses unless
+// the VM is a load balancer in the system account, and never terminates.
+func (d *Daemon) rebootSystemInstanceLocal(instanceID string) error {
+	if _, err := d.requireSystemInstanceFor("RebootSystemInstance", instanceID); err != nil {
+		return err
+	}
+	instance, exists := d.vmMgr.Get(instanceID)
+	if !exists {
+		return fmt.Errorf("%w: %s", systeminstance.ErrSystemInstanceNotFound, instanceID)
+	}
+	if instance.ManagedBy != tags.ManagedByELBv2 || instance.AccountID != awsidentifiers.GlobalAccountID {
+		slog.Warn("RebootSystemInstance: refused, not a load balancer system VM",
+			"instanceId", instanceID, "managedBy", instance.ManagedBy)
+		return fmt.Errorf("%w: %s", systeminstance.ErrSystemInstanceNotSystem, instanceID)
+	}
+	if err := d.vmMgr.Reboot(context.Background(), instanceID); err != nil {
+		slog.Error("RebootSystemInstance: vmMgr.Reboot failed", "instanceId", instanceID, "err", err)
+		return fmt.Errorf("reboot: %w", err)
+	}
+	slog.Info("RebootSystemInstance admitted", "instanceId", instanceID)
+	return nil
+}
+
 // prepareInstanceRelaunch makes this node's world match an instance record it is
 // about to launch from.
 //
@@ -1093,3 +1136,5 @@ func buildNetcfgBlob(nics []elbv2domain.NICConfig) (string, error) {
 	}
 	return sb.String(), nil
 }
+
+var _ systeminstance.SystemInstanceLauncher = (*Daemon)(nil)

spinifex/runtime/daemon/daemon_system_reboot_test.go

@@ -0,0 +1,209 @@
+package daemon
+
+import (
+	"strings"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/mulgadc/spinifex/spinifex/domains/ec2/systeminstance"
+	elbv2domain "github.com/mulgadc/spinifex/spinifex/domains/elbv2"
+	awsidentifiers "github.com/mulgadc/spinifex/spinifex/foundation/aws/identifiers"
+	"github.com/mulgadc/spinifex/spinifex/foundation/aws/tags"
+	"github.com/mulgadc/spinifex/spinifex/runtime/compute/qmp"
+	"github.com/mulgadc/spinifex/spinifex/runtime/compute/vm"
+	"github.com/nats-io/nats.go"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+// withMonitor gives a test VM a monitor client. A reboot ends the old process
+// and marks that client dead, which makes "Reboot ran" observable.
+func withMonitor(v *vm.VM) *qmp.QMPClient {
+	v.QMPClient = &qmp.QMPClient{}
+	return v.QMPClient
+}
+
+func monitorDead(q *qmp.QMPClient) bool {
+	q.Mu.Lock()
+	defer q.Mu.Unlock()
+	return q.Dead
+}
+
+// Verifies: ADR-0003:S7.
+func TestS7_LBRepairRefusesACustomerInstance(t *testing.T) {
+	cases := []struct {
+		name string
+		inst *vm.VM
+		loc  recordLocation
+	}{
+		{"running customer", &vm.VM{ID: "i-r-running", Status: vm.StateRunning, AccountID: testAccountID}, atLive},
+		{"stopped customer", &vm.VM{ID: "i-r-stopped", Status: vm.StateStopped, AccountID: testAccountID}, atStopped},
+		{"load balancer manager in a customer account", &vm.VM{ID: "i-r-lbcust", Status: vm.StateRunning, AccountID: testAccountID, ManagedBy: tags.ManagedByELBv2}, atLive},
+		{"customer account holding an EKS tag", &vm.VM{ID: "i-r-eks-cust", Status: vm.StateRunning, AccountID: testAccountID, ManagedBy: tags.ManagedByEKS}, atLive},
+		{"system VM that is not a load balancer", &vm.VM{ID: "i-r-eks", Status: vm.StateRunning, AccountID: awsidentifiers.GlobalAccountID, ManagedBy: tags.ManagedByEKS}, atLive},
+	}
+	for _, route := range []string{"local", "wrapper", "routed", "nats launcher"} {
+		t.Run(route, func(t *testing.T) {
+			d := guardDaemon(t)
+			terminated := countTerminations(d)
+			launcher := elbv2domain.NewNATSSystemInstanceLauncher(d.natsConn, 5*time.Second)
+
+			for _, c := range cases {
+				mon := withMonitor(c.inst)
+				stage(t, d, c.inst, c.loc)
+				var err error
+				switch route {
+				case "local":
+					err = d.rebootSystemInstanceLocal(c.inst.ID)
+				case "wrapper":
+					err = d.RebootSystemInstance(c.inst.ID)
+				case "routed":
+					require.NoError(t, d.subscribeSystemReboot(c.inst.ID))
+					err = d.rebootSystemInstanceRemote(c.inst.ID)
+				case "nats launcher":
+					require.NoError(t, d.subscribeSystemReboot(c.inst.ID))
+					err = launcher.RebootSystemInstance(c.inst.ID)
+				}
+				require.Error(t, err, c.name)
+				assert.Contains(t, err.Error(), systeminstance.ErrSystemInstanceNotSystem.Error(), c.name)
+				assert.NotContains(t, err.Error(), systeminstance.ErrSystemInstanceNotFound.Error(), c.name)
+				d.vmMgr.WaitForBackgroundWork()
+				assert.False(t, monitorDead(mon), "%s: Reboot must never be reached", c.name)
+			}
+			assert.Zero(t, terminated.Load())
+		})
+	}
+}
+
+// Verifies: ADR-0003:S7.
+func TestS7_LBRepairRebootsAndNeverTerminates(t *testing.T) {
+	for _, route := range []string{"wrapper", "routed", "nats launcher"} {
+		t.Run(route, func(t *testing.T) {
+			d := guardDaemon(t)
+			var mu sync.Mutex
+			var moves []vm.InstanceState
+			d.vmMgr.SetDeps(vm.Deps{NodeID: d.node, TransitionState: func(_ *vm.VM, to vm.InstanceState) error {
+				mu.Lock()
+				defer mu.Unlock()
+				moves = append(moves, to)
+				return nil
+			}})
+			eip := &sysInstReclaimingEIP{}
+			d.eipService = eip
+			lb := &vm.VM{ID: "i-lb-" + strings.ReplaceAll(route, " ", "-"), Status: vm.StateRunning, AccountID: awsidentifiers.GlobalAccountID, ManagedBy: tags.ManagedByELBv2}
+			mon := withMonitor(lb)
+			stage(t, d, lb, atLive)
+
+			var err error
+			switch route {
+			case "wrapper":
+				err = d.RebootSystemInstance(lb.ID)
+			case "routed":
+				require.NoError(t, d.subscribeSystemReboot(lb.ID))
+				err = d.rebootSystemInstanceRemote(lb.ID)
+			case "nats launcher":
+				require.NoError(t, d.subscribeSystemReboot(lb.ID))
+				err = elbv2domain.NewNATSSystemInstanceLauncher(d.natsConn, 5*time.Second).RebootSystemInstance(lb.ID)
+			}
+			require.NoError(t, err)
+
+			require.Eventually(t, func() bool { return monitorDead(mon) }, 5*time.Second, 10*time.Millisecond, "the owner restarts the guest")
+			d.vmMgr.WaitForBackgroundWork()
+			assert.NotEqual(t, vm.StateTerminated, d.vmMgr.Status(lb))
+			mu.Lock()
+			defer mu.Unlock()
+			assert.NotContains(t, moves, vm.StateShuttingDown, "a repair never terminates")
+			assert.NotContains(t, moves, vm.StateTerminated, "a repair never terminates")
+			assert.Empty(t, eip.released, "a repair never releases an address")
+			assert.Empty(t, eip.reclaimed)
+		})
+	}
+}
+
+func TestRebootSystemInstance_NoOwnerAndUnreadableRecord(t *testing.T) {
+	t.Run("no owner subscribed", func(t *testing.T) {
+		d := guardDaemon(t)
+		err := d.rebootSystemInstanceRemote("i-nobody-runs-this")
+		require.ErrorIs(t, err, systeminstance.ErrSystemInstanceNotFound)
+
+		err = elbv2domain.NewNATSSystemInstanceLauncher(d.natsConn, 2*time.Second).RebootSystemInstance("i-nobody-runs-this")
+		require.Error(t, err)
+		assert.ErrorIs(t, err, nats.ErrNoResponders, "an unreachable owner stays distinguishable")
+	})
+
+	t.Run("a record this node cannot read", func(t *testing.T) {
+		d := guardDaemon(t)
+		lb := &vm.VM{ID: "i-lb-unreadable", Status: vm.StateRunning, AccountID: awsidentifiers.GlobalAccountID, ManagedBy: tags.ManagedByELBv2}
+		mon := withMonitor(lb)
+		d.vmMgr.Insert(lb)
+		d.ownership = fakeOwnership{err: errInjected}
+
+		err := d.RebootSystemInstance(lb.ID)
+		require.ErrorIs(t, err, systeminstance.ErrSystemInstanceOwnershipUnknown)
+		d.vmMgr.WaitForBackgroundWork()
+		assert.False(t, monitorDead(mon))
+	})
+
+	t.Run("a VM that is not running", func(t *testing.T) {
+		d := guardDaemon(t)
+		lb := &vm.VM{ID: "i-lb-stopped", Status: vm.StateStopped, AccountID: awsidentifiers.GlobalAccountID, ManagedBy: tags.ManagedByELBv2}
+		stage(t, d, lb, atLive)
+		require.Error(t, d.RebootSystemInstance(lb.ID))
+	})
+}
+
+func TestRebootSystemInstanceRemote_ReplyAndConnectionFailures(t *testing.T) {
+	t.Run("no connection reads as an unreachable owner", func(t *testing.T) {
+		d := guardDaemon(t)
+		d.natsConn = nil
+		require.ErrorIs(t, d.rebootSystemInstanceRemote("i-lb"), systeminstance.ErrSystemInstanceNotFound)
+	})
+
+	t.Run("an undecodable reply is an error, not a reboot", func(t *testing.T) {
+		d := guardDaemon(t)
+		sub, err := d.natsConn.Subscribe(systemRebootSubject("i-garbled"), func(msg *nats.Msg) {
+			_ = msg.Respond([]byte("{not json"))
+		})
+		require.NoError(t, err)
+		defer func() { _ = sub.Unsubscribe() }()
+
+		err = d.rebootSystemInstanceRemote("i-garbled")
+		require.Error(t, err)
+		assert.NotErrorIs(t, err, systeminstance.ErrSystemInstanceNotFound)
+	})
+
+	t.Run("an owner that refuses is reported to the caller", func(t *testing.T) {
+		d := guardDaemon(t)
+		sub, err := d.natsConn.Subscribe(systemRebootSubject("i-refusing"), func(msg *nats.Msg) {
+			respondWithSystemRebootReply(msg, "not a system instance")
+		})
+		require.NoError(t, err)
+		defer func() { _ = sub.Unsubscribe() }()
+
+		require.Error(t, d.rebootSystemInstanceRemote("i-refusing"))
+	})
+}
+
+func TestHandleSystemRebootInstance_RejectsAMalformedSubject(t *testing.T) {
+	d := guardDaemon(t)
+	sub, err := d.natsConn.Subscribe("system.Elsewhere.i-lb", d.handleSystemRebootInstance)
+	require.NoError(t, err)
+	defer func() { _ = sub.Unsubscribe() }()
+
+	reply, err := d.natsConn.Request("system.Elsewhere.i-lb", nil, 2*time.Second)
+	require.NoError(t, err)
+	assert.Contains(t, string(reply.Data), "InvalidInstanceID.NotFound")
+	d.systemDispatchWg.Wait()
+}
+
+func TestSubscribeSystemReboot_IdempotentAndSkippedWithoutConnection(t *testing.T) {
+	d := guardDaemon(t)
+	require.NoError(t, d.subscribeSystemReboot("i-lb"))
+	require.NoError(t, d.subscribeSystemReboot("i-lb"))
+	assert.Contains(t, d.natsSubscriptions, systemRebootSubject("i-lb"))
+
+	d.natsConn = nil
+	require.NoError(t, d.subscribeSystemReboot("i-other"))
+	assert.NotContains(t, d.natsSubscriptions, systemRebootSubject("i-other"))
+}

spinifex/runtime/daemon/vm_adapters.go

@@ -739,6 +739,10 @@ func (d *Daemon) onInstanceUpHook() func(*vm.VM) error {
 				slog.Error("OnInstanceUp: failed to re-arm system terminate subscription",
 					"instanceId", instance.ID, "err", subErr)
 			}
+			if subErr := d.subscribeSystemRebootLocked(instance.ID); subErr != nil {
+				slog.Error("OnInstanceUp: failed to re-arm system reboot subscription",
+					"instanceId", instance.ID, "err", subErr)
+			}
 		}
 
 		// Re-claim GPU(s) after a daemon restart with a still-running QEMU
@@ -843,6 +847,13 @@ func (d *Daemon) onInstanceDownHook() func(string) {
 			}
 			delete(d.natsSubscriptions, terminateSubKey)
 		}
+		if sub, ok := d.natsSubscriptions[systemRebootSubject(instanceID)]; ok {
+			if err := sub.Unsubscribe(); err != nil {
+				slog.Error("OnInstanceDown: failed to unsubscribe system reboot topic",
+					"instanceId", instanceID, "err", err)
+			}
+			delete(d.natsSubscriptions, systemRebootSubject(instanceID))
+		}
 	}
 }
 

tests/e2e/eks/eks_test.go

@@ -242,6 +242,10 @@ func TestEKS(t *testing.T) {
 		runEBSCSIVolume(t, c, env, artifacts, fx)
 	})
 
+	t.Run("EndpointNICSelfHeal", func(t *testing.T) {
+		runEndpointNICSelfHeal(t, c, env, artifacts)
+	})
+
 	parent := t
 	t.Run("NodegroupDelete", func(t *testing.T) {
 		requireClusterReady(t, fx)
@@ -1060,8 +1064,23 @@ type clusterFixture struct {
 	BystanderProfile string
 }
 
+// clusterShape varies the cluster a fixture builds. The zero value is the
+// shared fixture: public endpoint only, in the suite's own VPC.
+type clusterShape struct {
+	vpcCIDR, subnetCIDR, publicSubnetCIDR string
+	privateAccess                         bool
+}
+
 func setupClusterFixture(t *testing.T, c *harness.AWSClient, env *harness.Env, artifacts string) *clusterFixture {
 	t.Helper()
+	return setupClusterFixtureShaped(t, c, env, artifacts, clusterShape{})
+}
+
+func setupClusterFixtureShaped(t *testing.T, c *harness.AWSClient, env *harness.Env, artifacts string, shape clusterShape) *clusterFixture {
+	t.Helper()
+	if shape.vpcCIDR == "" {
+		shape.vpcCIDR, shape.subnetCIDR, shape.publicSubnetCIDR = eksVPCCIDR, eksSubnetCIDR, eksPublicSubnetCIDR
+	}
 	fx := &clusterFixture{ClusterName: fmt.Sprintf("%s-%d", eksClusterPfx, time.Now().Unix())}
 
 	harness.Phase(t, "Resolving caller account")
@@ -1070,12 +1089,12 @@ func setupClusterFixture(t *testing.T, c *harness.AWSClient, env *harness.Env, a
 	fx.AccountID = aws.StringValue(ident.Account)
 	t.Logf("account: %s", fx.AccountID)
 
-	harness.Phase(t, "Creating VPC topology (%s)", eksVPCCIDR)
-	fx.VPCID = harness.CreateVPC(t, c, eksVPCCIDR)
+	harness.Phase(t, "Creating VPC topology (%s)", shape.vpcCIDR)
+	fx.VPCID = harness.CreateVPC(t, c, shape.vpcCIDR)
 	t.Cleanup(func() { harness.DeleteVPC(t, c, fx.VPCID) })
-	fx.SubnetID = harness.CreateSubnet(t, c, fx.VPCID, eksSubnetCIDR)
+	fx.SubnetID = harness.CreateSubnet(t, c, fx.VPCID, shape.subnetCIDR)
 	t.Cleanup(func() { harness.DeleteSubnet(t, c, fx.SubnetID) })
-	fx.Egress = harness.CreateWorkerEgress(t, c, fx.VPCID, fx.SubnetID, eksPublicSubnetCIDR)
+	fx.Egress = harness.CreateWorkerEgress(t, c, fx.VPCID, fx.SubnetID, shape.publicSubnetCIDR)
 	t.Cleanup(func() { harness.DeleteWorkerEgress(t, c, fx.Egress) })
 
 	harness.Phase(t, "Creating cluster %q", fx.ClusterName)
@@ -1088,7 +1107,8 @@ func setupClusterFixture(t *testing.T, c *harness.AWSClient, env *harness.Env, a
 			// Public access (default). The customer VPC has a NAT Gateway
 			// (createWorkerEgress), so the private worker SNATs out to reach
 			// the internet-facing NLB endpoint and pull public CSI images.
-			EndpointPublicAccess: aws.Bool(true),
+			EndpointPublicAccess:  aws.Bool(true),
+			EndpointPrivateAccess: aws.Bool(shape.privateAccess),
 		},
 	})
 	require.NoError(t, err, "create-cluster")

tests/e2e/eks/endpoint_nic_heal_test.go

@@ -0,0 +1,247 @@
+//go:build e2e
+
+package eks
+
+import (
+	"fmt"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/aws/aws-sdk-go/aws"
+	"github.com/aws/aws-sdk-go/service/ec2"
+	"github.com/aws/aws-sdk-go/service/eks"
+	"github.com/aws/aws-sdk-go/service/elbv2"
+	"github.com/mulgadc/spinifex/tests/e2e/harness"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+const (
+	// Workers must be Ready again this long after the NIC is disabled.
+	endpointHealBudget = 10 * time.Minute
+)
+
+// runEndpointNICSelfHeal disables the private-endpoint NIC of the cluster's
+// endpoint NLB, the failure that left every worker NotReady, and asserts the
+// platform repairs it: every worker is Ready again within ten minutes, the
+// cluster stays ACTIVE, and the NLB guest is the same instance.
+//
+// The shared cluster has no private endpoint, as in AWS where it is off by
+// default, so this check builds its own cluster with private access on.
+func runEndpointNICSelfHeal(t *testing.T, c *harness.AWSClient, env *harness.Env, artifacts string) {
+	fx := setupClusterFixtureShaped(t, c, env, artifacts, clusterShape{
+		vpcCIDR: "10.211.0.0/16", subnetCIDR: "10.211.1.0/24", publicSubnetCIDR: "10.211.2.0/24", privateAccess: true,
+	})
+	t.Cleanup(func() { deleteClusterBestEffort(t, c, fx) })
+	fx.Cluster = harness.WaitForEKSClusterActive(t, c, fx.ClusterName)
+
+	nodeRoleArn, _ := ensureNodeRole(t, c, fx)
+	createWorkerNodegroup(t, c, fx, "nic-heal-ng", nodeRoleArn, nil)
+
+	kc := harness.NewKubectl(t, writeKubeconfig(t, artifacts, fx.Cluster), getTokenEnv(t, env))
+	requireWorkersReady(t, kc, 5*time.Minute)
+
+	sys := harness.SystemAWSClient(t, env)
+	sysInstance, node := findEndpointNLBGuest(t, sys, fx)
+	harness.Detail(t, "endpoint_nlb_instance", sysInstance, "node", node.Name)
+
+	procBefore, err := harness.ProcessIdentity(node, sysInstance)
+	require.NoError(t, err, "the endpoint NLB's guest must be running on %s", node.Name)
+
+	// The NIC frontend name depends on how the guest was built and whether the
+	// ENI was attached at launch, so it is found from the guest's own NICs: the
+	// one that carries an interface of the customer's cluster VPC.
+	endpointNIC, endpointMAC := endpointNICName(t, c, sys, node, sysInstance, fx)
+	harness.Detail(t, "endpoint_eni_mac", endpointMAC, "nic", endpointNIC)
+
+	faultAt := time.Now()
+	harness.Step(t, "disable %s of %s on %s", endpointNIC, sysInstance, node.Name)
+	// A NIC that cannot be reached through QMP stops the step: the criterion
+	// then needs the person's decision rather than a weaker test.
+	require.NoError(t, harness.LBSetLink(t, node, sysInstance, endpointNIC, false),
+		"the private-endpoint NIC could not be disabled through QMP; stop and ask whether to revise the criterion")
+
+	harness.Step(t, "every worker is Ready again within %s", endpointHealBudget)
+	sawNotReady := false
+	harness.EventuallyErr(t, func() error {
+		cl, derr := c.EKS.DescribeCluster(&eks.DescribeClusterInput{Name: aws.String(fx.ClusterName)})
+		if derr != nil {
+			return fmt.Errorf("describe-cluster: %w", derr)
+		}
+		if s := aws.StringValue(cl.Cluster.Status); s != eks.ClusterStatusActive {
+			return fmt.Errorf("cluster is %s, want %s throughout", s, eks.ClusterStatusActive)
+		}
+		ready, rerr := workersReady(kc)
+		if rerr != nil {
+			sawNotReady = true
+			return rerr
+		}
+		if !ready {
+			sawNotReady = true
+			return fmt.Errorf("a worker is not Ready")
+		}
+		if !sawNotReady && time.Since(faultAt) < 2*time.Minute {
+			return fmt.Errorf("workers still Ready before the fault has had any effect")
+		}
+		return nil
+	}, endpointHealBudget, 10*time.Second)
+
+	procAfter, err := harness.ProcessIdentity(node, sysInstance)
+	require.NoError(t, err, "the endpoint NLB's guest must be running after the repair")
+	assert.NotEqual(t, procBefore, procAfter, "the endpoint NLB's guest process is unchanged, so nothing rebooted it")
+
+	again, _ := findEndpointNLBGuest(t, sys, fx)
+	assert.Equal(t, sysInstance, again, "the endpoint NLB changed guest instead of rebooting it")
+
+	cl, err := c.EKS.DescribeCluster(&eks.DescribeClusterInput{Name: aws.String(fx.ClusterName)})
+	require.NoError(t, err)
+	assert.Equal(t, eks.ClusterStatusActive, aws.StringValue(cl.Cluster.Status))
+
+	started := harness.DaemonLogLines(clusterNodes(t), faultAt.Add(-time.Minute), "ELBv2 data-path repair started", sysInstance)
+	total := 0
+	for _, lines := range started {
+		total += len(lines)
+	}
+	assert.Equalf(t, 1, total, "want exactly one repair of %s, got %d: %v", sysInstance, total, started)
+}
+
+// endpointNICName returns the guest NIC name and MAC of the endpoint NLB's
+// interface in the cluster's own VPC, the one workers reach the API server
+// through. It fails unless exactly one NIC of the guest is an interface of that
+// VPC, so a guest without the endpoint ENI is reported rather than guessed at.
+func endpointNICName(t *testing.T, c, sys *harness.AWSClient, node harness.Node, instanceID string, fx *clusterFixture) (string, string) {
+	t.Helper()
+	nics, err := harness.LBNICsByMAC(t, node, instanceID)
+	require.NoError(t, err, "read the NICs of the endpoint NLB's guest")
+	out, err := c.EC2.DescribeNetworkInterfaces(&ec2.DescribeNetworkInterfacesInput{
+		Filters: []*ec2.Filter{{Name: aws.String("vpc-id"), Values: []*string{aws.String(fx.VPCID)}}},
+	})
+	require.NoError(t, err, "describe the cluster VPC's interfaces")
+	var names, macs []string
+	for _, eni := range out.NetworkInterfaces {
+		mac := strings.ToLower(aws.StringValue(eni.MacAddress))
+		if name, ok := nics[mac]; ok {
+			names = append(names, name)
+			macs = append(macs, mac)
+		}
+	}
+	if len(names) != 1 {
+		logEndpointENIs(t, c, sys, instanceID, fx)
+	}
+	require.Lenf(t, names, 1, "want exactly one NIC of %s in cluster VPC %s, the guest has %v and the VPC matched %v", instanceID, fx.VPCID, nics, macs)
+	return names[0], macs[0]
+}
+
+// logEndpointENIs records every interface the system account sees on the
+// endpoint NLB's guest and every interface the customer sees in the cluster VPC,
+// so a missing endpoint NIC can be told from an invisible one.
+func logEndpointENIs(t *testing.T, c, sys *harness.AWSClient, instanceID string, fx *clusterFixture) {
+	t.Helper()
+	dump := func(who string, cl *harness.AWSClient, filter *ec2.Filter) {
+		out, err := cl.EC2.DescribeNetworkInterfaces(&ec2.DescribeNetworkInterfacesInput{Filters: []*ec2.Filter{filter}})
+		if err != nil {
+			t.Logf("%s view: describe-network-interfaces: %v", who, err)
+			return
+		}
+		for _, eni := range out.NetworkInterfaces {
+			attached := ""
+			if eni.Attachment != nil {
+				attached = aws.StringValue(eni.Attachment.InstanceId)
+			}
+			t.Logf("%s view: %s mac=%s subnet=%s vpc=%s status=%s attached=%q desc=%q", who,
+				aws.StringValue(eni.NetworkInterfaceId), aws.StringValue(eni.MacAddress), aws.StringValue(eni.SubnetId),
+				aws.StringValue(eni.VpcId), aws.StringValue(eni.Status), attached, aws.StringValue(eni.Description))
+		}
+	}
+	dump("system, attached to guest", sys, &ec2.Filter{Name: aws.String("attachment.instance-id"), Values: []*string{aws.String(instanceID)}})
+	dump("customer, cluster VPC", c, &ec2.Filter{Name: aws.String("vpc-id"), Values: []*string{aws.String(fx.VPCID)}})
+}
+
+// workersReady reports whether every registered node is Ready, and that at
+// least one node exists.
+func workersReady(kc *harness.Kubectl) (bool, error) {
+	out, err := kc.Run(30*time.Second, "get", "nodes", "-o",
+		`jsonpath={range .items[*]}{.metadata.name}={.status.conditions[?(@.type=="Ready")].status}{"\n"}{end}`)
+	if err != nil {
+		return false, fmt.Errorf("kubectl get nodes: %v\n%s", err, out)
+	}
+	lines := strings.Fields(out)
+	if len(lines) == 0 {
+		return false, fmt.Errorf("no worker has registered")
+	}
+	for _, l := range lines {
+		if !strings.HasSuffix(l, "=True") {
+			return false, nil
+		}
+	}
+	return true, nil
+}
+
+func requireWorkersReady(t *testing.T, kc *harness.Kubectl, budget time.Duration) {
+	t.Helper()
+	harness.EventuallyErr(t, func() error {
+		ready, err := workersReady(kc)
+		if err != nil {
+			return err
+		}
+		if !ready {
+			return fmt.Errorf("a worker is not Ready")
+		}
+		return nil
+	}, budget, 10*time.Second)
+}
+
+func clusterNodes(t *testing.T) *harness.Cluster {
+	t.Helper()
+	cluster, err := harness.ClusterFromEnv()
+	require.NoError(t, err, "this check needs node SSH to disable a NIC and read the repair line")
+	return cluster
+}
+
+// findEndpointNLBGuest returns the guest running the cluster's endpoint NLB and
+// the node it runs on. The NLB belongs to the system account, so it is found
+// there, by the cluster tag.
+func findEndpointNLBGuest(t *testing.T, sys *harness.AWSClient, fx *clusterFixture) (string, harness.Node) {
+	t.Helper()
+	lbs, err := sys.ELBv2.DescribeLoadBalancers(&elbv2.DescribeLoadBalancersInput{})
+	require.NoError(t, err, "describe load balancers as the system account")
+
+	for _, lb := range lbs.LoadBalancers {
+		if aws.StringValue(lb.Type) != elbv2.LoadBalancerTypeEnumNetwork {
+			continue
+		}
+		tagOut, terr := sys.ELBv2.DescribeTags(&elbv2.DescribeTagsInput{ResourceArns: []*string{lb.LoadBalancerArn}})
+		if terr != nil {
+			continue
+		}
+		owned := false
+		for _, td := range tagOut.TagDescriptions {
+			for _, tag := range td.Tags {
+				if strings.Contains(aws.StringValue(tag.Value), fx.ClusterName) {
+					owned = true
+				}
+			}
+		}
+		if !owned {
+			continue
+		}
+		parts := strings.Split(aws.StringValue(lb.LoadBalancerArn), "/")
+		desc := fmt.Sprintf("ELB %s/%s/%s", parts[len(parts)-3], parts[len(parts)-2], parts[len(parts)-1])
+		enis, eerr := sys.EC2.DescribeNetworkInterfaces(&ec2.DescribeNetworkInterfacesInput{
+			Filters: []*ec2.Filter{{Name: aws.String("description"), Values: []*string{aws.String(desc)}}},
+		})
+		require.NoError(t, eerr, "describe the endpoint NLB's interfaces")
+		for _, eni := range enis.NetworkInterfaces {
+			if eni.Attachment == nil || aws.StringValue(eni.Attachment.InstanceId) == "" {
+				continue
+			}
+			id := aws.StringValue(eni.Attachment.InstanceId)
+			host := harness.InstanceHostingNode(t, clusterNodes(t), id)
+			require.NotNilf(t, host, "no node is running the endpoint NLB's guest %s", id)
+			return id, *host
+		}
+	}
+	t.Fatalf("no endpoint NLB found for cluster %s in the system account", fx.ClusterName)
+	return "", harness.Node{}
+}

tests/e2e/harness/lb_fault.go

@@ -0,0 +1,179 @@
+//go:build e2e
+
+package harness
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"regexp"
+	"strings"
+	"testing"
+	"time"
+)
+
+// qmpSetLinkScript sends set_link to a guest's telemetry QMP socket. The daemon
+// holds the main QMP socket for the life of the guest; the telemetry socket is a
+// separate QMP server that accepts a second client between collector polls.
+const qmpSetLinkScript = `import json, socket, sys
+path = "/run/spinifex/qmp-telemetry-%s.sock" % sys.argv[1]
+s = socket.socket(socket.AF_UNIX)
+s.settimeout(5)
+s.connect(path)
+f = s.makefile("rw")
+f.readline()
+def rpc(c):
+    f.write(json.dumps(c) + "\n")
+    f.flush()
+    while True:
+        l = json.loads(f.readline())
+        if "return" in l or "error" in l:
+            return l
+rpc({"execute": "qmp_capabilities"})
+r = rpc({"execute": "set_link", "arguments": {"name": sys.argv[2], "up": sys.argv[3] == "up"}})
+print(json.dumps(r))
+sys.exit(0 if "return" in r else 1)
+`
+
+// qmpRxFilterScript prints query-rx-filter for every NIC frontend of a guest, so
+// a caller can map an ENI's MAC to the device name set_link needs.
+const qmpRxFilterScript = `import json, socket, sys
+path = "/run/spinifex/qmp-telemetry-%s.sock" % sys.argv[1]
+s = socket.socket(socket.AF_UNIX)
+s.settimeout(5)
+s.connect(path)
+f = s.makefile("rw")
+f.readline()
+def rpc(c):
+    f.write(json.dumps(c) + "\n")
+    f.flush()
+    while True:
+        l = json.loads(f.readline())
+        if "return" in l or "error" in l:
+            return l
+rpc({"execute": "qmp_capabilities"})
+r = rpc({"execute": "query-rx-filter"})
+print(json.dumps(r))
+sys.exit(0 if "return" in r else 1)
+`
+
+var (
+	instanceIDShape = regexp.MustCompile(`^i-[0-9a-f]+$`)
+	nicNameShape    = regexp.MustCompile(`^(net|virtio-net-pci\.)[0-9]+$`)
+)
+
+// LBSetLink sets one data NIC of a running load balancer guest up or down
+// through its telemetry QMP socket, retrying while the collector holds it.
+//
+// A guest's own interface names are net0..netN in the order the daemon attached
+// them; the data NIC is net0 and the management NIC is the last one.
+func LBSetLink(t *testing.T, node Node, instanceID, nic string, up bool) error {
+	t.Helper()
+	if !instanceIDShape.MatchString(instanceID) || !nicNameShape.MatchString(nic) {
+		return fmt.Errorf("refusing to inject on %q/%q: not an instance id and a netN or virtio-net-pci.N name", instanceID, nic)
+	}
+	state := "down"
+	if up {
+		state = "up"
+	}
+	cmd := fmt.Sprintf("sudo python3 - %s %s %s <<'PY'\n%sPY", instanceID, nic, state, qmpSetLinkScript)
+
+	var lastErr error
+	for attempt := 1; attempt <= 20; attempt++ {
+		ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
+		out, err := NewPeerSSH().Run(ctx, node.Addr, cmd)
+		cancel()
+		if err == nil {
+			Detail(t, "qmp_set_link", strings.TrimSpace(string(out)), "node", node.Name, "nic", nic, "state", state)
+			return nil
+		}
+		lastErr = err
+		time.Sleep(time.Second)
+	}
+	return fmt.Errorf("set_link %s %s on %s: %w", nic, state, node.Name, lastErr)
+}
+
+// LBNICsByMAC returns a guest's NIC frontend names keyed by lower-case main MAC,
+// read from the guest's own QMP query-rx-filter.
+func LBNICsByMAC(t *testing.T, node Node, instanceID string) (map[string]string, error) {
+	t.Helper()
+	if !instanceIDShape.MatchString(instanceID) {
+		return nil, fmt.Errorf("%q is not an instance id", instanceID)
+	}
+	cmd := fmt.Sprintf("sudo python3 - %s <<'PY'\n%sPY", instanceID, qmpRxFilterScript)
+	var lastErr error
+	for attempt := 1; attempt <= 20; attempt++ {
+		ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
+		out, err := NewPeerSSH().Run(ctx, node.Addr, cmd)
+		cancel()
+		if err == nil {
+			var resp struct {
+				Return []struct {
+					Name    string `json:"name"`
+					MainMAC string `json:"main-mac"`
+				} `json:"return"`
+			}
+			if jerr := json.Unmarshal([]byte(strings.TrimSpace(string(out))), &resp); jerr != nil {
+				return nil, fmt.Errorf("parse query-rx-filter: %w: %s", jerr, strings.TrimSpace(string(out)))
+			}
+			nics := map[string]string{}
+			for _, nic := range resp.Return {
+				nics[strings.ToLower(nic.MainMAC)] = nic.Name
+			}
+			return nics, nil
+		}
+		lastErr = err
+		time.Sleep(time.Second)
+	}
+	return nil, fmt.Errorf("query-rx-filter on %s: %w", node.Name, lastErr)
+}
+
+// ProcessIdentity returns the pid and start time of the QEMU process running an
+// instance on node. A reboot, a relaunch or a restart changes it, so equality
+// before and after is evidence the guest was left alone, and inequality is
+// evidence it was not.
+func ProcessIdentity(node Node, instanceID string) (string, error) {
+	if !instanceIDShape.MatchString(instanceID) {
+		return "", fmt.Errorf("%q is not an instance id", instanceID)
+	}
+	cmd := fmt.Sprintf(
+		"ps -eo pid=,lstart=,args= | grep -E '[q]emu.*%s' | head -1 | awk '{print $1, $2, $3, $4, $5, $6}'",
+		instanceID)
+	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
+	defer cancel()
+	out, err := NewPeerSSH().Run(ctx, node.Addr, cmd)
+	if err != nil {
+		return "", err
+	}
+	id := strings.TrimSpace(string(out))
+	if id == "" {
+		return "", fmt.Errorf("no qemu process for %s on %s", instanceID, node.Name)
+	}
+	return id, nil
+}
+
+// DaemonLogLines returns the spinifex-daemon journal lines on every node that
+// contain all of the given needles, since the given time.
+func DaemonLogLines(c *Cluster, since time.Time, needles ...string) map[string][]string {
+	grep := ""
+	for _, n := range needles {
+		grep += fmt.Sprintf(" | grep -F %s", ShellQuote(n))
+	}
+	out := map[string][]string{}
+	for _, node := range c.Nodes {
+		cmd := fmt.Sprintf("sudo journalctl -u spinifex-daemon --no-pager --since @%d%s || true", since.Unix(), grep)
+		ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second)
+		raw, err := NewPeerSSH().Run(ctx, node.Addr, cmd)
+		cancel()
+		if err != nil {
+			out[node.Name] = []string{"(unreadable: " + err.Error() + ")"}
+			continue
+		}
+		for _, line := range strings.Split(strings.TrimSpace(string(raw)), "\n") {
+			if strings.TrimSpace(line) != "" {
+				out[node.Name] = append(out[node.Name], line)
+			}
+		}
+	}
+	return out
+}

tests/e2e/lbrecovery/selfheal_test.go

@@ -0,0 +1,586 @@
+//go:build e2e
+
+package lbrecovery
+
+import (
+	"encoding/base64"
+	"fmt"
+	"slices"
+	"strconv"
+	"strings"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/aws/aws-sdk-go/aws"
+	"github.com/aws/aws-sdk-go/service/ec2"
+	"github.com/aws/aws-sdk-go/service/elbv2"
+	"github.com/mulgadc/spinifex/tests/e2e/harness"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+const (
+	// Older than every timeout in the repair path, so a repair that reaches a
+	// customer instance would find one the platform has long stopped expecting
+	// to change.
+	shTargetAge = 21 * time.Minute
+
+	// The requirement: traffic is back within five minutes of the fault.
+	shRecoveryBudget = 5 * time.Minute
+
+	// The data NIC of a load balancer guest; the management NIC is the last.
+	shDataNIC = "net0"
+
+	stateActive   = "active"
+	stateImpaired = "active_impaired"
+)
+
+type shKind struct {
+	name    string
+	lbType  string
+	scheme  string
+	tgProto string
+}
+
+var (
+	shInternetFacingNLB = shKind{"InternetFacingNLB", "network", "internet-facing", "TCP"}
+	shInternalALB       = shKind{"InternalALB", "application", "internal", "HTTP"}
+)
+
+// TestLBSelfHealsADeadDataNIC takes one data NIC of a running load balancer
+// guest away and asserts the platform notices, reports the load balancer as
+// impaired, reboots that guest once, and leaves everything else alone.
+//
+// Each kind is its own subtest because the guest, its ENIs and its backends are
+// different, but the assertions are the same: the customer's name and address,
+// the guest's identity, and every instance that is not the load balancer.
+func TestLBSelfHealsADeadDataNIC(t *testing.T) {
+	fix := requireFixture(t)
+	for _, kind := range []shKind{shInternetFacingNLB, shInternalALB} {
+		t.Run(kind.name, func(t *testing.T) { runLBSelfHeal(t, fix, kind) })
+	}
+}
+
+// shInstanceSnap is what a customer instance must still be after the repair.
+type shInstanceSnap struct {
+	id      string
+	launch  time.Time
+	process string
+	node    harness.Node
+}
+
+func runLBSelfHeal(t *testing.T, fix *Fixture, kind shKind) {
+	harness.Phase(t, "A %s repairs a dead data NIC", kind.name)
+	f := shSetup(t, fix, kind)
+	internetFacing := kind.scheme == "internet-facing"
+	cli := fix.AWS
+
+	// The unrelated customer instance shares the subnet and the security group
+	// with the backends. It is the witness that the repair touched only the
+	// load balancer's own guest.
+	bystander := shLaunchBystander(t, fix, f)
+
+	harness.Step(t, "wait until every customer instance is older than %s", shTargetAge)
+	customers := append(slices.Clone(f.TargetIDs), bystander)
+	var client string
+	if !internetFacing {
+		client = shLaunchInVPCClient(t, fix, f)
+		customers = append(customers, client)
+	}
+	shWaitOlderThan(t, cli, customers, shTargetAge)
+
+	if client != "" {
+		shLogClockSkew(t, cli, client)
+	}
+	before := shSnapshot(t, fix, customers)
+	lbBefore := shLBIdentity(t, fix, f)
+	lbProcBefore, err := harness.ProcessIdentity(f.Host, f.SysInstanceID)
+	require.NoError(t, err, "the load balancer's guest must be running on %s", f.Host.Name)
+
+	if internetFacing {
+		harness.Step(t, "prove the name serves traffic before anything is broken")
+		lbrRequireServing(t, fix, f.Host, harness.Node{}, f, "before the fault", 90*time.Second)
+	}
+	harness.WaitForTargetsHealthy(t, cli, f.TGArn, lbrTargets, "backends before the fault", targetsHealthyBudget)
+
+	states := newStateWatch(t, cli, f.LBArn)
+	defer states.stop()
+	var dns *dnsWatch
+	if internetFacing {
+		dns = newDNSWatch(t, fix, harness.Node{}, f)
+		defer dns.stop()
+	}
+
+	faultAt := time.Now()
+	harness.Step(t, "disable %s of %s on %s", shDataNIC, f.SysInstanceID, f.Host.Name)
+	require.NoError(t, harness.LBSetLink(t, f.Host, f.SysInstanceID, shDataNIC, false))
+
+	harness.Step(t, "the load balancer reports %s while the NIC is dead", stateImpaired)
+	require.Eventuallyf(t, func() bool { return states.seen(stateImpaired) }, shRecoveryBudget, 3*time.Second,
+		"the load balancer never reported %s after its data NIC was disabled; states seen: %v",
+		stateImpaired, states.sequence())
+
+	harness.Step(t, "the load balancer is repaired and active again within %s of the fault", shRecoveryBudget)
+	require.Eventuallyf(t, func() bool { return states.current() == stateActive && states.seen(stateImpaired) },
+		shRecoveryBudget-time.Since(faultAt), 3*time.Second,
+		"the load balancer was not active again %s after the fault; states seen: %v", shRecoveryBudget, states.sequence())
+
+	if internetFacing {
+		remaining := max(shRecoveryBudget-time.Since(faultAt), 10*time.Second)
+		lbrRequireServing(t, fix, f.Host, harness.Node{}, f, "after the repair", remaining)
+	} else {
+		harness.Step(t, "the in-VPC client reaches the internal ALB again within %s of the fault", shRecoveryBudget)
+		shRequireClientServed(t, cli, client, faultAt)
+	}
+	harness.WaitForTargetsHealthy(t, cli, f.TGArn, lbrTargets, "backends after the repair", targetsHealthyBudget)
+
+	states.stop()
+	// Never active between the first impaired report and the recovery: a
+	// load balancer with a dead NIC must not claim to be healthy.
+	assert.Equalf(t, []string{stateActive, stateImpaired, stateActive}, states.sequence(),
+		"the load balancer's state must go active, active_impaired, active and nothing else")
+	if dns != nil {
+		dns.stop()
+		dns.requireNameNeverBroke(t, f)
+	}
+
+	harness.Step(t, "same guest, same interfaces, same address")
+	assert.Equal(t, lbBefore, shLBIdentity(t, fix, f), "the repair must not replace the guest, its ENIs or its address")
+	lbProcAfter, err := harness.ProcessIdentity(f.Host, f.SysInstanceID)
+	require.NoError(t, err, "the load balancer's guest must be running on %s after the repair", f.Host.Name)
+	assert.NotEqual(t, lbProcBefore, lbProcAfter, "the guest process is unchanged, so nothing rebooted it")
+
+	harness.Step(t, "customer instances were not touched (ADR-0003:S7)")
+	assert.Equal(t, before, shSnapshot(t, fix, customers),
+		"a customer instance changed identity, launch time or process during a load balancer repair")
+
+	harness.Step(t, "exactly one repair line, with the fields an operator needs")
+	started := harness.DaemonLogLines(fix.Cluster, faultAt.Add(-time.Minute), "ELBv2 data-path repair started", f.LBID)
+	total := 0
+	var line string
+	for _, lines := range started {
+		total += len(lines)
+		if len(lines) > 0 {
+			line = lines[0]
+		}
+	}
+	require.Equalf(t, 1, total, "want exactly one repair for %s, got %d: %v", f.LBID, total, started)
+	for _, field := range []string{"lbArn", "lbId", "instanceId", "eniId", "mac", "failing_ms",
+		"consecutive_failures", "rx_packets", "tx_packets", "repair_attempt", "reason"} {
+		assert.Containsf(t, line, field, "the repair line lacks %s: %s", field, line)
+	}
+	assert.Containsf(t, line, f.SysInstanceID, "the repair line names another guest: %s", line)
+}
+
+const shClientMarker = "SHCLIENT"
+
+// shClientUserData curls the internal load balancer every five seconds from
+// inside the VPC and prints a timestamped verdict on the serial console, the
+// only channel back from a guest.
+func shClientUserData(dnsName string) string {
+	return fmt.Sprintf(`#!/bin/bash
+while true; do
+  if curl -fsS -m 3 -o /dev/null http://%s:%d/index.html 2>/dev/null; then
+    echo "%s $(date +%%s) OK" | tee /dev/console
+  else
+    echo "%s $(date +%%s) FAIL" | tee /dev/console
+  fi
+  sleep 5
+done
+`, dnsName, lbrHTTPPort, shClientMarker, shClientMarker)
+}
+
+func shLaunchInVPCClient(t *testing.T, fix *Fixture, f *albFixture) string {
+	t.Helper()
+	out, err := fix.AWS.EC2.RunInstances(&ec2.RunInstancesInput{
+		ImageId:          aws.String(f.AMIID),
+		InstanceType:     aws.String(f.InstanceType),
+		KeyName:          aws.String(f.KeyName),
+		SubnetId:         aws.String(f.SubnetID),
+		SecurityGroupIds: []*string{aws.String(f.SGID)},
+		MinCount:         aws.Int64(1),
+		MaxCount:         aws.Int64(1),
+		UserData:         aws.String(base64.StdEncoding.EncodeToString([]byte(shClientUserData(f.DNSName)))),
+	})
+	require.NoError(t, err, "launch the in-VPC client")
+	require.NotEmpty(t, out.Instances, "RunInstances returned no instance")
+	id := aws.StringValue(out.Instances[0].InstanceId)
+	t.Cleanup(func() { lbrTerminate(t, fix.AWS, []string{id}) })
+	harness.WaitForInstanceRunning(t, fix.AWS, id, 3*time.Minute)
+	harness.Detail(t, "in-VPC client", id)
+	return id
+}
+
+func shConsole(c *harness.AWSClient, id string) string {
+	out, err := c.EC2.GetConsoleOutput(&ec2.GetConsoleOutputInput{InstanceId: aws.String(id)})
+	if err != nil || aws.StringValue(out.Output) == "" {
+		return ""
+	}
+	raw, derr := base64.StdEncoding.DecodeString(aws.StringValue(out.Output))
+	if derr != nil {
+		return aws.StringValue(out.Output)
+	}
+	return string(raw)
+}
+
+// shClientVerdicts returns the client's verdict lines as unix seconds and
+// whether each succeeded.
+func shClientVerdicts(console string) (times []int64, oks []bool) {
+	for line := range strings.SplitSeq(console, "\n") {
+		fields := strings.Fields(line)
+		if len(fields) != 3 || fields[0] != shClientMarker {
+			continue
+		}
+		ts, err := strconv.ParseInt(fields[1], 10, 64)
+		if err != nil {
+			continue
+		}
+		times = append(times, ts)
+		oks = append(oks, fields[2] == "OK")
+	}
+	return times, oks
+}
+
+// shClockSkewLimit is the guest-to-host clock offset past which the client's
+// timestamps cannot back the timing assertion.
+const shClockSkewLimit = 10 * time.Second
+
+// shLogClockSkew compares the newest guest timestamp on the console with host
+// time. The offset includes up to one probe interval of console delay, so a
+// healthy pair reads a few seconds.
+func shLogClockSkew(t *testing.T, c *harness.AWSClient, client string) {
+	t.Helper()
+	times, _ := shClientVerdicts(shConsole(c, client))
+	if len(times) == 0 {
+		t.Logf("clock skew: the client has printed no verdict yet, so the guest clock is unmeasured")
+		return
+	}
+	last := time.Unix(times[len(times)-1], 0)
+	skew := time.Since(last)
+	harness.Detail(t, "guest clock offset from host", skew.Round(time.Second).String(), "guest_epoch", strconv.FormatInt(last.Unix(), 10))
+	if skew > shClockSkewLimit || skew < -shClockSkewLimit {
+		t.Logf("clock skew %s exceeds %s: the timing assertion is UNPROVEN", skew.Round(time.Second), shClockSkewLimit)
+	}
+}
+
+// shRequireClientServed fails unless the client's first success after the
+// fault was reported no later than one absolute deadline after the fault.
+func shRequireClientServed(t *testing.T, c *harness.AWSClient, client string, faultAt time.Time) {
+	t.Helper()
+	deadline := faultAt.Add(shRecoveryBudget)
+	var console string
+	for time.Now().Before(deadline.Add(15 * time.Second)) {
+		console = shConsole(c, client)
+		times, oks := shClientVerdicts(console)
+		failAt := int64(0)
+		for i, ts := range times {
+			if !oks[i] && ts >= faultAt.Unix() {
+				failAt = ts
+				break
+			}
+		}
+		for i, ts := range times {
+			if failAt != 0 && oks[i] && ts > failAt {
+				require.LessOrEqualf(t, ts, deadline.Unix(),
+					"the client's first success after its first failure came at %d, after the %s deadline %d",
+					ts, shRecoveryBudget, deadline.Unix())
+				harness.Detail(t, "first failure after the fault", time.Unix(failAt, 0).Sub(faultAt).Round(time.Second).String())
+				harness.Detail(t, "first success after that failure", time.Unix(ts, 0).Sub(faultAt).Round(time.Second).String())
+				return
+			}
+		}
+		time.Sleep(5 * time.Second)
+	}
+	lines := strings.Split(strings.TrimSpace(console), "\n")
+	tail := strings.Join(lines[max(0, len(lines)-15):], "\n")
+	times, oks := shClientVerdicts(console)
+	sawFail := false
+	for i, ts := range times {
+		if !oks[i] && ts >= faultAt.Unix() {
+			sawFail = true
+		}
+	}
+	if !sawFail {
+		require.Failf(t, "in-VPC client never failed after the fault",
+			"the client saw no failure after the fault, so it was not on the faulted path and the timing proves nothing; last console lines:\n%s", tail)
+	}
+	require.Failf(t, "in-VPC client never reached the internal ALB",
+		"no success after the first failure within %s of the fault; last console lines:\n%s", shRecoveryBudget, tail)
+}
+
+// stateWatch samples a load balancer's reported state continuously and keeps the
+// run of distinct states, so an interval in which it claimed to be healthy is
+// seen even if nothing was looking at that moment.
+type stateWatch struct {
+	stopOnce sync.Once
+	done     chan struct{}
+	finished chan struct{}
+
+	mu   sync.Mutex
+	runs []string
+}
+
+func newStateWatch(t *testing.T, c *harness.AWSClient, arn string) *stateWatch {
+	t.Helper()
+	w := &stateWatch{done: make(chan struct{}), finished: make(chan struct{})}
+	go func() {
+		defer close(w.finished)
+		ticker := time.NewTicker(2 * time.Second)
+		defer ticker.Stop()
+		for {
+			select {
+			case <-w.done:
+				return
+			case <-ticker.C:
+				out, err := c.ELBv2.DescribeLoadBalancers(&elbv2.DescribeLoadBalancersInput{
+					LoadBalancerArns: []*string{aws.String(arn)},
+				})
+				if err != nil || len(out.LoadBalancers) == 0 || out.LoadBalancers[0].State == nil {
+					continue
+				}
+				w.record(aws.StringValue(out.LoadBalancers[0].State.Code))
+			}
+		}
+	}()
+	return w
+}
+
+func (w *stateWatch) record(state string) {
+	w.mu.Lock()
+	defer w.mu.Unlock()
+	if len(w.runs) == 0 || w.runs[len(w.runs)-1] != state {
+		w.runs = append(w.runs, state)
+	}
+}
+
+func (w *stateWatch) stop() {
+	w.stopOnce.Do(func() {
+		close(w.done)
+		<-w.finished
+	})
+}
+
+func (w *stateWatch) sequence() []string {
+	w.mu.Lock()
+	defer w.mu.Unlock()
+	return slices.Clone(w.runs)
+}
+
+func (w *stateWatch) current() string {
+	w.mu.Lock()
+	defer w.mu.Unlock()
+	if len(w.runs) == 0 {
+		return ""
+	}
+	return w.runs[len(w.runs)-1]
+}
+
+func (w *stateWatch) seen(state string) bool {
+	w.mu.Lock()
+	defer w.mu.Unlock()
+	return slices.Contains(w.runs, state)
+}
+
+// shLBIdentity is what must survive the repair unchanged: the guest, its
+// interfaces and the address.
+func shLBIdentity(t *testing.T, fix *Fixture, f *albFixture) string {
+	t.Helper()
+	out, err := fix.AWS.EC2.DescribeNetworkInterfaces(&ec2.DescribeNetworkInterfacesInput{
+		Filters: []*ec2.Filter{{
+			Name:   aws.String("description"),
+			Values: []*string{aws.String(shENIDescription(f))},
+		}},
+	})
+	require.NoError(t, err, "describe the load balancer's interfaces")
+	ids := make([]string, 0, len(out.NetworkInterfaces))
+	for _, eni := range out.NetworkInterfaces {
+		attached, public := "", ""
+		if eni.Attachment != nil {
+			attached = aws.StringValue(eni.Attachment.InstanceId)
+		}
+		if eni.Association != nil {
+			public = aws.StringValue(eni.Association.PublicIp)
+		}
+		ids = append(ids, fmt.Sprintf("%s|%s|%s", aws.StringValue(eni.NetworkInterfaceId), attached, public))
+	}
+	slices.Sort(ids)
+	return strings.Join(ids, ",")
+}
+
+func shENIDescription(f *albFixture) string {
+	parts := strings.Split(f.LBArn, "/")
+	// arn:...:loadbalancer/<app|net>/<name>/<id>
+	return fmt.Sprintf("ELB %s/%s/%s", parts[len(parts)-3], parts[len(parts)-2], parts[len(parts)-1])
+}
+
+// shSnapshot records, per customer instance, what a reboot or a relaunch would
+// change.
+func shSnapshot(t *testing.T, fix *Fixture, ids []string) []shInstanceSnap {
+	t.Helper()
+	out, err := fix.AWS.EC2.DescribeInstances(&ec2.DescribeInstancesInput{InstanceIds: aws.StringSlice(ids)})
+	require.NoError(t, err, "describe customer instances")
+	launch := map[string]time.Time{}
+	for _, r := range out.Reservations {
+		for _, inst := range r.Instances {
+			require.Equalf(t, "running", aws.StringValue(inst.State.Name), "%s is not running", aws.StringValue(inst.InstanceId))
+			launch[aws.StringValue(inst.InstanceId)] = aws.TimeValue(inst.LaunchTime)
+		}
+	}
+	snaps := make([]shInstanceSnap, 0, len(ids))
+	for _, id := range ids {
+		host := harness.InstanceHostingNode(t, fix.Cluster, id)
+		require.NotNilf(t, host, "no node is running %s", id)
+		proc, perr := harness.ProcessIdentity(*host, id)
+		require.NoError(t, perr)
+		snaps = append(snaps, shInstanceSnap{id: id, launch: launch[id], process: proc, node: *host})
+	}
+	return snaps
+}
+
+func shWaitOlderThan(t *testing.T, c *harness.AWSClient, ids []string, age time.Duration) {
+	t.Helper()
+	out, err := c.EC2.DescribeInstances(&ec2.DescribeInstancesInput{InstanceIds: aws.StringSlice(ids)})
+	require.NoError(t, err, "describe customer instances")
+	var newest time.Time
+	for _, r := range out.Reservations {
+		for _, inst := range r.Instances {
+			if lt := aws.TimeValue(inst.LaunchTime); lt.After(newest) {
+				newest = lt
+			}
+		}
+	}
+	if wait := time.Until(newest.Add(age)); wait > 0 {
+		harness.Detail(t, "waiting", wait.Round(time.Second).String())
+		time.Sleep(wait)
+	}
+}
+
+func shLaunchBystander(t *testing.T, fix *Fixture, f *albFixture) string {
+	t.Helper()
+	id := lbrLaunchTarget(t, fix, f)
+	harness.WaitForInstanceRunning(t, fix.AWS, id, 3*time.Minute)
+	harness.Detail(t, "bystander", id)
+	return id
+}
+
+// shSetup builds the VPC, a load balancer of the given kind with a listener,
+// and two registered backends.
+func shSetup(t *testing.T, fix *Fixture, kind shKind) *albFixture {
+	t.Helper()
+	f := &albFixture{}
+	instanceType, arch := harness.DiscoverNanoInstanceType(t, fix.Harness)
+	f.InstanceType = instanceType
+	f.AMIID = harness.DiscoverUbuntuAMI(t, fix.Harness, arch)
+	f.KeyName, _ = harness.EnsureKeyPair(t, fix.Harness)
+
+	lbrCreateVPC(t, fix.AWS, f)
+	lbrCreateIGW(t, fix.AWS, f)
+	lbrCreateSubnet(t, fix.AWS, f)
+	lbrAddPublicRoute(t, fix.AWS, f)
+	lbrOpenDefaultSG(t, fix.AWS, f)
+
+	c := fix.AWS
+	tgIn := &elbv2.CreateTargetGroupInput{
+		Name:     aws.String("lbr-sh-" + strings.ToLower(kind.lbType[:3])),
+		Protocol: aws.String(kind.tgProto),
+		Port:     aws.Int64(lbrHTTPPort),
+		VpcId:    aws.String(f.VPCID),
+
+		HealthCheckIntervalSeconds: aws.Int64(5),
+		HealthyThresholdCount:      aws.Int64(2),
+		UnhealthyThresholdCount:    aws.Int64(2),
+	}
+	if kind.tgProto == "HTTP" {
+		tgIn.HealthCheckPath = aws.String("/index.html")
+	}
+	tg, err := c.ELBv2.CreateTargetGroup(tgIn)
+	require.NoError(t, err, "create target group")
+	f.TGArn = aws.StringValue(tg.TargetGroups[0].TargetGroupArn)
+	t.Cleanup(func() {
+		if _, derr := c.ELBv2.DeleteTargetGroup(&elbv2.DeleteTargetGroupInput{TargetGroupArn: aws.String(f.TGArn)}); derr != nil {
+			t.Logf("delete target group: %v", derr)
+		}
+	})
+
+	lbIn := &elbv2.CreateLoadBalancerInput{
+		Name:    aws.String("lbr-sh-" + strings.ToLower(kind.lbType[:3])),
+		Subnets: []*string{aws.String(f.SubnetID)},
+		Scheme:  aws.String(kind.scheme),
+		Type:    aws.String(kind.lbType),
+	}
+	if kind.lbType == "application" {
+		lbIn.SecurityGroups = []*string{aws.String(f.LBSGID)}
+	}
+	lbOut, err := c.ELBv2.CreateLoadBalancer(lbIn)
+	require.NoError(t, err, "create load balancer")
+	require.NotEmpty(t, lbOut.LoadBalancers, "CreateLoadBalancer returned nothing")
+	f.LBArn = aws.StringValue(lbOut.LoadBalancers[0].LoadBalancerArn)
+	f.DNSName = aws.StringValue(lbOut.LoadBalancers[0].DNSName)
+	parts := strings.Split(f.LBArn, "/")
+	f.LBID = parts[len(parts)-1]
+	t.Cleanup(func() { lbrDeleteLB(t, c, f) })
+
+	ls, err := c.ELBv2.CreateListener(&elbv2.CreateListenerInput{
+		LoadBalancerArn: aws.String(f.LBArn),
+		Protocol:        aws.String(kind.tgProto),
+		Port:            aws.Int64(lbrHTTPPort),
+		DefaultActions: []*elbv2.Action{{
+			Type:           aws.String("forward"),
+			TargetGroupArn: aws.String(f.TGArn),
+		}},
+	})
+	require.NoError(t, err, "create listener")
+	f.ListenerArn = aws.StringValue(ls.Listeners[0].ListenerArn)
+	t.Cleanup(func() { lbrDeleteListener(t, c, f) })
+
+	harness.WaitForLBActive(t, c, f.LBArn, kind.name, albActiveBudget)
+	shDiscoverPlacement(t, fix, f, kind)
+
+	for len(f.TargetIDs) < lbrTargets {
+		id := lbrLaunchTarget(t, fix, f)
+		harness.WaitForInstanceRunning(t, c, id, 3*time.Minute)
+		f.TargetIDs = append(f.TargetIDs, id)
+	}
+	lbrRegisterTargets(t, c, f)
+	harness.WaitForTargetsHealthy(t, c, f.TGArn, lbrTargets, kind.name+" targets", targetsHealthyBudget)
+	return f
+}
+
+// shDiscoverPlacement finds the load balancer's guest through the ENI it is
+// described by, because a system instance is hidden from DescribeInstances.
+func shDiscoverPlacement(t *testing.T, fix *Fixture, f *albFixture, kind shKind) {
+	t.Helper()
+	desc := shENIDescription(f)
+
+	var eni *ec2.NetworkInterface
+	harness.EventuallyErr(t, func() error {
+		out, err := fix.AWS.EC2.DescribeNetworkInterfaces(&ec2.DescribeNetworkInterfacesInput{
+			Filters: []*ec2.Filter{{Name: aws.String("description"), Values: []*string{aws.String(desc)}}},
+		})
+		if err != nil {
+			return err
+		}
+		if len(out.NetworkInterfaces) == 0 {
+			return fmt.Errorf("no ENI described as %q", desc)
+		}
+		eni = out.NetworkInterfaces[0]
+		if eni.Attachment == nil || aws.StringValue(eni.Attachment.InstanceId) == "" {
+			return fmt.Errorf("ENI %s is not attached to an instance yet", aws.StringValue(eni.NetworkInterfaceId))
+		}
+		if kind.scheme == "internet-facing" && (eni.Association == nil || aws.StringValue(eni.Association.PublicIp) == "") {
+			return fmt.Errorf("ENI %s has no public address yet", aws.StringValue(eni.NetworkInterfaceId))
+		}
+		return nil
+	}, 60*time.Second, 3*time.Second)
+
+	f.ENIID = aws.StringValue(eni.NetworkInterfaceId)
+	f.SysInstanceID = aws.StringValue(eni.Attachment.InstanceId)
+	if eni.Association != nil {
+		f.PublicIP = aws.StringValue(eni.Association.PublicIp)
+	}
+	host := harness.InstanceHostingNode(t, fix.Cluster, f.SysInstanceID)
+	require.NotNilf(t, host, "the load balancer is active but no node is running %s", f.SysInstanceID)
+	f.Host = *host
+	harness.Detail(t, "lb_instance", f.SysInstanceID, "lb_eni", f.ENIID, "lb_node", f.Host.Name, "lb_public_ip", f.PublicIP)
+}