diff --git a/lib/durable_server/lifecycle_manager.ex b/lib/durable_server/lifecycle_manager.ex index 3bb1893..8c33fe6 100644 --- a/lib/durable_server/lifecycle_manager.ex +++ b/lib/durable_server/lifecycle_manager.ex @@ -2116,7 +2116,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 @@ -2125,7 +2125,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 @@ -2165,6 +2165,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 -> %{} @@ -2562,38 +2571,46 @@ defmodule DurableServer.LifecycleManager do end defp discovery_skip?(%LifecycleManager{} = state, key, etag) do - case :ets.lookup(state.discovery_skip_table, key) do - [{^key, ^etag, :skip, _ts}] -> - true + case :ets.whereis(state.discovery_skip_table) do + :undefined -> + false - [{^key, ^etag, %Meta{} = meta, _ts}] -> - # Etag unchanged so stored state is identical, but time-dependent checks - # (placement gates, circuit breaker, node health) may have changed. - case CircuitBreaker.check_module_circuit_breaker(state.circuit_breaker, meta.module) do - {:circuit_open, _} -> + _ -> + case :ets.lookup(state.discovery_skip_table, key) do + [{^key, ^etag, :skip, _ts}] -> true - :ok -> - case appears_restartable?(state, meta) do - {:restartable, _claim_context} -> - # Now restartable — remove from cache, let async task do fresh GET - :ets.delete(state.discovery_skip_table, key) - false - - :transient -> - # Heartbeat/read uncertainty is not a stable "non-restartable" state. - # Drop the cache entry so the next async pass does a fresh read. - :ets.delete(state.discovery_skip_table, key) - false - - false -> + [{^key, ^etag, %Meta{} = meta, _ts}] -> + # Etag unchanged so stored state is identical, but time-dependent checks + # (placement gates, circuit breaker, node health) may have changed. + case CircuitBreaker.check_module_circuit_breaker(state.circuit_breaker, meta.module) do + {:circuit_open, _} -> true + + :ok -> + case appears_restartable?(state, meta) do + {:restartable, _claim_context} -> + # Now restartable — remove from cache, let async task do fresh GET + :ets.delete(state.discovery_skip_table, key) + false + + :transient -> + # Heartbeat/read uncertainty is not a stable "non-restartable" state. + # Drop the cache entry so the next async pass does a fresh read. + :ets.delete(state.discovery_skip_table, key) + false + + false -> + true + end end - end - _ -> - false + _ -> + false + end end + rescue + ArgumentError -> false end # Strip fields not needed for re-evaluation to reduce ETS memory. diff --git a/test/durable_server/lifecycle_test.exs b/test/durable_server/lifecycle_test.exs index fadda4d..6ca7ad8 100644 --- a/test/durable_server/lifecycle_test.exs +++ b/test/durable_server/lifecycle_test.exs @@ -3301,4 +3301,226 @@ 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), + pause_list: Keyword.get(opts, :pause_list, false) + }, + 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} = state, prefix, opts) do + # Heartbeat refresh also lists "#{prefix}__nodes/" during init. Only the + # discovery listing should block, otherwise the manager never finishes starting. + if state.pause_list and not String.contains?(prefix, "__nodes/") do + send(state.notify_pid, {:paused_in_discovery_list, self()}) + + receive do + :continue_list -> :ok + after + 10_000 -> :ok + end + end + + StorageBackend.list_all_objects_stream(delegate, prefix, opts) + end + + @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 worker does not crash when the skip table dies mid-fetch", + %{supervisor_name: supervisor_name, prefix: prefix, config: config} do + key = "skip-race-#{DurableServer.UUID.uuid4()}" + + {manager_pid, worker_pid, discovery_pid, skip_table} = + pause_discovery_worker(supervisor_name, prefix, config, key) + + worker_ref = Process.monitor(worker_pid) + + # 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 + assert Process.alive?(discovery_pid) + + Process.exit(manager_pid, :kill) + assert_eventually(fn -> :ets.whereis(skip_table) == :undefined end) + + # the worker outlives the manager and then writes the skip cache + assert Process.alive?(worker_pid) + + send(worker_pid, :continue) + + # without skip_cache_put/2 this exits with ArgumentError from :ets.insert/2 + assert_receive {:DOWN, ^worker_ref, :process, ^worker_pid, :normal}, 10_000 + assert :ets.whereis(skip_table) == :undefined + end + + test "orphaned discovery task does not crash when discovery_skip? reads a dead skip table", + %{supervisor_name: supervisor_name, prefix: prefix, config: config} do + key = "skip-read-race-#{DurableServer.UUID.uuid4()}" + delegate = DurableServer.Supervisor.__get_config__(supervisor_name).storage_backend + put_non_restartable_object(delegate, prefix, key, supervisor_name) + + {:ok, storage_backend} = + DurableServer.StorageBackend.init_backend(PauseOnFetchBackend, + delegate: delegate, + notify_pid: self(), + pause_key: key, + pause_list: true + ) + + {:ok, manager_pid} = + start_standalone_lifecycle_manager( + supervisor_name, + discovery_race_config(config, storage_backend) + ) + + send(manager_pid, :discover_and_restart) + + assert_receive {:paused_in_discovery_list, discovery_pid}, 10_000 + discovery_ref = Process.monitor(discovery_pid) + + manager_state = :sys.get_state(manager_pid) + skip_table = manager_state.discovery_skip_table + assert manager_state.current_discovery_task.pid == discovery_pid + 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) + assert Process.alive?(discovery_pid) + + # listing resumes only after the table is gone, so discovery_skip?/3 runs next + send(discovery_pid, :continue_list) + + # without the :ets.whereis/1 guard this task dies with ArgumentError from :ets.lookup/2 + # and the worker pause never arrives + assert_receive {:paused_in_discovery_worker, worker_pid}, 10_000 + worker_ref = Process.monitor(worker_pid) + assert Process.alive?(discovery_pid) + + send(worker_pid, :continue) + + assert_receive {:DOWN, ^worker_ref, :process, ^worker_pid, :normal}, 10_000 + assert_receive {:DOWN, ^discovery_ref, :process, ^discovery_pid, :normal}, 10_000 + assert :ets.whereis(skip_table) == :undefined + end + end + + defp pause_discovery_worker(supervisor_name, prefix, config, key) do + delegate = DurableServer.Supervisor.__get_config__(supervisor_name).storage_backend + put_non_restartable_object(delegate, prefix, key, supervisor_name) + + {:ok, storage_backend} = + DurableServer.StorageBackend.init_backend(PauseOnFetchBackend, + delegate: delegate, + notify_pid: self(), + pause_key: key + ) + + {:ok, manager_pid} = + start_standalone_lifecycle_manager( + supervisor_name, + discovery_race_config(config, storage_backend) + ) + + send(manager_pid, :discover_and_restart) + + assert_receive {:paused_in_discovery_worker, worker_pid}, 10_000 + + manager_state = :sys.get_state(manager_pid) + + {manager_pid, worker_pid, manager_state.current_discovery_task.pid, + manager_state.discovery_skip_table} + end + + defp discovery_race_config(config, storage_backend) do + 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) + end + + defp put_non_restartable_object(delegate, prefix, key, supervisor_name) do + 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) + end end