Compare terminal Jepsen cursors with origin stream heads

test/jepsen/README.md

@@ -41,6 +41,8 @@ terminal snapshots. The independent checker requires:
   node;
 - consistent registry, PG, cluster, claim, cursor, oplog, and remote-authority
   indexes inside every shard;
+- every admitted receiver stream at its independently captured origin head,
+  including streams with no writes;
 - no staged partial snapshot and no retained data for a retired origin;
 - coverage of delta batches, snapshot fallback, multi-chunk assembly, and
   registry conflict termination;
@@ -210,6 +212,22 @@ rather than only tokens; node/boot/owner/incarnation tokens still identify owner
 lifetimes without exporting raw PIDs. Old token-only histories are intentionally
 not accepted as exact metadata evidence.
 
+The cursor capture qualification uses three real Group nodes (two small peer
+VMs), passes their snapshot EDN to the checker, and covers pristine streams,
+explicit zero markers, ahead/missing cursors, cluster closure, instance restart
+and retirement.
+
+Terminal evidence now includes each origin's generation, active epochs and
+per-shard heads, plus every receiver cursor and its actual lane. The checker
+derives admission from both sides' active clusters rather than from the set of
+already-present cursors. A missing cursor is legal only at head zero; an extra
+cursor is illegal even at zero. Source heads must be fully applied, and stream
+evidence must remain unchanged across terminal observations. Generation and
+epoch identities are opaque external-term encodings so reference identity
+survives EDN transport. Old captures without stream evidence cannot qualify.
+All evidence is collected by snapshot clients; no Group shard makes a blocking
+cross-node call.
+
 `mix test.soak` runs that
 same PR gate, the complete mutation/live-checker qualification, and then
 `campaign.sh`. Chaos/mixed uses a sender/repair buffer of 32 and requires

test/jepsen/cursor_capture.exs

@@ -0,0 +1,128 @@
+alias Group.{JepsenCursorCapture, TestCluster}
+
+peers = TestCluster.start_peers(2, schedulers: 1)
+nodes = [node() | Enum.map(peers, &elem(&1, 1))]
+path = Path.expand("node.exs", __DIR__)
+rpc = fn target, function, args -> :erpc.call(target, JepsenCursorCapture, function, args) end
+
+await = fn fun ->
+  Enum.reduce_while(1..200, nil, fn _, _ ->
+    if fun.(),
+      do: {:halt, :ok},
+      else:
+        (
+          Process.sleep(25)
+          {:cont, nil}
+        )
+  end)
+  |> case do
+    :ok -> :ok
+    _ -> raise "capture did not converge"
+  end
+end
+
+try do
+  for target <- nodes, do: rpc.(target, :start, [path])
+
+  :ok =
+    await.(fn ->
+      Enum.all?(nodes, &(length(:erpc.call(&1, Group, :nodes, [:jepsen_group])) == 2))
+    end)
+
+  capture = fn -> Map.new(nodes, &{Atom.to_string(&1), rpc.(&1, :snapshot, [])}) end
+  pristine = capture.()
+
+  [origin, receiver, _survivor] = nodes
+  :ok = rpc.(origin, :write, [])
+  shard = :erlang.phash2({nil, "jepsen/registry/0"}, 2)
+  stream = rpc.(origin, :stream, [nil, shard])
+
+  :ok =
+    await.(fn ->
+      Enum.all?(tl(nodes), fn target ->
+        :erpc.call(target, Group.Replica.Data, :replica_cursor, [:jepsen_group, shard, stream]) ==
+          1
+      end)
+    end)
+
+  healthy = capture.()
+  :ok = rpc.(receiver, :freeze, [])
+
+  corruptions =
+    for {selected, value} <- [
+          {stream, 101},
+          {stream, :missing},
+          {rpc.(origin, :stream, ["red", 0]), 101}
+        ] do
+      old = rpc.(receiver, :set_cursor, [selected, value])
+      snapshot = capture.()
+      true = snapshot[Atom.to_string(receiver)].internal.healthy
+      :ok = rpc.(receiver, :restore_cursor, [selected, old])
+      snapshot
+    end
+
+  zero_stream = rpc.(origin, :stream, ["red", 1])
+  old = rpc.(receiver, :set_cursor, [zero_stream, 0])
+  zero = capture.()
+  :ok = rpc.(receiver, :restore_cursor, [zero_stream, old])
+  :ok = rpc.(receiver, :thaw, [])
+
+  for target <- nodes, do: :ok = :erpc.call(target, Group, :disconnect, [:jepsen_group, ["red"]])
+
+  :ok =
+    await.(fn ->
+      Enum.all?(capture.(), fn {_, snapshot} ->
+        Enum.all?(snapshot.streams.cursors, &(&1.stream.cluster != "red"))
+      end)
+    end)
+
+  closed = capture.()
+
+  :ok = rpc.(origin, :restart, [])
+  new_generation = rpc.(origin, :snapshot, []).streams.generation
+  false = new_generation == healthy[Atom.to_string(origin)].streams.generation
+
+  :ok =
+    await.(fn ->
+      Enum.all?(capture.(), fn {_, snapshot} ->
+        snapshot.internal.healthy and
+          Enum.all?(snapshot.streams.cursors, fn cursor ->
+            cursor.stream.origin != Atom.to_string(origin) or
+              cursor.stream.generation == new_generation
+          end)
+      end)
+    end)
+
+  restarted = capture.()
+  :ok = rpc.(origin, :stop_group, [])
+
+  for target <- tl(nodes), shard <- 0..1 do
+    :ok = :erpc.call(target, TestCluster, :expire_replica_lane, [:jepsen_group, shard, origin])
+  end
+
+  retired_capture = fn ->
+    Map.new(tl(nodes), &{Atom.to_string(&1), rpc.(&1, :snapshot, [[origin]])})
+  end
+
+  :ok =
+    await.(fn ->
+      Enum.all?(retired_capture.(), fn {_, snapshot} -> snapshot.internal.healthy end)
+    end)
+
+  retired = retired_capture.()
+
+  File.write!(
+    hd(System.argv()),
+    Group.Jepsen.EDN.encode(%{
+      pristine: pristine,
+      healthy: healthy,
+      zero: zero,
+      closed: closed,
+      restarted: restarted,
+      retired: retired,
+      corruptions: corruptions
+    })
+  )
+after
+  TestCluster.stop_peers(peers)
+end

test/jepsen/node.exs

@@ -1367,6 +1367,57 @@ end
 defmodule Group.Jepsen.Snapshot do
   @moduledoc false
 
+  alias Group.Replica.{Data, WireProtocol}
+
+  # Read each origin independently from the snapshot client, never by calling
+  # another node from a Group shard. Include pristine streams explicitly: an
+  # absent stream-meta row means head zero, not missing terminal evidence.
+  def stream_positions do
+    group = :jepsen_group
+    generation = Data.generation(group)
+    epochs = Data.local_cluster_epochs(group)
+    num_shards = Group.get_config(group).num_shards
+
+    heads =
+      for shard <- 0..(num_shards - 1), {cluster, epoch} <- epochs do
+        stream = WireProtocol.stream_id(group, node(), generation, shard, cluster, epoch)
+        {_floor, head, applied} = Data.replica_stream_head(group, shard, stream)
+        %{stream: stream_identity(stream), head: head, applied: applied}
+      end
+
+    cursors =
+      for shard <- 0..(num_shards - 1),
+          {stream, position} <- :ets.tab2list(Data.replica_cursor_table(group, shard)) do
+        %{stream: stream_identity(stream), position: position_value(position), lane: shard}
+      end
+
+    %{
+      origin: Atom.to_string(node()),
+      generation: identity(generation),
+      shards: num_shards,
+      epochs:
+        Map.new(epochs, fn {cluster, epoch} -> {cluster_name(cluster), identity(epoch)} end),
+      heads: Enum.sort(heads),
+      cursors: Enum.sort(cursors)
+    }
+  end
+
+  defp stream_identity(stream) do
+    %{
+      group: Atom.to_string(WireProtocol.stream_name(stream)),
+      origin: Atom.to_string(WireProtocol.stream_origin(stream)),
+      generation: identity(WireProtocol.stream_generation(stream)),
+      shard: WireProtocol.stream_shard(stream),
+      cluster: cluster_name(WireProtocol.stream_cluster(stream)),
+      epoch: identity(WireProtocol.stream_epoch(stream))
+    }
+  end
+
+  # References must retain their originating-node identity across EDN captures.
+  defp identity(term), do: term |> :erlang.term_to_binary() |> Base.encode64()
+  defp position_value(value) when is_integer(value), do: value
+  defp position_value(value), do: inspect(value)
+
   def capture(node_id, boot_id, key_count, clusters, retired_nodes) do
     owners = Group.Jepsen.Driver.owner_snapshots()
 
@@ -1418,6 +1469,7 @@ defmodule Group.Jepsen.Snapshot do
         transport_events: Group.Jepsen.Transport.Stats.snapshot(),
         transport_profile: Group.Jepsen.Transport.Control.profile(),
         internal: Group.Jepsen.Invariant.snapshot(retired_nodes),
+        streams: stream_positions(),
         registry: registry,
         pg: pg
       }

test/jepsen/src/group/jepsen/model.clj

@@ -1,5 +1,6 @@
 (ns group.jepsen.model
   (:require [clojure.set :as set]
+            [group.jepsen.streams :as streams]
             [jepsen.checker :as checker]
             [jepsen.history :as history]))
 
@@ -91,6 +92,7 @@
    :peers (set (:peers snapshot))
    :unexpected-deaths (set (:unexpected-deaths snapshot))
    :conflict-evidence (:conflict-evidence snapshot)
+   :streams (:streams snapshot)
    :view (normalize-view test snapshot)
    :internal (stable-internal snapshot)})
 
@@ -197,6 +199,7 @@
                           [node fingerprints]))))
               relevant-observations)
         expected (expected-state test relevant-snapshots)
+        stream-errors (streams/errors relevant-snapshots)
         expected-view (select-keys expected [:registry :pg])
         views (into {} (map (fn [[node snapshot]]
                               [node (normalize-view test snapshot)]))
@@ -296,6 +299,7 @@
                     delta-run-coverage?
                     (empty? transport-profile-mismatches)
                     (empty? internal-errors)
+                    (empty? stream-errors)
                     (empty? (:conflicts expected))
                     (empty? mismatches)
                     (empty? unexpected-deaths)
@@ -317,6 +321,7 @@
      :min-delta-run-records min-delta-run-records
      :transport-profile-mismatches transport-profile-mismatches
      :internal-invariant-errors internal-errors
+     :stream-position-errors stream-errors
      :max-group-operation-latency-ms (/ max-latency-us 1000.0)
      :group-operation-latency-limit-ms (/ latency-limit-us 1000.0)
      :live-owner-count (count live-tokens)

test/jepsen/src/group/jepsen/streams.clj

@@ -0,0 +1,57 @@
+(ns group.jepsen.streams
+  (:require [clojure.set :as set]))
+
+(defn natural? [n] (and (integer? n) (<= 0 n)))
+
+(defn evidence-valid? [{:keys [origin generation shards epochs heads cursors]}]
+  (and (string? origin) (string? generation)
+       (integer? shards) (< 0 shards)
+       (map? epochs) (= generation (get epochs "root"))
+       (every? string? (keys epochs)) (every? string? (vals epochs))
+       (sequential? heads) (sequential? cursors)
+       (= (count heads) (count (set (map :stream heads))))
+       (= (count cursors) (count (set (map :stream cursors))))
+       (= (set (map :stream heads))
+          (set (for [shard (range shards), [cluster epoch] epochs]
+                 {:group "jepsen_group" :origin origin :generation generation
+                  :shard shard :cluster cluster :epoch epoch})))
+       (every? #(and (natural? (:head %)) (= (:head %) (:applied %))) heads)
+       (every? #(and (natural? (:position %))
+                     (natural? (:lane %)) (< (:lane %) shards)
+                     (= (:lane %) (get-in % [:stream :shard]))) cursors)))
+
+(defn errors
+  "At stable quiescence every admitted remote stream equals its origin head.
+   Head zero permits an absent cursor or an explicit zero admission marker.
+   Admission comes from both origins' current active epochs, not from the
+   receiver's existing cursor set (which could itself be missing or stale)."
+  [snapshots]
+  (let [evidence (into {} (map (fn [[node snapshot]] [node (:streams snapshot)])) snapshots)
+        invalid (into {} (remove (comp evidence-valid? val)) evidence)
+        origins (map :origin (vals evidence))]
+    (if (or (seq invalid) (not= (count origins) (count (set origins))))
+      {:invalid-evidence invalid :duplicate-origins (not= (count origins) (count (set origins)))}
+      (into {}
+            (keep
+              (fn [[node receiver]]
+                (let [expected
+                      (into {}
+                            (for [[other sender] evidence
+                                  :when (not= other node)
+                                  {:keys [stream head]} (:heads sender)
+                                  :when (contains? (:epochs receiver) (:cluster stream))]
+                              [stream head]))
+                      actual (into {} (map (juxt :stream :position)) (:cursors receiver))
+                      mismatches
+                      (into {}
+                            (keep (fn [stream]
+                                    (let [head (get expected stream)
+                                          cursor (get actual stream 0)]
+                                      (when (not= head cursor)
+                                        [stream {:head head
+                                                 :cursor (get actual stream :missing)}]))))
+                            (set/union (set (keys expected)) (set (keys actual))))
+                      shard-counts (set (map :shards (vals evidence)))]
+                  (when (or (seq mismatches) (< 1 (count shard-counts)))
+                    [node {:positions mismatches :shard-counts shard-counts}]))))
+            evidence))))

test/jepsen/test/group/jepsen/cursor_capture_test.clj

@@ -0,0 +1,22 @@
+(ns group.jepsen.cursor-capture-test
+  (:require [clojure.edn :as edn]
+            [clojure.java.shell :as shell]
+            [clojure.test :refer :all]
+            [group.jepsen.streams :as streams]))
+
+(deftest ^:capture compares-real-three-node-cursors-with-independent-origin-heads
+  (let [output (java.io.File/createTempFile "group-cursors-" ".edn")]
+    (try
+      (let [run (shell/sh "env" "ERL_FLAGS=+S 2:2" "MIX_ENV=test" "mix" "run"
+                          "test/jepsen/cursor_capture.exs" (.getPath output)
+                          :dir "../..")]
+        (is (= 0 (:exit run)) (str (:out run) (:err run)))
+        (when (zero? (:exit run))
+          (let [captures (edn/read-string (slurp output))]
+            (doseq [kind [:pristine :healthy :zero :closed :restarted :retired]]
+              (is (empty? (streams/errors (get captures kind))) (str kind))
+              (is (every? #(true? (get-in % [:internal :healthy])) (vals (get captures kind)))))
+            (doseq [snapshots (:corruptions captures)]
+              (is (every? #(true? (get-in % [:internal :healthy])) (vals snapshots)))
+              (is (seq (streams/errors snapshots)))))))
+      (finally (.delete output)))))

test/jepsen/test/group/jepsen/model_test.clj

@@ -50,6 +50,12 @@
             :unexpected-deaths []
             :transport-events {}
             :transport-profile :distribution
+            :streams {:origin (str "group@" node) :generation node :shards 1
+                      :epochs {"root" node}
+                      :heads [{:stream {:group "jepsen_group" :origin (str "group@" node)
+                                        :generation node :shard 0 :cluster "root" :epoch node}
+                               :head 0 :applied 0}]
+                      :cursors []}
             :internal (healthy-internal)
             :registry registry
             :pg pg}}))