From b2127f7209396e2f01cd9194bfecfced72ab5f66 Mon Sep 17 00:00:00 2001 From: Jason Stiebs Date: Fri, 11 Sep 2026 12:34:52 -0500 Subject: [PATCH] Check terminal receiver cursors against independent origin heads --- mix.exs | 2 +- test/jepsen/README.md | 22 ++- test/jepsen/cursor_capture.exs | 128 ++++++++++++++++++ test/jepsen/node.exs | 56 +++++++- test/jepsen/project.clj | 2 + test/jepsen/src/group/jepsen/model.clj | 5 + test/jepsen/src/group/jepsen/streams.clj | 57 ++++++++ .../test/group/jepsen/cursor_capture_test.clj | 22 +++ test/jepsen/test/group/jepsen/model_test.clj | 6 + .../jepsen/test/group/jepsen/streams_test.clj | 85 ++++++++++++ test/support/jepsen_cursor_capture.ex | 102 ++++++++++++++ 11 files changed, 484 insertions(+), 3 deletions(-) create mode 100644 test/jepsen/cursor_capture.exs create mode 100644 test/jepsen/src/group/jepsen/streams.clj create mode 100644 test/jepsen/test/group/jepsen/cursor_capture_test.clj create mode 100644 test/jepsen/test/group/jepsen/streams_test.clj create mode 100644 test/support/jepsen_cursor_capture.ex diff --git a/mix.exs b/mix.exs index f9b9580..07439f6 100644 --- a/mix.exs +++ b/mix.exs @@ -64,7 +64,7 @@ defmodule Group.MixProject do defp aliases do [ - test: ["test", "cmd test/jepsen/checker.sh"], + test: ["test", "cmd test/jepsen/checker.sh", "cmd test/jepsen/lein.sh test :capture"], "test.soak": [ # Run the PR gate in a child VM. test_helper starts distribution, and # keeping that VM alive for the following `cmd` phases can retain a diff --git a/test/jepsen/README.md b/test/jepsen/README.md index 199c9e7..b094aab 100644 --- a/test/jepsen/README.md +++ b/test/jepsen/README.md @@ -40,6 +40,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; @@ -135,7 +137,25 @@ test/jepsen/checker.sh ``` At the repository root, `mix test` runs this pure checker after the complete -ExUnit, StreamData, and deterministic-chaos suite. `mix test.soak` runs that +ExUnit, StreamData, and deterministic-chaos suite, followed by executable +capture qualification (`test/jepsen/lein.sh test :capture`, requiring Elixir). +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 evidence that one repaired delta run contained at least two records; all other diff --git a/test/jepsen/cursor_capture.exs b/test/jepsen/cursor_capture.exs new file mode 100644 index 0000000..7715f77 --- /dev/null +++ b/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 diff --git a/test/jepsen/node.exs b/test/jepsen/node.exs index 65bd92a..e8eda37 100644 --- a/test/jepsen/node.exs +++ b/test/jepsen/node.exs @@ -1230,6 +1230,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() @@ -1280,6 +1331,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 } @@ -1565,4 +1617,6 @@ defmodule Group.Jepsen.Main do end end -Group.Jepsen.Main.run(System.argv()) +unless System.get_env("GROUP_JEPSEN_LIBRARY") == "1" do + Group.Jepsen.Main.run(System.argv()) +end diff --git a/test/jepsen/project.clj b/test/jepsen/project.clj index 05284a5..247c800 100644 --- a/test/jepsen/project.clj +++ b/test/jepsen/project.clj @@ -4,5 +4,7 @@ :license {:name "MIT"} :dependencies [[org.clojure/clojure "1.12.4"] [jepsen "0.3.13"]] + :test-selectors {:default (complement :capture) + :capture :capture} :main group.jepsen.core :jvm-opts ["-Xmx4g" "-Djava.awt.headless=true" "-server"]) diff --git a/test/jepsen/src/group/jepsen/model.clj b/test/jepsen/src/group/jepsen/model.clj index bc772aa..1f4c65b 100644 --- a/test/jepsen/src/group/jepsen/model.clj +++ b/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])) @@ -88,6 +89,7 @@ {:owners (set (:owners snapshot)) :peers (set (:peers snapshot)) :unexpected-deaths (set (:unexpected-deaths snapshot)) + :streams (:streams snapshot) :view (normalize-view test snapshot) :internal (stable-internal snapshot)}) @@ -121,6 +123,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)])) @@ -195,6 +198,7 @@ delta-run-coverage? (empty? transport-profile-mismatches) (empty? internal-errors) + (empty? stream-errors) (empty? (:conflicts expected)) (empty? mismatches) (empty? unexpected-deaths) @@ -213,6 +217,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) diff --git a/test/jepsen/src/group/jepsen/streams.clj b/test/jepsen/src/group/jepsen/streams.clj new file mode 100644 index 0000000..9c429af --- /dev/null +++ b/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)))) diff --git a/test/jepsen/test/group/jepsen/cursor_capture_test.clj b/test/jepsen/test/group/jepsen/cursor_capture_test.clj new file mode 100644 index 0000000..813db2e --- /dev/null +++ b/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))))) diff --git a/test/jepsen/test/group/jepsen/model_test.clj b/test/jepsen/test/group/jepsen/model_test.clj index 81e5263..d3c2d32 100644 --- a/test/jepsen/test/group/jepsen/model_test.clj +++ b/test/jepsen/test/group/jepsen/model_test.clj @@ -41,6 +41,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}})) diff --git a/test/jepsen/test/group/jepsen/streams_test.clj b/test/jepsen/test/group/jepsen/streams_test.clj new file mode 100644 index 0000000..05481a3 --- /dev/null +++ b/test/jepsen/test/group/jepsen/streams_test.clj @@ -0,0 +1,85 @@ +(ns group.jepsen.streams-test + (:require [clojure.test :refer :all] + [group.jepsen.model :as model] + [group.jepsen.model-test :as fixtures] + [group.jepsen.streams :as streams])) + +(defn evidence [origin generation epochs heads cursors] + {:origin origin :generation generation :shards 2 :epochs epochs + :heads (vec (for [[cluster epoch] epochs, shard (range 2)] + {:stream {:group "jepsen_group" :origin origin :generation generation + :shard shard :cluster cluster :epoch epoch} + :head (get heads [cluster shard] 0) + :applied (get heads [cluster shard] 0)})) + :cursors cursors}) + +(defn cursor [evidence cluster shard position] + {:stream (:stream (first (filter #(and (= cluster (get-in % [:stream :cluster])) + (= shard (get-in % [:stream :shard]))) + (:heads evidence)))) + :lane shard :position position}) + +(def origin (evidence "a" "g1" {"root" "g1" "red" "e1"} {["root" 0] 1} [])) +(def receiver (evidence "b" "g2" {"root" "g2" "red" "e2"} {} + [(cursor origin "root" 0 1)])) +(def survivor (evidence "c" "g3" {"root" "g3"} {} + [(cursor origin "root" 0 1)])) +(def baseline {"a" {:streams origin} "b" {:streams receiver} "c" {:streams survivor}}) + +(deftest accepts-exact-and-pristine-streams + (is (empty? (streams/errors baseline))) + (is (empty? (streams/errors + (update-in baseline ["b" :streams :cursors] + conj (cursor origin "red" 1 0)))))) + +(deftest rejects-out-of-range-missing-and-misplaced-cursors + (doseq [position [101 0 -1 "1"]] + (is (seq (streams/errors + (assoc-in baseline ["b" :streams :cursors 0 :position] position))))) + (is (seq (streams/errors (assoc-in baseline ["b" :streams :cursors] []))))) + +(deftest rejects-every-stale-stream-dimension + (doseq [[field wrong] [[:group "other"] [:origin "retired"] [:generation "old"] + [:epoch "closed"] [:cluster "blue"] [:shard 1]]] + (is (seq (streams/errors + (assoc-in baseline ["b" :streams :cursors 0 :stream field] wrong)))))) + +(deftest requires-complete-origin-evidence + (is (seq (streams/errors (update baseline "a" dissoc :streams)))) + (is (seq (streams/errors (assoc-in baseline ["a" :streams :heads] [])))) + (is (seq (streams/errors (assoc-in baseline ["a" :streams :heads 0 :applied] -1))))) + +(deftest closed-and-restarted-origins-cannot-retain-cursors + (let [with-red (update-in baseline ["b" :streams :cursors] + conj (cursor origin "red" 0 0)) + closed (evidence "a" "g1" {"root" "g1"} {["root" 0] 1} [])] + (is (empty? (streams/errors with-red))) + (is (seq (streams/errors (assoc-in with-red ["a" :streams] closed)))) + (is (empty? (streams/errors (assoc-in baseline ["a" :streams] closed))))) + (let [retired (dissoc baseline "a")] + (is (seq (streams/errors retired))) + (is (empty? (streams/errors + (into {} (map (fn [[node snapshot]] + [node (assoc-in snapshot [:streams :cursors] [])])) + retired))))) + (let [restarted (assoc-in baseline ["a" :streams] + (evidence "a" "new" {"root" "new"} {} []))] + (is (seq (streams/errors restarted))) + (is (empty? (streams/errors + (-> restarted + (assoc-in ["b" :streams :cursors] []) + (assoc-in ["c" :streams :cursors] []))))))) + +(deftest terminal-checker-requires-stable-stream-positions + (let [history (mapv #(fixtures/snapshot-op % (str "n" %) [] + (fixtures/empty-registry) (fixtures/empty-pg)) + [1 2 3]) + changed (assoc-in history [1 :value :streams :heads 0 :head] 1) + changed (assoc-in changed [1 :value :streams :heads 0 :applied] 1)] + (is (:valid? (model/analyze fixtures/test-map history))) + (is (false? (:valid? (model/analyze fixtures/test-map changed)))) + (is (seq (:stream-position-errors (model/analyze fixtures/test-map changed)))) + (let [two-rounds (concat history (map #(update % :index + 3) changed)) + result (model/analyze (assoc fixtures/test-map :terminal-snapshots-per-node 2) + two-rounds)] + (is (contains? (:unstable-terminal-observations result) "n2"))))) diff --git a/test/support/jepsen_cursor_capture.ex b/test/support/jepsen_cursor_capture.ex new file mode 100644 index 0000000..973dd51 --- /dev/null +++ b/test/support/jepsen_cursor_capture.ex @@ -0,0 +1,102 @@ +defmodule Group.JepsenCursorCapture do + @moduledoc false + @compile {:no_warn_undefined, + [ + Group.Jepsen.Driver, + Group.Jepsen.Driver.Supervisor, + Group.Jepsen.Transport.Stats, + Group.Jepsen.Snapshot + ]} + alias Group.Replica.{Data, WireProtocol} + + def start(path) do + System.put_env("GROUP_JEPSEN_LIBRARY", "1") + Code.require_file(path) + + for starter <- [ + fn -> Group.Jepsen.Transport.Stats.start_link([]) end, + &start_group/0, + fn -> + Group.Jepsen.Driver.Supervisor.start_link( + node_id: Atom.to_string(node()), + boot_id: "cursor-capture" + ) + end + ] do + {:ok, pid} = starter.() + Process.unlink(pid) + end + + :ok = Group.connect(:jepsen_group, ["red"]) + end + + defp start_group do + Group.start_link( + name: :jepsen_group, + shards: 2, + log: false, + replicated_anti_entropy_interval: 50, + replicated_peer_lease_timeout: 60_000 + ) + end + + def stop_group do + %{status: :ok} = Group.Jepsen.Driver.kill("owner") + Supervisor.stop(:jepsen_group_group_sup) + end + + def restart do + :ok = stop_group() + {:ok, pid} = start_group() + Process.unlink(pid) + :ok + end + + def snapshot(retired \\ []) do + Group.Jepsen.Snapshot.capture(Atom.to_string(node()), "cursor-capture", 1, ["red"], retired).snapshot + end + + def write do + %{status: :ok} = Group.Jepsen.Driver.mutate(:register, "owner", nil, 0, 1) + :ok + end + + def freeze do + for shard <- 0..1, do: :sys.suspend(Group.Replica.shard_name(:jepsen_group, shard)) + :ok + end + + def thaw do + for shard <- 0..1, do: :sys.resume(Group.Replica.shard_name(:jepsen_group, shard)) + :ok + end + + def stream(cluster, shard) do + WireProtocol.stream_id( + :jepsen_group, + node(), + Data.generation(:jepsen_group), + shard, + cluster, + Data.local_cluster_epoch(:jepsen_group, cluster) + ) + end + + def set_cursor(stream, position) do + table = Data.replica_cursor_table(:jepsen_group, WireProtocol.stream_shard(stream)) + old = :ets.lookup(table, stream) + + if position == :missing, + do: :ets.delete(table, stream), + else: :ets.insert(table, {stream, position}) + + old + end + + def restore_cursor(stream, old) do + set_cursor(stream, :missing) + table = Data.replica_cursor_table(:jepsen_group, WireProtocol.stream_shard(stream)) + :ets.insert(table, old) + :ok + end +end