Require receiver commits for Jepsen repair coverage

test/jepsen/Dockerfile.node

@@ -11,6 +11,7 @@ ENV MIX_ENV=prod
 COPY mix.exs mix.lock ./
 COPY lib ./lib
 COPY test/jepsen/node.exs ./test/jepsen/node.exs
+COPY test/jepsen/repair_coverage.exs ./test/jepsen/repair_coverage.exs
 COPY test/support/test_tcp_transport.ex ./test/support/test_tcp_transport.ex
 
 RUN mix local.hex --force \

test/jepsen/README.md

@@ -56,9 +56,65 @@ cannot masquerade as a current owner. Every history explicitly restarts one
 node after the deterministic conflict prelude, proving the checker does not
 mistake restart-sensitive instrumentation for missing protocol coverage.
 
+Registry-conflict exits are obligations, not trusted coverage counters. Before
+calling Group, each owner journals its registration attempt independently of
+Group's tables; it journals successful or rejected replies, unregisters, and
+cluster-intent removal as well. Drivers retain the victim token, key, and
+winner metadata from every conflict death. The checker reconstructs the
+victim's registrations, requires the reported winner to rank strictly higher
+by `{revision, token}`, and requires a matching historical winning claim in
+the same cluster and key. Only validated deaths count toward coverage.
+
+Registration calls interrupted by death or an indeterminate reply remain
+possible claims: Group may have installed them before the owner could record
+success. A definitive `:taken` reply excludes an attempt. Winning evidence is
+retained after unregister, death, and BEAM restart because delayed replicas
+can legitimately act on an older claim. Local journal order excludes winners
+first attempted after a death; the oracle does not invent a global clock or
+infer remote deletion delivery from wall time. This establishes independently
+witnessed possible winners, not the exact instant a replica learned a claim.
+
+The append-only conflict journal retains small operation records for the
+bounded campaign, not production ETS rows. Its path defaults to
+`/tmp/group-jepsen-conflict-evidence` inside each container and can be overridden
+with the driver's `:conflict_evidence_path` option. Corrupt or unreadable
+journals fail closed. At permanent retirement, the stopped-container collector
+archives and decodes the journal alongside unexpected deaths. The checker
+replays retired and surviving nodes' evidence together, without treating retired
+owners as live. Conflict archives are bounded to 64 MiB with ten-second command
+deadlines; missing or malformed archives fail the history.
+
+Each new history resets the running recorder through the harness socket after
+DB restart and before workload mutations. This clears disk and in-memory
+evidence together and initializes an empty journal, so repeated histories
+cannot inherit earlier conflict coverage.
+
+Checker qualification also loads the real Elixir
+Owner/Driver harness, injects valid and forged death reasons, exercises a
+register interrupted before its reply, and checks the emitted EDN with the
+Clojure lifecycle oracle.
+
+Repair coverage is receiver evidence, not transport admission. Before starting
+Group, the Jepsen VM compiles test-only wrappers around the existing delta-run
+and terminal snapshot-install functions. Each wrapper samples the same stream's
+cursor before and after its original body in one shard turn. Only positive
+committed advancement emits `applied-delta-run-records-peak`; a completed snapshot
+with a multi-chunk manifest emits `multi-chunk-snapshot-committed`. Duplicate
+prefixes do not contribute to the delta peak. Provisional chunks, missing
+terminal frames, rejected authority, and logical drops contribute neither.
+The wrapper installation fails if either expected boundary disappears.
+
+The receiver evidence owner persists increasing maxima outside the shard and
+reloads them after VM restart. A crash before persistence can lose evidence
+(failing coverage conservatively), but cannot create it. This instrumentation
+is shared by distribution, TCP, and chaos and changes no production BEAM or API.
+Sender `attempted-*` peaks remain diagnostics only. The pure checker reads the
+same EDN fixture that executable receiver regressions compare to live output.
+
 ## Requirements
 
 - Docker with Compose v2
+- Elixir/Mix with the repository dependencies installed (checker qualification)
 - Java 21 or newer
 - `curl`
 

test/jepsen/conflict_probe.exs

@@ -0,0 +1,146 @@
+# Invoked by the Clojure checker qualification. Load the actual harness without
+# starting its TCP server; no copied Owner/Driver logic lives in this probe.
+System.put_env("GROUP_JEPSEN_LIBRARY_ONLY", "1")
+Code.require_file("node.exs", __DIR__)
+Application.ensure_all_started(:group)
+Logger.configure(level: :emergency)
+
+defmodule Group.Jepsen.ConflictProbe do
+  alias Group.Jepsen.{ConflictEvidence, Driver, EDN}
+
+  def run do
+    directory = Path.join(__DIR__, ".cache/conflict-probe-#{System.unique_integer([:positive])}")
+    File.mkdir_p!(directory)
+
+    try do
+      {winner, rejected, winner_events, archive} = winner(directory)
+
+      cases = [
+        {"historical winner subsequently unregistered and died", true, 1, "jepsen/registry/0",
+         winner, false},
+        {"victim killed during register", true, 1, "jepsen/registry/0", winner, true},
+        {"forged key and rank", false, 99, "nonexistent", %{token: "invented", revision: -100},
+         false},
+        {"rightful winner killed", false, 99, "jepsen/registry/0", winner, false},
+        {"nonexistent winning claim", false, 1, "jepsen/registry/0",
+         %{token: "invented", revision: 100}, false},
+        {"definitively rejected winning claim", false, 1, "jepsen/registry/0", rejected, false}
+      ]
+
+      scenarios =
+        Enum.with_index(cases, fn {label, valid, revision, key, meta, pending}, index ->
+          {events, reset_events} = victim(directory, index, revision, key, meta, pending)
+
+          %{
+            label: label,
+            valid: valid,
+            archive: archive,
+            reset_evidence: reset_events,
+            snapshots: %{
+              "n1" => %{conflict_evidence: events},
+              "n2" => %{conflict_evidence: winner_events}
+            }
+          }
+        end)
+
+      IO.puts("CONFLICT-PROBE " <> EDN.encode(scenarios))
+    after
+      File.rm_rf!(directory)
+    end
+  end
+
+  defp start(directory, label, node_id) do
+    {:ok, group} = Group.start_link(name: :jepsen_group, shards: 1, log: false)
+    path = Path.join(directory, label)
+    {:ok, evidence} = ConflictEvidence.start_link(conflict_evidence_path: path)
+    {:ok, driver} = Driver.start_link(index: 0, node_id: node_id, boot_id: label)
+    {group, evidence, driver, path}
+  end
+
+  defp mutate(driver, operation, revision) do
+    GenServer.call(driver, {:mutate, operation, "owner", nil, 0, revision})
+  end
+
+  defp winner(directory) do
+    {group, evidence, driver, path} = start(directory, "winner", "n2")
+    %{status: :ok, owner: %{token: token}} = mutate(driver, :register, 10)
+
+    %{status: :fail, owner: %{token: rejected}} =
+      GenServer.call(driver, {:mutate, :register, "rejected", nil, 0, 100})
+
+    %{status: :ok} = mutate(driver, :unregister, 0)
+    %{status: :ok} = GenServer.call(driver, {:kill, "owner"})
+    %{status: :ok} = GenServer.call(driver, {:kill, "rejected"})
+    events = ConflictEvidence.snapshot()
+    archive = File.read!(path)
+    GenServer.stop(driver)
+    GenServer.stop(evidence)
+    Supervisor.stop(group)
+    {%{token: token, revision: 10}, %{token: rejected, revision: 100}, events, archive}
+  end
+
+  defp victim(directory, index, revision, key, winner, pending) do
+    {group, evidence, driver, path} = start(directory, "victim-#{index}", "n1")
+    %{status: :ok} = mutate(driver, :join, 0)
+    {pid, _token, _monitor, _cached} = :sys.get_state(driver).owners["owner"]
+    shard = Group.Replica.shard_for(:jepsen_group, nil, "jepsen/registry/0")
+
+    task =
+      if pending do
+        :ok = :sys.suspend(shard)
+        task = Task.async(fn -> mutate(driver, :register, revision) end)
+        wait(fn -> Enum.any?(ConflictEvidence.snapshot(), &(&1.kind == :register)) end)
+        task
+      else
+        %{status: :ok} = mutate(driver, :register, revision)
+        nil
+      end
+
+    Process.exit(pid, {:group_registry_conflict, key, winner})
+    if task, do: Task.await(task)
+    wait(fn -> Enum.any?(ConflictEvidence.snapshot(), &(&1.kind == :death)) end)
+    if pending, do: :sys.resume(shard)
+    events = ConflictEvidence.snapshot()
+    [] = GenServer.call(driver, :unexpected_deaths)
+    {:ok, []} = GenServer.call(driver, :owner_snapshots)
+    GenServer.stop(driver)
+    GenServer.stop(evidence)
+
+    # The exact registration/death obligations must survive a recorder restart.
+    {:ok, evidence} = ConflictEvidence.start_link(conflict_evidence_path: path)
+    ^events = ConflictEvidence.snapshot()
+    :ok = ConflictEvidence.reset()
+    [] = ConflictEvidence.snapshot()
+    "" = File.read!(path)
+    GenServer.stop(evidence)
+    {:ok, evidence} = ConflictEvidence.start_link(conflict_evidence_path: path)
+    [] = ConflictEvidence.snapshot()
+
+    1 =
+      ConflictEvidence.record(%{
+        kind: :register,
+        token: "next-history",
+        key: 0,
+        cluster: nil,
+        revision: 1
+      })
+
+    :ok = ConflictEvidence.reset()
+    reset_events = ConflictEvidence.snapshot()
+    GenServer.stop(evidence)
+    Supervisor.stop(group)
+    {events, reset_events}
+  end
+
+  defp wait(fun, remaining \\ 200)
+  defp wait(_fun, 0), do: raise("probe timed out")
+
+  defp wait(fun, remaining) do
+    unless fun.() do
+      Process.sleep(5)
+      wait(fun, remaining - 1)
+    end
+  end
+end
+
+Group.Jepsen.ConflictProbe.run()

test/jepsen/decode_conflict_evidence.exs

@@ -0,0 +1,6 @@
+System.put_env("GROUP_JEPSEN_LIBRARY_ONLY", "1")
+Code.require_file("node.exs", __DIR__)
+
+[path] = System.argv()
+events = path |> File.read!() |> Group.Jepsen.ConflictEvidence.decode()
+IO.puts("CONFLICT-EVIDENCE " <> Group.Jepsen.EDN.encode(events))

test/jepsen/fixtures/applied-repair-events.edn

@@ -0,0 +1 @@
+{:applied-delta-run-records-peak 2 :multi-chunk-snapshot-committed 1}

test/jepsen/metadata_capture.exs

@@ -1,9 +1,17 @@
 System.put_env("GROUP_JEPSEN_LIBRARY", "1")
+Code.require_file("repair_coverage.exs", __DIR__)
 Code.require_file("node.exs", __DIR__)
 
 alias Group.Jepsen.{Driver, EDN, Snapshot}
 alias Group.Replica.Data
 
+repair_path =
+  Path.join(
+    System.tmp_dir!(),
+    "group-jepsen-metadata-repair-#{System.unique_integer([:positive])}"
+  )
+
+{:ok, _} = Group.Jepsen.RepairCoverage.start_link(path: repair_path)
 {:ok, _} = Group.Jepsen.Transport.Stats.start_link([])
 
 {:ok, _} =
@@ -74,3 +82,4 @@ Enum.each(originals, fn {table, rows} -> :ets.insert(table, rows) end)
 false = reincarnated.token == owner.token
 
 File.write!(hd(System.argv()), EDN.encode(%{healthy: healthy, corruptions: corruptions}))
+File.rm(repair_path)