Preserve unexpected Jepsen operation failures

test/jepsen/harness_modules.exs

@@ -0,0 +1,14 @@
+# Require once so independent harness regressions share the same real modules.
+# Keep top-level executable startup, transport adapters, and unrelated modules
+# out of the test VM. Log paths are configured at runtime, never rewritten here.
+path = Path.join(__DIR__, "node.exs")
+{:__block__, metadata, forms} = path |> File.read!() |> Code.string_to_quoted!()
+modules = [Group.Jepsen.Transport.Stats, Group.Jepsen.Owner, Group.Jepsen.Driver]
+
+forms =
+  Enum.filter(forms, fn
+    {:defmodule, _, [{:__aliases__, _, parts}, _]} -> Module.concat(parts) in modules
+    _ -> false
+  end)
+
+Code.compile_quoted({:__block__, metadata, forms}, path)

test/jepsen/node.exs

@@ -428,17 +428,18 @@ defmodule Group.Jepsen.Owner do
   @moduledoc false
   use GenServer
 
-  def start(token), do: GenServer.start(__MODULE__, token)
+  def start(token, api \\ Group), do: GenServer.start(__MODULE__, {token, api})
 
   @impl true
-  def init(token), do: {:ok, %{token: token, registrations: %{}, memberships: %{}}}
+  def init({token, api}),
+    do: {:ok, %{token: token, api: api, registrations: %{}, memberships: %{}}}
 
   @impl true
   def handle_call({:mutate, :register, cluster, key, revision}, _from, state) do
     meta = %{token: state.token, revision: revision}
 
-    case safe_group_call(fn ->
-           Group.register(:jepsen_group, registry_key(key), meta, cluster_opts(cluster))
+    case safe_group_call(state.token, fn ->
+           state.api.register(:jepsen_group, registry_key(key), meta, cluster_opts(cluster))
          end) do
       :ok ->
         entry = %{cluster: cluster, key: key, revision: revision}
@@ -454,8 +455,8 @@ defmodule Group.Jepsen.Owner do
     owner_key = {cluster, key}
 
     if Map.has_key?(state.registrations, owner_key) do
-      case safe_group_call(fn ->
-             Group.unregister(:jepsen_group, registry_key(key), cluster_opts(cluster))
+      case safe_group_call(state.token, fn ->
+             state.api.unregister(:jepsen_group, registry_key(key), cluster_opts(cluster))
            end) do
         :ok ->
           state = %{state | registrations: Map.delete(state.registrations, owner_key)}
@@ -472,8 +473,8 @@ defmodule Group.Jepsen.Owner do
   def handle_call({:mutate, :join, cluster, key, revision}, _from, state) do
     meta = %{token: state.token, revision: revision}
 
-    case safe_group_call(fn ->
-           Group.join(:jepsen_group, pg_key(key), meta, cluster_opts(cluster))
+    case safe_group_call(state.token, fn ->
+           state.api.join(:jepsen_group, pg_key(key), meta, cluster_opts(cluster))
          end) do
       :ok ->
         entry = %{cluster: cluster, key: key, revision: revision}
@@ -489,8 +490,8 @@ defmodule Group.Jepsen.Owner do
     owner_key = {cluster, key}
 
     if Map.has_key?(state.memberships, owner_key) do
-      case safe_group_call(fn ->
-             Group.leave(:jepsen_group, pg_key(key), cluster_opts(cluster))
+      case safe_group_call(state.token, fn ->
+             state.api.leave(:jepsen_group, pg_key(key), cluster_opts(cluster))
            end) do
         :ok ->
           state = %{state | memberships: Map.delete(state.memberships, owner_key)}
@@ -529,12 +530,49 @@ defmodule Group.Jepsen.Owner do
 
   defp sort_entries(entries), do: Enum.sort_by(entries, &{&1.cluster || "", &1.key})
 
-  defp safe_group_call(fun) do
-    fun.()
+  defp safe_group_call(token, fun) do
+    case fun.() do
+      :ok ->
+        :ok
+
+      {:error, code}
+      when code in [:taken, :undefined, :not_owner, :not_in_group, :stale_cluster_epoch] ->
+        {:error, code}
+
+      other ->
+        unexpected(token, :return, other, [])
+    end
   rescue
-    exception -> {:error, {:exception, Exception.message(exception)}}
+    exception in ArgumentError ->
+      case __STACKTRACE__ do
+        [{Group, :validate_cluster_connected!, _, _} | _] ->
+          {:error, :not_connected}
+
+        stack ->
+          unexpected(token, :exception, exception, stack)
+      end
+
+    exception ->
+      unexpected(token, :exception, exception, __STACKTRACE__)
   catch
-    kind, reason -> {:error, {kind, reason}}
+    :exit, {reason, {GenServer, :call, _}}
+    when reason in [:timeout, :noproc, :normal, :shutdown] ->
+      {:error, {:indeterminate, reason}}
+
+    kind, reason ->
+      unexpected(token, kind, reason, __STACKTRACE__)
+  end
+
+  defp unexpected(token, kind, reason, stack) do
+    evidence = %{kind: kind, reason: inspect(reason), stack: inspect(stack)}
+
+    :ok =
+      Group.Jepsen.Driver.persist_unexpected_death(%{
+        token: token,
+        reason: "operation failure: " <> inspect(evidence)
+      })
+
+    {:error, {:unexpected, evidence}}
   end
 
   defp cluster_opts(nil), do: []
@@ -611,6 +649,7 @@ defmodule Group.Jepsen.Driver do
      %{
        node_id: Keyword.fetch!(opts, :node_id),
        boot_id: Keyword.fetch!(opts, :boot_id),
+       api: Keyword.get(opts, :api, Group),
        owners: %{},
        monitors: %{},
        incarnations: %{},
@@ -629,12 +668,12 @@ defmodule Group.Jepsen.Driver do
            put_owner_state(state, logical_owner, pid, owner_state)}
 
         {:error, reason, owner_state} ->
-          {:reply, %{status: :fail, error: inspect(reason), owner: owner_state},
-           put_owner_state(state, logical_owner, pid, owner_state)}
+          response = Map.merge(failure_response(reason), %{owner: owner_state})
+          {:reply, response, put_owner_state(state, logical_owner, pid, owner_state)}
       end
     catch
       :exit, reason ->
-        {:reply, %{status: :unknown, error: inspect(reason)}, state}
+        {:reply, %{status: :unknown, code: :indeterminate, error: inspect(reason)}, state}
     end
   end
 
@@ -700,6 +739,28 @@ defmodule Group.Jepsen.Driver do
     end
   end
 
+  def handle_call(:unexpected_deaths, _from, state) do
+    {:reply, state.unexpected_deaths, state}
+  end
+
+  defp failure_response({:unexpected, evidence}),
+    do: %{status: :fail, code: :unexpected, error: evidence}
+
+  defp failure_response({:indeterminate, reason}),
+    do: %{status: :unknown, code: :indeterminate, error: inspect(reason)}
+
+  defp failure_response(code)
+       when code in [
+              :taken,
+              :undefined,
+              :not_owner,
+              :not_owned,
+              :not_connected,
+              :not_in_group,
+              :stale_cluster_epoch
+            ],
+       do: %{status: :fail, code: code, error: inspect(code)}
+
   defp live_owner_snapshot(pid) do
     if Process.alive?(pid) do
       try do
@@ -712,10 +773,6 @@ defmodule Group.Jepsen.Driver do
     end
   end
 
-  def handle_call(:unexpected_deaths, _from, state) do
-    {:reply, state.unexpected_deaths, state}
-  end
-
   @impl true
   def handle_info({:DOWN, monitor_ref, :process, _pid, reason}, state) do
     case Map.pop(state.monitors, monitor_ref) do
@@ -757,7 +814,7 @@ defmodule Group.Jepsen.Driver do
   defp start_owner(state, logical_owner) do
     incarnation = Map.get(state.incarnations, logical_owner, 0) + 1
     token = "#{state.node_id}/#{state.boot_id}/#{logical_owner}/#{incarnation}"
-    {:ok, pid} = Group.Jepsen.Owner.start(token)
+    {:ok, pid} = Group.Jepsen.Owner.start(token, state.api)
     monitor_ref = Process.monitor(pid)
     owner_state = %{token: token, registrations: [], memberships: []}
 
@@ -788,12 +845,15 @@ defmodule Group.Jepsen.Driver do
   defp driver(logical_owner), do: name(:erlang.phash2(logical_owner, @driver_count))
   defp name(index), do: :"group_jepsen_driver_#{index}"
 
-  defp persist_unexpected_death(%{token: token, reason: reason}) do
-    File.write(@unexpected_death_log, token <> "\t" <> reason <> "\n", [:append])
+  def persist_unexpected_death(%{token: token, reason: reason}) do
+    File.write(unexpected_death_log(), token <> "\t" <> reason <> "\n", [:append])
   end
 
+  defp unexpected_death_log,
+    do: System.get_env("GROUP_JEPSEN_UNEXPECTED_DEATH_LOG", @unexpected_death_log)
+
   defp persisted_unexpected_deaths do
-    case File.read(@unexpected_death_log) do
+    case File.read(unexpected_death_log()) do
       {:ok, contents} ->
         contents
         |> String.split("\n", trim: true)

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

@@ -96,6 +96,12 @@
        (remove history/invoke?)
        (keep #(get-in % [:value :response :latency-us]))))
 
+(defn unexpected-operation-failures [history]
+  (->> history
+       (remove history/invoke?)
+       (filter #(= :unexpected (get-in % [:value :response :code])))
+       vec))
+
 (defn analyze [test history]
   (let [observations (snapshots-by-node history)
         snapshots (latest-snapshots history)
@@ -169,6 +175,7 @@
                           [node internal]))))
               relevant-snapshots)
         unexpected-deaths (->> relevant-snapshots vals (mapcat :unexpected-deaths) set)
+        operation-failures (unexpected-operation-failures history)
         live-tokens (set (keys (:owners expected)))
         actual-tokens (->> views
                            vals
@@ -198,6 +205,7 @@
                     (empty? (:conflicts expected))
                     (empty? mismatches)
                     (empty? unexpected-deaths)
+                    (empty? operation-failures)
                     (empty? orphaned)
                     (empty? missing-live)
                     (not latency-violation?))]
@@ -219,6 +227,7 @@
      :live-registry-conflicts (:conflicts expected)
      :mismatched-views mismatches
      :unexpected-owner-deaths unexpected-deaths
+     :unexpected-operation-failures operation-failures
      :orphaned-owner-tokens orphaned
      :missing-live-owner-tokens missing-live
      :expected expected-view}))

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

@@ -94,6 +94,23 @@
                  (snapshot-op 2 "n3" survivors [] (empty-registry) (empty-pg))]]
     (is (:valid? (model/analyze permanent-test history)))))
 
+(deftest operation-failures-survive-owner-and-vm-retirement
+  (let [survivors ["n2" "n3"]
+        test (assoc test-map :terminal-nodes survivors)
+        snapshots [(snapshot-op 2 "n2" survivors [] (empty-registry) (empty-pg))
+                   (snapshot-op 3 "n3" survivors [] (empty-registry) (empty-pg))]
+        failure (fn [code error]
+                  {:index 1 :process 0 :type :fail :f :register
+                   :value {:node "n1" :response {:code code :error error}}})]
+    (doseq [code [:taken :undefined :not-owner :not-owned :not-in-group
+                 :stale-cluster-epoch :indeterminate]]
+      (is (:valid? (model/analyze test (cons (failure code "expected") snapshots)))))
+    (doseq [kind [:exception :throw :error :return]]
+      (let [op (failure :unexpected {:kind kind :reason "implementation bug"})
+            result (model/analyze test (cons op snapshots))]
+        (is (false? (:valid? result)))
+        (is (= [op] (:unexpected-operation-failures result)))))))
+
 (deftest rejects-zombies-missing-live-owners-and-divergence
   (let [live (owner "live" [(registration nil 0 1)] [])
         stale-registry (assoc-in (empty-registry) ["root" 0] "dead")

test/jepsen_failure_classification_test.exs

@@ -0,0 +1,81 @@
+defmodule Group.JepsenFailureClassificationTest do
+  use ExUnit.Case, async: false
+  @moduletag :local
+  Code.require_file("jepsen/harness_modules.exs", __DIR__)
+
+  defmodule API do
+    def register(_, _, %{revision: revision}, _) do
+      case revision do
+        0 -> {:error, :taken}
+        1 -> raise "implementation bug"
+        2 -> throw(:implementation_bug)
+        3 -> :erlang.error(:implementation_bug)
+        4 -> exit({:timeout, {GenServer, :call, [:replica, :write, 5_000]}})
+        5 -> {:error, :invented_error}
+      end
+    end
+  end
+
+  test "actual Group admission and ownership faults retain stable codes" do
+    start_supervised!({Group, name: :jepsen_group, shards: 1, log: false})
+
+    driver =
+      start_supervised!({Group.Jepsen.Driver, [index: 0, node_id: "n1", boot_id: "boot"]})
+
+    assert %{status: :fail, code: :not_connected} =
+             GenServer.call(driver, {:mutate, :register, "one", "closed", 0, 0})
+
+    assert %{status: :fail, code: :not_owned} =
+             GenServer.call(driver, {:mutate, :unregister, "one", nil, 0, 0})
+
+    assert %{status: :ok} =
+             GenServer.call(driver, {:mutate, :register, "one", nil, 0, 0})
+
+    assert %{status: :fail, code: :taken} =
+             GenServer.call(driver, {:mutate, :register, "two", nil, 0, 0})
+
+    for owner <- ["one", "two"], do: GenServer.call(driver, {:kill, owner})
+  end
+
+  @tag :tmp_dir
+  test "actual Owner/Driver boundary preserves unexpected failures independently of owner life",
+       %{tmp_dir: tmp_dir} do
+    old = System.get_env("GROUP_JEPSEN_UNEXPECTED_DEATH_LOG")
+    path = Path.join(tmp_dir, "evidence")
+    System.put_env("GROUP_JEPSEN_UNEXPECTED_DEATH_LOG", path)
+
+    on_exit(fn ->
+      if old,
+        do: System.put_env("GROUP_JEPSEN_UNEXPECTED_DEATH_LOG", old),
+        else: System.delete_env("GROUP_JEPSEN_UNEXPECTED_DEATH_LOG")
+    end)
+
+    driver =
+      start_supervised!(
+        {Group.Jepsen.Driver, [index: 0, node_id: "n1", boot_id: "boot", api: API]}
+      )
+
+    mutate = fn revision ->
+      GenServer.call(driver, {:mutate, :register, "owner", nil, 0, revision})
+    end
+
+    assert %{status: :fail, code: :taken} = mutate.(0)
+    assert %{status: :unknown, code: :indeterminate} = mutate.(4)
+    refute File.exists?(path)
+
+    for revision <- [1, 2, 3, 5] do
+      assert %{status: :fail, code: :unexpected, error: %{reason: reason}} = mutate.(revision)
+      assert reason != ""
+    end
+
+    assert :ok =
+             GenServer.call(driver, {:kill, "owner"})
+             |> then(fn
+               %{status: :ok} -> :ok
+             end)
+
+    assert Process.alive?(driver)
+    assert length(File.read!(path) |> String.split("\n", trim: true)) == 4
+    assert {:ok, []} = GenServer.call(driver, :owner_snapshots)
+  end
+end