Validate Jepsen conflict deaths against independent claim evidence

mix.exs

@@ -35,7 +35,7 @@ defmodule Group.MixProject do
   end
 
   def cli do
-    [preferred_envs: ["test.soak": :test]]
+    [preferred_envs: ["test.exunit": :test, "test.soak": :test]]
   end
 
   defp deps do
@@ -64,7 +64,10 @@ 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"],
+      # Mutation targets must measure ExUnit, not the independent JVM gate.
+      # Invoke the task module directly so the `test` alias is not expanded.
+      "test.exunit": [&Mix.Tasks.Test.run/1],
       "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

test/jepsen/README.md

@@ -32,7 +32,8 @@ The replica lane is selectable without changing the workload or checker:
 After faults stop, every surviving node reconnects and the harness takes two
 terminal snapshots. The independent checker requires:
 
-- exact, identical public registry and PG views on every survivor;
+- exact, identical public registry and PG metadata on every survivor, including
+  each revision acknowledged by its owner;
 - every live owner claim to be visible, and no dead owner token to remain;
 - deterministic resolution of registry conflicts with no unexpected owner
   deaths;
@@ -55,9 +56,48 @@ 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.
+
 ## Requirements
 
 - Docker with Compose v2
+- Elixir/Mix with the repository dependencies installed (checker qualification)
 - Java 21 or newer
 - `curl`
 
@@ -109,9 +149,18 @@ count, concurrency, keys, owners, and recovery time.
 Run mutation qualification plus live positive- and negative-checker tests:
 
 ```bash
-test/jepsen/qualify.sh
+elixir test/jepsen/qualify.exs
 ```
 
+`qualify.sh` remains a thin compatibility launcher for the same Elixir script.
+The script owns artifact creation, command sequencing, and result validation.
+Each live run still uses GNU `timeout` with a three-minute deadline and a
+30-second TERM/KILL grace period, and streams its output to a separate log.
+Qualification requires both the expected exit status and a fresh checker record
+confirming that the intended corruption was actually detected; a failed command
+alone never counts. Executable ExUnit regressions exercise this runner with
+stubbed external commands as part of normal `mix test`, without Docker.
+
 This runs every mutation defined by `test/mutation/run.exs`, then verifies that
 a healthy live history is accepted and deliberately injected owner-death,
 internal-index, stranded snapshot-cursor, registry claim/projection, and
@@ -135,7 +184,16 @@ 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 capture qualification mutates real owners repeatedly, corrupts materialized
+metadata while preserving internal index consistency, and passes the real
+snapshot EDN to the independent checker. Public values retain metadata maps
+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.
+
+`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

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/invariant_qualification_test.exs

@@ -0,0 +1,90 @@
+# MIX_ENV=test mix run --no-start test/jepsen/invariant_qualification_test.exs
+# Compile only the actual oracle modules, not the node entrypoint. Redirect
+# the injection marker into the workspace so this probe needs no Docker or
+# machine-global files.
+ExUnit.start()
+
+defmodule Group.Jepsen.InvariantQualificationTest do
+  use ExUnit.Case, async: false
+
+  alias Group.Jepsen.Invariant
+  alias Group.Replica.Data
+
+  @compile {:no_warn_undefined, [Group.Jepsen.Invariant, Group.Jepsen.EDN]}
+
+  setup_all do
+    work =
+      Path.expand(
+        "../../tmp/invariant-qualification-#{System.unique_integer([:positive])}",
+        __DIR__
+      )
+
+    File.mkdir_p!(work)
+    marker = Path.join(work, "cursor-marker")
+    on_exit(fn -> File.rm_rf!(work) end)
+    {:ok, _} = Application.ensure_all_started(:group)
+
+    {:__block__, _, expressions} =
+      __DIR__
+      |> Path.join("node.exs")
+      |> File.read!()
+      |> Code.string_to_quoted!()
+
+    modules =
+      Enum.filter(expressions, fn
+        {:defmodule, _, [{:__aliases__, _, [:Group, :Jepsen, name]}, _]} ->
+          name in [:InvariantViolation, :Invariant, :ConflictResolver, :EDN]
+
+        _ ->
+          false
+      end)
+
+    ast =
+      Macro.prewalk({:__block__, [], modules}, fn
+        "/tmp/group-jepsen-cursor-marker-corruption" -> marker
+        node -> node
+      end)
+
+    Code.compile_quoted(ast)
+    {:ok, marker: marker}
+  end
+
+  setup %{marker: marker} do
+    File.rm(marker)
+    start_supervised!({Group, name: :jepsen_group, shards: 1, log: false})
+    :ok
+  end
+
+  test "arming with no remote cursor reports no injection or invariant evidence", %{
+    marker: marker
+  } do
+    File.write!(marker, "enabled\n")
+    snapshot = Invariant.snapshot([])
+    refute snapshot.healthy
+    assert snapshot.snapshot_staging_count == -1
+    assert snapshot.injected_corruptions == []
+    assert snapshot.failed_invariants == []
+  end
+
+  test "a real marker insertion reports its exact invariant", %{marker: marker} do
+    File.write!(marker, "enabled\n")
+    :ets.insert(Data.replica_cursor_table(:jepsen_group, 0), {:probe_stream, 1})
+    snapshot = Invariant.snapshot([])
+    refute snapshot.healthy
+    assert snapshot.injected_corruptions == [:"cursor-marker"]
+    assert snapshot.failed_invariants == [:cursor_snapshot_marker]
+    assert Group.Jepsen.EDN.encode(snapshot) =~ ":cursor-snapshot-marker"
+  end
+
+  test "an unrelated index failure cannot masquerade as a cursor marker" do
+    :ets.insert(
+      Data.reg_by_pid_table(:jepsen_group, 0),
+      {{self(), nil, "qualification-probe"}, %{}, 0, node()}
+    )
+
+    snapshot = Invariant.snapshot([])
+    refute snapshot.healthy
+    assert snapshot.injected_corruptions == []
+    assert snapshot.failed_invariants == [:registry_dual_indexes]
+  end
+end