fix: prod review of 14 September — IPsec follower reads, stopped-instance EIP probes, audit client IP and JetStream formation divergence (#1029)

* fix: read NB_Global.ipsec through the NB remote list so followers and compute nodes answer quietly * fix: stop probing and nudging the EIP datapath of a stopped instance's port * fix: log the real client IP, request ID and duration_ms in the awsgw access line * test: mark the UI proxy header test as in-package * fix: discard pre-formation JetStream data when forming a multi-node cluster A node that ran as its own single-node cluster carries KV streams into formation. nats-server adopts a local stream with the same name as the clustered one, and raft catch-up only fetches sequences beyond a follower's last, so the conflicting prefix is never repaired and replicas diverge permanently. - spx admin join removes the local JetStream store once formation succeeds. A store holding streams now needs --force, and the identity-discard warning names it. - spx admin init --nodes >= 2 refuses a store that holds streams unless --discard-jetstream is given, and removes it once every node has joined. Keeping it as the seed is unsafe: stream leadership ignores who holds data, and unassigned local streams are deleted as orphans. - Neither removes a store while `spx service nats` runs, --force included. - install-node.sh passes --discard-jetstream and waits for every `spx service` process to exit before formation. * feat: add spx admin kv digest for per-replica JetStream consistency checks Replicas of a stream can hold different content while raft reports them current, and nothing in the cluster detects it. A direct get over NATS is no way to inspect one replica: a client's request goes to a random member of the direct-get queue group, routed replicas included. kv digest copies this node's $G streams tree (without consumer state) to a temporary directory, opens the copy in a private in-process server with no listener, and prints first/last sequence, message count and a SHA-256 over every stored message per stream. Running it on each node and diffing the output shows which streams diverge; --seqs locates the first differing sequence. The live store is only ever read. * fix: let spx admin kv digest read replicated streams A replicated stream's meta.inf carries num_replicas > 1, which a standalone server refuses to recover, so the digest could not read any production store. Rewrite the copy's config to one replica with no expiry and a recomputed checksum, and test it against a three-node cluster and an expired bucket. * test: use assert.ErrorIs for the removed JetStream store * test: mark the JetStream store and digest tests as in-package * fix: stop the process-group test misreading a reaped child as alive The test polled until the child was gone, then checked once more. A task being reaped shows state X for an instant, which processRunning counted as running, so that second look failed about 1 run in 50 under -race. * fix: wait for the recovery retry loop before removing a test's data dir restoreInstances starts the retry loop untracked, and it rewrites the local state file when it ends. A write landing during t.TempDir cleanup failed TestRestoreInstances_* with RemoveAll: directory not empty.
2026-10-15 09:01:31 +00:00 · 2026-09-14 15:39:03 -05:00
parent d7c2a202fc
commit 455c97a2e7
26 changed files with 1745 additions and 78 deletions
@@ -358,6 +358,7 @@ func init() {

 	// Flags for admin init
 	adminInitCmd.Flags().Bool("force", false, "Force re-initialization (overwrites existing config)")
+	adminInitCmd.Flags().Bool("discard-jetstream", false, "With --nodes >= 2, remove this node's existing JetStream streams once every node has joined")
 	adminInitCmd.Flags().String("region", "ap-southeast-2", "Mulga region to create")
 	adminInitCmd.Flags().String("az", "ap-southeast-2a", "Mulga AZ to create")
 	adminInitCmd.Flags().String("node", "node1", "Node name, increment for additional nodes (default, node1)")
@@ -1313,6 +1314,15 @@ func runAdminInit(cmd *cobra.Command, args []string) {
 		os.Exit(0)
 	}

+	if spxRoot == "" {
+		spxRoot = DefaultDataDir()
+	}
+	discardJetStream, _ := cmd.Flags().GetBool("discard-jetstream")
+	if err := checkInitJetStreamStore(jetStreamStoreDir(spxRoot), nodes, discardJetStream); err != nil {
+		fmt.Fprintf(os.Stderr, "❌ Error: %v\n", err)
+		os.Exit(1)
+	}
+
 	// Preserve the previously-captured operator email across --force re-inits
 	// when --email is omitted (e.g. reset-dev-env.sh workflows). Without this
 	// a reset would silently blank the address.
@@ -1845,6 +1855,17 @@ func runAdminInitMultiNode(cmd *cobra.Command, accessKey, secretKey, accountID,

 	fmt.Printf("✅ All %d nodes joined!\n", expectedNodes)

+	// The pre-start check refused a store with streams unless discard was asked
+	// for, so anything left here is removed now that the cluster is committed.
+	if discard, _ := cmd.Flags().GetBool("discard-jetstream"); discard {
+		storeDir := jetStreamStoreDir(spxRoot)
+		if err := discardJetStreamStore(storeDir); err != nil {
+			fmt.Fprintf(os.Stderr, "❌ Error: %v\n", err)
+			os.Exit(1)
+		}
+		fmt.Printf("✅ Discarded this node's pre-formation JetStream store: %s\n", storeDir)
+	}
+
 	// Build cluster topology from formation data
 	allNodes := fs.Nodes()
 	clusterRoutes := formation.BuildClusterRoutes(allNodes)
@@ -2002,19 +2023,27 @@ func joinRetryable(err error, statusCode int) bool {
 }

 // checkJoinPreconditions rejects a join that would silently destroy this node's
-// existing cluster identity. Joining adopts the primary's CA and master key,
-// overwriting whatever is here: correct on a freshly installed node, and on one
-// that has been in service it orphans every fragment and volume sealed under the
-// old key. runAdminInit guards the same way before re-initializing.
-func checkJoinPreconditions(configDir string, force bool) error {
+// identity or JetStream data; --force accepts that loss. A running NATS service
+// is refused even with --force, because the store cannot be removed under it.
+func checkJoinPreconditions(configDir, storeDir string, force bool) error {
+	if err := checkStoreDiscardable(); err != nil {
+		return err
+	}
 	if force {
 		return nil
 	}
 	tomlPath := filepath.Join(configDir, "spinifex.toml")
-	if !admin.FileExists(tomlPath) {
-		return nil
+	if admin.FileExists(tomlPath) {
+		return fmt.Errorf("this node is already initialized: %s", tomlPath)
 	}
-	return fmt.Errorf("this node is already initialized: %s", tomlPath)
+	streams, err := localStreams(storeDir)
+	if err != nil {
+		return err
+	}
+	if len(streams) > 0 {
+		return fmt.Errorf("this node's JetStream store %s holds %d stream(s)", storeDir, len(streams))
+	}
+	return nil
 }

 // joinDiscardsIdentityMsg spells out what a forced join throws away. Kept out of
@@ -2023,6 +2052,8 @@ const joinDiscardsIdentityMsg = `Joining will discard this node's own cluster id
  - CA certificate and key
  - master key, and any data sealed under it
  - viperblock key, and any volumes encrypted under it
+  - JetStream store (<data-dir>/nats/jetstream): every stream and KV bucket
+    this node holds, so it takes the cluster's copy instead of diverging from it

 That is safe on a freshly installed node — an ISO install initializes a
 single-node cluster at first boot, and nothing has been sealed under these keys
@@ -2103,7 +2134,11 @@ func runAdminJoin(cmd *cobra.Command, args []string) {
 	// Checked before any network call so a node that will not join says so
 	// immediately. Unlike init this exits non-zero: a node that did not join
 	// must not look like success to a provisioning script.
-	if err := checkJoinPreconditions(configDir, force); err != nil {
+	if dataDir == "" {
+		dataDir = DefaultDataDir()
+	}
+	storeDir := jetStreamStoreDir(dataDir)
+	if err := checkJoinPreconditions(configDir, storeDir, force); err != nil {
 		fmt.Fprintf(os.Stderr, "⚠️  %v\n\n%s\n", err, joinDiscardsIdentityMsg)
 		os.Exit(1)
 	}
@@ -2113,11 +2148,6 @@ func runAdminJoin(cmd *cobra.Command, args []string) {
 		clusterBind = bindIP
 	}

-	// Set default data directory
-	if dataDir == "" {
-		dataDir = DefaultDataDir()
-	}
-
 	fmt.Println("🚀 Joining Spinifex cluster...")
 	fmt.Printf("Node: %s\n", node)
 	fmt.Printf("Leader: %s\n", leaderHost)
@@ -2298,6 +2328,14 @@ func runAdminJoin(cmd *cobra.Command, args []string) {
 		os.Exit(1)
 	}

+	// Removed only once formation has succeeded, so a failed join leaves the
+	// node's single-node cluster intact.
+	if err := discardJetStreamStore(storeDir); err != nil {
+		fmt.Fprintf(os.Stderr, "❌ Error: %v\n", err)
+		os.Exit(1)
+	}
+	fmt.Printf("✅ Discarded this node's pre-formation JetStream store: %s\n", storeDir)
+
 	// Set up config directory
 	if configDir == "" {
 		configDir = DefaultConfigDir()
@@ -0,0 +1,371 @@
+package cmd
+
+import (
+	"bytes"
+	"context"
+	"crypto/sha256"
+	"encoding/binary"
+	"encoding/hex"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"io"
+	"io/fs"
+	"os"
+	"path/filepath"
+	"slices"
+	"sort"
+	"strings"
+	"time"
+
+	"github.com/minio/highwayhash"
+	"github.com/nats-io/nats-server/v2/server"
+	"github.com/nats-io/nats.go"
+	"github.com/nats-io/nats.go/jetstream"
+	"github.com/spf13/cobra"
+)
+
+var kvCmd = &cobra.Command{
+	Use:   "kv",
+	Short: "Inspect JetStream KV state",
+}
+
+var kvDigestCmd = &cobra.Command{
+	Use:   "digest",
+	Short: "Print a digest of every stream held by this node's JetStream replica",
+	Long: `Read this node's own copy of every JetStream stream and print one line per
+stream: name, first and last sequence, message count, and a SHA-256 over every
+stored message (sequence, subject, timestamp, headers, data).
+
+Run it on every node and compare the output. Replicas of one stream must
+produce identical lines; any difference means the replicas have diverged, and
+--seqs prints a hash per sequence to find where.
+
+It never touches the live store. A direct get over NATS may be answered by any
+replica, so instead the streams tree is copied to --work-dir and read by a
+private in-process server with no listener, which is removed afterwards. The
+copy's stream config is set to one replica with no expiry so it can be loaded.
+
+Writes landing during the copy, or applied but not yet flushed to disk, can
+make a replica's tail differ, so gate API writes and re-run before calling a
+tail difference divergence.`,
+	Run: runKVDigest,
+}
+
+func init() {
+	adminCmd.AddCommand(kvCmd)
+	kvCmd.AddCommand(kvDigestCmd)
+	kvDigestCmd.Flags().String("store-dir", "", "JetStream store to read (default <spinifex-dir>/nats/jetstream)")
+	kvDigestCmd.Flags().String("work-dir", os.TempDir(), "Directory to hold the temporary copy")
+	kvDigestCmd.Flags().StringSlice("stream", nil, "Only these streams (repeatable)")
+	kvDigestCmd.Flags().Bool("seqs", false, "Also print a hash per sequence")
+}
+
+// streamDigest is one stream's content as held by one replica.
+type streamDigest struct {
+	Name     string
+	FirstSeq uint64
+	LastSeq  uint64
+	Msgs     uint64
+	Digest   string
+	Seqs     []seqDigest
+}
+
+// seqDigest is the hash of one stored sequence, or "deleted" when the replica
+// has no message at that sequence.
+type seqDigest struct {
+	Seq  uint64
+	Hash string
+}
+
+func runKVDigest(cmd *cobra.Command, _ []string) {
+	storeDir, _ := cmd.Flags().GetString("store-dir")
+	if storeDir == "" {
+		spxRoot, _ := cmd.Flags().GetString("spinifex-dir")
+		if spxRoot == "" {
+			spxRoot = DefaultDataDir()
+		}
+		storeDir = jetStreamStoreDir(spxRoot)
+	}
+	workParent, _ := cmd.Flags().GetString("work-dir")
+	only, _ := cmd.Flags().GetStringSlice("stream")
+	withSeqs, _ := cmd.Flags().GetBool("seqs")
+
+	work, err := os.MkdirTemp(workParent, "spx-kv-digest-")
+	if err != nil {
+		fmt.Fprintf(os.Stderr, "kv digest: %v\n", err)
+		os.Exit(1)
+	}
+	defer os.RemoveAll(work)
+
+	copiedAt := time.Now().UTC()
+	skipped, err := copyStreamTree(storeDir, filepath.Join(work, "jetstream"))
+	if err != nil {
+		fmt.Fprintf(os.Stderr, "kv digest: %v\n", err)
+		os.RemoveAll(work)
+		os.Exit(1)
+	}
+
+	ctx, cancel := context.WithTimeout(cmd.Context(), 10*time.Minute)
+	defer cancel()
+	digests, err := digestStore(ctx, work, only, withSeqs)
+	if err != nil {
+		fmt.Fprintf(os.Stderr, "kv digest: %v\n", err)
+		cancel()
+		os.RemoveAll(work)
+		os.Exit(1)
+	}
+
+	host, _ := os.Hostname()
+	writeDigests(cmd.OutOrStdout(), host, storeDir, copiedAt, skipped, digests)
+}
+
+// writeDigests prints the report. Only the "#" lines carry node-specific
+// values, so `grep -v '^#'` output from two replicas diffs cleanly.
+func writeDigests(w io.Writer, host, storeDir string, copiedAt time.Time, skipped []string, digests []streamDigest) {
+	fmt.Fprintf(w, "# host=%s store=%s copied_at=%s\n", host, storeDir, copiedAt.Format(time.RFC3339))
+	for _, s := range skipped {
+		fmt.Fprintf(w, "# skipped account stream (not read): %s\n", s)
+	}
+	for _, d := range digests {
+		fmt.Fprintf(w, "%s\tfirst=%d\tlast=%d\tmsgs=%d\tsha256=%s\n", d.Name, d.FirstSeq, d.LastSeq, d.Msgs, d.Digest)
+		for _, s := range d.Seqs {
+			fmt.Fprintf(w, "%s\tseq=%d\t%s\n", d.Name, s.Seq, s.Hash)
+		}
+	}
+}
+
+// globalAccountDir is the account Spinifex's token-authenticated clients use.
+// Streams under any other account are listed but not read.
+const globalAccountDir = "$G"
+
+// copyStreamTree copies <storeDir>/$G/streams to <dst>/$G/streams, leaving out
+// consumer state (obs), and returns streams found under other accounts.
+func copyStreamTree(storeDir, dst string) (skipped []string, err error) {
+	streams, err := localStreams(storeDir)
+	if err != nil {
+		return nil, err
+	}
+	if len(streams) == 0 {
+		return nil, fmt.Errorf("no streams found under %s", storeDir)
+	}
+	for _, s := range streams {
+		if !strings.HasPrefix(s, globalAccountDir+"/") {
+			skipped = append(skipped, s)
+		}
+	}
+
+	src := filepath.Join(storeDir, globalAccountDir, "streams")
+	out := filepath.Join(dst, globalAccountDir, "streams")
+	err = filepath.WalkDir(src, func(path string, d fs.DirEntry, walkErr error) error {
+		if walkErr != nil {
+			return walkErr
+		}
+		rel, err := filepath.Rel(src, path)
+		if err != nil {
+			return err
+		}
+		// <stream>/obs holds consumers, which are not stream content.
+		if parts := strings.Split(rel, string(filepath.Separator)); d.IsDir() && len(parts) == 2 && parts[1] == "obs" {
+			return filepath.SkipDir
+		}
+		target := filepath.Join(out, rel)
+		if d.IsDir() {
+			return os.MkdirAll(target, 0o700)
+		}
+		if !d.Type().IsRegular() {
+			return nil
+		}
+		return copyFile(path, target)
+	})
+	if err != nil {
+		return nil, fmt.Errorf("copy streams from %s: %w", src, err)
+	}
+	entries, err := os.ReadDir(out)
+	if err != nil {
+		return nil, err
+	}
+	for _, e := range entries {
+		if e.IsDir() {
+			if err := rewriteStreamMetaStandalone(filepath.Join(out, e.Name())); err != nil {
+				return nil, err
+			}
+		}
+	}
+	return skipped, nil
+}
+
+// rewriteStreamMetaStandalone makes a copied stream loadable by a standalone
+// server, which refuses replicas > 1 and would expire aged messages on
+// recovery. meta.sum is highwayhash64 keyed by sha256 of the directory name.
+func rewriteStreamMetaStandalone(streamDir string) error {
+	metaPath := filepath.Join(streamDir, "meta.inf")
+	raw, err := os.ReadFile(metaPath)
+	if errors.Is(err, os.ErrNotExist) {
+		return nil
+	}
+	if err != nil {
+		return err
+	}
+	dec := json.NewDecoder(bytes.NewReader(raw))
+	dec.UseNumber()
+	var meta map[string]any
+	if err := dec.Decode(&meta); err != nil {
+		return fmt.Errorf("parse %s: %w", metaPath, err)
+	}
+	meta["num_replicas"] = 1
+	meta["max_age"] = 0
+	delete(meta, "allow_msg_ttl")
+	delete(meta, "subject_delete_marker_ttl")
+	delete(meta, "placement")
+	out, err := json.Marshal(meta)
+	if err != nil {
+		return err
+	}
+	key := sha256.Sum256([]byte(filepath.Base(streamDir)))
+	hh, err := highwayhash.New64(key[:])
+	if err != nil {
+		return err
+	}
+	hh.Write(out)
+	if err := os.WriteFile(metaPath, out, 0o600); err != nil {
+		return err
+	}
+	return os.WriteFile(filepath.Join(streamDir, "meta.sum"), []byte(hex.EncodeToString(hh.Sum(nil))), 0o600)
+}
+
+func copyFile(src, dst string) error {
+	in, err := os.Open(src)
+	if err != nil {
+		return err
+	}
+	defer in.Close()
+	out, err := os.OpenFile(dst, os.O_CREATE|os.O_WRONLY|os.O_TRUNC, 0o600)
+	if err != nil {
+		return err
+	}
+	if _, err := io.Copy(out, in); err != nil {
+		out.Close()
+		return err
+	}
+	return out.Close()
+}
+
+// digestStore opens storeParent (holding jetstream/) in a standalone server
+// that accepts in-process connections only, and digests each stream.
+func digestStore(ctx context.Context, storeParent string, only []string, withSeqs bool) ([]streamDigest, error) {
+	ns, err := server.NewServer(&server.Options{
+		ServerName: "spx-kv-digest",
+		JetStream:  true,
+		StoreDir:   storeParent,
+		DontListen: true,
+		NoLog:      true,
+		NoSigs:     true,
+	})
+	if err != nil {
+		return nil, fmt.Errorf("create reader server: %w", err)
+	}
+	ns.Start()
+	defer func() {
+		ns.Shutdown()
+		ns.WaitForShutdown()
+	}()
+	if !ns.ReadyForConnections(30 * time.Second) {
+		return nil, errors.New("reader server did not become ready")
+	}
+
+	nc, err := nats.Connect("", nats.InProcessServer(ns))
+	if err != nil {
+		return nil, fmt.Errorf("connect to reader server: %w", err)
+	}
+	defer nc.Close()
+	js, err := jetstream.New(nc)
+	if err != nil {
+		return nil, err
+	}
+
+	var names []string
+	lister := js.StreamNames(ctx)
+	for name := range lister.Name() {
+		if len(only) == 0 || slices.Contains(only, name) {
+			names = append(names, name)
+		}
+	}
+	if err := lister.Err(); err != nil {
+		return nil, fmt.Errorf("list streams: %w", err)
+	}
+	sort.Strings(names)
+
+	digests := make([]streamDigest, 0, len(names))
+	for _, name := range names {
+		d, err := digestStream(ctx, js, name, withSeqs)
+		if err != nil {
+			return nil, err
+		}
+		digests = append(digests, d)
+	}
+	return digests, nil
+}
+
+func digestStream(ctx context.Context, js jetstream.JetStream, name string, withSeqs bool) (streamDigest, error) {
+	s, err := js.Stream(ctx, name)
+	if err != nil {
+		return streamDigest{}, fmt.Errorf("open stream %s: %w", name, err)
+	}
+	info, err := s.Info(ctx)
+	if err != nil {
+		return streamDigest{}, fmt.Errorf("stream info %s: %w", name, err)
+	}
+	st := info.State
+	d := streamDigest{Name: name, FirstSeq: st.FirstSeq, LastSeq: st.LastSeq, Msgs: st.Msgs}
+
+	whole := sha256.New()
+	for seq := st.FirstSeq; st.Msgs > 0 && seq <= st.LastSeq; seq++ {
+		msg, err := s.GetMsg(ctx, seq)
+		var h string
+		switch {
+		case errors.Is(err, jetstream.ErrMsgNotFound):
+			h = "deleted"
+		case err != nil:
+			return streamDigest{}, fmt.Errorf("read %s seq %d: %w", name, seq, err)
+		default:
+			h = hashStoredMsg(msg)
+		}
+		fmt.Fprintf(whole, "%d:%s\n", seq, h)
+		if withSeqs {
+			d.Seqs = append(d.Seqs, seqDigest{Seq: seq, Hash: h})
+		}
+	}
+	d.Digest = hex.EncodeToString(whole.Sum(nil))
+	return d, nil
+}
+
+// hashStoredMsg hashes everything a replica stores for one sequence. Header
+// keys are sorted so the hash does not depend on map iteration order.
+func hashStoredMsg(m *jetstream.RawStreamMsg) string {
+	h := sha256.New()
+	var n [8]byte
+	field := func(b []byte) {
+		binary.BigEndian.PutUint64(n[:], uint64(len(b)))
+		h.Write(n[:])
+		h.Write(b)
+	}
+	binary.BigEndian.PutUint64(n[:], m.Sequence)
+	h.Write(n[:])
+	field([]byte(m.Subject))
+	binary.BigEndian.PutUint64(n[:], uint64(m.Time.UnixNano()))
+	h.Write(n[:])
+	keys := make([]string, 0, len(m.Header))
+	for k := range m.Header {
+		keys = append(keys, k)
+	}
+	sort.Strings(keys)
+	for _, k := range keys {
+		field([]byte(k))
+		for _, v := range m.Header[k] {
+			field([]byte(v))
+		}
+	}
+	field(m.Data)
+	return hex.EncodeToString(h.Sum(nil))[:16]
+}
@@ -0,0 +1,294 @@
+//test:in-package: exercises the unexported digest helpers against a real clustered
+// store, which the cobra command only reaches through a live node.
+
+package cmd
+
+import (
+	"bytes"
+	"context"
+	"fmt"
+	"net"
+	"os"
+	"path/filepath"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/nats-io/nats-server/v2/server"
+	"github.com/nats-io/nats.go"
+	"github.com/nats-io/nats.go/jetstream"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+// kvServer runs a standalone JetStream server on storeParent and returns a KV
+// bucket on it plus a stop function.
+func kvServer(t *testing.T, storeParent, bucket string) (jetstream.KeyValue, func()) {
+	t.Helper()
+	ns, err := server.NewServer(&server.Options{
+		ServerName: "kv-test", JetStream: true, StoreDir: storeParent,
+		DontListen: true, NoLog: true, NoSigs: true,
+	})
+	require.NoError(t, err)
+	ns.Start()
+	require.True(t, ns.ReadyForConnections(10*time.Second))
+	nc, err := nats.Connect("", nats.InProcessServer(ns))
+	require.NoError(t, err)
+	js, err := jetstream.New(nc)
+	require.NoError(t, err)
+	kv, err := js.CreateKeyValue(context.Background(), jetstream.KeyValueConfig{Bucket: bucket, History: 5})
+	require.NoError(t, err)
+	return kv, func() {
+		nc.Close()
+		ns.Shutdown()
+		ns.WaitForShutdown()
+	}
+}
+
+func digestOf(t *testing.T, storeDir string, withSeqs bool) []streamDigest {
+	t.Helper()
+	work := t.TempDir()
+	_, err := copyStreamTree(storeDir, filepath.Join(work, "jetstream"))
+	require.NoError(t, err)
+	d, err := digestStore(context.Background(), work, nil, withSeqs)
+	require.NoError(t, err)
+	return d
+}
+
+func TestKVDigest(t *testing.T) {
+	ctx := context.Background()
+	parent := t.TempDir()
+	kv, stop := kvServer(t, parent, "test")
+	for _, kvp := range [][2]string{{"a", "1"}, {"b", "2"}, {"a", "3"}} {
+		_, err := kv.PutString(ctx, kvp[0], kvp[1])
+		require.NoError(t, err)
+	}
+	require.NoError(t, kv.Delete(ctx, "b"))
+	store := filepath.Join(parent, "jetstream")
+
+	// Copied while the server is live, exactly as it runs on a node.
+	live := digestOf(t, store, true)
+	stop()
+
+	require.Len(t, live, 1)
+	assert.Equal(t, "KV_test", live[0].Name)
+	assert.Equal(t, uint64(1), live[0].FirstSeq)
+	assert.Equal(t, uint64(4), live[0].LastSeq)
+	assert.Equal(t, uint64(4), live[0].Msgs)
+	require.Len(t, live[0].Seqs, 4)
+
+	t.Run("stable across reads and a stopped server", func(t *testing.T) {
+		again := digestOf(t, store, false)
+		assert.Equal(t, live[0].Digest, again[0].Digest)
+	})
+
+	t.Run("source store is left untouched", func(t *testing.T) {
+		_, err := os.Stat(filepath.Join(store, "$G", "streams", "KV_test"))
+		assert.NoError(t, err)
+	})
+
+	// Same keys and values written separately differ in timestamps, which is
+	// exactly what separates two nodes' pre-formation writes.
+	t.Run("independently written store differs", func(t *testing.T) {
+		other := t.TempDir()
+		kv2, stop2 := kvServer(t, other, "test")
+		for _, kvp := range [][2]string{{"a", "1"}, {"b", "2"}, {"a", "3"}} {
+			_, err := kv2.PutString(ctx, kvp[0], kvp[1])
+			require.NoError(t, err)
+		}
+		require.NoError(t, kv2.Delete(ctx, "b"))
+		stop2()
+		d := digestOf(t, filepath.Join(other, "jetstream"), false)
+		assert.NotEqual(t, live[0].Digest, d[0].Digest)
+	})
+
+	t.Run("report lines", func(t *testing.T) {
+		var buf bytes.Buffer
+		writeDigests(&buf, "node1", store, time.Unix(0, 0), []string{"ACC/S"}, live)
+		out := buf.String()
+		assert.Contains(t, out, "# host=node1 ")
+		assert.Contains(t, out, "# skipped account stream (not read): ACC/S")
+		assert.Contains(t, out, "KV_test\tfirst=1\tlast=4\tmsgs=4\tsha256="+live[0].Digest)
+		assert.Equal(t, 4, strings.Count(out, "KV_test\tseq="))
+	})
+}
+
+func TestKVDigestStreamFilterAndDeletedSeq(t *testing.T) {
+	ctx := context.Background()
+	parent := t.TempDir()
+	// The KV bucket is only there to be excluded by the stream filter.
+	_, stop := kvServer(t, parent, "other")
+	stop()
+
+	// KV streams deny message deletes, so an interior gap needs a plain stream.
+	ns, err := server.NewServer(&server.Options{JetStream: true, StoreDir: parent, DontListen: true, NoLog: true, NoSigs: true})
+	require.NoError(t, err)
+	ns.Start()
+	require.True(t, ns.ReadyForConnections(10*time.Second))
+	nc, err := nats.Connect("", nats.InProcessServer(ns))
+	require.NoError(t, err)
+	js, err := jetstream.New(nc)
+	require.NoError(t, err)
+	s, err := js.CreateStream(ctx, jetstream.StreamConfig{Name: "ONE", Subjects: []string{"one.>"}})
+	require.NoError(t, err)
+	for _, v := range []string{"v1", "v2", "v3"} {
+		_, err = js.Publish(ctx, "one.k", []byte(v))
+		require.NoError(t, err)
+	}
+	require.NoError(t, s.DeleteMsg(ctx, 2))
+	nc.Close()
+	ns.Shutdown()
+	ns.WaitForShutdown()
+
+	work := t.TempDir()
+	_, err = copyStreamTree(filepath.Join(parent, "jetstream"), filepath.Join(work, "jetstream"))
+	require.NoError(t, err)
+	d, err := digestStore(ctx, work, []string{"ONE"}, true)
+	require.NoError(t, err)
+	require.Len(t, d, 1)
+	assert.Equal(t, "ONE", d[0].Name)
+	assert.Equal(t, uint64(2), d[0].Msgs)
+	assert.Equal(t, []seqDigest{{1, d[0].Seqs[0].Hash}, {2, "deleted"}, {3, d[0].Seqs[2].Hash}}, d[0].Seqs)
+}
+
+func freePort(t *testing.T) int {
+	t.Helper()
+	l, err := net.Listen("tcp", "127.0.0.1:0")
+	require.NoError(t, err)
+	defer l.Close()
+	return l.Addr().(*net.TCPAddr).Port
+}
+
+// A replicated stream's meta.inf carries num_replicas > 1, which a standalone
+// server refuses to load; this is the case every production node presents.
+func TestKVDigestClusteredReplicas(t *testing.T) {
+	ctx := context.Background()
+	const n = 3
+	ports := make([]int, n)
+	routes := make([]string, n)
+	for i := range ports {
+		ports[i] = freePort(t)
+		routes[i] = fmt.Sprintf("nats://127.0.0.1:%d", ports[i])
+	}
+	stores := make([]string, n)
+	servers := make([]*server.Server, n)
+	for i := range servers {
+		stores[i] = t.TempDir()
+		ns, err := server.NewServer(&server.Options{
+			ServerName: fmt.Sprintf("s%d", i+1), JetStream: true, StoreDir: stores[i],
+			Host: "127.0.0.1", Port: -1, NoLog: true, NoSigs: true,
+			Cluster: server.ClusterOpts{Name: "digest", Host: "127.0.0.1", Port: ports[i]},
+			Routes:  server.RoutesFromStr(strings.Join(routes, ",")),
+		})
+		require.NoError(t, err)
+		ns.Start()
+		servers[i] = ns
+	}
+	t.Cleanup(func() {
+		for _, ns := range servers {
+			ns.Shutdown()
+			ns.WaitForShutdown()
+		}
+	})
+
+	nc, err := nats.Connect("", nats.InProcessServer(servers[0]))
+	require.NoError(t, err)
+	defer nc.Close()
+	js, err := jetstream.New(nc)
+	require.NoError(t, err)
+
+	var kv jetstream.KeyValue
+	require.Eventually(t, func() bool {
+		kv, err = js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: "rep", History: 3, Replicas: n, TTL: time.Hour})
+		return err == nil
+	}, 30*time.Second, 250*time.Millisecond, "clustered KV create: %v", err)
+	for i := range 5 {
+		_, err := kv.PutString(ctx, fmt.Sprintf("k%d", i%2), fmt.Sprintf("v%d", i))
+		require.NoError(t, err)
+	}
+	s, err := js.Stream(ctx, "KV_rep")
+	require.NoError(t, err)
+	require.Eventually(t, func() bool {
+		info, err := s.Info(ctx)
+		if err != nil || info.Cluster == nil || len(info.Cluster.Replicas) != n-1 {
+			return false
+		}
+		for _, r := range info.Cluster.Replicas {
+			if !r.Current || r.Lag != 0 {
+				return false
+			}
+		}
+		return true
+	}, 30*time.Second, 250*time.Millisecond)
+
+	// Current means applied, not flushed; shutdown flushes every block to disk.
+	nc.Close()
+	for _, ns := range servers {
+		ns.Shutdown()
+		ns.WaitForShutdown()
+	}
+
+	var first string
+	for i, store := range stores {
+		d := digestOf(t, filepath.Join(store, "jetstream"), false)
+		require.Len(t, d, 1, "replica %d", i+1)
+		assert.Equal(t, uint64(5), d[0].Msgs, "replica %d", i+1)
+		if i == 0 {
+			first = d[0].Digest
+			continue
+		}
+		assert.Equal(t, first, d[0].Digest, "replica %d matches replica 1", i+1)
+	}
+}
+
+// Recovery applies MaxAge, so without the rewrite an expired bucket copy
+// would read as empty and mask whatever the replica holds.
+func TestKVDigestIgnoresExpiry(t *testing.T) {
+	ctx := context.Background()
+	parent := t.TempDir()
+	ns, err := server.NewServer(&server.Options{JetStream: true, StoreDir: parent, DontListen: true, NoLog: true, NoSigs: true})
+	require.NoError(t, err)
+	ns.Start()
+	require.True(t, ns.ReadyForConnections(10*time.Second))
+	nc, err := nats.Connect("", nats.InProcessServer(ns))
+	require.NoError(t, err)
+	js, err := jetstream.New(nc)
+	require.NoError(t, err)
+	kv, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: "ttl", TTL: time.Second})
+	require.NoError(t, err)
+	_, err = kv.PutString(ctx, "k", "v")
+	require.NoError(t, err)
+
+	work := t.TempDir()
+	_, err = copyStreamTree(filepath.Join(parent, "jetstream"), filepath.Join(work, "jetstream"))
+	require.NoError(t, err)
+	nc.Close()
+	ns.Shutdown()
+	ns.WaitForShutdown()
+
+	time.Sleep(1500 * time.Millisecond)
+	d, err := digestStore(ctx, work, nil, false)
+	require.NoError(t, err)
+	require.Len(t, d, 1)
+	assert.Equal(t, uint64(1), d[0].Msgs)
+}
+
+func TestCopyStreamTree(t *testing.T) {
+	store := seedStore(t, "KV_a")
+	require.NoError(t, os.MkdirAll(filepath.Join(store, "$G", "streams", "KV_a", "obs", "c1"), 0o755))
+	require.NoError(t, os.WriteFile(filepath.Join(store, "$G", "streams", "KV_a", "obs", "c1", "o.dat"), []byte("x"), 0o600))
+	require.NoError(t, os.MkdirAll(filepath.Join(store, "ACME", "streams", "ORDERS"), 0o755))
+
+	dst := filepath.Join(t.TempDir(), "jetstream")
+	skipped, err := copyStreamTree(store, dst)
+	require.NoError(t, err)
+	assert.Equal(t, []string{"ACME/ORDERS"}, skipped)
+
+	_, err = os.Stat(filepath.Join(dst, "$G", "streams", "KV_a", "msgs", "1.blk"))
+	assert.NoError(t, err, "message blocks are copied")
+	_, err = os.Stat(filepath.Join(dst, "$G", "streams", "KV_a", "obs"))
+	assert.True(t, os.IsNotExist(err), "consumer state is left out")
+
+	_, err = copyStreamTree(filepath.Join(t.TempDir(), "absent"), dst)
+	assert.Error(t, err)
+}
@@ -1289,6 +1289,9 @@ func TestAMIVolumeSizeGiB(t *testing.T) {
 }

 func TestCheckJoinPreconditions(t *testing.T) {
+	withFakeProc(t, nil)
+	noStore := filepath.Join(t.TempDir(), "nats", "jetstream")
+
 	writeToml := func(t *testing.T) string {
 		t.Helper()
 		dir := t.TempDir()
@@ -1299,13 +1302,13 @@ func TestCheckJoinPreconditions(t *testing.T) {
 	}

 	t.Run("uninitialized node joins", func(t *testing.T) {
-		if err := checkJoinPreconditions(t.TempDir(), false); err != nil {
+		if err := checkJoinPreconditions(t.TempDir(), noStore, false); err != nil {
 			t.Fatalf("want nil, got %v", err)
 		}
 	})

 	t.Run("initialized node is refused", func(t *testing.T) {
-		err := checkJoinPreconditions(writeToml(t), false)
+		err := checkJoinPreconditions(writeToml(t), noStore, false)
 		if err == nil {
 			t.Fatal("want refusal, got nil")
 		}
@@ -1317,7 +1320,7 @@ func TestCheckJoinPreconditions(t *testing.T) {
 	// The guidance is the whole point of the guard: an operator must be able to
 	// tell what a forced join would destroy without reading the source.
 	t.Run("guidance names what is discarded", func(t *testing.T) {
-		for _, want := range []string{"CA certificate", "master key", "viperblock key", "--force"} {
+		for _, want := range []string{"CA certificate", "master key", "viperblock key", "JetStream store", "--force"} {
 			if !strings.Contains(joinDiscardsIdentityMsg, want) {
 				t.Errorf("guidance missing %q", want)
 			}
@@ -1325,10 +1328,27 @@ func TestCheckJoinPreconditions(t *testing.T) {
 	})

 	t.Run("force overrides", func(t *testing.T) {
-		if err := checkJoinPreconditions(writeToml(t), true); err != nil {
+		if err := checkJoinPreconditions(writeToml(t), noStore, true); err != nil {
 			t.Fatalf("want nil with force, got %v", err)
 		}
 	})
+
+	// A store with streams and no config is still data the join would destroy.
+	t.Run("uninitialized node with streams is refused", func(t *testing.T) {
+		err := checkJoinPreconditions(t.TempDir(), seedStore(t, "KV_a"), false)
+		if err == nil || !strings.Contains(err.Error(), "JetStream store") {
+			t.Fatalf("want JetStream refusal, got %v", err)
+		}
+	})
+
+	// --force accepts the loss of data, never a wipe under a live server.
+	t.Run("running NATS is refused even with force", func(t *testing.T) {
+		withFakeProc(t, map[string]string{"4242": "/usr/local/bin/spx\x00service\x00nats\x00start\x00"})
+		err := checkJoinPreconditions(writeToml(t), seedStore(t, "KV_a"), true)
+		if !errors.Is(err, errNATSRunning) {
+			t.Fatalf("want errNATSRunning, got %v", err)
+		}
+	})
 }

 func TestJoinRetryable(t *testing.T) {
@@ -0,0 +1,139 @@
+package cmd
+
+import (
+	"bytes"
+	"errors"
+	"fmt"
+	"os"
+	"path/filepath"
+	"sort"
+	"strings"
+)
+
+// jetStreamStoreDir is where the NATS service keeps its JetStream data under a
+// node's data directory. The nats.conf template sets store_dir to
+// <data_dir>/nats/, and nats-server appends "jetstream" itself.
+func jetStreamStoreDir(dataDir string) string {
+	return filepath.Join(dataDir, "nats", "jetstream")
+}
+
+// localStreams lists the streams a JetStream store holds on disk, as
+// "<account>/<stream>". A missing store holds nothing.
+func localStreams(storeDir string) ([]string, error) {
+	accounts, err := os.ReadDir(storeDir)
+	if errors.Is(err, os.ErrNotExist) {
+		return nil, nil
+	}
+	if err != nil {
+		return nil, fmt.Errorf("read JetStream store %s: %w", storeDir, err)
+	}
+	var streams []string
+	for _, acc := range accounts {
+		if !acc.IsDir() {
+			continue
+		}
+		entries, err := os.ReadDir(filepath.Join(storeDir, acc.Name(), "streams"))
+		if errors.Is(err, os.ErrNotExist) {
+			continue
+		}
+		if err != nil {
+			return nil, fmt.Errorf("read streams for account %s: %w", acc.Name(), err)
+		}
+		for _, e := range entries {
+			if e.IsDir() {
+				streams = append(streams, acc.Name()+"/"+e.Name())
+			}
+		}
+	}
+	sort.Strings(streams)
+	return streams, nil
+}
+
+// procRoot is overridable in tests so the process scan can run against a fake
+// /proc tree.
+var procRoot = "/proc"
+
+// natsRunningLocally reports whether this host runs the Spinifex NATS service,
+// found as a process whose command line is `spx service nats ...`. Removing a
+// store under a live server races JetStream, which rewrites its index files.
+func natsRunningLocally() (bool, error) {
+	entries, err := os.ReadDir(procRoot)
+	if err != nil {
+		return false, fmt.Errorf("scan processes: %w", err)
+	}
+	for _, e := range entries {
+		if !e.IsDir() || strings.TrimLeft(e.Name(), "0123456789") != "" {
+			continue
+		}
+		raw, err := os.ReadFile(filepath.Join(procRoot, e.Name(), "cmdline"))
+		if err != nil || len(raw) == 0 {
+			continue
+		}
+		if isSpxNATSService(bytes.Split(bytes.TrimRight(raw, "\x00"), []byte{0})) {
+			return true, nil
+		}
+	}
+	return false, nil
+}
+
+// isSpxNATSService matches `spx service nats` with any leading path on spx.
+func isSpxNATSService(argv [][]byte) bool {
+	return len(argv) >= 3 &&
+		filepath.Base(string(argv[0])) == "spx" &&
+		string(argv[1]) == "service" &&
+		string(argv[2]) == "nats"
+}
+
+// errNATSRunning is returned whenever a store would be discarded with the NATS
+// service still up. No flag overrides it: the wipe would race the live server.
+var errNATSRunning = errors.New("the NATS service is still running on this node; stop it first (sudo systemctl stop spinifex.target, then confirm `pgrep -af 'spx service'` prints nothing)")
+
+// checkStoreDiscardable fails if the NATS service is running locally.
+func checkStoreDiscardable() error {
+	running, err := natsRunningLocally()
+	if err != nil {
+		return err
+	}
+	if running {
+		return errNATSRunning
+	}
+	return nil
+}
+
+// discardJetStreamStore removes a node's JetStream store before it enters a
+// multi-node cluster. nats-server adopts a same-named local stream and raft
+// catch-up never repairs the prefix it already holds.
+func discardJetStreamStore(storeDir string) error {
+	if err := checkStoreDiscardable(); err != nil {
+		return err
+	}
+	if err := os.RemoveAll(storeDir); err != nil {
+		return fmt.Errorf("remove JetStream store %s: %w", storeDir, err)
+	}
+	if _, err := os.Stat(storeDir); !errors.Is(err, os.ErrNotExist) {
+		return fmt.Errorf("JetStream store %s still present after removal", storeDir)
+	}
+	return nil
+}
+
+// checkInitJetStreamStore refuses a multi-node init over a store holding streams
+// unless discard was asked for. Keeping it as a seed is not safe: leadership
+// ignores who holds data, and unassigned local streams are deleted as orphans.
+func checkInitJetStreamStore(storeDir string, nodes int, discard bool) error {
+	if nodes < 2 {
+		return nil
+	}
+	streams, err := localStreams(storeDir)
+	if err != nil {
+		return err
+	}
+	if len(streams) == 0 {
+		return nil
+	}
+	if !discard {
+		return fmt.Errorf("this node's JetStream store %s holds %d stream(s) from before formation; "+
+			"a multi-node cluster cannot adopt them consistently. "+
+			"Re-run with --discard-jetstream to remove them once every node has joined", storeDir, len(streams))
+	}
+	return checkStoreDiscardable()
+}
@@ -0,0 +1,140 @@
+//test:in-package: swaps the unexported procRoot for a fake /proc tree and asserts on
+// the store helpers the init and join commands call.
+
+package cmd
+
+import (
+	"os"
+	"path/filepath"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+// withFakeProc points the process scan at a temp tree holding the given
+// pid -> cmdline entries (NUL-separated argv) for the duration of the test.
+func withFakeProc(t *testing.T, procs map[string]string) {
+	t.Helper()
+	root := t.TempDir()
+	for pid, cmdline := range procs {
+		dir := filepath.Join(root, pid)
+		require.NoError(t, os.MkdirAll(dir, 0o755))
+		require.NoError(t, os.WriteFile(filepath.Join(dir, "cmdline"), []byte(cmdline), 0o644))
+	}
+	prev := procRoot
+	procRoot = root
+	t.Cleanup(func() { procRoot = prev })
+}
+
+// seedStore builds a JetStream store with the named streams under account $G,
+// laid out the way nats-server writes it, and returns the store path.
+func seedStore(t *testing.T, streams ...string) string {
+	t.Helper()
+	store := filepath.Join(t.TempDir(), "nats", "jetstream")
+	for _, s := range streams {
+		dir := filepath.Join(store, "$G", "streams", s, "msgs")
+		require.NoError(t, os.MkdirAll(dir, 0o755))
+		require.NoError(t, os.WriteFile(filepath.Join(dir, "1.blk"), []byte("x"), 0o600))
+	}
+	return store
+}
+
+func TestJetStreamStoreDir(t *testing.T) {
+	assert.Equal(t, "/var/lib/spinifex/nats/jetstream", jetStreamStoreDir("/var/lib/spinifex"))
+}
+
+func TestLocalStreams(t *testing.T) {
+	t.Run("missing store holds nothing", func(t *testing.T) {
+		got, err := localStreams(filepath.Join(t.TempDir(), "absent"))
+		require.NoError(t, err)
+		assert.Empty(t, got)
+	})
+
+	t.Run("lists streams across accounts, sorted", func(t *testing.T) {
+		store := seedStore(t, "KV_b", "KV_a")
+		require.NoError(t, os.MkdirAll(filepath.Join(store, "$SYS", "_js_", "_meta_"), 0o755))
+		require.NoError(t, os.WriteFile(filepath.Join(store, "$G", "streams", "stray-file"), nil, 0o600))
+
+		got, err := localStreams(store)
+		require.NoError(t, err)
+		assert.Equal(t, []string{"$G/KV_a", "$G/KV_b"}, got)
+	})
+}
+
+func TestNATSRunningLocally(t *testing.T) {
+	tests := []struct {
+		name  string
+		procs map[string]string
+		want  bool
+	}{
+		{"no processes", nil, false},
+		{"installed service", map[string]string{"10": "/usr/local/bin/spx\x00service\x00nats\x00start\x00"}, true},
+		{"bare spx on PATH", map[string]string{"10": "spx\x00service\x00nats\x00start\x00"}, true},
+		// The join itself is an spx process and must not count.
+		{"spx admin join", map[string]string{"10": "spx\x00admin\x00join\x00--force\x00"}, false},
+		{"other spx service", map[string]string{"10": "spx\x00service\x00predastore\x00start\x00"}, false},
+		{"nats-server binary", map[string]string{"10": "nats-server\x00service\x00nats\x00"}, false},
+		{"kernel thread, empty cmdline", map[string]string{"2": ""}, false},
+		{"non-pid entry ignored", map[string]string{"self": "spx\x00service\x00nats\x00"}, false},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			withFakeProc(t, tt.procs)
+			got, err := natsRunningLocally()
+			require.NoError(t, err)
+			assert.Equal(t, tt.want, got)
+		})
+	}
+}
+
+func TestDiscardJetStreamStore(t *testing.T) {
+	t.Run("removes the store", func(t *testing.T) {
+		withFakeProc(t, nil)
+		store := seedStore(t, "KV_a")
+		require.NoError(t, discardJetStreamStore(store))
+		_, err := os.Stat(store)
+		assert.ErrorIs(t, err, os.ErrNotExist)
+	})
+
+	t.Run("missing store is not an error", func(t *testing.T) {
+		withFakeProc(t, nil)
+		assert.NoError(t, discardJetStreamStore(filepath.Join(t.TempDir(), "absent")))
+	})
+
+	t.Run("refuses under a running NATS and leaves the store", func(t *testing.T) {
+		withFakeProc(t, map[string]string{"7": "spx\x00service\x00nats\x00start\x00"})
+		store := seedStore(t, "KV_a")
+		assert.ErrorIs(t, discardJetStreamStore(store), errNATSRunning)
+		_, err := os.Stat(store)
+		assert.NoError(t, err)
+	})
+}
+
+func TestCheckInitJetStreamStore(t *testing.T) {
+	withFakeProc(t, nil)
+
+	t.Run("single-node init keeps its store", func(t *testing.T) {
+		assert.NoError(t, checkInitJetStreamStore(seedStore(t, "KV_a"), 1, false))
+	})
+
+	t.Run("multi-node init over an empty store", func(t *testing.T) {
+		assert.NoError(t, checkInitJetStreamStore(filepath.Join(t.TempDir(), "absent"), 3, false))
+	})
+
+	t.Run("multi-node init over streams is refused without the flag", func(t *testing.T) {
+		err := checkInitJetStreamStore(seedStore(t, "KV_a", "KV_b"), 4, false)
+		require.Error(t, err)
+		assert.Contains(t, err.Error(), "2 stream(s)")
+		assert.Contains(t, err.Error(), "--discard-jetstream")
+	})
+
+	t.Run("multi-node init over streams proceeds with the flag", func(t *testing.T) {
+		assert.NoError(t, checkInitJetStreamStore(seedStore(t, "KV_a"), 4, true))
+	})
+
+	t.Run("flag does not override a running NATS", func(t *testing.T) {
+		withFakeProc(t, map[string]string{"7": "spx\x00service\x00nats\x00start\x00"})
+		assert.ErrorIs(t, checkInitJetStreamStore(seedStore(t, "KV_a"), 4, true), errNATSRunning)
+	})
+}
@@ -95,8 +95,8 @@ Operational commands for inspecting cluster state. These fan out NATS requests t

 | Command | Flags | Description |
 |---------|-------|-------------|
-| `spx admin init` | `--nodes`, `--node`, `--bind`, `--advertise` (external IP off-host clients dial; auto-detected from the default route), `--port`, `--region`, `--az`, `--cluster-name`, `--cluster-bind`, `--cluster-routes`, `--predastore-nodes`, `--predastore-compaction-interval` (seconds; test clusters set a short one), `--services`, `--formation-timeout`, `--token-ttl`, `--force`, `--skip-host-dns`, `--email` (operator address for update and security notices), `--no-telemetry`, `--gpu-passthrough`, `--ipsec` (default true; encrypt intra-AZ Geneve via OVN native IPsec cluster-wide — disable only for a trusted single-rack lab), `--host` (leader to join; falls back to multicast discovery), `--external-mode` (`pool` \| `nat` routed NAT for non-bridgeable uplinks, single-node; pair with `setup-ovn.sh --nat-uplink`; note: system instances (ECS/EKS/load-balancer agents) are unsupported in nat v1 — they require `pool` mode), `--external-iface` (WAN NIC for br-external, auto-detected), `--external-source` (`dhcp` \| `static`), `--external-pool` (`start-end`), `--external-prefix-len` (default 24), `--external-gateway` (auto-detected), `--external-bind-bridge` (default `br-wan` when `--external-source=dhcp`), `--gateway-ip` (OVN gateway router's external IP for SNAT; defaults to the pool's range start, required for `nat` mode without DHCP) | Generates root IAM credentials (AKIA-prefixed access key + secret) → creates master.key (AES-256, 32 bytes, 0600) → writes bootstrap.json (consumed on first start) → generates CA + server TLS certificates → generates join token (written to `join-token` file, displayed in join command) → creates NATS config with auth token → writes spinifex.toml, awsgw.toml, predastore.toml → configures AWS CLI `spx` profile → creates directory structure under `~/spinifex/` → points the host resolver at this node's local northstar so the Spinifex zones (ELB/EKS names) resolve from the node itself (opt out with `--skip-host-dns`). `--force` on an already-initialized node is idempotent for crypto: it preserves the existing keys (master.key, predastore/viperblock encryption keys), system + admin credentials, and CA, and refreshes only the config files and the CA-signed server certificate (so a changed bind IP / SANs is picked up without breaking already-joined nodes or CA-baked AMIs). A genuine clean slate requires removing the data dirs first. |
-| `spx admin join` | `--host` (required), `--node` (required), `--token` (required), `--bind`, `--advertise` (defaults to `--bind`, or the auto-detected WAN IP when `--bind` is 0.0.0.0), `--port`, `--region`, `--az`, `--cluster-bind`, `--cluster-routes`, `--data-dir`, `--services`, `--skip-host-dns`, `--email`, `--no-telemetry`, `--predastore-compaction-interval`, `--join-timeout` (how long to keep retrying while the formation server is unreachable, default 20m), `--force` (join even though this node is already initialized, **discarding its own CA and master key**) | Connects to leader node with join token (Authorization: Bearer header) → retrieves cluster configuration → configures local node to join cluster and participate in distributed operations → points the host resolver at this node's local northstar so the Spinifex zones resolve from the node itself (opt out with `--skip-host-dns`) |
+| `spx admin init` | `--nodes`, `--node`, `--bind`, `--advertise` (external IP off-host clients dial; auto-detected from the default route), `--port`, `--region`, `--az`, `--cluster-name`, `--cluster-bind`, `--cluster-routes`, `--predastore-nodes`, `--predastore-compaction-interval` (seconds; test clusters set a short one), `--services`, `--formation-timeout`, `--token-ttl`, `--force`, `--discard-jetstream` (with `--nodes` >= 2, remove this node's existing JetStream streams once every node has joined; without it a multi-node init over a store that holds streams is refused, and it never proceeds while NATS is running), `--skip-host-dns`, `--email` (operator address for update and security notices), `--no-telemetry`, `--gpu-passthrough`, `--ipsec` (default true; encrypt intra-AZ Geneve via OVN native IPsec cluster-wide — disable only for a trusted single-rack lab), `--host` (leader to join; falls back to multicast discovery), `--external-mode` (`pool` \| `nat` routed NAT for non-bridgeable uplinks, single-node; pair with `setup-ovn.sh --nat-uplink`; note: system instances (ECS/EKS/load-balancer agents) are unsupported in nat v1 — they require `pool` mode), `--external-iface` (WAN NIC for br-external, auto-detected), `--external-source` (`dhcp` \| `static`), `--external-pool` (`start-end`), `--external-prefix-len` (default 24), `--external-gateway` (auto-detected), `--external-bind-bridge` (default `br-wan` when `--external-source=dhcp`), `--gateway-ip` (OVN gateway router's external IP for SNAT; defaults to the pool's range start, required for `nat` mode without DHCP) | Generates root IAM credentials (AKIA-prefixed access key + secret) → creates master.key (AES-256, 32 bytes, 0600) → writes bootstrap.json (consumed on first start) → generates CA + server TLS certificates → generates join token (written to `join-token` file, displayed in join command) → creates NATS config with auth token → writes spinifex.toml, awsgw.toml, predastore.toml → configures AWS CLI `spx` profile → creates directory structure under `~/spinifex/` → points the host resolver at this node's local northstar so the Spinifex zones (ELB/EKS names) resolve from the node itself (opt out with `--skip-host-dns`). `--force` on an already-initialized node is idempotent for crypto: it preserves the existing keys (master.key, predastore/viperblock encryption keys), system + admin credentials, and CA, and refreshes only the config files and the CA-signed server certificate (so a changed bind IP / SANs is picked up without breaking already-joined nodes or CA-baked AMIs). A genuine clean slate requires removing the data dirs first. |
+| `spx admin join` | `--host` (required), `--node` (required), `--token` (required), `--bind`, `--advertise` (defaults to `--bind`, or the auto-detected WAN IP when `--bind` is 0.0.0.0), `--port`, `--region`, `--az`, `--cluster-bind`, `--cluster-routes`, `--data-dir`, `--services`, `--skip-host-dns`, `--email`, `--no-telemetry`, `--predastore-compaction-interval`, `--join-timeout` (how long to keep retrying while the formation server is unreachable, default 20m), `--force` (join even though this node is already initialized or holds JetStream streams, **discarding its own CA, master key and JetStream store**; the store is removed only after formation succeeds, and the join is refused while NATS is running even with `--force`) | Connects to leader node with join token (Authorization: Bearer header) → retrieves cluster configuration → configures local node to join cluster and participate in distributed operations → points the host resolver at this node's local northstar so the Spinifex zones resolve from the node itself (opt out with `--skip-host-dns`) |

 ### Version

@@ -108,6 +108,7 @@ Operational commands for inspecting cluster state. These fan out NATS requests t

 | Command | Flags | Description |
 |---------|-------|-------------|
+| `spx admin kv digest` | `--store-dir` (default `<spinifex-dir>/nats/jetstream`), `--work-dir` (holds the temporary copy, default the system temp dir), `--stream` (repeatable filter), `--seqs` (also print a hash per sequence) | Read-only per-replica consistency check. Copies this node's `$G` streams tree (without consumer state) to a temporary directory, opens the copy in a private in-process NATS server with no listener, and prints one line per stream with first/last sequence, message count and a SHA-256 over every stored message (sequence, subject, timestamp, headers, data). Run it on every node with `sudo` and diff the output with the `#` lines removed: replicas of one stream must match. A direct get over NATS is not used because any replica may answer it. Writes landing during the copy can change a replica's tail, so gate API writes or re-run before calling a tail difference divergence. |
 | `spx admin cluster shutdown` | `--force` (shutdown even if nodes don't respond), `--timeout` (max wait per phase, default 120s), `--dry-run` (print phase plan without executing) | Performs coordinated, phased shutdown of entire cluster. Phases execute in order: GATE (stop API/UI) → DRAIN (stop VMs) → STORAGE (stop viperblock) → PERSIST (stop predastore) → INFRA (stop NATS/daemon). Each phase waits for all nodes to ACK before proceeding. Uses JetStream state tracking. |
 | `spx admin cluster drain-dhcp` | `--timeout` (reply-collection window, default 30s) | Asks each vpcd to DHCPRELEASE every external-pool DHCP lease it currently holds, returning them to the upstream DHCP server. Run on teardown before stopping services — an env reset otherwise strands held leases upstream until their TTL expires, eventually exhausting the upstream scope. Best-effort: warns and exits 0 if the cluster is already down. |
 | `spx admin node drain --local` | `--local` (drain the local node only, required), `--timeout` (max wait per phase, default 120s), `--unless-restarting` (drain on a real host shutdown or a plain target stop, and skip only when a restart of the stack is already queued; unset, it always drains) | Runs the GATE and DRAIN phases against the local node only: powers down its guests via QMP and unmounts their volumes (flushing the viperblock WAL) while every service is still up. STORAGE/PERSIST/INFRA are left to systemd's ordered teardown. Wired as the `ExecStop` of `spinifex-shutdown.service`, so `systemctl stop spinifex.target` and host reboot/poweroff drain guests before any storage service stops. |
@@ -235,7 +235,9 @@ sudo ovn-sbctl show

 Run init and join **concurrently** — init blocks until all nodes have joined.

-`--force` is in every command below so the sequence is identical whichever way you installed. It does the work on ISO-installed servers, which arrive as their own single-node cluster: joining replaces that server's CA and master key with server 1's, and `--force` is the confirmation. On `spx admin init` it is idempotent — existing keys, credentials and CA are preserved, and only the config files and server certificate are refreshed. On a freshly installed server there is nothing to lose either way.
+`--force` is in every command below so the sequence is identical whichever way you installed. It does the work on ISO-installed servers, which arrive as their own single-node cluster: joining replaces that server's CA and master key with server 1's and removes its JetStream store, and `--force` is the confirmation. On `spx admin init` it is idempotent — existing keys, credentials and CA are preserved, and only the config files and server certificate are refreshed. On a freshly installed server there is nothing to lose either way.
+
+`--discard-jetstream` on server 1 removes the streams its single-node cluster wrote, once every server has joined. Without it init refuses to form a cluster over them, because NATS would adopt each server's own copy as a replica and those copies never converge. Stop `spinifex.target` on every server first and confirm `pgrep -af 'spx service'` prints nothing: init and join both refuse while NATS is still running.

 > [!WARNING]
 > Do not point these commands at a server that has already been in service. Joining discards its master key, orphaning every volume and fragment sealed under it. That is what `--force` overrides, and it is unrecoverable.
@@ -243,7 +245,7 @@ Run init and join **concurrently** — init blocks until all nodes have joined.
 **Server 1 — initialize:**

 ```bash
-sudo spx admin init --force \
+sudo spx admin init --force --discard-jetstream \
  --node node1 --nodes 3 \
  --bind $SPINIFEX_NODE1 --cluster-bind $SPINIFEX_NODE1 \
  --port 4432 --region $AWS_REGION --az $AWS_AZ
@@ -21,6 +21,7 @@ require (
 	github.com/jackc/pgx/v5 v5.11.0
 	github.com/klauspost/cpuid/v2 v2.4.0
 	github.com/miekg/dns v1.1.73
+	github.com/minio/highwayhash v1.0.4
 	github.com/mulgadc/bluebottle v1.19.0
 	github.com/mulgadc/northstar v1.19.0
 	github.com/mulgadc/predastore v1.19.0
@@ -146,7 +147,6 @@ require (
 	github.com/mdlayher/packet v1.1.2 // indirect
 	github.com/mdlayher/socket v0.5.1 // indirect
 	github.com/minio/crc64nvme v1.1.1 // indirect
-	github.com/minio/highwayhash v1.0.4 // indirect
 	github.com/moby/locker v1.0.1 // indirect
 	github.com/moby/sys/capability v0.4.0 // indirect
 	github.com/moby/sys/mountinfo v0.7.2 // indirect
@@ -368,6 +368,22 @@ for host in "${HOSTS[@]}"; do
    on "$host" "sudo systemctl stop spinifex.target" || fail "$host: could not stop spinifex.target"
 done

+# Init and join remove each node's JetStream store and refuse while NATS still
+# runs. A stopped target is not proof of that: its storage units can outlive it.
+# Matched on the service argv so the ssh shell running this cannot match itself.
+for host in "${HOSTS[@]}"; do
+    $DRY_RUN && { echo "  [dry-run] $host: wait for every 'spx service' process to exit"; continue; }
+    on "$host" "for u in spinifex-predastore spinifex-viperblock spinifex-nats; do
+            systemctl is-active --quiet \$u && sudo systemctl stop \$u
+        done
+        for _ in \$(seq 1 45); do
+            pgrep -f '^(/usr/local/bin/)?spx service ' >/dev/null || exit 0
+            sleep 2
+        done
+        pgrep -af '^(/usr/local/bin/)?spx service ' >&2
+        exit 1" || fail "$host: spx services still running 90s after stopping spinifex.target"
+done
+
 # --- Disarm the host firewall ----------------------------------------------
 #
 # The policy lives in the kernel and outlives the services, so stopping the
@@ -529,6 +545,7 @@ done
 # target.
 init_args=(
    --force
+    --discard-jetstream
    --node "${NODE_NAMES[0]}"
    --nodes "$N"
    --bind "${LAN_IPS[0]}"
@@ -577,8 +594,8 @@ else
 fi

 # --force: each joining node arrived from the ISO already initialized, with its
-# own CA and master key. Joining replaces them with the init node's, and
-# `spx admin join` refuses to do that silently.
+# own CA, master key and JetStream store. Joining replaces the keys with the init
+# node's and removes the store, and `spx admin join` refuses to do that silently.
 #
 # Every join must run concurrently. `spx admin join` registers and then blocks
 # until the formation server has all N nodes, so joining one at a time deadlocks
@@ -0,0 +1,17 @@
+package daemon_test
+
+import (
+	"testing"
+
+	"github.com/mulgadc/spinifex/spinifex/daemon"
+	"github.com/mulgadc/spinifex/spinifex/network/reconcile"
+	"github.com/stretchr/testify/assert"
+)
+
+// vpcd reads instance records to tell a stopped guest's port from a broken one,
+// and cannot import this package to learn where they live. If these drift apart
+// it silently probes every stopped instance's port again.
+func TestInstanceRecordLocationMatchesVPCD(t *testing.T) {
+	assert.Equal(t, daemon.InstanceStateBucket, reconcile.InstanceRecordBucket)
+	assert.Equal(t, daemon.InstanceRecordPrefix, reconcile.InstanceRecordPrefix)
+}
@@ -4,12 +4,14 @@ import (
 	"context"
 	"log/slog"
 	"net/http"
+	"regexp"
 	"sync"
 	"time"

 	"go.opentelemetry.io/otel/attribute"
 	"go.opentelemetry.io/otel/trace"

+	"github.com/mulgadc/spinifex/spinifex/otelsetup"
 	"github.com/mulgadc/spinifex/spinifex/utils"
 )

@@ -26,6 +28,7 @@ const ctxAudit contextKey = "gateway.audit"
 type requestAudit struct {
 	mu            sync.Mutex
 	clientIP      string
+	requestID     string
 	accessKeyID   string
 	accountID     string
 	region        string
@@ -40,13 +43,27 @@ type requestAudit struct {
 // logging happens to be enabled.
 func requestAuditMiddleware(next http.Handler) http.Handler {
 	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
-		audit := &requestAudit{clientIP: utils.ClientIP(r.RemoteAddr)}
+		audit := &requestAudit{clientIP: utils.RequestClientIP(r), requestID: requestIDFrom(r)}
 		ctx := context.WithValue(r.Context(), ctxAudit, audit)
 		next.ServeHTTP(w, r.WithContext(ctx))
 		audit.annotate(ctx)
 	})
 }

+// requestIDPattern is what a correlation ID may look like. nginx's $request_id is
+// 32 hex characters; anything else a client sends is dropped rather than logged.
+var requestIDPattern = regexp.MustCompile(`^[A-Za-z0-9._:-]{1,128}$`)
+
+// requestIDFrom returns the X-Request-ID the edge stamped, so an awsgw line can
+// be joined to the edge access-log line that produced it.
+func requestIDFrom(r *http.Request) string {
+	id := r.Header.Get("X-Request-ID")
+	if !requestIDPattern.MatchString(id) {
+		return ""
+	}
+	return id
+}
+
 // auditFrom returns the request's audit record, or nil when there is none. Every
 // method below is nil-safe, so a handler called directly by a test behaves
 // exactly as it did before this record existed.
@@ -164,12 +181,13 @@ func (a *requestAudit) pairs() []auditField {
 		{"action", "aws.action", a.action},
 		{"principalType", "aws.principal_type", a.principalType},
 		{"authError", "aws.auth_error", a.authError},
+		{"requestID", "http.request.header.x-request-id", a.requestID},
 	}
 }

 // logRequest emits the access log line for a finished request, carrying the
 // audit fields alongside the HTTP result.
 func logRequest(r *http.Request, status int, duration time.Duration) {
-	attrs := []any{"method", r.Method, "path", r.URL.Path, "status", status, "duration", duration}
+	attrs := []any{"method", r.Method, "path", r.URL.Path, "status", status, "duration_ms", otelsetup.Millis(duration)}
 	slog.InfoContext(r.Context(), "request", append(attrs, auditFrom(r.Context()).fields()...)...)
 }
@@ -1,10 +1,14 @@
 package gateway

 import (
+	"bytes"
 	"context"
 	"fmt"
+	"log/slog"
+	"net"
 	"net/http"
 	"net/http/httptest"
+	"strings"
 	"testing"
 	"time"

@@ -62,7 +66,7 @@ func TestRequestAuditCarriesNoSecret(t *testing.T) {
 	}
 	assert.Equal(t, []string{
 		"sourceIP", "accessKeyID", "accountID", "region", "service", "action",
-		"principalType", "authError",
+		"principalType", "authError", "requestID",
 	}, keys)
 }

@@ -170,3 +174,83 @@ func TestRequestAuditRecordsRateLimitLockout(t *testing.T) {
 	assert.Equal(t, "10.15.8.13", audit().clientIP)
 	assert.Equal(t, awserrors.ErrorRequestLimitExceeded, audit().authError)
 }
+
+// sourceIP must not be something a direct caller can choose. X-Real-IP is the
+// edge's word only on a loopback connection; from anywhere else both it and
+// X-Forwarded-For are ignored.
+func TestRequestAuditClientIPIgnoresSpoofedHeadersFromNonLoopback(t *testing.T) {
+	handler, audit := auditRouter(t, nil)
+
+	for _, remote := range []string{"203.0.113.9:5555", "10.2.0.3:5555", "[2001:db8::9]:5555"} {
+		req := httptest.NewRequest(http.MethodGet, "/", nil)
+		req.RemoteAddr = remote
+		req.Header.Set("X-Real-IP", "127.0.0.1")
+		req.Header.Set("X-Forwarded-For", "198.51.100.1")
+		handler.ServeHTTP(httptest.NewRecorder(), req)
+
+		require.NotNil(t, audit())
+		host, _, err := net.SplitHostPort(remote)
+		require.NoError(t, err)
+		assert.Equal(t, host, audit().clientIP, "spoofed headers from %s must not be logged as the source", remote)
+	}
+}
+
+// Through the edge (or the console proxy) the connection is loopback, and the
+// edge's X-Real-IP is the real client.
+func TestRequestAuditClientIPTrustsRealIPFromLoopback(t *testing.T) {
+	handler, audit := auditRouter(t, nil)
+
+	req := httptest.NewRequest(http.MethodGet, "/", nil)
+	req.RemoteAddr = "127.0.0.1:40000"
+	req.Header.Set("X-Real-IP", "198.51.100.7")
+	req.Header.Set("X-Forwarded-For", "203.0.113.1")
+	handler.ServeHTTP(httptest.NewRecorder(), req)
+
+	require.NotNil(t, audit())
+	assert.Equal(t, "198.51.100.7", audit().clientIP)
+}
+
+// The edge's request ID joins an awsgw line to its access-log line; a value that
+// does not look like an ID is dropped rather than written into the log.
+func TestRequestAuditRecordsRequestID(t *testing.T) {
+	handler, audit := auditRouter(t, nil)
+
+	cases := map[string]string{
+		"4f1c2b7e9a0d4e6f8b3c5a7d9e1f2a3b": "4f1c2b7e9a0d4e6f8b3c5a7d9e1f2a3b",
+		"":                                 "",
+		"has space":                        "",
+		strings.Repeat("a", 129):           "",
+	}
+	for header, want := range cases {
+		req := httptest.NewRequest(http.MethodGet, "/", nil)
+		req.RemoteAddr = "127.0.0.1:40000"
+		if header != "" {
+			req.Header.Set("X-Request-ID", header)
+		}
+		handler.ServeHTTP(httptest.NewRecorder(), req)
+
+		require.NotNil(t, audit())
+		assert.Equal(t, want, audit().requestID, "X-Request-ID %q", header)
+	}
+}
+
+// slog renders a bare time.Duration as unitless nanoseconds, so the access line
+// carries duration_ms instead.
+func TestLogRequestDurationInMilliseconds(t *testing.T) {
+	buf := &bytes.Buffer{}
+	prev := slog.Default()
+	slog.SetDefault(slog.New(slog.NewTextHandler(buf, &slog.HandlerOptions{Level: slog.LevelInfo})))
+	t.Cleanup(func() { slog.SetDefault(prev) })
+
+	req := httptest.NewRequest(http.MethodGet, "/", nil)
+	audit := &requestAudit{clientIP: "198.51.100.7", requestID: "abc123"}
+	req = req.WithContext(context.WithValue(req.Context(), ctxAudit, audit))
+
+	logRequest(req, http.StatusOK, 1500*time.Millisecond)
+
+	out := buf.String()
+	assert.Contains(t, out, "duration_ms=1500")
+	assert.NotContains(t, out, "duration=")
+	assert.Contains(t, out, "sourceIP=198.51.100.7")
+	assert.Contains(t, out, "requestID=abc123")
+}
@@ -37,28 +37,41 @@ func EnableIPSecEncapsulation() error {
 	return nil
 }

-// errNBUnreachable marks the one failure that means "this node has no local NB
-// DB" rather than "the read broke". A permission change on the socket, a missing
-// binary or a timed-out transaction are all real faults on a management node,
-// and lumping them in with this would silence them.
-var errNBUnreachable = errors.New("no local OVN NB DB")
+// errNBUnreachable marks the one failure that means "no NB DB answered" rather
+// than "the read broke". A permission change on the socket, a missing binary or
+// a timed-out transaction are real faults, and lumping them in would silence them.
+var errNBUnreachable = errors.New("no OVN NB DB reachable")

 // nbUnreachablePatterns are what ovsdb's client library prints when it cannot
-// open or complete a handshake on the socket.
+// open or complete a handshake with any of its remotes.
 var nbUnreachablePatterns = []string{
 	"database connection failed",
 	"connection refused",
 	"no such file or directory",
 }

-// GetNBGlobalIPSec reads NB_Global.ipsec from the local OVN NB DB. The error
-// doubles as the reachability answer: a present socket file says nothing about
-// whether the database behind it accepts connections yet.
+// nbctlArgs targets nbAddr (the cluster remote list; empty is the local socket).
+// Reads accept any connected member, so a raft follower answers instead of
+// refusing; writes stay leader-only so ovn-nbctl walks the remotes to the leader.
+func nbctlArgs(nbAddr string, leaderOnly bool, cmd ...string) []string {
+	var args []string
+	if nbAddr != "" {
+		args = append(args, "--db="+nbAddr)
+	}
+	if !leaderOnly {
+		args = append(args, "--no-leader-only")
+	}
+	return append(append(args, "--timeout=5"), cmd...)
+}
+
+// GetNBGlobalIPSec reads NB_Global.ipsec. The error doubles as the reachability
+// answer: a present socket file says nothing about whether the database behind
+// it accepts connections yet.
 //
 // Reads stdout alone. ovn-nbctl writes vlog lines to stderr on a successful run,
 // and folding those into the value parses a live "true" as false.
-func GetNBGlobalIPSec() (bool, error) {
-	cmd := utils.SudoCommand("ovn-nbctl", "--timeout=5", "get", "NB_Global", ".", "ipsec")
+func GetNBGlobalIPSec(nbAddr string) (bool, error) {
+	cmd := utils.SudoCommand("ovn-nbctl", nbctlArgs(nbAddr, false, "get", "NB_Global", ".", "ipsec")...)
 	var stderr bytes.Buffer
 	cmd.Stderr = &stderr
 	out, err := cmd.Output()
@@ -83,15 +96,15 @@ func isNBUnreachable(msg string) bool {
 	return false
 }

-// SetNBGlobalIPSec writes NB_Global.ipsec on the local OVN NB DB, triggering
-// ovn-controller to add options:remote_name to Geneve tunnels for strongSwan.
-// Only the management node has a reachable NB DB; callers gate on that.
-func SetNBGlobalIPSec(enable bool) error {
+// SetNBGlobalIPSec writes NB_Global.ipsec, triggering ovn-controller to add
+// options:remote_name to Geneve tunnels for strongSwan. Callers elect a single
+// writer, so this never races another node's write.
+func SetNBGlobalIPSec(nbAddr string, enable bool) error {
 	val := "false"
 	if enable {
 		val = "true"
 	}
-	out, err := utils.SudoCommand("ovn-nbctl", "--timeout=5", "set", "NB_Global", ".", "ipsec="+val).CombinedOutput()
+	out, err := utils.SudoCommand("ovn-nbctl", nbctlArgs(nbAddr, true, "set", "NB_Global", ".", "ipsec="+val)...).CombinedOutput()
 	if err != nil {
 		return fmt.Errorf("set NB_Global ipsec=%s: %s: %w", val, strings.TrimSpace(string(out)), err)
 	}
@@ -36,12 +36,21 @@ type IPSecNodeStatus struct {
 	// Ready reports that this node's local IPsec configuration is complete.
 	Ready bool

-	// NBReachable reports that this node can read and write NB_Global. Only
-	// management nodes can, and which nodes those are is not in the config, so
-	// they say so here and every node elects the same writer from the answer.
+	// NBReachable reports that this node read NB_Global this pass, over its
+	// configured NB remotes or its local socket. Reachability can differ per node,
+	// so each says so here and every node elects the same writer from the answer.
 	NBReachable bool
 }

+// nbAddrFor returns this node's configured NB remote list, or "" for the local
+// socket. The list reaches the cluster from a node that runs no NB DB at all.
+func nbAddrFor(clusterConfig *config.ClusterConfig) string {
+	if clusterConfig == nil {
+		return ""
+	}
+	return clusterConfig.Nodes[clusterConfig.Node].VPCD.OVNNBAddr
+}
+
 // IPSecBarrier carries each node's IPsec state across the cluster.
 // NB_Global.ipsec is cluster-wide, so asserting it from one node's local
 // knowledge is what lets a partially configured mesh black-hole guest traffic.
@@ -156,7 +165,8 @@ func DisableOVNIPSec(ctx context.Context, clusterConfig *config.ClusterConfig, b
 		}
 	}

-	current, err := GetNBGlobalIPSec()
+	nbAddr := nbAddrFor(clusterConfig)
+	current, err := GetNBGlobalIPSec(nbAddr)
 	if err != nil {
 		return skipUnlessNBReadable(err)
 	}
@@ -164,15 +174,15 @@ func DisableOVNIPSec(ctx context.Context, clusterConfig *config.ClusterConfig, b
 		return nil
 	}
 	slog.Info("ipsec: switched off for this cluster, releasing the cluster-wide encryption requirement")
-	return SetNBGlobalIPSec(false)
+	return SetNBGlobalIPSec(nbAddr, false)
 }

-// skipUnlessNBReadable turns "this node has no local NB DB" into a quiet skip
-// and leaves every other failure — a socket it may not open, a missing binary,
-// a timed-out transaction — as a real error, so the pass retries and says so.
+// skipUnlessNBReadable turns "no NB DB answered" into a quiet skip and leaves
+// every other failure — a socket it may not open, a missing binary, a timed-out
+// transaction — as a real error, so the pass retries and says so.
 func skipUnlessNBReadable(err error) error {
 	if errors.Is(err, errNBUnreachable) {
-		slog.Debug("ipsec: no local OVN NB DB, leaving NB_Global to a management node", "err", err)
+		slog.Debug("ipsec: no OVN NB DB reachable, leaving NB_Global to a node that can read it", "err", err)
 		return nil
 	}
 	return err
@@ -232,7 +242,8 @@ func EnableOVNIPSec(ctx context.Context, configPath string, clusterConfig *confi

 	// Probed before the local half, because the answer is published alongside it:
 	// which nodes can write NB_Global is not in the config, so they have to say.
-	current, nbErr := GetNBGlobalIPSec()
+	nbAddr := nbAddrFor(clusterConfig)
+	current, nbErr := GetNBGlobalIPSec(nbAddr)
 	if nbErr != nil && !errors.Is(nbErr, errNBUnreachable) {
 		return nbErr
 	}
@@ -256,7 +267,7 @@ func EnableOVNIPSec(ctx context.Context, configPath string, clusterConfig *confi
 	if nbErr != nil {
 		return skipUnlessNBReadable(nbErr)
 	}
-	return reconcileNBGlobalIPSec(ctx, clusterConfig, barrier, current)
+	return reconcileNBGlobalIPSec(ctx, clusterConfig, barrier, nbAddr, current)
 }

 // configureLocalIPSec does this node's own half: point OVS at the peer cert and
@@ -298,12 +309,12 @@ func publishStatus(ctx context.Context, clusterConfig *config.ClusterConfig, bar
 // and ovs-monitor-ipsec materialises no strongSwan connections; with it, a
 // chassis that has not finished its own setup silently drops guest traffic. So
 // it tracks the slowest live chassis, not this one.
-func reconcileNBGlobalIPSec(ctx context.Context, clusterConfig *config.ClusterConfig, barrier IPSecBarrier, current bool) error {
+func reconcileNBGlobalIPSec(ctx context.Context, clusterConfig *config.ClusterConfig, barrier IPSecBarrier, nbAddr string, current bool) error {
 	if barrier == nil || clusterConfig == nil {
 		if current {
 			return nil
 		}
-		return SetNBGlobalIPSec(true)
+		return SetNBGlobalIPSec(nbAddr, true)
 	}

 	cluster, err := barrier.Cluster(ctx, nodeNames(clusterConfig))
@@ -313,7 +324,7 @@ func reconcileNBGlobalIPSec(ctx context.Context, clusterConfig *config.ClusterCo
 		return fmt.Errorf("read IPsec cluster state: %w", err)
 	}

-	// One writer. Every management node can reach the NB DB, so without this they
+	// One writer. Every node that reads NB_Global can also write it, so without this they
 	// race on one cluster-global row from snapshots taken at different instants,
 	// and two that disagree for a single pass flip the flag — tearing down and
 	// rebuilding every strongSwan connection in the cluster on each flip.
@@ -329,7 +340,7 @@ func reconcileNBGlobalIPSec(ctx context.Context, clusterConfig *config.ClusterCo
 		return nil
 	case len(pending) == 0:
 		slog.Info("ipsec: every live chassis reports a complete configuration, requiring encryption cluster-wide")
-		return SetNBGlobalIPSec(true)
+		return SetNBGlobalIPSec(nbAddr, true)
 	case !current:
 		slog.Info("ipsec: holding encryption off until every chassis is configured", "pending", pending)
 		return nil
@@ -338,7 +349,7 @@ func reconcileNBGlobalIPSec(ctx context.Context, clusterConfig *config.ClusterCo
 	// Plaintext Geneve is where the cluster sat before IPsec was asked for, and it
 	// is both recoverable and visible. A black hole is neither.
 	slog.Error("ipsec: retracting cluster-wide encryption, chassis are unconfigured and cross-chassis guest traffic would black-hole", "pending", pending)
-	if err := SetNBGlobalIPSec(false); err != nil {
+	if err := SetNBGlobalIPSec(nbAddr, false); err != nil {
 		slog.Error("ipsec: NB_Global still demands encryption that unconfigured chassis cannot perform, cross-chassis guest traffic is being dropped now",
 			"pending", pending, "err", err)
 		return err
@@ -347,8 +358,8 @@ func reconcileNBGlobalIPSec(ctx context.Context, clusterConfig *config.ClusterCo
 }

 // nbGlobalWriter picks the one node that may write NB_Global this pass: the
-// first live management node in name order. Every node computes it from the same
-// published set, so they agree without a lock.
+// first live node in name order that could read it. Every node computes it from
+// the same published set, so they agree without a lock.
 func nbGlobalWriter(cluster map[string]IPSecNodeStatus) string {
 	var writer string
 	for _, node := range slices.Sorted(maps.Keys(cluster)) {
@@ -33,15 +33,15 @@ type recordingSudo struct {
 	activeOutput  string
 	enabledOutput map[string]string
 	activePerUnit map[string]string
-	// nbUnreachable stands in for a node whose ovn-central is not answering, or
-	// which is not a management node at all. The socket file exists in both
-	// cases, which is why its presence was never the right question.
+	// nbUnreachable stands in for a node that no NB DB answered, over its
+	// remotes or its local socket. A socket file can exist either way, which is
+	// why its presence was never the right question.
 	nbUnreachable bool
 	// nbIPSec is what NB_Global.ipsec currently reads as.
 	nbIPSec string
-	// nbError is a failure that is not "there is no local NB DB here" — a
-	// permission change on the socket, a missing binary, a timed-out
-	// transaction. On a management node each of those is a real fault.
+	// nbError is a failure that is not "no NB DB answered" — a permission
+	// change on the socket, a missing binary, a timed-out transaction. Each of
+	// those is a real fault.
 	nbError string
 }

@@ -122,7 +122,7 @@ func (f *fakeBarrier) Cluster(_ context.Context, _ []string) (map[string]IPSecNo
 }

 // allReady is the steady state: every chassis has finished its local setup and
-// node1 is the management node that owns the NB_Global write.
+// node1 is the first node that reads NB_Global, so it owns the write.
 func allReady() *fakeBarrier {
 	return &fakeBarrier{cluster: map[string]IPSecNodeStatus{
 		"node1": {Ready: true, NBReachable: true},
@@ -521,7 +521,7 @@ func (f *flakyBarrier) succeeded() bool {
 	return f.calls > f.failFor
 }

-// Every management node can reach the NB DB, so without a single elected writer
+// Every node that reads NB_Global can write it, so without a single elected writer
 // two of them disagreeing for one pass flip the flag and rebuild every
 // strongSwan connection in the cluster.
 func TestEnableOVNIPSec_OnlyTheElectedWriterTouchesNBGlobal(t *testing.T) {
@@ -542,8 +542,8 @@ func TestEnableOVNIPSec_OnlyTheElectedWriterTouchesNBGlobal(t *testing.T) {
 		"a node that is not the elected writer must not race the flag")
 }

-// The elected writer is the first live management node, not merely the first
-// node: a hypervisor cannot write the flag and must not veto the one that can.
+// The elected writer is the first live node that could read NB_Global, not
+// merely the first node: one that cannot reach NB must not veto one that can.
 func TestNBGlobalWriter_SkipsNodesWithoutANBDB(t *testing.T) {
 	writer := nbGlobalWriter(map[string]IPSecNodeStatus{
 		"node1": {Ready: true},
@@ -606,8 +606,8 @@ func TestReconcileOVNIPSec_DisabledIsANoOpWhenAlreadyReleased(t *testing.T) {
 	assert.Empty(t, recorder.nbctlWrites())
 }

-// A management node whose NB read fails for any reason other than "there is no
-// local DB here" has a real fault. Swallowing it was how the node that holds
+// A node whose NB read fails for any reason other than "no NB DB answered" has
+// a real fault. Swallowing it was how the node that holds
 // the flag could go blind and never retract.
 func TestEnableOVNIPSec_NBReadFailureFailsThePass(t *testing.T) {
 	recorder := &recordingSudo{nbError: "ovn-nbctl: unix:/var/run/ovn/ovnnb_db.sock: permission denied"}
@@ -621,7 +621,7 @@ func TestEnableOVNIPSec_NBReadFailureFailsThePass(t *testing.T) {
 	assert.Empty(t, recorder.nbctlWrites())
 }

-// A node with no local NB DB is the ordinary hypervisor case, not a fault.
+// No NB DB answering is skipped quietly, not a fault, and the node stands down.
 func TestEnableOVNIPSec_NoLocalNBDBIsNotAnError(t *testing.T) {
 	recorder := &recordingSudo{nbUnreachable: true}
 	t.Cleanup(utils.SetSudoCommandForTest(recorder.stub))
@@ -645,7 +645,59 @@ func TestGetNBGlobalIPSec_IgnoresStderrChatter(t *testing.T) {
 		return exec.Command("sh", "-c", `echo "ovsdb-idl|WARN|reconnecting" >&2; echo true`)
 	}))

-	value, err := GetNBGlobalIPSec()
+	value, err := GetNBGlobalIPSec("")
 	require.NoError(t, err)
 	assert.True(t, value)
 }
+
+// nbctlReads returns the ovn-nbctl invocations that read NB_Global.
+func (r *recordingSudo) nbctlReads() [][]string {
+	var out [][]string
+	for _, run := range r.runs {
+		if run[0] == "ovn-nbctl" && slices.Contains(run, "get") {
+			out = append(out, run)
+		}
+	}
+	return out
+}
+
+// A raft follower refuses a leader-only read on its own socket, which logged an
+// ERR every minute and read as "no NB DB here". The read must accept any
+// connected member; the write must still reach the leader.
+func TestNBctlArgs_ReadsAcceptFollowersWritesStayLeaderOnly(t *testing.T) {
+	const nb = "tcp:10.0.0.1:6641,tcp:10.0.0.2:6641"
+
+	assert.Equal(t, []string{"--no-leader-only", "--timeout=5", "get", "NB_Global", ".", "ipsec"},
+		nbctlArgs("", false, "get", "NB_Global", ".", "ipsec"))
+	assert.Equal(t, []string{"--db=" + nb, "--no-leader-only", "--timeout=5", "get", "NB_Global", ".", "ipsec"},
+		nbctlArgs(nb, false, "get", "NB_Global", ".", "ipsec"))
+	assert.Equal(t, []string{"--db=" + nb, "--timeout=5", "set", "NB_Global", ".", "ipsec=false"},
+		nbctlArgs(nb, true, "set", "NB_Global", ".", "ipsec=false"))
+}
+
+// A node that runs no NB DB reaches the cluster through its configured remotes,
+// so both the read and the retraction carry the list rather than the local socket.
+func TestReconcileOVNIPSec_UsesTheConfiguredNBRemotes(t *testing.T) {
+	const nb = "tcp:10.0.0.1:6641,tcp:10.0.0.2:6641"
+	recorder := &recordingSudo{nbIPSec: "true"}
+	t.Cleanup(utils.SetSudoCommandForTest(recorder.stub))
+
+	cfg := multiNodeIPSecConfig(false)
+	cfg.Nodes["node1"] = config.Config{VPCD: config.VPCDConfig{OVNNBAddr: nb}}
+
+	require.NoError(t, ReconcileOVNIPSec(t.Context(), "/etc/spinifex/spinifex.toml", cfg, allReady()))
+
+	assert.Equal(t, [][]string{{"ovn-nbctl", "--db=" + nb, "--no-leader-only", "--timeout=5", "get", "NB_Global", ".", "ipsec"}},
+		recorder.nbctlReads())
+	assert.Equal(t, [][]string{{"ovn-nbctl", "--db=" + nb, "--timeout=5", "set", "NB_Global", ".", "ipsec=false"}},
+		recorder.nbctlWrites())
+}
+
+// Only "nothing answered" is unreachable. The local socket being absent is that
+// case; a follower that answers is not, and never produces this output.
+func TestIsNBUnreachable(t *testing.T) {
+	assert.True(t, isNBUnreachable("ovn-nbctl: unix:/var/run/ovn/ovnnb_db.sock: database connection failed (No such file or directory)"))
+	assert.True(t, isNBUnreachable("ovn-nbctl: tcp:10.0.0.1:6641: database connection failed (Connection refused)"))
+	assert.False(t, isNBUnreachable("ovn-nbctl: unix:/var/run/ovn/ovnnb_db.sock: permission denied"))
+	assert.False(t, isNBUnreachable("ovn-nbctl: transaction timed out"))
+}
@@ -58,6 +58,8 @@ func SeedNexthopMAC(ctx context.Context, runner Runner, nbAddr, lrpName, nexthop
 		}
 	}

+	// Writes with --no-leader-only and relies on a follower forwarding the
+	// transaction. NB_Global.ipsec writes stay leader-only instead (ipsec.go).
 	var db []string
 	if nbAddr != "" {
 		db = []string{"--db=" + nbAddr, "--no-leader-only"}
@@ -707,6 +707,9 @@ func (r *reconciler) applyEIPs(ctx context.Context, intent IntentState, _ Actual
 	sem := make(chan struct{}, guestPortDatapathConcurrency)
 	var wg sync.WaitGroup
 	for _, spec := range specs {
+		if r.guestPortIdle(spec, intent.IdlePorts) {
+			continue
+		}
 		select {
 		case <-ctx.Done():
 			wg.Wait()
@@ -719,6 +722,60 @@ func (r *reconciler) applyEIPs(ctx context.Context, intent IntentState, _ Actual
 		})
 	}
 	wg.Wait()
+	r.forgetIdlePorts(specs)
+}
+
+// guestPortIdle reports whether spec's port belongs to a stopped or terminated
+// instance, logging at Info only when that changes. Entering idle drops any held
+// backoff, so the port is probed on the first pass after the instance starts.
+func (r *reconciler) guestPortIdle(spec policy.EIPSpec, idle map[string]string) bool {
+	if spec.PortName == "" {
+		return false
+	}
+	instanceID, isIdle := idle[spec.PortName]
+
+	r.portBackoffMu.Lock()
+	_, wasIdle := r.idlePorts[spec.PortName]
+	switch {
+	case isIdle && !wasIdle:
+		if r.idlePorts == nil {
+			r.idlePorts = map[string]struct{}{}
+		}
+		r.idlePorts[spec.PortName] = struct{}{}
+		delete(r.portBackoff, spec.PortName)
+	case !isIdle && wasIdle:
+		delete(r.idlePorts, spec.PortName)
+	}
+	r.portBackoffMu.Unlock()
+
+	logKV := []any{"vpc_id", spec.VPCID, "lsp", spec.PortName, "eni_id", eniIDFromPort(spec.PortName),
+		"external_ip", spec.ExternalIP, "logical_ip", spec.LogicalIP}
+	switch {
+	case isIdle && !wasIdle:
+		slog.Info("reconcile/apply: guest port's instance is not running; not probing its datapath",
+			append(logKV, "instance_id", instanceID)...)
+	case isIdle:
+		slog.Debug("reconcile/apply: guest port's instance still not running; not probing", logKV...)
+	case wasIdle:
+		slog.Info("reconcile/apply: guest port's instance is no longer stopped; probing its datapath again", logKV...)
+	}
+	return isIdle
+}
+
+// forgetIdlePorts drops idle marks for ports that no longer carry a public IP, so
+// the set tracks live EIPs rather than every port ever seen.
+func (r *reconciler) forgetIdlePorts(specs []policy.EIPSpec) {
+	live := make(map[string]struct{}, len(specs))
+	for _, spec := range specs {
+		live[spec.PortName] = struct{}{}
+	}
+	r.portBackoffMu.Lock()
+	defer r.portBackoffMu.Unlock()
+	for lsp := range r.idlePorts {
+		if _, ok := live[lsp]; !ok {
+			delete(r.idlePorts, lsp)
+		}
+	}
 }

 // floatingIPSpecs is every dnat_and_snat the datapath must carry: user EIPs
@@ -0,0 +1,132 @@
+package reconcile
+
+//test:in-package — drives applyEIPs and reads the unexported idle and backoff
+//state the skip is recorded in.
+
+import (
+	"bytes"
+	"context"
+	"log/slog"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/mulgadc/spinifex/spinifex/network/policy"
+)
+
+// captureReconcileLogs swaps the default logger for one writing to a buffer at
+// Info, so a test can count what an operator would see in the journal.
+func captureReconcileLogs(t *testing.T) *bytes.Buffer {
+	t.Helper()
+	buf := &bytes.Buffer{}
+	prev := slog.Default()
+	slog.SetDefault(slog.New(slog.NewTextHandler(buf, &slog.HandlerOptions{Level: slog.LevelInfo})))
+	t.Cleanup(func() { slog.SetDefault(prev) })
+	return buf
+}
+
+func idleEIPIntent(idle bool) IntentState {
+	intent := IntentState{EIPs: map[string]policy.EIPSpec{
+		"10.0.0.4": {VPCID: "vpc-a", ExternalIP: "192.0.2.46", LogicalIP: "10.0.0.4", PortName: "port-eni-1"},
+	}}
+	if idle {
+		intent.IdlePorts = map[string]string{"port-eni-1": "i-stopped"}
+	}
+	return intent
+}
+
+// A customer-stopped instance keeps its ENI and EIP but has no tap, so its port
+// can never bind. Passes must keep the DNAT row and never probe, nudge or file
+// an ERROR, and say so once rather than every pass.
+func TestApplyEIPs_StoppedInstancePortIsNotProbed(t *testing.T) {
+	withFastGuestPortBounds(t)
+	logs := captureReconcileLogs(t)
+
+	f := &fakeClaimVerifier{guestUpAfter: -1}
+	nat := &countingNAT{}
+	r := &reconciler{gwClaim: f, nat: nat, ovn: ovnWithGuestLSP(t, "port-eni-1")}
+
+	for range 3 {
+		r.applyEIPs(context.Background(), idleEIPIntent(true), ActualState{}, &passResult{})
+	}
+
+	if got := nat.count(); got != 3 {
+		t.Errorf("AddEIP called %d times, want 3: the DNAT row must still be asserted", got)
+	}
+	if f.guestChecks != 0 || f.nudges != 0 || f.sbChecks != 0 {
+		t.Errorf("probes=%d nudges=%d sb_checks=%d, want all 0 for a stopped instance's port",
+			f.guestChecks, f.nudges, f.sbChecks)
+	}
+	out := logs.String()
+	if n := strings.Count(out, "guest port's instance is not running"); n != 1 {
+		t.Errorf("idle notice logged %d times over 3 passes, want 1:\n%s", n, out)
+	}
+	for _, noisy := range []string{"level=WARN", "level=ERROR"} {
+		if strings.Contains(out, noisy) {
+			t.Errorf("stopped instance's port produced %s:\n%s", noisy, out)
+		}
+	}
+}
+
+// Starting the instance must converge on the first pass without a nudge, and
+// the resume is logged once.
+func TestApplyEIPs_StartedInstanceConvergesWithoutNudge(t *testing.T) {
+	withFastGuestPortBounds(t)
+	logs := captureReconcileLogs(t)
+
+	f := &fakeClaimVerifier{guestUpAfter: 0}
+	r := &reconciler{gwClaim: f, nat: &countingNAT{}, ovn: ovnWithGuestLSP(t, "port-eni-1")}
+
+	r.applyEIPs(context.Background(), idleEIPIntent(true), ActualState{}, &passResult{})
+	r.applyEIPs(context.Background(), idleEIPIntent(false), ActualState{}, &passResult{})
+	r.applyEIPs(context.Background(), idleEIPIntent(false), ActualState{}, &passResult{})
+
+	if f.guestChecks != 2 {
+		t.Errorf("probes = %d, want 2: one per pass once the instance is running", f.guestChecks)
+	}
+	if f.nudges != 0 {
+		t.Errorf("nudges = %d, want 0: a started instance's bound port needs no recompute", f.nudges)
+	}
+	out := logs.String()
+	if n := strings.Count(out, "no longer stopped; probing its datapath again"); n != 1 {
+		t.Errorf("resume notice logged %d times, want 1:\n%s", n, out)
+	}
+	if strings.Contains(out, "did not converge") {
+		t.Errorf("started instance logged a convergence failure:\n%s", out)
+	}
+}
+
+// Failures filed while the instance was stopping must not hold back the probe
+// once it starts: backoff caps at 30 minutes, which would hide a real outage.
+func TestApplyEIPs_IdleClearsHeldBackoff(t *testing.T) {
+	withFastGuestPortBounds(t)
+	withFastGuestPortBackoff(t, time.Hour, time.Hour)
+	captureReconcileLogs(t)
+
+	f := &fakeClaimVerifier{guestUpAfter: 0}
+	r := &reconciler{gwClaim: f, nat: &countingNAT{}, ovn: ovnWithGuestLSP(t, "port-eni-1")}
+	r.recordPortFailure("port-eni-1")
+
+	r.applyEIPs(context.Background(), idleEIPIntent(true), ActualState{}, &passResult{})
+	r.applyEIPs(context.Background(), idleEIPIntent(false), ActualState{}, &passResult{})
+
+	if f.guestChecks != 1 {
+		t.Errorf("probes = %d, want 1: the backoff from before the stop must not survive it", f.guestChecks)
+	}
+}
+
+// An EIP that is released while its instance is stopped must not leave its idle
+// mark behind to swallow a later notice for the same port.
+func TestApplyEIPs_ForgetsIdleMarkWhenEIPGoes(t *testing.T) {
+	withFastGuestPortBounds(t)
+	captureReconcileLogs(t)
+
+	r := &reconciler{gwClaim: &fakeClaimVerifier{guestUpAfter: 0}, nat: &countingNAT{}, ovn: ovnWithGuestLSP(t, "port-eni-1")}
+
+	r.applyEIPs(context.Background(), idleEIPIntent(true), ActualState{}, &passResult{})
+	r.applyEIPs(context.Background(), IntentState{}, ActualState{}, &passResult{})
+
+	if len(r.idlePorts) != 0 {
+		t.Errorf("idle marks = %v, want none once the port carries no public IP", r.idlePorts)
+	}
+}
@@ -19,6 +19,7 @@ import (
 	"github.com/mulgadc/spinifex/spinifex/network/policy"
 	"github.com/mulgadc/spinifex/spinifex/network/topology"
 	"github.com/mulgadc/spinifex/spinifex/utils"
+	"github.com/mulgadc/spinifex/spinifex/vm"
 	"github.com/nats-io/nats.go/jetstream"
 )

@@ -35,8 +36,18 @@ type IntentState struct {
 	IGWRoutes   map[string]SubnetEgressIntent // per (subnet, dest) IGW egress reroute
 	NATGWRoutes map[string]SubnetEgressIntent // per (subnet, dest) NATGW egress reroute
 	DropGates   map[string]SubnetEgressIntent // per (subnet, dest=0.0.0.0/0) drop policy
+	// IdlePorts maps a public-IP guest LSP to its instance when that instance is
+	// stopped or terminated. Its port has no tap, so the datapath probe skips it.
+	IdlePorts map[string]string
 }

+// The instance record space the daemon owns. Duplicated rather than imported
+// because daemon imports this package; a daemon test pins the two together.
+const (
+	InstanceRecordBucket = "spinifex-instance-state"
+	InstanceRecordPrefix = "i."
+)
+
 // SubnetEgressIntent is a per-subnet default-route policy entry for the VPC LR,
 // keyed by (subnet, destCIDR) pointing at an IGW or NAT gateway.
 type SubnetEgressIntent struct {
@@ -64,6 +75,7 @@ func LoadIntentFromKV(ctx context.Context, js jetstream.JetStream, localAZ strin
 		IGWRoutes:   make(map[string]SubnetEgressIntent),
 		NATGWRoutes: make(map[string]SubnetEgressIntent),
 		DropGates:   make(map[string]SubnetEgressIntent),
+		IdlePorts:   make(map[string]string),
 	}

 	localVPCs, err := loadVPCs(ctx, js, localAZ, intent.VPCs)
@@ -76,7 +88,8 @@ func LoadIntentFromKV(ctx context.Context, js jetstream.JetStream, localAZ strin
 	if err := loadSGs(ctx, js, localVPCs, intent.SGs); err != nil {
 		return IntentState{}, err
 	}
-	if err := loadPorts(ctx, js, localVPCs, intent.Ports); err != nil {
+	portInstances := make(map[string]string)
+	if err := loadPorts(ctx, js, localVPCs, intent.Ports, portInstances); err != nil {
 		return IntentState{}, err
 	}
 	if err := loadIGWs(ctx, js, localVPCs, intent.IGWs); err != nil {
@@ -85,6 +98,7 @@ func LoadIntentFromKV(ctx context.Context, js jetstream.JetStream, localAZ strin
 	if err := loadEIPs(ctx, js, localVPCs, intent.EIPs); err != nil {
 		return IntentState{}, err
 	}
+	loadIdlePorts(ctx, js, intent, portInstances)
 	routeTables, err := loadRouteTables(ctx, js, localVPCs)
 	if err != nil {
 		return IntentState{}, err
@@ -348,7 +362,7 @@ func loadSGs(ctx context.Context, js jetstream.JetStream, localVPCs map[string]s
 	return nil
 }

-func loadPorts(ctx context.Context, js jetstream.JetStream, localVPCs map[string]struct{}, out map[string]topology.PortSpec) error {
+func loadPorts(ctx context.Context, js jetstream.JetStream, localVPCs map[string]struct{}, out map[string]topology.PortSpec, instances map[string]string) error {
 	kv, err := js.KeyValue(ctx, handlers_ec2_vpc.KVBucketENIs)
 	if err != nil {
 		slog.Debug("reconcile/intent: ENI bucket not available, skipping", "err", err)
@@ -401,10 +415,71 @@ func loadPorts(ctx context.Context, js jetstream.JetStream, localVPCs map[string
 			PublicIP:     publicIP,
 			SuppressDHCP: rec.SuppressDHCP,
 		}
+		if rec.InstanceId != "" {
+			instances[rec.NetworkInterfaceId] = rec.InstanceId
+		}
 	}
 	return nil
 }

+// loadIdlePorts fills intent.IdlePorts for the ports the EIP datapath probe would
+// visit. Anything it cannot read is left out, so the port is probed as before.
+func loadIdlePorts(ctx context.Context, js jetstream.JetStream, intent IntentState, instances map[string]string) {
+	candidates := make(map[string]struct{}, len(intent.EIPs))
+	for id, p := range intent.Ports {
+		if p.PublicIP.IsValid() {
+			candidates[topology.Port(id)] = struct{}{}
+		}
+	}
+	for _, e := range intent.EIPs {
+		if e.PortName != "" {
+			candidates[e.PortName] = struct{}{}
+		}
+	}
+	if len(candidates) == 0 {
+		return
+	}
+
+	kv, err := js.KeyValue(ctx, InstanceRecordBucket)
+	if err != nil {
+		slog.Debug("reconcile/intent: instance record bucket not available, probing every guest port", "err", err)
+		return
+	}
+	for lsp := range candidates {
+		instanceID := instances[eniIDFromPort(lsp)]
+		if instanceID == "" {
+			continue
+		}
+		entry, err := kv.Get(ctx, InstanceRecordPrefix+instanceID)
+		if err != nil {
+			if !errors.Is(err, jetstream.ErrKeyNotFound) {
+				slog.Warn("reconcile/intent: instance record read failed", "instance_id", instanceID, "err", err)
+			}
+			continue
+		}
+		var rec vm.InstanceRecord
+		if err := json.Unmarshal(entry.Value(), &rec); err != nil {
+			slog.Warn("reconcile/intent: instance record unmarshal failed", "instance_id", instanceID, "err", err)
+			continue
+		}
+		if instanceHasNoTap(&rec) {
+			intent.IdlePorts[lsp] = instanceID
+		}
+	}
+}
+
+// instanceHasNoTap reports an instance nothing will relaunch. A DRAIN also leaves
+// StateStopped, but restore relaunches those, so desired state has to agree.
+func instanceHasNoTap(rec *vm.InstanceRecord) bool {
+	switch rec.Status.Status {
+	case vm.StateTerminated:
+		return true
+	case vm.StateStopped:
+		return rec.Spec.DesiredState == vm.DesiredStopped
+	}
+	return false
+}
+
 func loadIGWs(ctx context.Context, js jetstream.JetStream, localVPCs map[string]struct{}, out map[string]external.IGWSpec) error {
 	kv, err := js.KeyValue(ctx, handlers_ec2_igw.KVBucketIGW)
 	if err != nil {
@@ -15,6 +15,7 @@ import (
 	handlers_ec2_vpc "github.com/mulgadc/spinifex/spinifex/handlers/ec2/vpc"
 	"github.com/mulgadc/spinifex/spinifex/network/policy"
 	"github.com/mulgadc/spinifex/spinifex/testutil"
+	"github.com/mulgadc/spinifex/spinifex/vm"
 	"github.com/nats-io/nats.go/jetstream"
 )

@@ -556,6 +557,69 @@ func TestLoadIntentFromKV_SGRuleFieldMapping(t *testing.T) {
 	}
 }

+// Only an instance nothing will relaunch marks its public-IP port idle. A DRAIN
+// stop is relaunched by restore, and a missing record is unknown, so both stay
+// probed; a port with no public IP is never read at all.
+func TestLoadIntentFromKV_IdlePortsFromInstanceState(t *testing.T) {
+	js := startKV(t)
+
+	testutil.SeedKV(t, js, handlers_ec2_vpc.KVBucketVPCs, map[string][]byte{
+		"acct/vpc-a": mustJSON(t, handlers_ec2_vpc.VPCRecord{
+			VpcId: "vpc-a", CidrBlock: "10.0.0.0/16", AZ: "us-east-1a", CreatedAt: time.Now(),
+		}),
+	})
+	eni := func(id, ip, publicIP, instanceID string) []byte {
+		return mustJSON(t, handlers_ec2_vpc.ENIRecord{
+			NetworkInterfaceId: id, SubnetId: "subnet-a", VpcId: "vpc-a",
+			PrivateIpAddress: ip, MacAddress: "02:00:00:00:00:" + ip[len(ip)-2:],
+			PublicIpAddress: publicIP, InstanceId: instanceID, Status: "in-use", CreatedAt: time.Now(),
+		})
+	}
+	testutil.SeedKV(t, js, handlers_ec2_vpc.KVBucketENIs, map[string][]byte{
+		"acct/eni-stopped": eni("eni-stopped", "10.0.1.11", "192.0.2.11", "i-stopped"),
+		"acct/eni-term":    eni("eni-term", "10.0.1.12", "192.0.2.12", "i-term"),
+		"acct/eni-drain":   eni("eni-drain", "10.0.1.13", "192.0.2.13", "i-drain"),
+		"acct/eni-running": eni("eni-running", "10.0.1.14", "192.0.2.14", "i-running"),
+		"acct/eni-gone":    eni("eni-gone", "10.0.1.15", "192.0.2.15", "i-gone"),
+		"acct/eni-eip":     eni("eni-eip", "10.0.1.16", "", "i-eip"),
+		"acct/eni-private": eni("eni-private", "10.0.1.17", "", "i-private"),
+	})
+	testutil.SeedKV(t, js, handlers_ec2_eip.KVBucketEIPs, map[string][]byte{
+		"acct/eipassoc-a": mustJSON(t, handlers_ec2_eip.EIPRecord{
+			AllocationId: "eipalloc-a", PublicIp: "203.0.113.16", PrivateIp: "10.0.1.16",
+			VpcId: "vpc-a", ENIId: "eni-eip", State: "associated",
+		}),
+	})
+	record := func(status vm.InstanceState, desired vm.DesiredState) []byte {
+		return mustJSON(t, &vm.InstanceRecord{
+			Spec:   vm.InstanceSpec{DesiredState: desired},
+			Status: vm.InstanceStatus{Status: status},
+		})
+	}
+	testutil.SeedKV(t, js, InstanceRecordBucket, map[string][]byte{
+		InstanceRecordPrefix + "i-stopped": record(vm.StateStopped, vm.DesiredStopped),
+		InstanceRecordPrefix + "i-term":    record(vm.StateTerminated, vm.DesiredRunning),
+		InstanceRecordPrefix + "i-drain":   record(vm.StateStopped, vm.DesiredRunning),
+		InstanceRecordPrefix + "i-running": record(vm.StateRunning, vm.DesiredRunning),
+		InstanceRecordPrefix + "i-eip":     record(vm.StateStopped, vm.DesiredStopped),
+		InstanceRecordPrefix + "i-private": record(vm.StateStopped, vm.DesiredStopped),
+	})
+
+	intent, err := LoadIntentFromKV(context.Background(), js, "us-east-1a")
+	if err != nil {
+		t.Fatalf("LoadIntentFromKV: %v", err)
+	}
+
+	want := map[string]string{
+		"port-eni-stopped": "i-stopped",
+		"port-eni-term":    "i-term",
+		"port-eni-eip":     "i-eip",
+	}
+	if !reflect.DeepEqual(intent.IdlePorts, want) {
+		t.Errorf("IdlePorts = %v, want %v", intent.IdlePorts, want)
+	}
+}
+
 func mustJSON(t *testing.T, v any) []byte {
 	t.Helper()
 	b, err := json.Marshal(v)
@@ -154,6 +154,8 @@ type reconciler struct {
 	// is gone stops paying the full nudge sequence every cycle.
 	portBackoffMu sync.Mutex
 	portBackoff   map[string]portBackoffState
+	// Guest ports last seen idle, so the skip is logged once per state change.
+	idlePorts map[string]struct{}

 	// Serialises sb-cluster-state-reset. It is cluster-wide and re-syncs every
 	// chassis, so concurrent guest-port probes must not each fire one.
@@ -14,6 +14,7 @@ import (
 	"time"

 	"github.com/mulgadc/bluebottle/pkg/tlsconfig"
+	"github.com/mulgadc/spinifex/spinifex/utils"
 	"go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp"
 )

@@ -59,6 +60,11 @@ func newReverseProxy(backendHost, pathPrefix string, transport http.RoundTripper
 			pr.SetURL(target)
 			pr.Out.Host = target.Host

+			// The backend trusts X-Real-IP from loopback, which this hop always is.
+			// The UI also listens publicly, so re-derive it rather than forward a
+			// client's own claim. X-Request-ID passes through unchanged.
+			pr.Out.Header.Set("X-Real-IP", utils.RequestClientIP(pr.In))
+
 			// Strip the prefix from both path forms. RawPath must survive: a
 			// client that percent-encodes a segment (a ':' in a bedrock
 			// modelId) signs it, so dropping it re-encodes and breaks SigV4.
@@ -0,0 +1,59 @@
+//test:in-package: drives the unexported newReverseProxy, which the package exports only through
+// the full UI server; building that here would test far more than the header rule.
+
+package spinifexui
+
+import (
+	"net/http"
+	"net/http/httptest"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+// proxyHeaders sends one request through the awsgw proxy from remoteAddr and
+// returns the headers the backend received.
+func proxyHeaders(t *testing.T, remoteAddr string, set map[string]string) http.Header {
+	t.Helper()
+	var got http.Header
+	backend := httptest.NewTLSServer(http.HandlerFunc(func(_ http.ResponseWriter, r *http.Request) {
+		got = r.Header.Clone()
+	}))
+	t.Cleanup(backend.Close)
+
+	transport, ok := backend.Client().Transport.(*http.Transport)
+	require.True(t, ok, "expected *http.Transport")
+	proxy := newReverseProxy(backend.Listener.Addr().String(), "/proxy/awsgw", transport)
+
+	req := httptest.NewRequest(http.MethodPost, "/proxy/awsgw/", nil)
+	req.RemoteAddr = remoteAddr
+	for k, v := range set {
+		req.Header.Set(k, v)
+	}
+	proxy.ServeHTTP(httptest.NewRecorder(), req)
+	require.NotNil(t, got, "backend was not reached")
+	return got
+}
+
+// Behind nginx the UI's own connection is loopback, so the edge's X-Real-IP and
+// X-Request-ID reach awsgw unchanged.
+func TestNewReverseProxy_PassesEdgeHeadersFromLoopback(t *testing.T) {
+	got := proxyHeaders(t, "127.0.0.1:50000", map[string]string{
+		"X-Real-IP":    "198.51.100.7",
+		"X-Request-ID": "4f1c2b7e9a0d4e6f8b3c5a7d9e1f2a3b",
+	})
+	assert.Equal(t, "198.51.100.7", got.Get("X-Real-IP"))
+	assert.Equal(t, "4f1c2b7e9a0d4e6f8b3c5a7d9e1f2a3b", got.Get("X-Request-ID"))
+}
+
+// The UI listens publicly. A client talking to it directly must not be able to
+// hand awsgw a forged X-Real-IP over the loopback hop, where awsgw trusts it.
+func TestNewReverseProxy_OverwritesSpoofedRealIPFromPublicClient(t *testing.T) {
+	got := proxyHeaders(t, "203.0.113.9:50000", map[string]string{
+		"X-Real-IP":       "10.0.0.1",
+		"X-Forwarded-For": "127.0.0.1",
+	})
+	assert.Equal(t, "203.0.113.9", got.Get("X-Real-IP"))
+	assert.Empty(t, got.Get("X-Forwarded-For"), "Rewrite drops inbound X-Forwarded-For")
+}
@@ -14,9 +14,11 @@ import (
 	"log/slog"
 	"net"
 	"net/http"
+	"net/netip"
 	"os"
 	"os/signal"
 	"reflect"
+	"strings"
 	"time"

 	"github.com/aws/aws-sdk-go/aws"
@@ -393,6 +395,21 @@ func HashMAC(id string) string {
 	return net.HardwareAddr(b).String()
 }

+// RequestClientIP is the caller's address: X-Real-IP when the connection is from
+// loopback, where only a local proxy (the nginx edge, the console) can have set
+// it, and RemoteAddr otherwise, so a direct client cannot choose what is logged.
+func RequestClientIP(r *http.Request) string {
+	ip := ClientIP(r.RemoteAddr)
+	if addr, err := netip.ParseAddr(ip); err != nil || !addr.IsLoopback() {
+		return ip
+	}
+	realIP, err := netip.ParseAddr(strings.TrimSpace(r.Header.Get("X-Real-IP")))
+	if err != nil {
+		return ip
+	}
+	return realIP.String()
+}
+
 // ClientIP returns the IP from a RemoteAddr, stripping the port. Handles both
 // IPv4 and IPv6, and tolerates a RemoteAddr that is already a bare IP (no port).
 func ClientIP(remoteAddr string) string {
@@ -1807,3 +1807,39 @@ func TestWithRequestID_PointerToStruct(t *testing.T) {
 	require.Equal(t, "req-456", v.FieldByName("RequestId").String())
 	require.Equal(t, []string{"b"}, v.FieldByName("Names").Interface())
 }
+
+// X-Real-IP is honoured only on a loopback connection, where the local edge set
+// it. From anywhere else it, and X-Forwarded-For, are the client's own claim.
+func TestRequestClientIP(t *testing.T) {
+	tests := []struct {
+		name       string
+		remoteAddr string
+		realIP     string
+		xff        string
+		expected   string
+	}{
+		{"loopback trusts X-Real-IP", "127.0.0.1:40000", "198.51.100.7", "", "198.51.100.7"},
+		{"IPv6 loopback trusts X-Real-IP", "[::1]:40000", "2001:db8::7", "", "2001:db8::7"},
+		{"4in6 loopback trusts X-Real-IP", "[::ffff:127.0.0.1]:40000", "198.51.100.7", "", "198.51.100.7"},
+		{"loopback without header", "127.0.0.1:40000", "", "", "127.0.0.1"},
+		{"loopback with garbage header", "127.0.0.1:40000", "not-an-ip", "", "127.0.0.1"},
+		{"loopback ignores X-Forwarded-For", "127.0.0.1:40000", "", "203.0.113.1", "127.0.0.1"},
+		{"spoofed X-Real-IP from public client", "203.0.113.9:5555", "10.0.0.1", "", "203.0.113.9"},
+		{"spoofed X-Forwarded-For from public client", "203.0.113.9:5555", "", "127.0.0.1", "203.0.113.9"},
+		{"spoofed both from private client", "10.2.0.3:5555", "127.0.0.1", "127.0.0.1", "10.2.0.3"},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			r := httptest.NewRequest(http.MethodGet, "/", nil)
+			r.RemoteAddr = tt.remoteAddr
+			if tt.realIP != "" {
+				r.Header.Set("X-Real-IP", tt.realIP)
+			}
+			if tt.xff != "" {
+				r.Header.Set("X-Forwarded-For", tt.xff)
+			}
+			assert.Equal(t, tt.expected, RequestClientIP(r))
+		})
+	}
+}