diff --git a/docs/COMMANDS.md b/docs/COMMANDS.md index a98d42437..68aff755c 100644 --- a/docs/COMMANDS.md +++ b/docs/COMMANDS.md @@ -109,7 +109,7 @@ Operational commands for inspecting cluster state. These fan out NATS requests t | Command | Flags | Description | |---------|-------|-------------| | `spx admin kv digest` | `--store-dir` (default `/nats/jetstream`), `--work-dir` (holds the temporary copy, default the system temp dir), `--stream` (repeatable filter), `--seqs` (also print a hash per sequence), `--json` (print one JSON document, the input `spx admin kv compare` reads) | 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. Prefer `--json --seqs` on every node followed by `spx admin kv compare`, which tells a diverged replica from one that is merely behind. | -| `spx admin kv compare ...` | — (`-` reads any number of digests from stdin) | Reads the `spx admin kv digest --json --seqs` output of every node and checks that each stream's replicas hold the same content. A replica that is merely behind is not divergence; these are: the same sequence holding different content on two replicas, a message missing from one replica that nothing on it explains (no later write to the same subject, no expiry under `max_age` or a message TTL, no stream limit dropping its head), a stream held by fewer or more nodes than its replica count, replicas disagreeing on that count, or no sequence range common to every replica. Sequences past the slowest replica's last are not compared. Exits 0 when every stream is consistent, 1 on divergence, 2 on bad input. Digests are taken from live stores seconds apart, so re-run before trusting a failure that a write landing between two copies could explain. `install-node.sh` gates cluster formation on it. | +| `spx admin kv compare ...` | — (`-` reads any number of digests from stdin) | Reads the `spx admin kv digest --json --seqs` output of every node and checks that each stream's replicas hold the same content. A replica that is merely behind is not divergence; these are: the same sequence holding different content on two replicas, a message missing from one replica that nothing on it explains (no later write to the same subject, no expiry under `max_age` or a message TTL, no stream limit dropping its head), a stream held by fewer or more nodes than its replica count, replicas disagreeing on that count, or no sequence range common to every replica. A stream with a `max_age` whose replicas were copied at different times and share no sequence range is reported as `skipped` (volatile) rather than diverged, since a short-lived bucket such as a leader lease turns over between copies; any sequence two of its replicas share must still match, and each message a replica lacks must be superseded, expired, or written after that replica was copied. Sequences past the slowest replica's last are not compared. Exits 0 when every stream is consistent, 1 on divergence, 2 on bad input. Digests are taken from live stores seconds apart, so re-run before trusting a failure that a write landing between two copies could explain. `install-node.sh` gates cluster formation on it. | | `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. | diff --git a/spinifex/operator/cli/admin_kv_compare.go b/spinifex/operator/cli/admin_kv_compare.go index 97cb0517e..62a140a01 100644 --- a/spinifex/operator/cli/admin_kv_compare.go +++ b/spinifex/operator/cli/admin_kv_compare.go @@ -32,6 +32,12 @@ A replica that is merely behind is not divergent, so only these fail: - replicas disagree on the replica count; - no sequence range is common to every replica, so nothing can be checked. +A stream with a max_age whose replicas, copied at different times, share no +sequence range is reported as skipped (volatile) rather than diverged: a short +lived bucket such as a leader lease turns over between two copies. Sequences +two of its replicas do share must still hold the same content, and each +message a replica lacks must be superseded, expired, or written after its copy. + 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. @@ -110,6 +116,8 @@ type streamVerdict struct { Holders []string Problems []string Notes []string + // Volatile says why a stream that turned over between copies was not compared. + Volatile string } type kvCompareResult struct { @@ -228,6 +236,16 @@ func compareStream(name string, hosts []string, views []*replicaView) streamVerd } } if lo > hi { + maxAge, spread := views[0].d.MaxAge, copySpread(views) + if maxAge > 0 && spread > 0 { + mismatches, unexplained := turnoverFindings(views) + v.Problems = append(v.Problems, mismatches...) + if len(unexplained) == 0 { + v.Volatile = fmt.Sprintf("volatile: max_age %s and copies %s apart share no sequence range, so it turned over between them; not compared", maxAge, spread) + return v + } + v.Problems = append(v.Problems, unexplained...) + } v.Problems = append(v.Problems, "no sequence range is common to every replica, so nothing could be compared") return v } @@ -253,18 +271,8 @@ func compareStream(name string, hosts []string, views []*replicaView) streamVerd 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, " "))) + if p := contentMismatch(seq, present); p != "" { + problems = append(problems, p) continue } msg := present[0].bySeq[seq] @@ -277,10 +285,7 @@ func compareStream(name string, hosts []string, views []*replicaView) streamVerd 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...) + v.Problems = append(v.Problems, capFindings(problems)...) note := fmt.Sprintf("seqs %d-%d compared", lo, hi) if explained > 0 { @@ -299,6 +304,84 @@ func compareStream(name string, hosts []string, views []*replicaView) streamVerd return v } +// contentMismatch describes seq holding different content on the replicas +// present at it, or returns "" when they agree. +func contentMismatch(seq uint64, present []*replicaView) string { + byHash := map[string][]string{} + for _, r := range present { + h := r.bySeq[seq].Hash + byHash[h] = append(byHash[h], r.host) + } + if len(byHash) < 2 { + return "" + } + var parts []string + for h, hs := range byHash { + parts = append(parts, strings.Join(hs, ",")+"="+h) + } + sort.Strings(parts) + return fmt.Sprintf("seq %d %s holds different content: %s", seq, present[0].bySeq[seq].Subject, strings.Join(parts, " ")) +} + +// turnoverFindings compares every sequence held by two or more replicas, since +// no timing explains two values at one sequence, and lists each absence that +// neither absenceExplained nor a write after that replica's copy accounts for. +func turnoverFindings(views []*replicaView) (mismatches, unexplained []string) { + seqSet := map[uint64]struct{}{} + for _, r := range views { + for s := range r.bySeq { + seqSet[s] = struct{}{} + } + } + for _, seq := range slices.Sorted(maps.Keys(seqSet)) { + 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 + } + if p := contentMismatch(seq, present); p != "" { + mismatches = append(mismatches, p) + continue + } + msg := present[0].bySeq[seq] + for _, r := range absent { + if absenceExplained(r, msg) || (seq > r.d.LastSeq && msg.Time.After(r.copiedAt)) { + continue + } + unexplained = append(unexplained, fmt.Sprintf("seq %d %s is on %s but missing from %s, with no later write to that subject, expiry, limit or copy order to explain it", + seq, msg.Subject, hostsOf(present), r.host)) + } + } + return capFindings(mismatches), capFindings(unexplained) +} + +func capFindings(problems []string) []string { + if len(problems) > maxFindingsPerStream { + return append(problems[:maxFindingsPerStream], fmt.Sprintf("... and %d more", len(problems)-maxFindingsPerStream)) + } + return problems +} + +// copySpread is how far apart the earliest and latest replica copies were taken. +func copySpread(views []*replicaView) time.Duration { + first, last := views[0].copiedAt, views[0].copiedAt + for _, r := range views[1:] { + if r.copiedAt.Before(first) { + first = r.copiedAt + } + if r.copiedAt.After(last) { + last = r.copiedAt + } + } + return last.Sub(first) +} + // 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. @@ -338,12 +421,16 @@ func writeCompareResult(w io.Writer, nodes []nodeDigest, res kvCompareResult) { fmt.Fprintf(w, " %s@%s", n.Host, n.CopiedAt.Format(time.TimeOnly)) } fmt.Fprintln(w) - bad := 0 + bad, volatile := 0, 0 for _, v := range res.Streams { status := "ok" - if len(v.Problems) > 0 { + switch { + case len(v.Problems) > 0: status = "DIVERGED" bad++ + case v.Volatile != "": + status = "skipped" + volatile++ } fmt.Fprintf(w, "%-9s %s [%s]\n", status, v.Name, strings.Join(v.Holders, ",")) for _, p := range v.Problems { @@ -352,6 +439,13 @@ func writeCompareResult(w io.Writer, nodes []nodeDigest, res kvCompareResult) { for _, n := range v.Notes { fmt.Fprintf(w, " %s\n", n) } + if v.Volatile != "" { + fmt.Fprintf(w, " %s\n", v.Volatile) + } } - fmt.Fprintf(w, "%d streams: %d consistent, %d diverged\n", len(res.Streams), len(res.Streams)-bad, bad) + fmt.Fprintf(w, "%d streams: %d consistent, %d diverged", len(res.Streams), len(res.Streams)-bad-volatile, bad) + if volatile > 0 { + fmt.Fprintf(w, ", %d skipped as volatile", volatile) + } + fmt.Fprintln(w) } diff --git a/spinifex/operator/cli/admin_kv_compare_test.go b/spinifex/operator/cli/admin_kv_compare_test.go index a9031f9d2..95264cdd7 100644 --- a/spinifex/operator/cli/admin_kv_compare_test.go +++ b/spinifex/operator/cli/admin_kv_compare_test.go @@ -160,6 +160,83 @@ func TestKVCompareNoCommonRange(t *testing.T) { assert.Contains(t, report, "no sequence range is common") } +// A 60s lease bucket renewed every few seconds turns over between digests +// taken a couple of seconds apart, so its replicas share no window. +func TestKVCompareVolatileStream(t *testing.T) { + t1, t2, t3 := t0, t0.Add(2*time.Second), t0.Add(4*time.Second) + lease := func(maxAge time.Duration, msgs ...msg) stream { + return stream{name: "KV_bedrock-leader", replicas: 3, maxAge: maxAge, msgs: msgs} + } + early := kvs(msg{3984, "$KV.bedrock-leader.leader", "h3984", t1.Add(-time.Second)}) + late := kvs(msg{3985, "$KV.bedrock-leader.leader", "h3985", t2.Add(-time.Second)}, msg{3986, "$KV.bedrock-leader.leader", "h3986", t2}) + + t.Run("adjacent windows of a max_age stream are skipped, not diverged", func(t *testing.T) { + divergent, report, err := cli.CompareKVDigests( + doc(t, "node1", t1, lease(time.Minute, early...)), + doc(t, "node2", t2, lease(time.Minute, late...)), + doc(t, "node3", t3, lease(time.Minute, late...))) + require.NoError(t, err) + assert.False(t, divergent, report) + assert.Contains(t, report, "skipped KV_bedrock-leader") + assert.Contains(t, report, "volatile: max_age 1m0s") + assert.Contains(t, report, "1 streams: 0 consistent, 0 diverged, 1 skipped as volatile") + }) + + t.Run("a sequence two replicas share must still match", func(t *testing.T) { + mid := kvs(msg{3984, "$KV.bedrock-leader.leader", "other", t1.Add(-time.Second)}, late[0]) + divergent, report, err := cli.CompareKVDigests( + doc(t, "node1", t1, lease(time.Minute, early...)), + doc(t, "node2", t2, lease(time.Minute, mid...)), + doc(t, "node3", t3, lease(time.Minute, late[1]))) + require.NoError(t, err) + assert.True(t, divergent, report) + assert.Contains(t, report, "seq 3984 $KV.bedrock-leader.leader holds different content: node1=h3984 node2=other") + }) + + t.Run("a long-lived bucket with no common window is still diverged", func(t *testing.T) { + divergent, report, err := cli.CompareKVDigests( + doc(t, "node1", t1, lease(0, early...)), + doc(t, "node2", t2, lease(0, late...)), + doc(t, "node3", t3, lease(0, late...))) + require.NoError(t, err) + assert.True(t, divergent, report) + assert.Contains(t, report, "no sequence range is common") + }) + + t.Run("a long max_age stream whose missing messages nothing explains is still diverged", func(t *testing.T) { + old := kvs(msg{1, "a", "h1", t1.Add(-time.Minute)}, msg{2, "b", "h2", t1.Add(-time.Minute)}) + newer := kvs(msg{10, "c", "h10", t1.Add(-time.Minute)}, msg{11, "d", "h11", t1.Add(-time.Minute)}) + divergent, report, err := cli.CompareKVDigests( + doc(t, "node1", t1, lease(30*24*time.Hour, old...)), + doc(t, "node2", t2, lease(30*24*time.Hour, newer...)), + doc(t, "node3", t3, lease(30*24*time.Hour, newer...))) + require.NoError(t, err) + assert.True(t, divergent, report) + assert.Contains(t, report, "no sequence range is common") + assert.NotContains(t, report, "volatile") + }) + + t.Run("a replica copied last but holding the older window is still diverged", func(t *testing.T) { + divergent, report, err := cli.CompareKVDigests( + doc(t, "node1", t1, lease(time.Minute, late...)), + doc(t, "node2", t2, lease(time.Minute, late...)), + doc(t, "node3", t3, lease(time.Minute, early...))) + require.NoError(t, err) + assert.True(t, divergent, report) + assert.Contains(t, report, "seq 3985 $KV.bedrock-leader.leader is on node1,node2 but missing from node3") + }) + + t.Run("digests copied at the same instant are not excused", func(t *testing.T) { + divergent, report, err := cli.CompareKVDigests( + doc(t, "node1", t1, lease(time.Minute, early...)), + doc(t, "node2", t1, lease(time.Minute, late...)), + doc(t, "node3", t1, lease(time.Minute, late...))) + require.NoError(t, err) + assert.True(t, divergent, report) + 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})}