Require completed checker evidence for Jepsen qualification

test/jepsen/README.md

@@ -109,9 +109,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

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

test/jepsen/node.exs

@@ -860,6 +860,10 @@ defmodule Group.Jepsen.Cluster do
   defp result({:error, reason}), do: %{status: :fail, error: inspect(reason)}
 end
 
+defmodule Group.Jepsen.InvariantViolation do
+  defexception [:message, :invariant]
+end
+
 defmodule Group.Jepsen.Invariant do
   @moduledoc false
 
@@ -868,9 +872,9 @@ defmodule Group.Jepsen.Invariant do
   def snapshot(retired_nodes) do
     config = Group.get_config(:jepsen_group)
     shards = 0..(config.num_shards - 1)
-    maybe_inject_cursor_marker_corruption(shards)
+    injected_corruptions = maybe_inject_cursor_marker_corruption(shards)
 
-    errors =
+    failures =
       check("dual indexes", &assert_dual_indexes/0) ++
         check("registry claims", &assert_registry_claims/0) ++
         check("oplog", &assert_oplogs/0) ++
@@ -889,8 +893,10 @@ defmodule Group.Jepsen.Invariant do
       end)
 
     %{
-      healthy: errors == [] and staging_count == 0,
-      errors: errors,
+      healthy: failures == [] and staging_count == 0,
+      errors: Enum.map(failures, & &1.message),
+      failed_invariants: Enum.flat_map(failures, &List.wrap(&1.invariant)),
+      injected_corruptions: injected_corruptions,
       snapshot_staging_count: staging_count,
       oplog_entries: oplog_entries,
       oplog_max_entries_per_shard: config.replicated_oplog_max_entries,
@@ -905,6 +911,8 @@ defmodule Group.Jepsen.Invariant do
       %{
         healthy: false,
         errors: ["invariant snapshot failed: #{Exception.message(exception)}"],
+        failed_invariants: [],
+        injected_corruptions: [],
         snapshot_staging_count: -1
       }
   end
@@ -928,19 +936,27 @@ defmodule Group.Jepsen.Invariant do
             {stream, {:snapshot_installing, 1}}
           )
 
+          [:"cursor-marker"]
+
         nil ->
           raise "no remote replica cursor available for corruption"
       end
+    else
+      []
     end
   end
 
   defp check(label, fun) do
     fun.()
     []
   rescue
-    exception -> ["#{label}: #{Exception.message(exception)}"]
+    exception in Group.Jepsen.InvariantViolation ->
+      [%{invariant: exception.invariant, message: "#{label}: #{Exception.message(exception)}"}]
+
+    exception ->
+      [%{invariant: nil, message: "#{label}: #{Exception.message(exception)}"}]
   catch
-    kind, reason -> ["#{label}: #{inspect({kind, reason})}"]
+    kind, reason -> [%{invariant: nil, message: "#{label}: #{inspect({kind, reason})}"}]
   end
 
   defp assert_dual_indexes do
@@ -973,7 +989,13 @@ defmodule Group.Jepsen.Invariant do
           {cluster, key, pid, meta, time, origin}
         end)
 
-      assert_equal!(reg_key, reg_pid, "registry dual indexes shard #{shard}")
+      assert_equal!(
+        reg_key,
+        reg_pid,
+        "registry dual indexes shard #{shard}",
+        :registry_dual_indexes
+      )
+
       assert_equal!(pg_key, pg_pid, "PG dual indexes shard #{shard}")
 
       expected_counts =
@@ -1074,7 +1096,7 @@ defmodule Group.Jepsen.Invariant do
           {cluster, key, pid, meta, time, origin}
         end)
 
-      assert_equal!(expected, visible, "registry projection shard #{shard}")
+      assert_equal!(expected, visible, "registry projection shard #{shard}", :registry_projection)
     end)
   end
 
@@ -1140,6 +1162,12 @@ defmodule Group.Jepsen.Invariant do
       Data.replica_cursor_table(:jepsen_group, shard)
       |> :ets.tab2list()
       |> Enum.each(fn {stream, seq} ->
+        if match?({:snapshot_installing, _}, seq) do
+          raise Group.Jepsen.InvariantViolation,
+            invariant: :cursor_snapshot_marker,
+            message: "cursor contains uncommitted snapshot marker #{inspect({stream, seq})}"
+        end
+
         origin = WireProtocol.stream_origin(stream)
         cluster = WireProtocol.stream_cluster(stream)
 
@@ -1202,10 +1230,13 @@ defmodule Group.Jepsen.Invariant do
     Enum.each(0..(num_shards - 1), fun)
   end
 
-  defp assert_equal!(left, right, label) do
+  defp assert_equal!(left, right, label, invariant \\ nil) do
     if left != right do
-      raise "#{label}: left-only=#{inspect(MapSet.difference(left, right))} " <>
-              "right-only=#{inspect(MapSet.difference(right, left))}"
+      raise Group.Jepsen.InvariantViolation,
+        invariant: invariant,
+        message:
+          "#{label}: left-only=#{inspect(MapSet.difference(left, right))} " <>
+            "right-only=#{inspect(MapSet.difference(right, left))}"
     end
   end
 
@@ -1439,7 +1470,7 @@ defmodule Group.Jepsen.Wire do
   defp corrupt("internal-index") do
     table = Group.Replica.Data.reg_by_pid_table(:jepsen_group, 0)
     :ets.insert(table, {{self(), nil, "jepsen/registry/corrupt"}, %{}, 0, node()})
-    %{status: :ok}
+    %{status: :ok, injected: :"internal-index"}
   end
 
   defp corrupt("cursor-marker") do
@@ -1483,7 +1514,7 @@ defmodule Group.Jepsen.Wire do
       end)
 
     if corrupted do
-      %{status: :ok}
+      %{status: :ok, injected: :"registry-projection"}
     else
       %{status: :fail, error: "no visible registry claim available for corruption"}
     end

test/jepsen/qualify.exs

@@ -0,0 +1,99 @@
+defmodule Group.Jepsen.Qualification do
+  @moduledoc false
+
+  # An intentionally corrupted history must be invalid, but still demonstrate
+  # that its intended corruption was injected and detected by the checker.
+  @checks [
+    {"none", true},
+    {"unexpected-death", false},
+    {"internal-index", false},
+    {"cursor-marker", false},
+    {"registry-projection", false},
+    {"terminal-unavailable", false}
+  ]
+
+  def run do
+    repo = Path.expand("../..", __DIR__)
+    cache = Path.join(__DIR__, ".cache")
+    File.mkdir_p!(cache)
+    suffix = Base.url_encode64(:crypto.strong_rand_bytes(12), padding: false)
+    artifacts = Path.join(cache, "qualification.#{suffix}")
+    File.mkdir!(artifacts)
+    File.chmod!(artifacts, 0o700)
+    IO.puts("qualification artifacts: #{artifacts}")
+
+    {_output, status} =
+      System.cmd("mix", ["run", "test/mutation/run.exs"],
+        cd: repo,
+        into: IO.stream(),
+        stderr_to_stdout: true
+      )
+
+    if status != 0, do: System.halt(status)
+
+    Enum.each(@checks, fn {corruption, expected_valid?} ->
+      qualify!(repo, artifacts, corruption, expected_valid?)
+    end)
+
+    IO.puts("mutation and live checker qualification passed")
+  end
+
+  defp qualify!(repo, artifacts, corruption, expected_valid?) do
+    log = Path.join(artifacts, "#{corruption}.log")
+    result = Path.join(artifacts, "#{corruption}.result")
+
+    # Keep the existing process-tree deadline and TERM/KILL grace period.
+    # Streaming output to disk avoids retaining a live history on this VM's heap.
+    {_output, status} =
+      System.cmd(
+        "timeout",
+        [
+          "--signal=TERM",
+          "--kill-after=30",
+          "180",
+          Path.join(__DIR__, "run.sh"),
+          "test",
+          "--no-ssh",
+          "--nodes",
+          "n1,n2,n3",
+          "--concurrency",
+          "2n",
+          "--time-limit",
+          "6",
+          "--fault-interval",
+          "1",
+          "--recovery-time",
+          "5",
+          "--transport",
+          "distribution",
+          "--scenario",
+          "mixed",
+          "--corruption",
+          corruption
+        ],
+        cd: repo,
+        env: [
+          {"GROUP_JEPSEN_SKIP_CHECKER", "1"},
+          {"GROUP_JEPSEN_QUALIFICATION_RESULT", result}
+        ],
+        into: File.stream!(log, [:write, :binary]),
+        stderr_to_stdout: true
+      )
+
+    expected_status = if expected_valid?, do: 0, else: 1
+
+    unless status == expected_status do
+      raise "#{corruption}: expected exit #{expected_status}, got #{status}; see #{log}"
+    end
+
+    expected_record = "group-qualification-v1\t#{corruption}\t#{expected_valid?}\ttrue\n"
+
+    unless File.read!(result) == expected_record do
+      raise "#{corruption}: missing or mismatched checker evidence; see #{result} and #{log}"
+    end
+
+    IO.puts("qualified #{corruption} (#{log})")
+  end
+end
+
+Group.Jepsen.Qualification.run()