From 57a93fdee3ea65561422afbc34a9f53ebccb7e52 Mon Sep 17 00:00:00 2001 From: Josh Siviour Date: Sun, 2 Aug 2026 10:29:07 +1000 Subject: [PATCH] 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 | 45 +++++++--- build/systemd/spinifex-nats-watchdog.service | 15 ++-- cmd/spinifex/cmd/admin.go | 2 + cmd/spinifex/cmd/admin_jsprobe.go | 94 ++++++++++++++++++++ cmd/spinifex/cmd/admin_jsprobe_test.go | 75 ++++++++++++++++ 5 files changed, 215 insertions(+), 16 deletions(-) create mode 100644 cmd/spinifex/cmd/admin_jsprobe.go create mode 100644 cmd/spinifex/cmd/admin_jsprobe_test.go diff --git a/build/scripts/nats-js-watchdog.sh b/build/scripts/nats-js-watchdog.sh index bccd71c30..49cbab8fe 100755 --- a/build/scripts/nats-js-watchdog.sh +++ b/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" diff --git a/build/systemd/spinifex-nats-watchdog.service b/build/systemd/spinifex-nats-watchdog.service index e97c83a5c..f414f5ca2 100644 --- a/build/systemd/spinifex-nats-watchdog.service +++ b/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 diff --git a/cmd/spinifex/cmd/admin.go b/cmd/spinifex/cmd/admin.go index 9253cfaea..487746652 100644 --- a/cmd/spinifex/cmd/admin.go +++ b/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) diff --git a/cmd/spinifex/cmd/admin_jsprobe.go b/cmd/spinifex/cmd/admin_jsprobe.go new file mode 100644 index 000000000..268f5da15 --- /dev/null +++ b/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 +} diff --git a/cmd/spinifex/cmd/admin_jsprobe_test.go b/cmd/spinifex/cmd/admin_jsprobe_test.go new file mode 100644 index 000000000..4b3c810cc --- /dev/null +++ b/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] == '_') +}