fix(nats): detect the JetStream write latch the health endpoint cannot see (#645)

The watchdog probed only healthz?js-enabled-only=true, which reports the
server-wide JetStream state. A per-message-block write error is recorded on
the block and returned on every later write, and nothing clears it short of
a restart -- the server-wide state is never touched, so healthz stays 200
while writes fail indefinitely after the disk recovers.

spx admin node js-probe round-trips a canary key through the cluster-state
bucket, which is the only operation that observes the latch. It probes that
bucket rather than one of its own because a dedicated bucket is a separate
stream that can be placed elsewhere and stay writable.

build/scripts/nats-js-watchdog.sh

@@ -1,21 +1,32 @@
 #!/bin/sh
 # JetStream liveness watchdog for spinifex-nats.
 #
-# NATS can disable JetStream at runtime (e.g. a filesystem permission error while
-# flushing stream state) and keep the process alive serving KV as read-only. That
-# is not a process failure, so Restart=on-failure never fires and KV-backed
-# services (awsgw IAM bucket init, etc.) stall until NATS is restarted by hand.
+# NATS can stop accepting JetStream writes at runtime while the process stays
+# alive and keeps serving reads. That is not a process failure, so
+# Restart=on-failure never fires and KV-backed services (awsgw IAM bucket init,
+# etc.) stall until NATS is restarted by hand.
 #
-# This probes the NATS monitoring endpoint's JS health and restarts the service
-# only when JetStream is sustainedly unhealthy while the service is running. The
-# multi-sample window avoids restarting during transient startup/quorum 503s.
+# Two distinct latches produce that symptom, and they need different probes:
+#
+#   1. Server-wide. handleOutOfSpace calls ShutdownJetStream and sets an
+#      internal disabled flag that only a restart clears. healthz reports it.
+#   2. Per message block. filestore records a write error on the block and
+#      returns it on every later write; nothing clears it short of a restart.
+#      healthz does NOT report it, because the server-wide state is untouched —
+#      only an actual write observes it.
+#
+# So a health check alone misses case 2 entirely. Both probes run here, and a
+# sustained failure of either restarts the service. The multi-sample window
+# avoids restarting during transient startup/quorum 503s.
 
 set -eu
 
 HEALTH_URL="http://127.0.0.1:8222/healthz?js-enabled-only=true"
 SERVICE="spinifex-nats.service"
+SPX="${SPX_BIN:-/usr/local/bin/spx}"
 SAMPLES=3        # consecutive failing samples required before acting
 SAMPLE_GAP=2     # seconds between samples
+PROBE_TIMEOUT=5  # seconds for one canary round-trip
 
 # Only act on a running service: a stopped/activating unit is systemd's to manage,
 # and try-restart on it would either no-op or fight an in-flight (re)start.
@@ -30,15 +41,29 @@ if ! command -v curl >/dev/null 2>&1; then
     exit 0
 fi
 
-# A single 200 means JetStream is healthy — exit fast on the common case.
 js_healthy() {
     curl -fsS -o /dev/null --max-time 5 "$HEALTH_URL" 2>/dev/null
 }
 
+# Round-trips a canary key through the cluster-state bucket. Absent spx this
+# degrades to the health check alone rather than restarting on a missing tool.
+js_writable() {
+    if [ ! -x "$SPX" ]; then
+        return 0
+    fi
+    "$SPX" admin node js-probe --timeout "${PROBE_TIMEOUT}s" >/dev/null 2>&1
+}
+
+failure=""
 i=1
 while [ "$i" -le "$SAMPLES" ]; do
     if js_healthy; then
-        exit 0
+        if js_writable; then
+            exit 0
+        fi
+        failure="JetStream accepts no writes (canary round-trip failed) while $HEALTH_URL reports healthy"
+    else
+        failure="JetStream unhealthy at $HEALTH_URL"
     fi
     if [ "$i" -lt "$SAMPLES" ]; then
         sleep "$SAMPLE_GAP"
@@ -46,5 +71,5 @@ while [ "$i" -le "$SAMPLES" ]; do
     i=$((i + 1))
 done
 
-echo "nats-js-watchdog: JetStream unhealthy on $SAMPLES consecutive probes ($HEALTH_URL); restarting $SERVICE" >&2
+echo "nats-js-watchdog: $failure on $SAMPLES consecutive probes; restarting $SERVICE" >&2
 systemctl try-restart "$SERVICE"

build/systemd/spinifex-nats-watchdog.service

@@ -5,13 +5,16 @@ After=spinifex-nats.service
 
 [Service]
 Type=oneshot
-# Runs as root: the probe is a local curl to the NATS monitoring port, and the
-# remedy is `systemctl try-restart spinifex-nats` when JetStream is disabled while
-# the process stays alive (which Restart=on-failure cannot catch).
+# Runs as root: the probes are a local curl to the NATS monitoring port and a
+# canary KV round-trip via spx, and the remedy is `systemctl try-restart
+# spinifex-nats` when JetStream stops accepting writes while the process stays
+# alive (which Restart=on-failure cannot catch).
 ExecStart=/var/lib/spinifex/nats-js-watchdog.sh
-# Sized for the probe loop's own worst case (3 samples x 5s curl + 2 gaps x 2s
-# ~= 19s): a stop mid-cycle lets the script finish rather than cutting it off.
-TimeoutStopSec=20
+# Sized for the probe loop's own worst case (3 samples x (5s curl + 5s canary)
+# + 2 gaps x 2s = 34s): a stop mid-cycle lets the script finish rather than
+# cutting it off. Only the failing path runs that long; a healthy node exits on
+# the first sample.
+TimeoutStopSec=45
 
 # Hardening (kept compatible with talking to PID1 over /run/systemd).
 NoNewPrivileges=yes

cmd/spinifex/cmd/admin.go

@@ -281,6 +281,8 @@ func init() {
 	nodeCmd.AddCommand(nodeDrainCmd)
 	nodeDrainCmd.Flags().Bool("local", false, "Drain the local node only (required)")
 	nodeDrainCmd.Flags().Duration("timeout", 120*time.Second, "Maximum time to wait per phase")
+	nodeCmd.AddCommand(nodeJSProbeCmd)
+	nodeJSProbeCmd.Flags().Duration("timeout", 10*time.Second, "Maximum time to wait for the canary round-trip")
 
 	adminCmd.AddCommand(imagesCmd)
 	imagesCmd.AddCommand(imagesImportCmd)

cmd/spinifex/cmd/admin_jsprobe.go

@@ -0,0 +1,94 @@
+package cmd
+
+import (
+	"context"
+	"fmt"
+	"os"
+	"time"
+
+	"github.com/nats-io/nats.go/jetstream"
+	"github.com/spf13/cobra"
+
+	"github.com/mulgadc/spinifex/spinifex/daemon"
+)
+
+// canaryKey is written and read back by the JetStream write probe. It lives in
+// the same bucket the control plane depends on, under a reserved prefix no
+// other consumer reads.
+const canaryKey = "_watchdog.canary"
+
+var nodeJSProbeCmd = &cobra.Command{
+	Use:   "js-probe",
+	Short: "Verify JetStream still accepts writes on this node",
+	Long: `Round-trip a canary key through the cluster-state KV bucket and exit non-zero
+if the write or the read-back fails.
+
+This exists because JetStream can latch a write error per message block and keep
+serving reads: nats-server's filestore records the failure on the block and
+returns it on every later write, and nothing clears it short of a restart. The
+HTTP health endpoint reports the server-wide JetStream state, which that latch
+never touches, so only an actual write observes it.`,
+	Run: runNodeJSProbe,
+}
+
+func runNodeJSProbe(cmd *cobra.Command, _ []string) {
+	timeout, _ := cmd.Flags().GetDuration("timeout")
+	if timeout <= 0 {
+		timeout = 10 * time.Second
+	}
+
+	ctx, cancel := context.WithTimeout(cmd.Context(), timeout)
+	defer cancel()
+
+	if err := probeJetStreamWrite(ctx); err != nil {
+		fmt.Fprintf(os.Stderr, "js-probe: %v\n", err)
+		os.Exit(1)
+	}
+	fmt.Println("js-probe: ok")
+}
+
+// probeJetStreamWrite writes a canary value and reads it back, returning the
+// first error. It deliberately probes the existing cluster-state bucket rather
+// than a bucket of its own: a dedicated bucket is a separate stream that can be
+// placed on a different node, so it could stay writable while the bucket the
+// control plane actually depends on is wedged.
+func probeJetStreamWrite(ctx context.Context) error {
+	_, nc, err := loadConfigAndConnect()
+	if err != nil {
+		return fmt.Errorf("connect: %w", err)
+	}
+	defer nc.Close()
+
+	js, err := jetstream.New(nc)
+	if err != nil {
+		return fmt.Errorf("jetstream: %w", err)
+	}
+
+	kv, err := js.KeyValue(ctx, daemon.ClusterStateBucket)
+	if err != nil {
+		return fmt.Errorf("open bucket %s: %w", daemon.ClusterStateBucket, err)
+	}
+	return canaryRoundTrip(ctx, kv)
+}
+
+// canaryRoundTrip writes a unique value and reads it back, returning the first
+// error. The write is what matters: a latched block returns its stored error on
+// every subsequent write, so this is the only operation that observes it.
+func canaryRoundTrip(ctx context.Context, kv jetstream.KeyValue) error {
+	want := []byte(time.Now().UTC().Format(time.RFC3339Nano))
+	if _, err := kv.Put(ctx, canaryKey, want); err != nil {
+		return fmt.Errorf("write canary: %w", err)
+	}
+
+	entry, err := kv.Get(ctx, canaryKey)
+	if err != nil {
+		return fmt.Errorf("read back canary: %w", err)
+	}
+
+	// A stale value means the write was accepted but not durable, which is the
+	// same operational problem as an outright write failure.
+	if string(entry.Value()) != string(want) {
+		return fmt.Errorf("canary read back stale: wrote %q, read %q", want, entry.Value())
+	}
+	return nil
+}

cmd/spinifex/cmd/admin_jsprobe_test.go

@@ -0,0 +1,75 @@
+package cmd
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	"github.com/nats-io/nats.go/jetstream"
+	"github.com/stretchr/testify/require"
+
+	"github.com/mulgadc/spinifex/spinifex/testutil"
+)
+
+func newCanaryBucket(t *testing.T) (jetstream.KeyValue, jetstream.JetStream) {
+	t.Helper()
+	_, _, js := testutil.StartTestJetStream(t)
+
+	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+	defer cancel()
+
+	kv, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: "canary-test"})
+	require.NoError(t, err)
+	return kv, js
+}
+
+func TestCanaryRoundTrip_HealthyBucketPasses(t *testing.T) {
+	kv, _ := newCanaryBucket(t)
+
+	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+	defer cancel()
+
+	require.NoError(t, canaryRoundTrip(ctx, kv))
+}
+
+// Each probe must write a distinct value, otherwise a bucket that silently
+// stopped accepting writes would still read back the previous run's value and
+// the probe would pass while JetStream was wedged.
+func TestCanaryRoundTrip_WritesADistinctValueEachRun(t *testing.T) {
+	kv, _ := newCanaryBucket(t)
+
+	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+	defer cancel()
+
+	require.NoError(t, canaryRoundTrip(ctx, kv))
+	first, err := kv.Get(ctx, canaryKey)
+	require.NoError(t, err)
+
+	require.NoError(t, canaryRoundTrip(ctx, kv))
+	second, err := kv.Get(ctx, canaryKey)
+	require.NoError(t, err)
+
+	require.NotEqual(t, string(first.Value()), string(second.Value()))
+	require.Greater(t, second.Revision(), first.Revision())
+}
+
+// A deleted bucket stands in for the unreachable-store case: the probe must
+// report an error rather than treating a failed write as success.
+func TestCanaryRoundTrip_FailsWhenBucketIsGone(t *testing.T) {
+	kv, js := newCanaryBucket(t)
+
+	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+	defer cancel()
+
+	require.NoError(t, js.DeleteKeyValue(ctx, "canary-test"))
+
+	err := canaryRoundTrip(ctx, kv)
+	require.Error(t, err)
+	require.Contains(t, err.Error(), "canary")
+}
+
+// The canary key is namespaced so it cannot collide with a real control-plane
+// key in the shared cluster-state bucket.
+func TestCanaryKeyIsNamespaced(t *testing.T) {
+	require.True(t, len(canaryKey) > 0 && canaryKey[0] == '_')
+}