fix: stop lbagent heartbeats on positive LB deletion, and actually enable the JetStream watchdog (#643)

* fix(lbagent): stop heartbeating a load balancer once it is confirmed deleted

A LoadBalancerNotFound response is the gateway's positive confirmation that
the store has no record for the LB, not a connectivity failure — unlike a
dial error, timeout, or 5xx, which leave its existence unknown and must keep
retrying. Past the startup registration grace window (or after any prior
successful heartbeat), it can no longer be the record's own replication
race, so the agent now stops rather than retrying forever and filling the
log with heartbeat noise after teardown.

* fix(nats): enable the JetStream ENOSPC-latch watchdog timer

nats-server disables JetStream process-wide on ENOSPC (server/jetstream.go
handleOutOfSpace -> ShutdownJetStream) and never re-enables it itself —
that decision is nats-server's own internal state, not spinifex's, so a
restart is the only way back. The spinifex-nats-watchdog.timer + service
already probe /healthz?js-enabled-only=true and try-restart on a sustained
failure, but the timer's WantedBy=timers.target never took effect: neither
setup.sh nor firstboot enabled it, so the watchdog shipped inert and a full
disk needed a manual restart forever. Enable --now it alongside the rest of
the unit set.

cmd/installer/firstboot/firstboot.go

@@ -197,6 +197,9 @@ fi
 
 # Enable services to start, on reboot
 systemctl enable spinifex.target spinifex-banner.service
+# WantedBy=timers.target only self-activates once enabled — without this the
+# JetStream ENOSPC-latch watchdog never runs and a full disk needs a manual restart.
+systemctl enable --now spinifex-nats-watchdog.timer
 
 # Mark complete only after every step above has succeeded. Until this point,
 # any failure (set -e) leaves the marker absent and the next reboot retries

cmd/installer/firstboot/firstboot_test.go

@@ -114,6 +114,25 @@ func readScript(t *testing.T, root string) string {
 	return string(b)
 }
 
+// TestWriteScriptEnablesNatsWatchdogTimer pins that firstboot activates the
+// JetStream ENOSPC-latch watchdog timer. The unit file's WantedBy=timers.target
+// only takes effect once enabled — without this line the watchdog is dropped
+// onto disk but never runs, and a full disk needs a manual nats restart forever.
+func TestWriteScriptEnablesNatsWatchdogTimer(t *testing.T) {
+	root := t.TempDir()
+	makeRootDirs(t, root)
+
+	cfg := Config{Hostname: "test-node", ClusterRole: "init"}
+	if err := Write(root, cfg); err != nil {
+		t.Fatalf("Write: %v", err)
+	}
+	content := readScript(t, root)
+
+	if !strings.Contains(content, "systemctl enable --now spinifex-nats-watchdog.timer") {
+		t.Error("firstboot script must enable --now spinifex-nats-watchdog.timer")
+	}
+}
+
 func TestWriteScriptCallbackAfterDoneMarker(t *testing.T) {
 	root := t.TempDir()
 	makeRootDirs(t, root)

scripts/setup.sh

@@ -667,6 +667,10 @@ EOF
     fi
     $SUDO systemctl daemon-reload
     $SUDO systemctl enable spinifex.target
+    # The unit file alone does not self-activate (WantedBy=timers.target only
+    # takes effect once enabled) — without this the JetStream ENOSPC-latch
+    # watchdog is inert and a full disk requires a manual restart forever.
+    $SUDO systemctl enable --now spinifex-nats-watchdog.timer
     info "Systemd units installed and enabled (per-service users)"
 }
 

spinifex/lbagent/agent.go

@@ -10,6 +10,7 @@ import (
 	"crypto/tls"
 	"encoding/hex"
 	"encoding/xml"
+	"errors"
 	"fmt"
 	"io"
 	"log/slog"
@@ -25,6 +26,7 @@ import (
 
 	"github.com/mulgadc/spinifex/internal/gwsign"
 	"github.com/mulgadc/spinifex/internal/tlsconfig"
+	"github.com/mulgadc/spinifex/spinifex/awserrors"
 )
 
 const (
@@ -182,6 +184,15 @@ func (a *Agent) Stop() {
 
 // tick runs one heartbeat cycle: send health, check config hash, fetch if changed.
 func (a *Agent) tick() {
+	// A stopped agent must never heartbeat again, even if Start's select
+	// races a pending ticker tick against a just-closed stopCh — the whole
+	// point of stopping on a deleted LB is zero further noise.
+	select {
+	case <-a.stopCh:
+		return
+	default:
+	}
+
 	// nginx (NLB) has no per-server stats socket — the agent actively probes
 	// the delivered health targets instead of reading HAProxy stats.
 	var servers []ServerStatus
@@ -211,6 +222,21 @@ func (a *Agent) tick() {
 				"gateway", a.gatewayURL, "elapsed", time.Since(a.startedAt).Round(time.Second))
 			return
 		}
+
+		// LoadBalancerNotFound is a positive response from the gateway (the
+		// store round-tripped and confirmed no record), not a connectivity
+		// failure — distinct from a dial error, timeout, or 5xx, all of which
+		// leave the LB's existence unknown and must keep retrying. Past the
+		// startup grace window it can no longer be the record's own
+		// replication race, so it means the LB was deleted: stop heartbeating
+		// it rather than retrying forever.
+		var gwErr *gatewayError
+		if errors.As(err, &gwErr) && gwErr.code == awserrors.ErrorELBv2LoadBalancerNotFound {
+			slog.Info("Load balancer no longer exists, stopping agent", "lbId", a.lbID, "gateway", a.gatewayURL)
+			a.Stop()
+			return
+		}
+
 		slog.Error("Heartbeat failed", "err", err, "gateway", a.gatewayURL, "region", a.region)
 		return
 	}
@@ -372,12 +398,43 @@ func (a *Agent) signedPost(params url.Values) ([]byte, error) {
 	}
 
 	if resp.StatusCode != http.StatusOK {
-		return nil, fmt.Errorf("gateway returned %d: %s", resp.StatusCode, string(respBody))
+		return nil, &gatewayError{statusCode: resp.StatusCode, code: parseAWSErrorCode(respBody), body: string(respBody)}
 	}
 
 	return respBody, nil
 }
 
+// gatewayError wraps a non-200 gateway response, carrying the parsed AWS
+// error code (if any) alongside the raw status/body so callers can
+// distinguish a specific, unambiguous rejection (e.g. LoadBalancerNotFound)
+// from a generic or ambiguous one.
+type gatewayError struct {
+	statusCode int
+	code       string
+	body       string
+}
+
+func (e *gatewayError) Error() string {
+	return fmt.Sprintf("gateway returned %d: %s", e.statusCode, e.body)
+}
+
+// awsErrorEnvelope is the minimal shape of the query-protocol error response
+// needed to recover the AWS error <Code> (e.g. "LoadBalancerNotFound").
+type awsErrorEnvelope struct {
+	XMLName xml.Name `xml:"ErrorResponse"`
+	Code    string   `xml:"Error>Code"`
+}
+
+// parseAWSErrorCode extracts the AWS error <Code> from a gateway error body.
+// Returns "" if the body does not parse as the expected envelope.
+func parseAWSErrorCode(body []byte) string {
+	var env awsErrorEnvelope
+	if err := xml.Unmarshal(body, &env); err != nil {
+		return ""
+	}
+	return env.Code
+}
+
 // writeCertFiles writes each delivered TLS PEM to its path (0600) under certDir.
 // Paths that escape certDir are rejected to prevent traversal writes.
 func (a *Agent) writeCertFiles(certDir string, certs []certFile) error {

spinifex/lbagent/agent_test.go

@@ -695,14 +695,17 @@ func TestHeartbeat_StartupFailureLogsBelowError(t *testing.T) {
 }
 
 func TestHeartbeat_FailureAfterGraceLogsError(t *testing.T) {
-	gw := notFoundGateway(t)
+	// An ambiguous/generic failure (not a positive not-found signal) that
+	// persists past the grace window is the blackhole fingerprint — it must
+	// escalate to ERROR so the console telemetry catches it.
+	gw := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
+		http.Error(w, "internal error", http.StatusInternalServerError)
+	}))
 	defer gw.Close()
 
 	agent := newTestAgent(t, gw.URL)
 	logs := captureSlog(t)
 
-	// Past the grace window with no prior success — the blackhole fingerprint —
-	// must escalate so the console telemetry catches it.
 	agent.startedAt = time.Now().Add(-2 * registrationGrace)
 	agent.tick()
 
@@ -710,19 +713,71 @@ func TestHeartbeat_FailureAfterGraceLogsError(t *testing.T) {
 	if !strings.Contains(out, "level=ERROR") || !strings.Contains(out, "Heartbeat failed") {
 		t.Errorf("post-grace heartbeat failure did not escalate to ERROR:\n%s", out)
 	}
+	if isStopped(agent) {
+		t.Error("an ambiguous/generic failure must not stop the agent")
+	}
+}
+
+// TestHeartbeat_NotFoundWithinGraceDoesNotStop pins the startup-race carve-out:
+// a LoadBalancerNotFound seen before any successful heartbeat and inside the
+// grace window is still ambiguous (the record may not have replicated to this
+// gateway node yet), so the agent must keep retrying rather than stop.
+func TestHeartbeat_NotFoundWithinGraceDoesNotStop(t *testing.T) {
+	gw := notFoundGateway(t)
+	defer gw.Close()
+
+	agent := newTestAgent(t, gw.URL)
+	agent.tick()
+
+	if isStopped(agent) {
+		t.Error("a LoadBalancerNotFound within the startup grace window must not stop the agent")
+	}
+}
+
+// TestHeartbeat_NotFoundAfterGraceStopsAgent pins the fix: once the startup
+// grace window has passed with no prior success, a LoadBalancerNotFound is no
+// longer a replication race — the gateway has given a positive signal that no
+// such LB exists — so the agent must stop rather than heartbeat it forever.
+func TestHeartbeat_NotFoundAfterGraceStopsAgent(t *testing.T) {
+	gw := notFoundGateway(t)
+	defer gw.Close()
+
+	agent := newTestAgent(t, gw.URL)
+	logs := captureSlog(t)
+
+	agent.startedAt = time.Now().Add(-2 * registrationGrace)
+	agent.tick()
+
+	if !isStopped(agent) {
+		t.Error("a LoadBalancerNotFound past the grace window must stop the agent")
+	}
+	out := logs()
+	if strings.Contains(out, "level=ERROR") {
+		t.Errorf("a confirmed LB deletion is expected teardown, not an error:\n%s", out)
+	}
+}
+
+// isStopped reports whether agent's poll loop has been signalled to stop.
+func isStopped(agent *Agent) bool {
+	select {
+	case <-agent.stopCh:
+		return true
+	default:
+		return false
+	}
 }
 
 func TestHeartbeat_FailureAfterFirstSuccessLogsError(t *testing.T) {
-	// A gateway that answers healthily once, then only ever 400s.
+	// A gateway that answers healthily once, then fails ambiguously (a
+	// transient/unreachable-flavoured error, not a positive not-found).
 	var beats atomic.Int32
-	gw := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
+	gw := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
 		w.Header().Set("Content-Type", "text/xml")
 		if beats.Add(1) == 1 {
 			fmt.Fprint(w, `<LBAgentHeartbeatResponse><LBAgentHeartbeatResult><Status>active</Status><ConfigHash></ConfigHash></LBAgentHeartbeatResult></LBAgentHeartbeatResponse>`)
 			return
 		}
-		w.WriteHeader(http.StatusBadRequest)
-		fmt.Fprint(w, `<ErrorResponse><Error><Code>LoadBalancerNotFound</Code></Error></ErrorResponse>`)
+		http.Error(w, "internal error", http.StatusInternalServerError)
 	}))
 	defer gw.Close()
 
@@ -742,4 +797,80 @@ func TestHeartbeat_FailureAfterFirstSuccessLogsError(t *testing.T) {
 	if !strings.Contains(out, "level=ERROR") || !strings.Contains(out, "Heartbeat failed") {
 		t.Errorf("post-success heartbeat failure did not log ERROR:\n%s", out)
 	}
+	if isStopped(agent) {
+		t.Error("an ambiguous/transient failure must not stop the agent, even after a prior success")
+	}
+}
+
+// TestHeartbeat_NotFoundAfterSuccessStopsAgent is the primary bug scenario:
+// an LB that heartbeated successfully (proving it once existed and the
+// gateway was reachable) is then deleted. The gateway's LoadBalancerNotFound
+// is a positive, unambiguous signal — not a connectivity failure — so the
+// agent must stop rather than retry indefinitely and fill the log with
+// heartbeat noise after teardown.
+func TestHeartbeat_NotFoundAfterSuccessStopsAgent(t *testing.T) {
+	var beats atomic.Int32
+	gw := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
+		w.Header().Set("Content-Type", "text/xml")
+		if beats.Add(1) == 1 {
+			fmt.Fprint(w, `<LBAgentHeartbeatResponse><LBAgentHeartbeatResult><Status>active</Status><ConfigHash></ConfigHash></LBAgentHeartbeatResult></LBAgentHeartbeatResponse>`)
+			return
+		}
+		w.WriteHeader(http.StatusBadRequest)
+		fmt.Fprint(w, `<ErrorResponse><Error><Code>LoadBalancerNotFound</Code></Error></ErrorResponse>`)
+	}))
+	defer gw.Close()
+
+	agent := newTestAgent(t, gw.URL)
+	logs := captureSlog(t)
+
+	agent.tick() // succeeds
+	agent.tick() // LB now deleted
+
+	if !isStopped(agent) {
+		t.Fatal("agent must stop once the gateway confirms the LB is gone")
+	}
+
+	out := logs()
+	if strings.Contains(out, "level=ERROR") {
+		t.Errorf("a confirmed LB deletion is expected teardown, not an error:\n%s", out)
+	}
+
+	// A further tick (as if the poll loop raced one more cycle before
+	// noticing stopCh) must not re-log a "not found" transition or panic —
+	// idempotent stop, no additional noise.
+	preLen := len(logs())
+	agent.tick()
+	if len(logs()) != preLen {
+		t.Errorf("ticking a stopped agent logged additional heartbeat noise: %q", logs()[preLen:])
+	}
+}
+
+// TestHeartbeat_TransientErrorDuringOperationDoesNotStop is the negative
+// control the fix depends on: a NATS blip or control-plane restart surfaces
+// as a dial failure or 5xx, never as LoadBalancerNotFound, and must never be
+// treated as "deleted" — a healthy agent must keep retrying through it.
+func TestHeartbeat_TransientErrorDuringOperationDoesNotStop(t *testing.T) {
+	var beats atomic.Int32
+	gw := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
+		w.Header().Set("Content-Type", "text/xml")
+		if beats.Add(1) == 1 {
+			fmt.Fprint(w, `<LBAgentHeartbeatResponse><LBAgentHeartbeatResult><Status>active</Status><ConfigHash></ConfigHash></LBAgentHeartbeatResult></LBAgentHeartbeatResponse>`)
+			return
+		}
+		// Simulates a control-plane restart / NATS blip: unavailable, not deleted.
+		http.Error(w, "service unavailable", http.StatusServiceUnavailable)
+	}))
+	defer gw.Close()
+
+	agent := newTestAgent(t, gw.URL)
+
+	agent.tick() // succeeds
+	for range 5 {
+		agent.tick() // repeated transient failures
+	}
+
+	if isStopped(agent) {
+		t.Fatal("repeated transient/unreachable errors must never stop the agent")
+	}
 }