From 655949f00504f8f97c24f60b174bbb186a1a6ae4 Mon Sep 17 00:00:00 2001 From: Jason Stiebs Date: Fri, 11 Sep 2026 12:41:34 -0500 Subject: [PATCH 1/2] Handle monitored owner deaths during Jepsen cleanup and snapshots --- test/jepsen/node.exs | 82 ++++++++-------- test/jepsen_cleanup_deaths_test.exs | 140 ++++++++++++++++++++++++++++ 2 files changed, 177 insertions(+), 45 deletions(-) create mode 100644 test/jepsen_cleanup_deaths_test.exs diff --git a/test/jepsen/node.exs b/test/jepsen/node.exs index 65bd92a..76db12b 100644 --- a/test/jepsen/node.exs +++ b/test/jepsen/node.exs @@ -508,7 +508,7 @@ defmodule Group.Jepsen.Owner do registrations = drop_cluster(state.registrations, cluster) memberships = drop_cluster(state.memberships, cluster) state = %{state | registrations: registrations, memberships: memberships} - {:reply, :ok, state} + {:reply, snapshot(state), state} end def handle_call(:snapshot, _from, state), do: {:reply, snapshot(state), state} @@ -582,7 +582,13 @@ defmodule Group.Jepsen.Driver do def kill(logical_owner), do: GenServer.call(driver(logical_owner), {:kill, logical_owner}) def drop_cluster(cluster) do - Enum.each(names(), &GenServer.call(&1, {:drop_cluster, cluster}, 10_000)) + Enum.each(names(), fn driver -> + case GenServer.call(driver, {:drop_cluster, cluster}, 30_000) do + :ok -> :ok + {:error, reason} -> raise "owner cluster cleanup failed: #{inspect(reason)}" + end + end) + :ok end @@ -661,55 +667,41 @@ defmodule Group.Jepsen.Driver do end def handle_call({:drop_cluster, cluster}, _from, state) do - Enum.each(state.owners, fn {_logical_owner, {pid, _token, _monitor, _owner_state}} -> - if Process.alive?(pid), do: GenServer.call(pid, {:drop_cluster, cluster}, 10_000) - end) - - owners = - Map.new(state.owners, fn {logical_owner, {pid, token, monitor, _owner_state}} -> - owner_state = if Process.alive?(pid), do: GenServer.call(pid, :snapshot), else: nil - {logical_owner, {pid, token, monitor, owner_state}} - end) - - {:reply, :ok, %{state | owners: owners}} + case refresh_owners(state, {:drop_cluster, cluster}) do + {:ok, _snapshots, state} -> {:reply, :ok, state} + {:error, reason, state} -> {:reply, {:error, reason}, state} + end end def handle_call(:owner_snapshots, _from, state) do - result = - Enum.reduce_while(state.owners, {[], %{}}, fn - {logical_owner, {pid, token, monitor_ref, cached}}, {snapshots, acc} -> - case live_owner_snapshot(pid) do - {:ok, owner_state} -> - {:cont, - { - [owner_state | snapshots], - Map.put(acc, logical_owner, {pid, token, monitor_ref, owner_state}) - }} - - {:error, reason} -> - {:halt, {:error, {logical_owner, token, reason, cached}}} - end - end) - - case result do - {:error, reason} -> - {:reply, {:error, reason}, state} - - {owners, refreshed} -> - {:reply, {:ok, Enum.reverse(owners)}, %{state | owners: refreshed}} + case refresh_owners(state, :snapshot) do + {:ok, snapshots, state} -> {:reply, {:ok, snapshots}, state} + {:error, reason, state} -> {:reply, {:error, reason}, state} end end - defp live_owner_snapshot(pid) do - if Process.alive?(pid) do - try do - {:ok, GenServer.call(pid, :snapshot, 10_000)} - catch - :exit, reason -> {:error, reason} - end - else - {:error, :not_alive} - end + # Cleanup returns its snapshot in the same Owner turn. There is no second + # unguarded call, and every confirmed death uses the normal monitor path. + defp refresh_owners(state, request) do + Enum.reduce_while(state.owners, {:ok, [], state}, fn + {logical_owner, {pid, token, monitor_ref, cached}}, {:ok, snapshots, state} -> + try do + owner_state = GenServer.call(pid, request, 10_000) + state = put_owner_state(state, logical_owner, pid, owner_state) + {:cont, {:ok, [owner_state | snapshots], state}} + catch + :exit, reason -> + # A timeout is not proof of death. Only the tracked monitor can + # remove this owner, preserving every unrelated owner and monitor. + receive do + {:DOWN, ^monitor_ref, :process, ^pid, _death_reason} = down -> + {:noreply, state} = handle_info(down, state) + {:cont, {:ok, snapshots, state}} + after + 0 -> {:halt, {:error, {logical_owner, token, reason, cached}, state}} + end + end + end) end def handle_call(:unexpected_deaths, _from, state) do diff --git a/test/jepsen_cleanup_deaths_test.exs b/test/jepsen_cleanup_deaths_test.exs new file mode 100644 index 0000000..9d3dfbc --- /dev/null +++ b/test/jepsen_cleanup_deaths_test.exs @@ -0,0 +1,140 @@ +defmodule Group.JepsenCleanupDeathsTest do + use ExUnit.Case, async: false + + @path Path.expand("jepsen/node.exs", __DIR__) + @logs Path.expand("../tmp/jepsen-cleanup", __DIR__) + File.mkdir_p!(@logs) + + unless Code.ensure_loaded?(Group.Jepsen.Owner) do + @path + |> File.read!() + |> String.replace("Group.Jepsen.Main.run(System.argv())", "") + # Isolate the executable's durable log paths, not its Owner/Driver code. + |> String.replace("/tmp/group-jepsen-", @logs <> "/group-jepsen-") + |> Code.compile_string(@path) + end + + setup 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"]}) + + for logical <- ["loser", "survivor"] do + assert %{status: :ok} = + GenServer.call(driver, {:mutate, :join, logical, nil, 0, 1}) + end + + owners = :sys.get_state(driver).owners + {loser, _, _, _} = owners["loser"] + {survivor, survivor_token, _, _} = owners["survivor"] + + on_exit(fn -> + for pid <- [loser, survivor], Process.alive?(pid), do: Process.exit(pid, :kill) + end) + + %{driver: driver, loser: loser, survivor: survivor, token: survivor_token} + end + + for request <- [{:drop_cluster, "red"}, :owner_snapshots] do + @request request + test "monitored conflict death during #{inspect(request)} preserves other owners", ctx do + :ok = :sys.suspend(ctx.loser) + task = Task.async(fn -> GenServer.call(ctx.driver, @request, 15_000) end) + expected = if @request == :owner_snapshots, do: :snapshot, else: @request + + Group.LocalCase.wait_until(fn -> + {:messages, messages} = Process.info(ctx.loser, :messages) + Enum.any?(messages, &match?({:"$gen_call", _, ^expected}, &1)) + end) + + Process.exit( + ctx.loser, + {:group_registry_conflict, "jepsen/registry/0", %{token: "winner", revision: 2}} + ) + + response = Task.await(task, 15_000) + + if @request == :owner_snapshots, + do: assert(match?({:ok, [_]}, response)), + else: assert(response == :ok) + + assert Process.alive?(ctx.driver) + assert Process.alive?(ctx.survivor) + + assert {:ok, [%{token: token, memberships: [_]}]} = + GenServer.call(ctx.driver, :owner_snapshots) + + assert token == ctx.token + state = :sys.get_state(ctx.driver) + assert Map.keys(state.owners) == ["survivor"] + assert map_size(state.monitors) == 1 + assert state.unexpected_deaths == [] + + Group.LocalCase.wait_until(fn -> + Group.members(:jepsen_group, "jepsen/pg/0") == + [{ctx.survivor, %{token: ctx.token, revision: 1}}] + end) + end + end + + test "death queued before snapshot is consumed without replacing the Driver", ctx do + :ok = :sys.suspend(ctx.driver) + task = Task.async(fn -> GenServer.call(ctx.driver, :owner_snapshots) end) + + Group.LocalCase.wait_until(fn -> + {:messages, messages} = Process.info(ctx.driver, :messages) + Enum.any?(messages, &match?({:"$gen_call", _, :owner_snapshots}, &1)) + end) + + ref = Process.monitor(ctx.loser) + + Process.exit( + ctx.loser, + {:group_registry_conflict, "jepsen/registry/0", %{token: "winner", revision: 2}} + ) + + assert_receive {:DOWN, ^ref, :process, _, _} + :ok = :sys.resume(ctx.driver) + assert {:ok, [%{token: token}]} = Task.await(task) + assert token == ctx.token + assert Process.alive?(ctx.driver) + end + + test "unexpected owner death still leaves failure evidence", ctx do + :ok = :sys.suspend(ctx.loser) + task = Task.async(fn -> GenServer.call(ctx.driver, {:drop_cluster, "red"}, 15_000) end) + + Group.LocalCase.wait_until(fn -> + {:messages, messages} = Process.info(ctx.loser, :messages) + Enum.any?(messages, &match?({:"$gen_call", _, {:drop_cluster, "red"}}, &1)) + end) + + Process.exit(ctx.loser, :implementation_bug) + assert :ok = Task.await(task, 15_000) + + assert [%{reason: ":implementation_bug"}] = + GenServer.call(ctx.driver, :unexpected_deaths) + + assert Process.alive?(ctx.survivor) + assert Process.alive?(ctx.driver) + end + + test "a live owner's timeout is reported without deleting any owner state", ctx do + :ok = :sys.suspend(ctx.loser) + + assert {:error, {"loser", _, {:timeout, _}, _}} = + GenServer.call(ctx.driver, {:drop_cluster, "red"}, 15_000) + + assert Process.alive?(ctx.driver) + assert Process.alive?(ctx.loser) + assert Process.alive?(ctx.survivor) + state = :sys.get_state(ctx.driver) + assert map_size(state.owners) == 2 + assert map_size(state.monitors) == 2 + assert state.unexpected_deaths == [] + :ok = :sys.resume(ctx.loser) + assert {:ok, snapshots} = GenServer.call(ctx.driver, :owner_snapshots) + assert length(snapshots) == 2 + end +end From cf05e96638cec14b3be4ddee443210bb56608bed Mon Sep 17 00:00:00 2001 From: Jason Stiebs Date: Fri, 11 Sep 2026 16:12:59 -0500 Subject: [PATCH 2/2] Isolate Jepsen cleanup harness loading and runtime log paths --- test/jepsen/harness_modules.exs | 14 ++++++++++++++ test/jepsen/node.exs | 22 ++++++++++++++-------- test/jepsen_cleanup_deaths_test.exs | 25 ++++++++++++------------- 3 files changed, 40 insertions(+), 21 deletions(-) create mode 100644 test/jepsen/harness_modules.exs diff --git a/test/jepsen/harness_modules.exs b/test/jepsen/harness_modules.exs new file mode 100644 index 0000000..4b46a6a --- /dev/null +++ b/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) diff --git a/test/jepsen/node.exs b/test/jepsen/node.exs index 76db12b..7a60950 100644 --- a/test/jepsen/node.exs +++ b/test/jepsen/node.exs @@ -19,7 +19,7 @@ defmodule Group.Jepsen.Transport.Stats do def increment_persistent(event, amount \\ 1) do increment(event, amount) - File.write!(@persistent_event_log, "#{event}\t#{amount}\n", [:append]) + File.write!(persistent_event_log(), "#{event}\t#{amount}\n", [:append]) :ok end @@ -52,8 +52,11 @@ defmodule Group.Jepsen.Transport.Stats do {:ok, %{}} end + defp persistent_event_log, + do: System.get_env("GROUP_JEPSEN_PERSISTENT_EVENT_LOG", @persistent_event_log) + defp persistent_events do - case File.read(@persistent_event_log) do + case File.read(persistent_event_log()) do {:ok, contents} -> contents |> String.split("\n", trim: true) @@ -680,6 +683,10 @@ defmodule Group.Jepsen.Driver do end end + def handle_call(:unexpected_deaths, _from, state) do + {:reply, state.unexpected_deaths, state} + end + # Cleanup returns its snapshot in the same Owner turn. There is no second # unguarded call, and every confirmed death uses the normal monitor path. defp refresh_owners(state, request) do @@ -704,10 +711,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 @@ -781,11 +784,14 @@ defmodule Group.Jepsen.Driver do 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]) + 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) diff --git a/test/jepsen_cleanup_deaths_test.exs b/test/jepsen_cleanup_deaths_test.exs index 9d3dfbc..2194f97 100644 --- a/test/jepsen_cleanup_deaths_test.exs +++ b/test/jepsen_cleanup_deaths_test.exs @@ -1,20 +1,19 @@ defmodule Group.JepsenCleanupDeathsTest do use ExUnit.Case, async: false + @moduletag :local + @moduletag :tmp_dir + Code.require_file("jepsen/harness_modules.exs", __DIR__) - @path Path.expand("jepsen/node.exs", __DIR__) - @logs Path.expand("../tmp/jepsen-cleanup", __DIR__) - File.mkdir_p!(@logs) - - unless Code.ensure_loaded?(Group.Jepsen.Owner) do - @path - |> File.read!() - |> String.replace("Group.Jepsen.Main.run(System.argv())", "") - # Isolate the executable's durable log paths, not its Owner/Driver code. - |> String.replace("/tmp/group-jepsen-", @logs <> "/group-jepsen-") - |> Code.compile_string(@path) - end + setup %{tmp_dir: tmp_dir} do + for variable <- ["GROUP_JEPSEN_UNEXPECTED_DEATH_LOG", "GROUP_JEPSEN_PERSISTENT_EVENT_LOG"] do + previous = System.get_env(variable) + System.put_env(variable, Path.join(tmp_dir, variable)) + + on_exit(fn -> + if previous, do: System.put_env(variable, previous), else: System.delete_env(variable) + end) + end - setup do start_supervised!({Group, name: :jepsen_group, shards: 1, log: false}) driver =