diff --git a/lib/durable_server/lifecycle_manager.ex b/lib/durable_server/lifecycle_manager.ex index 7dc521c..24e392b 100644 --- a/lib/durable_server/lifecycle_manager.ex +++ b/lib/durable_server/lifecycle_manager.ex @@ -2076,7 +2076,7 @@ defmodule DurableServer.LifecycleManager do :skip -> # permanently non-restartable — cache without meta now = System.monotonic_time(:millisecond) - :ets.insert(state.discovery_skip_table, {key, etag, :skip, now}) + skip_cache_put(state, {key, etag, :skip, now}) clear_restart_gate_state(state, key) :noop @@ -2085,7 +2085,7 @@ defmodule DurableServer.LifecycleManager do # health check) — cache trimmed meta for re-evaluation next round now = System.monotonic_time(:millisecond) trimmed = trim_meta_for_cache(meta) - :ets.insert(state.discovery_skip_table, {key, etag, trimmed, now}) + skip_cache_put(state, {key, etag, trimmed, now}) clear_restart_gate_state(state, key) :noop @@ -2125,6 +2125,15 @@ defmodule DurableServer.LifecycleManager do :ok end + defp skip_cache_put(%LifecycleManager{} = state, entry) do + case :ets.whereis(state.discovery_skip_table) do + :undefined -> :ok + _ -> :ets.insert(state.discovery_skip_table, entry) + end + rescue + ArgumentError -> :ok + end + defp discovery_diag_snapshot(%LifecycleManager{} = state) do case :ets.whereis(state.discovery_diag_table) do :undefined -> %{} diff --git a/test/durable_server/lifecycle_test.exs b/test/durable_server/lifecycle_test.exs index 3285bf2..febc018 100644 --- a/test/durable_server/lifecycle_test.exs +++ b/test/durable_server/lifecycle_test.exs @@ -3040,4 +3040,143 @@ defmodule DurableServer.LifecycleTest do GenServer.stop(manager_pid) end end + + defmodule PauseOnFetchBackend do + @behaviour DurableServer.StorageBackend + + alias DurableServer.StorageBackend + + @impl true + def init_backend(opts) do + delegate = Keyword.fetch!(opts, :delegate) + + {:ok, + %{ + state: %{ + delegate: delegate, + notify_pid: Keyword.fetch!(opts, :notify_pid), + pause_key: Keyword.fetch!(opts, :pause_key) + }, + defaults: StorageBackend.defaults(delegate), + features: StorageBackend.features(delegate) + }} + end + + @impl true + def ensure_ready(%{delegate: delegate}), do: StorageBackend.ensure_ready(delegate) + + @impl true + def get_object(%{delegate: delegate, notify_pid: notify_pid, pause_key: pause_key}, key, opts) do + if String.ends_with?(key, pause_key) do + send(notify_pid, {:paused_in_discovery_worker, self()}) + + receive do + :continue -> :ok + after + 10_000 -> :ok + end + end + + StorageBackend.get_object(delegate, key, opts) + end + + @impl true + def list_all_objects_stream(%{delegate: delegate}, prefix, opts), + do: StorageBackend.list_all_objects_stream(delegate, prefix, opts) + + @impl true + def put_object(%{delegate: delegate}, key, data, opts), + do: StorageBackend.put_object(delegate, key, data, opts) + + @impl true + def delete_object(%{delegate: delegate}, key), do: StorageBackend.delete_object(delegate, key) + + @impl true + def try_claim(%{delegate: delegate}, key, body), + do: StorageBackend.try_claim(delegate, key, body) + + @impl true + def update_object(%{delegate: delegate}, key, update_fn, opts), + do: StorageBackend.update_object(delegate, key, update_fn, opts) + + @impl true + def encode(%{delegate: delegate}, data), do: StorageBackend.encode(delegate, data) + + @impl true + def decode(%{delegate: delegate}, data), do: StorageBackend.decode(delegate, data) + end + + describe "discovery shutdown race" do + test "orphaned discovery task writes to a dead skip table when the manager stops mid-round", + %{supervisor_name: supervisor_name, prefix: prefix, config: config} do + key = "skip-race-#{DurableServer.UUID.uuid4()}" + delegate = DurableServer.Supervisor.__get_config__(supervisor_name).storage_backend + + stored_state = %DurableServer.StoredState{ + vsn: 1, + state: %{"count" => 1}, + meta: %Meta{ + key: key, + prefix: prefix, + supervisor: supervisor_name, + module: TestServer, + permanent: false, + status: :running, + node_str: "unreachable@test", + node_ref: System.unique_integer([:positive]), + pid: self(), + last_heartbeat_at: System.system_time(:millisecond) - 60_000 + } + } + + assert {:ok, %{etag: _}} = + DurableServer.StorageBackend.put_object(delegate, "#{prefix}#{key}", stored_state) + + {:ok, storage_backend} = + DurableServer.StorageBackend.init_backend(PauseOnFetchBackend, + delegate: delegate, + notify_pid: self(), + pause_key: key + ) + + test_config = + config + |> Map.put(:object_store, storage_backend) + |> Map.put(:storage_backend, storage_backend) + |> Map.put(:initial_discovery_delay_ms, 60_000) + |> Map.put(:discovery_interval_ms, 60_000) + + {:ok, manager_pid} = start_standalone_lifecycle_manager(supervisor_name, test_config) + + send(manager_pid, :discover_and_restart) + + assert_receive {:paused_in_discovery_worker, worker_pid}, 10_000 + worker_ref = Process.monitor(worker_pid) + + manager_state = :sys.get_state(manager_pid) + skip_table = manager_state.discovery_skip_table + %Task{pid: discovery_pid} = manager_state.current_discovery_task + + # the skip table is owned by the manager, so it dies with the manager + assert :ets.info(skip_table, :owner) == manager_pid + + # async_nolink means the discovery task is not linked to the manager + {:links, discovery_links} = Process.info(discovery_pid, :links) + refute manager_pid in discovery_links + + Process.exit(manager_pid, :kill) + assert_eventually(fn -> :ets.whereis(skip_table) == :undefined end) + + # the discovery task and its worker outlive the manager + assert Process.alive?(discovery_pid) + assert Process.alive?(worker_pid) + + send(worker_pid, :continue) + + # without the skip-table guard this exits {:badarg, [{:ets, :insert, [skip_table, ...]}]} + # from lib/durable_server/lifecycle_manager.ex:2079 + assert_receive {:DOWN, ^worker_ref, :process, ^worker_pid, :normal}, 10_000 + assert :ets.whereis(skip_table) == :undefined + end + end end