feat: verify JetStream replicas agree after formation, discard stores by default (#1031)

spx admin init now discards the node's pre-formation JetStream store by default for --nodes >= 2; --discard-jetstream=false restores the refusal. spx admin kv digest gains --json with stream replica, retention and limit settings and per-sequence subject, time and TTL. The new spx admin kv compare reads every node's digest and fails only on real divergence: different content at one sequence, a missing message nothing explains, or wrong replica placement. install-node.sh gates formation on it and gains --ipsec on|off, verified on every node. The multi-node guide documents both.

cmd/spinifex/cmd/admin.go

@@ -358,7 +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().Bool("discard-jetstream", true, "With --nodes >= 2, remove this node's existing JetStream streams once every node has joined; =false refuses to form over a non-empty store instead. Ignored with --nodes 1")
 	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)")

cmd/spinifex/cmd/admin_kv.go

@@ -15,6 +15,7 @@ import (
 	"path/filepath"
 	"slices"
 	"sort"
+	"strconv"
 	"strings"
 	"time"
 
@@ -39,7 +40,9 @@ 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.
+--seqs prints a hash per sequence to find where. For a check that tells a
+diverged replica from one that is merely behind, run it with --json --seqs on
+every node and pass the results to "spx admin kv compare".
 
 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
@@ -59,25 +62,54 @@ func init() {
 	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")
+	kvDigestCmd.Flags().Bool("json", false, "Print one JSON document, the input to \"spx admin kv compare\"")
+}
+
+// nodeDigest is one node's digest report, the unit "kv compare" reads.
+type nodeDigest struct {
+	Host     string         `json:"host"`
+	Store    string         `json:"store"`
+	CopiedAt time.Time      `json:"copied_at"`
+	Skipped  []string       `json:"skipped,omitempty"`
+	Streams  []streamDigest `json:"streams"`
 }
 
 // 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
+	streamMeta
+
+	Name     string      `json:"name"`
+	FirstSeq uint64      `json:"first_seq"`
+	LastSeq  uint64      `json:"last_seq"`
+	Msgs     uint64      `json:"msgs"`
+	Digest   string      `json:"sha256"`
+	Seqs     []seqDigest `json:"seqs,omitempty"`
+}
+
+// streamMeta is the part of the live stream config that decides which replica
+// differences are legitimate. It is read before the copy is rewritten.
+type streamMeta struct {
+	Replicas    int           `json:"replicas,omitempty"`
+	Retention   string        `json:"retention,omitempty"`
+	MaxAge      time.Duration `json:"max_age_ns,omitempty"`
+	AllowMsgTTL bool          `json:"allow_msg_ttl,omitempty"`
+	MaxMsgs     int64         `json:"max_msgs,omitempty"`
+	MaxBytes    int64         `json:"max_bytes,omitempty"`
 }
 
 // seqDigest is the hash of one stored sequence, or "deleted" when the replica
-// has no message at that sequence.
+// has no message at that sequence. Subject, time and TTL are empty then.
 type seqDigest struct {
-	Seq  uint64
-	Hash string
+	Seq     uint64        `json:"seq"`
+	Hash    string        `json:"hash"`
+	Subject string        `json:"subject,omitempty"`
+	Time    time.Time     `json:"time,omitzero"`
+	TTL     time.Duration `json:"ttl_ns,omitempty"`
 }
 
+// deletedHash marks a sequence the replica holds no message for.
+const deletedHash = "deleted"
+
 func runKVDigest(cmd *cobra.Command, _ []string) {
 	storeDir, _ := cmd.Flags().GetString("store-dir")
 	if storeDir == "" {
@@ -90,6 +122,7 @@ func runKVDigest(cmd *cobra.Command, _ []string) {
 	workParent, _ := cmd.Flags().GetString("work-dir")
 	only, _ := cmd.Flags().GetStringSlice("stream")
 	withSeqs, _ := cmd.Flags().GetBool("seqs")
+	asJSON, _ := cmd.Flags().GetBool("json")
 
 	work, err := os.MkdirTemp(workParent, "spx-kv-digest-")
 	if err != nil {
@@ -99,7 +132,7 @@ func runKVDigest(cmd *cobra.Command, _ []string) {
 	defer os.RemoveAll(work)
 
 	copiedAt := time.Now().UTC()
-	skipped, err := copyStreamTree(storeDir, filepath.Join(work, "jetstream"))
+	skipped, metas, err := copyStreamTree(storeDir, filepath.Join(work, "jetstream"))
 	if err != nil {
 		fmt.Fprintf(os.Stderr, "kv digest: %v\n", err)
 		os.RemoveAll(work)
@@ -108,7 +141,7 @@ func runKVDigest(cmd *cobra.Command, _ []string) {
 
 	ctx, cancel := context.WithTimeout(cmd.Context(), 10*time.Minute)
 	defer cancel()
-	digests, err := digestStore(ctx, work, only, withSeqs)
+	digests, err := digestStore(ctx, work, only, withSeqs, metas)
 	if err != nil {
 		fmt.Fprintf(os.Stderr, "kv digest: %v\n", err)
 		cancel()
@@ -117,6 +150,14 @@ func runKVDigest(cmd *cobra.Command, _ []string) {
 	}
 
 	host, _ := os.Hostname()
+	if asJSON {
+		enc := json.NewEncoder(cmd.OutOrStdout())
+		if err := enc.Encode(nodeDigest{Host: host, Store: storeDir, CopiedAt: copiedAt, Skipped: skipped, Streams: digests}); err != nil {
+			fmt.Fprintf(os.Stderr, "kv digest: %v\n", err)
+			os.Exit(1)
+		}
+		return
+	}
 	writeDigests(cmd.OutOrStdout(), host, storeDir, copiedAt, skipped, digests)
 }
 
@@ -140,14 +181,15 @@ func writeDigests(w io.Writer, host, storeDir string, copiedAt time.Time, skippe
 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) {
+// consumer state (obs). It returns streams found under other accounts and each
+// copied stream's config as the live server held it, keyed by stream name.
+func copyStreamTree(storeDir, dst string) (skipped []string, metas map[string]streamMeta, err error) {
 	streams, err := localStreams(storeDir)
 	if err != nil {
-		return nil, err
+		return nil, nil, err
 	}
 	if len(streams) == 0 {
-		return nil, fmt.Errorf("no streams found under %s", storeDir)
+		return nil, nil, fmt.Errorf("no streams found under %s", storeDir)
 	}
 	for _, s := range streams {
 		if !strings.HasPrefix(s, globalAccountDir+"/") {
@@ -179,39 +221,58 @@ func copyStreamTree(storeDir, dst string) (skipped []string, err error) {
 		return copyFile(path, target)
 	})
 	if err != nil {
-		return nil, fmt.Errorf("copy streams from %s: %w", src, err)
+		return nil, nil, fmt.Errorf("copy streams from %s: %w", src, err)
 	}
 	entries, err := os.ReadDir(out)
 	if err != nil {
-		return nil, err
+		return nil, nil, err
 	}
+	metas = make(map[string]streamMeta, len(entries))
 	for _, e := range entries {
 		if e.IsDir() {
-			if err := rewriteStreamMetaStandalone(filepath.Join(out, e.Name())); err != nil {
-				return nil, err
+			meta, err := rewriteStreamMetaStandalone(filepath.Join(out, e.Name()))
+			if err != nil {
+				return nil, nil, err
 			}
+			metas[e.Name()] = meta
 		}
 	}
-	return skipped, nil
+	return skipped, metas, 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 {
+// It returns the settings as they were before the rewrite.
+func rewriteStreamMetaStandalone(streamDir string) (streamMeta, error) {
 	metaPath := filepath.Join(streamDir, "meta.inf")
 	raw, err := os.ReadFile(metaPath)
 	if errors.Is(err, os.ErrNotExist) {
-		return nil
+		return streamMeta{}, nil
 	}
 	if err != nil {
-		return err
+		return streamMeta{}, err
+	}
+	var orig struct {
+		Replicas    int           `json:"num_replicas"`
+		Retention   string        `json:"retention"`
+		MaxAge      time.Duration `json:"max_age"`
+		AllowMsgTTL bool          `json:"allow_msg_ttl"`
+		MaxMsgs     int64         `json:"max_msgs"`
+		MaxBytes    int64         `json:"max_bytes"`
+	}
+	if err := json.Unmarshal(raw, &orig); err != nil {
+		return streamMeta{}, fmt.Errorf("parse %s: %w", metaPath, err)
+	}
+	kept := streamMeta{
+		Replicas: orig.Replicas, Retention: orig.Retention, MaxAge: orig.MaxAge,
+		AllowMsgTTL: orig.AllowMsgTTL, MaxMsgs: orig.MaxMsgs, MaxBytes: orig.MaxBytes,
 	}
 	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)
+		return streamMeta{}, fmt.Errorf("parse %s: %w", metaPath, err)
 	}
 	meta["num_replicas"] = 1
 	meta["max_age"] = 0
@@ -220,18 +281,18 @@ func rewriteStreamMetaStandalone(streamDir string) error {
 	delete(meta, "placement")
 	out, err := json.Marshal(meta)
 	if err != nil {
-		return err
+		return streamMeta{}, err
 	}
 	key := sha256.Sum256([]byte(filepath.Base(streamDir)))
 	hh, err := highwayhash.New64(key[:])
 	if err != nil {
-		return err
+		return streamMeta{}, err
 	}
 	hh.Write(out)
 	if err := os.WriteFile(metaPath, out, 0o600); err != nil {
-		return err
+		return streamMeta{}, err
 	}
-	return os.WriteFile(filepath.Join(streamDir, "meta.sum"), []byte(hex.EncodeToString(hh.Sum(nil))), 0o600)
+	return kept, os.WriteFile(filepath.Join(streamDir, "meta.sum"), []byte(hex.EncodeToString(hh.Sum(nil))), 0o600)
 }
 
 func copyFile(src, dst string) error {
@@ -253,7 +314,7 @@ func copyFile(src, dst string) error {
 
 // 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) {
+func digestStore(ctx context.Context, storeParent string, only []string, withSeqs bool, metas map[string]streamMeta) ([]streamDigest, error) {
 	ns, err := server.NewServer(&server.Options{
 		ServerName: "spx-kv-digest",
 		JetStream:  true,
@@ -302,6 +363,7 @@ func digestStore(ctx context.Context, storeParent string, only []string, withSeq
 		if err != nil {
 			return nil, err
 		}
+		d.streamMeta = metas[name]
 		digests = append(digests, d)
 	}
 	return digests, nil
@@ -322,18 +384,21 @@ func digestStream(ctx context.Context, js jetstream.JetStream, name string, with
 	whole := sha256.New()
 	for seq := st.FirstSeq; st.Msgs > 0 && seq <= st.LastSeq; seq++ {
 		msg, err := s.GetMsg(ctx, seq)
-		var h string
+		sd := seqDigest{Seq: seq}
 		switch {
 		case errors.Is(err, jetstream.ErrMsgNotFound):
-			h = "deleted"
+			sd.Hash = deletedHash
 		case err != nil:
 			return streamDigest{}, fmt.Errorf("read %s seq %d: %w", name, seq, err)
 		default:
-			h = hashStoredMsg(msg)
+			sd.Hash = hashStoredMsg(msg)
+			sd.Subject = msg.Subject
+			sd.Time = msg.Time.UTC()
+			sd.TTL = msgTTL(msg.Header.Get(jetstream.MsgTTLHeader))
 		}
-		fmt.Fprintf(whole, "%d:%s\n", seq, h)
+		fmt.Fprintf(whole, "%d:%s\n", seq, sd.Hash)
 		if withSeqs {
-			d.Seqs = append(d.Seqs, seqDigest{Seq: seq, Hash: h})
+			d.Seqs = append(d.Seqs, sd)
 		}
 	}
 	d.Digest = hex.EncodeToString(whole.Sum(nil))
@@ -369,3 +434,18 @@ func hashStoredMsg(m *jetstream.RawStreamMsg) string {
 	field(m.Data)
 	return hex.EncodeToString(h.Sum(nil))[:16]
 }
+
+// msgTTL parses a Nats-TTL header: a Go duration or whole seconds. "never", an
+// absent header or an unparseable one yield zero, meaning no expiry.
+func msgTTL(v string) time.Duration {
+	if v == "" {
+		return 0
+	}
+	if d, err := time.ParseDuration(v); err == nil && d > 0 {
+		return d
+	}
+	if n, err := strconv.ParseInt(v, 10, 64); err == nil && n > 0 {
+		return time.Duration(n) * time.Second
+	}
+	return 0
+}

cmd/spinifex/cmd/admin_kv_compare.go

@@ -0,0 +1,360 @@
+package cmd
+
+import (
+	"encoding/json"
+	"errors"
+	"fmt"
+	"io"
+	"os"
+	"slices"
+	"sort"
+	"strings"
+	"time"
+
+	"github.com/spf13/cobra"
+)
+
+var kvCompareCmd = &cobra.Command{
+	Use:   "compare <digest.json>...",
+	Short: "Compare the kv digests of every node and report replicas that have diverged",
+	Long: `Read the "spx admin kv digest --json --seqs" output of every node in the
+cluster, one file per node ("-" reads any number of them from stdin), and
+check that every stream's replicas hold the same content.
+
+A replica that is merely behind is not divergent, so only these fail:
+
+  - the same sequence holds different content on two replicas;
+  - a message held by one replica is missing from another, and nothing on
+    that replica explains it: no later write to the same subject, no expiry
+    under max_age or a message TTL, and no stream limit dropping its head;
+  - a stream is held by fewer nodes than its replica count, or by more;
+  - replicas disagree on the replica count;
+  - no sequence range is common to every replica, so nothing can be checked.
+
+Sequences past the slowest replica's last one are not compared. Digests are
+taken from live stores a few seconds apart, so re-run before trusting a
+failure that could be a write landing between two copies.
+
+Exits 0 when every stream is consistent, 1 on divergence, 2 on bad input.`,
+	Args: cobra.MinimumNArgs(1),
+	Run:  runKVCompare,
+}
+
+func init() {
+	kvCmd.AddCommand(kvCompareCmd)
+}
+
+func runKVCompare(cmd *cobra.Command, args []string) {
+	var nodes []nodeDigest
+	for _, arg := range args {
+		var r io.Reader
+		if arg == "-" {
+			r = cmd.InOrStdin()
+		} else {
+			f, err := os.Open(arg)
+			if err != nil {
+				fmt.Fprintf(os.Stderr, "kv compare: %v\n", err)
+				os.Exit(2)
+			}
+			defer f.Close()
+			r = f
+		}
+		got, err := loadNodeDigests(r, arg)
+		if err != nil {
+			fmt.Fprintf(os.Stderr, "kv compare: %v\n", err)
+			os.Exit(2)
+		}
+		nodes = append(nodes, got...)
+	}
+	res, err := compareNodeDigests(nodes)
+	if err != nil {
+		fmt.Fprintf(os.Stderr, "kv compare: %v\n", err)
+		os.Exit(2)
+	}
+	writeCompareResult(cmd.OutOrStdout(), nodes, res)
+	if res.divergent() {
+		os.Exit(1)
+	}
+}
+
+// loadNodeDigests reads every JSON digest document in r.
+func loadNodeDigests(r io.Reader, source string) ([]nodeDigest, error) {
+	dec := json.NewDecoder(r)
+	var out []nodeDigest
+	for {
+		var d nodeDigest
+		err := dec.Decode(&d)
+		if errors.Is(err, io.EOF) {
+			break
+		}
+		if err != nil {
+			return nil, fmt.Errorf("%s: document %d: %w", source, len(out)+1, err)
+		}
+		if d.Host == "" {
+			return nil, fmt.Errorf("%s: document %d has no host; is it \"kv digest --json\" output?", source, len(out)+1)
+		}
+		out = append(out, d)
+	}
+	if len(out) == 0 {
+		return nil, fmt.Errorf("%s: no digest documents", source)
+	}
+	return out, nil
+}
+
+// maxFindingsPerStream caps what one stream prints; the rest are counted.
+const maxFindingsPerStream = 5
+
+type streamVerdict struct {
+	Name     string
+	Holders  []string
+	Problems []string
+	Notes    []string
+}
+
+type kvCompareResult struct {
+	Streams []streamVerdict
+}
+
+func (r kvCompareResult) divergent() bool {
+	return slices.ContainsFunc(r.Streams, func(v streamVerdict) bool { return len(v.Problems) > 0 })
+}
+
+// replicaView is one node's copy of one stream, indexed for the per-sequence rules.
+type replicaView struct {
+	host          string
+	copiedAt      time.Time
+	d             *streamDigest
+	bySeq         map[uint64]seqDigest
+	lastBySubject map[string]uint64
+}
+
+func compareNodeDigests(nodes []nodeDigest) (kvCompareResult, error) {
+	if len(nodes) < 2 {
+		return kvCompareResult{}, fmt.Errorf("need the digests of at least two nodes, got %d", len(nodes))
+	}
+	hosts := make([]string, 0, len(nodes))
+	byName := map[string][]*replicaView{}
+	for i := range nodes {
+		n := &nodes[i]
+		if slices.Contains(hosts, n.Host) {
+			return kvCompareResult{}, fmt.Errorf("two digests are from host %s", n.Host)
+		}
+		hosts = append(hosts, n.Host)
+		for j := range n.Streams {
+			d := &n.Streams[j]
+			v := &replicaView{host: n.Host, copiedAt: n.CopiedAt, d: d,
+				bySeq: make(map[uint64]seqDigest, len(d.Seqs)), lastBySubject: map[string]uint64{}}
+			for _, s := range d.Seqs {
+				v.bySeq[s.Seq] = s
+				if s.Hash != deletedHash && s.Seq > v.lastBySubject[s.Subject] {
+					v.lastBySubject[s.Subject] = s.Seq
+				}
+			}
+			byName[d.Name] = append(byName[d.Name], v)
+		}
+	}
+	names := make([]string, 0, len(byName))
+	for name := range byName {
+		names = append(names, name)
+	}
+	sort.Strings(names)
+
+	var res kvCompareResult
+	for _, name := range names {
+		res.Streams = append(res.Streams, compareStream(name, hosts, byName[name]))
+	}
+	return res, nil
+}
+
+func compareStream(name string, hosts []string, views []*replicaView) streamVerdict {
+	v := streamVerdict{Name: name}
+	for _, r := range views {
+		v.Holders = append(v.Holders, r.host)
+	}
+
+	counts := map[int][]string{}
+	replicas := 0
+	for _, r := range views {
+		if n := r.d.Replicas; n > 0 {
+			counts[n] = append(counts[n], r.host)
+			replicas = max(replicas, n)
+		}
+	}
+	if len(counts) > 1 {
+		var parts []string
+		for n, hs := range counts {
+			parts = append(parts, fmt.Sprintf("%d on %s", n, strings.Join(hs, ",")))
+		}
+		sort.Strings(parts)
+		v.Problems = append(v.Problems, "replicas disagree on the replica count: "+strings.Join(parts, "; "))
+	}
+	switch {
+	case replicas > 0 && len(views) < replicas:
+		missing := slices.DeleteFunc(slices.Clone(hosts), func(h string) bool { return slices.Contains(v.Holders, h) })
+		msg := fmt.Sprintf("held by %d of %d replicas; missing from %s", len(views), replicas, strings.Join(missing, ","))
+		if len(hosts) < replicas {
+			msg += fmt.Sprintf(" (only %d digests given)", len(hosts))
+		}
+		v.Problems = append(v.Problems, msg)
+	case replicas > 0 && len(views) > replicas:
+		v.Problems = append(v.Problems, fmt.Sprintf("held by %d nodes but configured for %d replicas; a node outside the stream's group still holds a copy", len(views), replicas))
+	}
+	if len(views) < 2 {
+		v.Notes = append(v.Notes, "single copy, nothing to compare")
+		return v
+	}
+
+	identical := true
+	for _, r := range views[1:] {
+		if r.d.Digest != views[0].d.Digest || r.d.FirstSeq != views[0].d.FirstSeq || r.d.LastSeq != views[0].d.LastSeq {
+			identical = false
+			break
+		}
+	}
+	if identical {
+		v.Notes = append(v.Notes, fmt.Sprintf("identical, seqs %d-%d", views[0].d.FirstSeq, views[0].d.LastSeq))
+		return v
+	}
+	for _, r := range views {
+		if r.d.Msgs > 0 && len(r.d.Seqs) == 0 {
+			v.Problems = append(v.Problems, "contents differ and "+r.host+"'s digest has no per-sequence detail; re-run digest with --seqs")
+			return v
+		}
+	}
+
+	lo, hi := views[0].d.FirstSeq, views[0].d.LastSeq
+	seqSet := map[uint64]struct{}{}
+	for _, r := range views {
+		lo, hi = max(lo, r.d.FirstSeq), min(hi, r.d.LastSeq)
+		for s := range r.bySeq {
+			seqSet[s] = struct{}{}
+		}
+	}
+	if lo > hi {
+		v.Problems = append(v.Problems, "no sequence range is common to every replica, so nothing could be compared")
+		return v
+	}
+	seqs := make([]uint64, 0, len(seqSet))
+	for s := range seqSet {
+		if s <= hi {
+			seqs = append(seqs, s)
+		}
+	}
+	slices.Sort(seqs)
+
+	var problems []string
+	explained := 0
+	for _, seq := range seqs {
+		var present, absent []*replicaView
+		for _, r := range views {
+			if e, ok := r.bySeq[seq]; ok && e.Hash != deletedHash {
+				present = append(present, r)
+			} else {
+				absent = append(absent, r)
+			}
+		}
+		if len(present) == 0 {
+			continue
+		}
+		byHash := map[string][]string{}
+		for _, r := range present {
+			h := r.bySeq[seq].Hash
+			byHash[h] = append(byHash[h], r.host)
+		}
+		if len(byHash) > 1 {
+			var parts []string
+			for h, hs := range byHash {
+				parts = append(parts, strings.Join(hs, ",")+"="+h)
+			}
+			sort.Strings(parts)
+			problems = append(problems, fmt.Sprintf("seq %d %s holds different content: %s", seq, present[0].bySeq[seq].Subject, strings.Join(parts, " ")))
+			continue
+		}
+		msg := present[0].bySeq[seq]
+		for _, r := range absent {
+			if absenceExplained(r, msg) {
+				explained++
+				continue
+			}
+			problems = append(problems, fmt.Sprintf("seq %d %s is on %s but missing from %s, with no later write to that subject, expiry or limit to explain it",
+				seq, msg.Subject, hostsOf(present), r.host))
+		}
+	}
+	if len(problems) > maxFindingsPerStream {
+		problems = append(problems[:maxFindingsPerStream], fmt.Sprintf("... and %d more", len(problems)-maxFindingsPerStream))
+	}
+	v.Problems = append(v.Problems, problems...)
+
+	note := fmt.Sprintf("seqs %d-%d compared", lo, hi)
+	if explained > 0 {
+		note += fmt.Sprintf(", %d superseded or expired messages missing from some replicas", explained)
+	}
+	var ahead []string
+	for _, r := range views {
+		if r.d.LastSeq > hi {
+			ahead = append(ahead, fmt.Sprintf("%s +%d", r.host, r.d.LastSeq-hi))
+		}
+	}
+	if len(ahead) > 0 {
+		note += ", tail not compared: " + strings.Join(ahead, " ")
+	}
+	v.Notes = append(v.Notes, note)
+	return v
+}
+
+// absenceExplained reports whether a replica lacking msg removed it legitimately:
+// the subject was written again later, the message aged out before that replica
+// was copied, or a stream limit dropped it from a head the replica has moved past.
+func absenceExplained(r *replicaView, msg seqDigest) bool {
+	if msg.Subject == "" {
+		return false
+	}
+	if r.lastBySubject[msg.Subject] > msg.Seq {
+		return true
+	}
+	meta := r.d.streamMeta
+	if meta.Retention != "" && meta.Retention != "limits" {
+		return true
+	}
+	if !msg.Time.IsZero() {
+		if meta.MaxAge > 0 && msg.Time.Add(meta.MaxAge).Before(r.copiedAt) {
+			return true
+		}
+		if msg.TTL > 0 && msg.Time.Add(msg.TTL).Before(r.copiedAt) {
+			return true
+		}
+	}
+	return msg.Seq < r.d.FirstSeq && (meta.MaxMsgs > 0 || meta.MaxBytes > 0)
+}
+
+func hostsOf(views []*replicaView) string {
+	hs := make([]string, len(views))
+	for i, r := range views {
+		hs[i] = r.host
+	}
+	return strings.Join(hs, ",")
+}
+
+func writeCompareResult(w io.Writer, nodes []nodeDigest, res kvCompareResult) {
+	fmt.Fprintf(w, "compared %d digests:", len(nodes))
+	for _, n := range nodes {
+		fmt.Fprintf(w, " %s@%s", n.Host, n.CopiedAt.Format(time.TimeOnly))
+	}
+	fmt.Fprintln(w)
+	bad := 0
+	for _, v := range res.Streams {
+		status := "ok"
+		if len(v.Problems) > 0 {
+			status = "DIVERGED"
+			bad++
+		}
+		fmt.Fprintf(w, "%-9s %s  [%s]\n", status, v.Name, strings.Join(v.Holders, ","))
+		for _, p := range v.Problems {
+			fmt.Fprintf(w, "    ! %s\n", p)
+		}
+		for _, n := range v.Notes {
+			fmt.Fprintf(w, "      %s\n", n)
+		}
+	}
+	fmt.Fprintf(w, "%d streams: %d consistent, %d diverged\n", len(res.Streams), len(res.Streams)-bad, bad)
+}

cmd/spinifex/cmd/admin_kv_compare_test.go

@@ -0,0 +1,233 @@
+package cmd_test
+
+import (
+	"encoding/json"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/mulgadc/spinifex/cmd/spinifex/cmd"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+var t0 = time.Date(2026, 9, 15, 10, 0, 0, 0, time.UTC)
+
+type msg struct {
+	seq     uint64
+	subject string
+	hash    string // "deleted" for a gap
+	at      time.Time
+}
+
+type stream struct {
+	name     string
+	replicas int
+	maxAge   time.Duration
+	maxMsgs  int64
+	msgs     []msg
+	noSeqs   bool
+}
+
+// doc renders one node's digest the way "kv digest --json --seqs" prints it.
+func doc(t *testing.T, host string, copiedAt time.Time, streams ...stream) string {
+	t.Helper()
+	var out []map[string]any
+	for _, s := range streams {
+		var seqs []map[string]any
+		var first, last, count uint64
+		digest := host + ":" // only equality matters to compare
+		for _, m := range s.msgs {
+			if first == 0 {
+				first = m.seq
+			}
+			last = m.seq
+			e := map[string]any{"seq": m.seq, "hash": m.hash}
+			if m.hash != "deleted" {
+				count++
+				e["subject"] = m.subject
+				at := m.at
+				if at.IsZero() {
+					at = t0
+				}
+				e["time"] = at
+			}
+			digest += m.hash + ","
+			seqs = append(seqs, e)
+		}
+		sd := map[string]any{"name": s.name, "first_seq": first, "last_seq": last, "msgs": count, "sha256": digest}
+		if s.replicas > 0 {
+			sd["replicas"] = s.replicas
+		}
+		if s.maxAge > 0 {
+			sd["max_age_ns"] = s.maxAge
+		}
+		if s.maxMsgs > 0 {
+			sd["max_msgs"] = s.maxMsgs
+		}
+		if !s.noSeqs {
+			sd["seqs"] = seqs
+		}
+		out = append(out, sd)
+	}
+	raw, err := json.Marshal(map[string]any{"host": host, "store": "/s", "copied_at": copiedAt, "streams": out})
+	require.NoError(t, err)
+	return string(raw)
+}
+
+func kvs(msgs ...msg) []msg { return msgs }
+
+func TestKVCompareIdentical(t *testing.T) {
+	s := stream{name: "KV_a", replicas: 3, msgs: kvs(msg{1, "k", "h1", t0}, msg{2, "k", "h2", t0})}
+	// The digest field embeds the host, so this exercises the per-seq path too.
+	divergent, report, err := cmd.CompareKVDigests(doc(t, "n1", t0, s), doc(t, "n2", t0, s), doc(t, "n3", t0, s))
+	require.NoError(t, err)
+	assert.False(t, divergent, report)
+	assert.Contains(t, report, "1 streams: 1 consistent, 0 diverged")
+}
+
+// The prod failure: each node wrote its own copy before formation, so the same
+// sequence holds different content on different replicas.
+func TestKVCompareSameSeqDifferentContent(t *testing.T) {
+	a := stream{name: "KV_acct", replicas: 2, msgs: kvs(msg{1, "$KV.acct.000000000002", "aaaa", t0})}
+	b := stream{name: "KV_acct", replicas: 2, msgs: kvs(msg{1, "$KV.acct.000000000002", "bbbb", t0})}
+	divergent, report, err := cmd.CompareKVDigests(doc(t, "node1", t0, a), doc(t, "node2", t0, b))
+	require.NoError(t, err)
+	assert.True(t, divergent)
+	assert.Contains(t, report, "DIVERGED  KV_acct")
+	assert.Contains(t, report, "seq 1 $KV.acct.000000000002 holds different content: node1=aaaa node2=bbbb")
+}
+
+func TestKVCompareMissingMessage(t *testing.T) {
+	full := kvs(msg{1, "k", "h1", t0}, msg{2, "j", "h2", t0}, msg{3, "k", "h3", t0})
+
+	t.Run("superseded by a later write to the subject", func(t *testing.T) {
+		lagging := stream{name: "KV_a", replicas: 2, msgs: full}
+		purged := stream{name: "KV_a", replicas: 2, msgs: kvs(msg{1, "", "deleted", time.Time{}}, full[1], full[2])}
+		divergent, report, err := cmd.CompareKVDigests(doc(t, "n1", t0, lagging), doc(t, "n2", t0, purged))
+		require.NoError(t, err)
+		assert.False(t, divergent, report)
+		assert.Contains(t, report, "1 superseded or expired")
+	})
+
+	t.Run("nothing explains it", func(t *testing.T) {
+		have := stream{name: "KV_a", replicas: 2, msgs: full}
+		lost := stream{name: "KV_a", replicas: 2, msgs: kvs(full[0], msg{2, "", "deleted", time.Time{}}, full[2])}
+		divergent, report, err := cmd.CompareKVDigests(doc(t, "n1", t0, have), doc(t, "n2", t0, lost))
+		require.NoError(t, err)
+		assert.True(t, divergent)
+		assert.Contains(t, report, "seq 2 j is on n1 but missing from n2")
+	})
+
+	t.Run("aged out under max_age before the replica was copied", func(t *testing.T) {
+		old := t0.Add(-time.Hour)
+		have := stream{name: "KV_lease", replicas: 2, maxAge: time.Minute, msgs: kvs(msg{1, "l", "h1", old}, msg{2, "m", "h2", t0})}
+		gone := stream{name: "KV_lease", replicas: 2, maxAge: time.Minute, msgs: kvs(msg{2, "m", "h2", t0})}
+		divergent, report, err := cmd.CompareKVDigests(doc(t, "n1", t0, have), doc(t, "n2", t0, gone))
+		require.NoError(t, err)
+		assert.False(t, divergent, report)
+
+		have.maxAge, gone.maxAge = 0, 0
+		divergent, _, err = cmd.CompareKVDigests(doc(t, "n1", t0, have), doc(t, "n2", t0, gone))
+		require.NoError(t, err)
+		assert.True(t, divergent, "without max_age the same gap is unexplained")
+	})
+
+	t.Run("head dropped by a stream limit", func(t *testing.T) {
+		behind := stream{name: "S", replicas: 2, maxMsgs: 2, msgs: kvs(msg{1, "a", "h1", t0}, msg{2, "b", "h2", t0})}
+		ahead := stream{name: "S", replicas: 2, maxMsgs: 2, msgs: kvs(msg{2, "b", "h2", t0}, msg{3, "c", "h3", t0})}
+		divergent, report, err := cmd.CompareKVDigests(doc(t, "n1", t0, behind), doc(t, "n2", t0, ahead))
+		require.NoError(t, err)
+		assert.False(t, divergent, report)
+	})
+}
+
+func TestKVCompareTailLag(t *testing.T) {
+	behind := stream{name: "KV_a", replicas: 2, msgs: kvs(msg{1, "k", "h1", t0}, msg{2, "k", "h2", t0})}
+	ahead := stream{name: "KV_a", replicas: 2, msgs: kvs(msg{1, "k", "h1", t0}, msg{2, "k", "h2", t0}, msg{3, "k", "h3", t0}, msg{4, "k", "h4", t0})}
+	divergent, report, err := cmd.CompareKVDigests(doc(t, "n1", t0, behind), doc(t, "n2", t0, ahead))
+	require.NoError(t, err)
+	assert.False(t, divergent, report)
+	assert.Contains(t, report, "tail not compared: n2 +2")
+}
+
+func TestKVCompareNoCommonRange(t *testing.T) {
+	a := stream{name: "KV_a", replicas: 2, msgs: kvs(msg{1, "k", "h1", t0})}
+	b := stream{name: "KV_a", replicas: 2, msgs: kvs(msg{5, "k", "h5", t0})}
+	divergent, report, err := cmd.CompareKVDigests(doc(t, "n1", t0, a), doc(t, "n2", t0, b))
+	require.NoError(t, err)
+	assert.True(t, divergent)
+	assert.Contains(t, report, "no sequence range is common")
+}
+
+func TestKVCompareReplicaPlacement(t *testing.T) {
+	s := stream{name: "KV_a", replicas: 3, msgs: kvs(msg{1, "k", "h1", t0})}
+	other := stream{name: "KV_b", replicas: 1, msgs: kvs(msg{1, "k", "h1", t0})}
+
+	t.Run("fewer holders than replicas", func(t *testing.T) {
+		divergent, report, err := cmd.CompareKVDigests(doc(t, "n1", t0, s), doc(t, "n2", t0, s), doc(t, "n3", t0, other))
+		require.NoError(t, err)
+		assert.True(t, divergent)
+		assert.Contains(t, report, "held by 2 of 3 replicas; missing from n3")
+	})
+
+	t.Run("more holders than replicas", func(t *testing.T) {
+		divergent, report, err := cmd.CompareKVDigests(doc(t, "n1", t0, other), doc(t, "n2", t0, other))
+		require.NoError(t, err)
+		assert.True(t, divergent)
+		assert.Contains(t, report, "held by 2 nodes but configured for 1 replicas")
+	})
+
+	t.Run("replica counts disagree", func(t *testing.T) {
+		r1 := s
+		r1.replicas = 1
+		divergent, report, err := cmd.CompareKVDigests(doc(t, "n1", t0, s), doc(t, "n2", t0, r1))
+		require.NoError(t, err)
+		assert.True(t, divergent)
+		assert.Contains(t, report, "replicas disagree on the replica count")
+	})
+
+	t.Run("single replica stream on one node", func(t *testing.T) {
+		divergent, report, err := cmd.CompareKVDigests(doc(t, "n1", t0, other), doc(t, "n2", t0))
+		require.NoError(t, err)
+		assert.False(t, divergent, report)
+	})
+}
+
+func TestKVCompareNeedsSeqs(t *testing.T) {
+	a := stream{name: "KV_a", replicas: 2, noSeqs: true, msgs: kvs(msg{1, "k", "h1", t0})}
+	b := stream{name: "KV_a", replicas: 2, noSeqs: true, msgs: kvs(msg{1, "k", "h2", t0})}
+	divergent, report, err := cmd.CompareKVDigests(doc(t, "n1", t0, a), doc(t, "n2", t0, b))
+	require.NoError(t, err)
+	assert.True(t, divergent)
+	assert.Contains(t, report, "re-run digest with --seqs")
+}
+
+func TestKVCompareInput(t *testing.T) {
+	s := stream{name: "KV_a", replicas: 2, msgs: kvs(msg{1, "k", "h1", t0})}
+
+	// Several documents in one stream, as "-" reads them from stdin.
+	divergent, _, err := cmd.CompareKVDigests(doc(t, "n1", t0, s) + "\n" + doc(t, "n2", t0, s))
+	require.NoError(t, err)
+	assert.False(t, divergent)
+
+	_, _, err = cmd.CompareKVDigests(doc(t, "n1", t0, s))
+	assert.ErrorContains(t, err, "at least two nodes")
+
+	_, _, err = cmd.CompareKVDigests(doc(t, "n1", t0, s), doc(t, "n1", t0, s))
+	assert.ErrorContains(t, err, "two digests are from host n1")
+
+	_, _, err = cmd.CompareKVDigests(`KV_a	first=1	last=1`)
+	assert.Error(t, err, "text digest output is rejected")
+
+	_, _, err = cmd.CompareKVDigests(`{"streams":[]}`)
+	assert.ErrorContains(t, err, "no host")
+
+	_, _, err = cmd.CompareKVDigests("  ")
+	assert.ErrorContains(t, err, "no digest documents")
+}
+
+func TestAdminInitDiscardsJetStreamByDefault(t *testing.T) {
+	assert.Equal(t, "true", cmd.AdminInitFlagDefault("discard-jetstream"))
+	assert.True(t, strings.HasPrefix(cmd.AdminInitFlagDefault("ipsec"), "true"))
+}

cmd/spinifex/cmd/admin_kv_test.go

@@ -48,9 +48,9 @@ func kvServer(t *testing.T, storeParent, bucket string) (jetstream.KeyValue, fun
 func digestOf(t *testing.T, storeDir string, withSeqs bool) []streamDigest {
 	t.Helper()
 	work := t.TempDir()
-	_, err := copyStreamTree(storeDir, filepath.Join(work, "jetstream"))
+	_, metas, err := copyStreamTree(storeDir, filepath.Join(work, "jetstream"))
 	require.NoError(t, err)
-	d, err := digestStore(context.Background(), work, nil, withSeqs)
+	d, err := digestStore(context.Background(), work, nil, withSeqs, metas)
 	require.NoError(t, err)
 	return d
 }
@@ -98,8 +98,19 @@ func TestKVDigest(t *testing.T) {
 		}
 		require.NoError(t, kv2.Delete(ctx, "b"))
 		stop2()
-		d := digestOf(t, filepath.Join(other, "jetstream"), false)
+		d := digestOf(t, filepath.Join(other, "jetstream"), true)
 		assert.NotEqual(t, live[0].Digest, d[0].Digest)
+
+		// Two stores adopted as replicas of one stream: same subjects and
+		// sequences, different content, which compare must call divergent.
+		res, err := compareNodeDigests([]nodeDigest{
+			{Host: "node1", CopiedAt: time.Now(), Streams: live},
+			{Host: "node2", CopiedAt: time.Now(), Streams: d},
+		})
+		require.NoError(t, err)
+		require.True(t, res.divergent())
+		require.Len(t, res.Streams, 1)
+		assert.Contains(t, strings.Join(res.Streams[0].Problems, "\n"), "seq 1 $KV.test.a holds different content")
 	})
 
 	t.Run("report lines", func(t *testing.T) {
@@ -141,14 +152,21 @@ func TestKVDigestStreamFilterAndDeletedSeq(t *testing.T) {
 	ns.WaitForShutdown()
 
 	work := t.TempDir()
-	_, err = copyStreamTree(filepath.Join(parent, "jetstream"), filepath.Join(work, "jetstream"))
+	_, metas, err := copyStreamTree(filepath.Join(parent, "jetstream"), filepath.Join(work, "jetstream"))
 	require.NoError(t, err)
-	d, err := digestStore(ctx, work, []string{"ONE"}, true)
+	d, err := digestStore(ctx, work, []string{"ONE"}, true, metas)
 	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)
+	require.Len(t, d[0].Seqs, 3)
+	assert.Equal(t, seqDigest{Seq: 2, Hash: "deleted"}, d[0].Seqs[1])
+	for _, i := range []int{0, 2} {
+		assert.Equal(t, "one.k", d[0].Seqs[i].Subject)
+		assert.False(t, d[0].Seqs[i].Time.IsZero())
+	}
+	assert.Equal(t, "limits", d[0].Retention)
+	assert.Equal(t, 1, d[0].Replicas)
 }
 
 func freePort(t *testing.T) int {
@@ -229,16 +247,23 @@ func TestKVDigestClusteredReplicas(t *testing.T) {
 	}
 
 	var first string
+	var nodes []nodeDigest
 	for i, store := range stores {
-		d := digestOf(t, filepath.Join(store, "jetstream"), false)
+		d := digestOf(t, filepath.Join(store, "jetstream"), true)
 		require.Len(t, d, 1, "replica %d", i+1)
 		assert.Equal(t, uint64(5), d[0].Msgs, "replica %d", i+1)
+		assert.Equal(t, n, d[0].Replicas, "live replica count survives the standalone rewrite")
+		assert.Equal(t, time.Hour, d[0].MaxAge)
+		nodes = append(nodes, nodeDigest{Host: fmt.Sprintf("s%d", i+1), CopiedAt: time.Now(), Streams: d})
 		if i == 0 {
 			first = d[0].Digest
 			continue
 		}
 		assert.Equal(t, first, d[0].Digest, "replica %d matches replica 1", i+1)
 	}
+	res, err := compareNodeDigests(nodes)
+	require.NoError(t, err)
+	assert.False(t, res.divergent(), "%+v", res)
 }
 
 // Recovery applies MaxAge, so without the rewrite an expired bucket copy
@@ -260,14 +285,14 @@ func TestKVDigestIgnoresExpiry(t *testing.T) {
 	require.NoError(t, err)
 
 	work := t.TempDir()
-	_, err = copyStreamTree(filepath.Join(parent, "jetstream"), filepath.Join(work, "jetstream"))
+	_, metas, 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)
+	d, err := digestStore(ctx, work, nil, false, metas)
 	require.NoError(t, err)
 	require.Len(t, d, 1)
 	assert.Equal(t, uint64(1), d[0].Msgs)
@@ -280,7 +305,7 @@ func TestCopyStreamTree(t *testing.T) {
 	require.NoError(t, os.MkdirAll(filepath.Join(store, "ACME", "streams", "ORDERS"), 0o755))
 
 	dst := filepath.Join(t.TempDir(), "jetstream")
-	skipped, err := copyStreamTree(store, dst)
+	skipped, _, err := copyStreamTree(store, dst)
 	require.NoError(t, err)
 	assert.Equal(t, []string{"ACME/ORDERS"}, skipped)
 
@@ -289,6 +314,6 @@ func TestCopyStreamTree(t *testing.T) {
 	_, 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)
+	_, _, err = copyStreamTree(filepath.Join(t.TempDir(), "absent"), dst)
 	assert.Error(t, err)
 }

cmd/spinifex/cmd/export_test.go

@@ -4,8 +4,10 @@
 package cmd
 
 import (
+	"bytes"
 	"errors"
 	"strconv"
+	"strings"
 	"time"
 
 	handlers_iam "github.com/mulgadc/spinifex/spinifex/handlers/iam"
@@ -146,3 +148,28 @@ var PrintAccountTable = printAccountTable
 
 // AccountSummaries exposes accountSummaries for testing.
 var AccountSummaries = accountSummaries
+
+// CompareKVDigests runs "kv compare" over digest JSON documents and returns
+// whether any stream diverged, plus the printed report.
+func CompareKVDigests(docs ...string) (bool, string, error) {
+	var nodes []nodeDigest
+	for i, doc := range docs {
+		got, err := loadNodeDigests(strings.NewReader(doc), "doc"+strconv.Itoa(i))
+		if err != nil {
+			return false, "", err
+		}
+		nodes = append(nodes, got...)
+	}
+	res, err := compareNodeDigests(nodes)
+	if err != nil {
+		return false, "", err
+	}
+	var buf bytes.Buffer
+	writeCompareResult(&buf, nodes, res)
+	return res.divergent(), buf.String(), nil
+}
+
+// AdminInitFlagDefault returns the registered default of an admin init flag.
+func AdminInitFlagDefault(name string) string {
+	return adminInitCmd.Flags().Lookup(name).DefValue
+}

cmd/spinifex/cmd/jetstream_store.go

@@ -117,7 +117,7 @@ func discardJetStreamStore(storeDir string) error {
 }
 
 // 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
+// when discard was turned off. 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 {
@@ -133,7 +133,7 @@ func checkInitJetStreamStore(storeDir string, nodes int, discard bool) error {
 	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))
+			"re-run without --discard-jetstream=false to remove them once every node has joined", storeDir, len(streams))
 	}
 	return checkStoreDiscardable()
 }

docs/install/install-multi-node/README.md

@@ -237,7 +237,7 @@ 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 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.
+Every server also discards the JetStream store its single-node cluster wrote. Server 1 removes its own once every server has joined, and each joining server removes its own when it joins. Keeping them is never safe: NATS would adopt each server's copy as a replica of the cluster's stream, and those copies never converge. This is automatic whenever `--nodes` is 2 or more. `--discard-jetstream=false` makes init refuse to form over a non-empty store instead, for a server whose store you want to inspect first. 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.
@@ -245,7 +245,7 @@ Run init and join **concurrently** — init blocks until all nodes have joined.
 **Server 1 — initialize:**
 
 ```bash
-sudo spx admin init --force --discard-jetstream \
+sudo spx admin init --force \
   --node node1 --nodes 3 \
   --bind $SPINIFEX_NODE1 --cluster-bind $SPINIFEX_NODE1 \
   --port 4432 --region $AWS_REGION --az $AWS_AZ
@@ -253,6 +253,8 @@ sudo spx admin init --force --discard-jetstream \
 
 `--nodes 3` is the number of servers init waits for. Set it to your total node count if you are building a larger cluster.
 
+IPsec encrypts the Geneve overlay between servers and is on by default. Joining servers take the setting from server 1, so it is chosen once, on init. On servers that share a trusted private link, `--ipsec=false` leaves the overlay unencrypted in exchange for considerably higher throughput between instances.
+
 The init output displays the join command including the token:
 
 ```
@@ -368,6 +370,32 @@ aws ec2 describe-instance-types
 
 A list of instance types means the gateway, IAM and the cluster behind them are all working.
 
+**4. Every server holds the same cluster state.**
+
+Services report Ready even over a cluster whose JetStream replicas disagree, so check the replicas themselves. On **every server**, take a digest of its local store:
+
+```bash
+sudo spx admin kv digest --json --seqs > kv-digest-$(hostname).json
+```
+
+Copy the files to one server and compare them:
+
+```bash
+spx admin kv compare kv-digest-*.json
+```
+
+```
+compared 3 digests: node1@10:02:11 node2@10:02:14 node3@10:02:18
+ok        KV_spinifex-iam-users  [node1,node2,node3]
+      identical, seqs 1-14
+...
+42 streams: 42 consistent, 0 diverged
+```
+
+The last line must report `0 diverged`, and the command exits non-zero otherwise. The digests are copied from live stores a few seconds apart, so a write landing between two copies can briefly show as a difference: take fresh digests and compare again before acting on one. A difference that persists means the cluster adopted a server's old store as a replica. Do not put that cluster in service — reset the servers and form it again.
+
+A replica that is merely behind is not reported, and neither is a message that one server has already replaced with a newer write to the same key. `spx admin kv compare --help` lists exactly what counts as divergence.
+
 **Congratulations! Your Spinifex cluster is installed.**
 
 Continue to [Setting Up Your Cluster](/docs/setting-up-your-cluster) to import an AMI, create a VPC, and launch your first instance.

scripts/install-node.sh

@@ -32,6 +32,9 @@
 #   --identity FILE          SSH private key
 #   --hosts-file FILE        One host per line; blank lines and # comments ignored
 #   --token-ttl D            Join token validity (default: 30m)
+#   --ipsec on|off           Encrypt Geneve between nodes with OVN IPsec
+#                            (default: on). Joining nodes take it from the first
+#                            host; off is for nodes on a trusted private link.
 #   --wipe                   Reset every host to its pre-install state first.
 #                            Destroys all data on ALL hosts, including the first.
 #                            Confirmed separately from --yes; unattended runs
@@ -77,6 +80,7 @@ AZ=""
 EMAIL=""
 PORT=4432
 TOKEN_TTL="30m"
+IPSEC="on"
 RUN_SMOKE=false
 WIPE=false
 MANAGE_FIREWALL=true
@@ -96,6 +100,12 @@ FIREWALL_TIMEOUT=180
 # How long to wait for every ovn-controller to register its chassis.
 CHASSIS_TIMEOUT=120
 
+# How many times to compare the nodes' JetStream digests, and the pause between.
+# Digests are copied from live stores seconds apart, so one round can catch a
+# write that has reached some replicas and not yet the others.
+KV_ROUNDS=5
+KV_RETRY_SLEEP=15
+
 FIREWALL_HELPER="/usr/local/lib/spinifex/spinifex-firewall-apply"
 FIREWALL_PEERS="/etc/spinifex/firewall/peers.nft"
 
@@ -114,6 +124,7 @@ while [[ $# -gt 0 ]]; do
         --user)                SSH_USER="$2"; shift 2 ;;
         --identity)            IDENTITY="$2"; shift 2 ;;
         --token-ttl)           TOKEN_TTL="$2"; shift 2 ;;
+        --ipsec)               IPSEC="$2"; shift 2 ;;
         --hosts-file)
             while IFS= read -r line; do
                 line="${line%%#*}"
@@ -127,7 +138,7 @@ while [[ $# -gt 0 ]]; do
         --no-firewall)  MANAGE_FIREWALL=false; shift ;;
         --yes|-y)   ASSUME_YES=true; shift ;;
         --dry-run)  DRY_RUN=true; shift ;;
-        -h|--help)  sed -n '2,58p' "${BASH_SOURCE[0]}" | sed 's/^# \{0,1\}//'; exit 0 ;;
+        -h|--help)  sed -n '2,61p' "${BASH_SOURCE[0]}" | sed 's/^# \{0,1\}//'; exit 0 ;;
         -*)         echo "ERROR: unknown option: $1" >&2; exit 2 ;;
         *)          HOSTS+=("$1"); shift ;;
     esac
@@ -161,6 +172,12 @@ for required in "--external-pool:$EXT_POOL" "--external-gateway:$EXT_GATEWAY" "-
          --external-gateway 216.218.163.97 --external-prefix-len 27"
 done
 
+case "$IPSEC" in
+    on)  IPSEC_ENABLED=true ;;
+    off) IPSEC_ENABLED=false ;;
+    *)   fail "--ipsec must be on or off, got: $IPSEC" ;;
+esac
+
 [[ "$EXT_POOL" =~ ^[0-9.]+-[0-9.]+$ ]] || fail "--external-pool must be START-END, got: $EXT_POOL"
 [[ "$EXT_PREFIX" =~ ^[0-9]+$ ]] && [ "$EXT_PREFIX" -ge 1 ] && [ "$EXT_PREFIX" -le 32 ] ||
     fail "--external-prefix-len must be 1-32, got: $EXT_PREFIX"
@@ -542,10 +559,12 @@ done
 # --advertise must be explicit whenever --bind is pinned. resolveAdvertiseIP
 # echoes a non-wildcard --bind straight back, so binding services to the lan
 # plane without this publishes the internal address as the node's public dial
-# target.
+# target. --discard-jetstream is init's default for a formation, spelled out
+# because a node that kept its own store would never converge with the rest.
 init_args=(
     --force
     --discard-jetstream
+    --ipsec="$IPSEC_ENABLED"
     --node "${NODE_NAMES[0]}"
     --nodes "$N"
     --bind "${LAN_IPS[0]}"
@@ -732,6 +751,26 @@ done
 
 log "  OVN Southbound: $chassis chassis registered"
 
+# Joiners copy the IPsec setting from the init node's formation response, so a
+# node that disagrees here got its config from somewhere else.
+ipsec_mismatch=""
+for i in $(seq 0 $((N - 1))); do
+    got=$(out "${HOSTS[$i]}" "sudo grep -oP '^ipsec_enabled\s*=\s*\K\w+' /etc/spinifex/spinifex.toml")
+    got="${got//[$'\r\n']/}"
+    [ "$got" = "$IPSEC_ENABLED" ] || ipsec_mismatch+=" ${HOSTS[$i]}=${got:-unset}"
+done
+[ -z "$ipsec_mismatch" ] ||
+    fail "ipsec_enabled should be $IPSEC_ENABLED on every node, found:$ipsec_mismatch"
+
+if ! $IPSEC_ENABLED; then
+    nb_remotes=""
+    for i in $(seq 0 $((DB_NODES - 1))); do nb_remotes+="${nb_remotes:+,}tcp:${LAN_IPS[$i]}:6641"; done
+    nb_ipsec=$(out "${HOSTS[0]}" "sudo ovn-nbctl --db=$nb_remotes --no-leader-only get NB_Global . ipsec")
+    [ "${nb_ipsec//[$'\r\n']/}" = "false" ] ||
+        fail "--ipsec off, but NB_Global.ipsec is '${nb_ipsec//[$'\r\n']/}' — Geneve would demand encryption no node provides"
+fi
+log "  IPsec: $IPSEC on every node"
+
 # --- Re-arm the host firewall ----------------------------------------------
 #
 # The daemon derives the peer sets from cluster membership, so starting the
@@ -826,6 +865,48 @@ else
     log "               are still scoped to themselves and will drop cluster traffic."
 fi
 
+# --- Verify the JetStream replicas agree ------------------------------------
+#
+# Every node discarded its own store on the way in, so each stream's replicas
+# must hold the same content. A formation that adopted a stale store still
+# reports Ready nodes; only reading the replicas themselves shows it.
+
+echo ""
+log "checking that every node holds the same JetStream state"
+kv_rc=1
+kv_report=""
+for round in $(seq 1 "$KV_ROUNDS"); do
+    digests=""
+    for host in "${HOSTS[@]}"; do
+        d=$(out "$host" "sudo spx admin kv digest --json --seqs")
+        [ -n "$d" ] || fail "$host: spx admin kv digest produced no output — run it there to see why"
+        digests+="$d"$'\n'
+    done
+    kv_rc=0
+    kv_report=$(ssh "${SSH_OPTS[@]}" "$SSH_USER@${HOSTS[0]}" "sudo spx admin kv compare -" <<<"$digests" 2>&1) ||
+        kv_rc=$?
+    [ "$kv_rc" -eq 1 ] || break
+    [ "$round" -lt "$KV_ROUNDS" ] || break
+    log "  round $round of $KV_ROUNDS: replicas differ, checking again in ${KV_RETRY_SLEEP}s"
+    sleep "$KV_RETRY_SLEEP"
+done
+
+case "$kv_rc" in
+    0)
+        log "  $(tail -1 <<<"$kv_report")"
+        ;;
+    1)
+        grep -E '^DIVERGED|^    !|streams:' <<<"$kv_report" | sed 's/^/  /' >&2
+        fail "JetStream replicas still disagree after $KV_ROUNDS rounds.
+       The cluster has adopted inconsistent state and must not be put in service.
+       Reset these nodes and form the cluster again."
+        ;;
+    *)
+        sed 's/^/  | /' <<<"$kv_report" >&2
+        fail "could not compare the JetStream digests (spx admin kv compare exited $kv_rc)"
+        ;;
+esac
+
 # The pool is the one thing the operator supplied by hand and the one thing
 # nothing else validates, so print what actually landed. dns_servers in
 # particular is detected rather than passed, and is wrong often enough to be