diff --git a/CHANGELOG.md b/CHANGELOG.md index 325fd8f..a6a8105 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,6 @@ ## Unreleased +- Reserve up to one second (half the budget for short calls) inside remote placement RPC deadlines and bound readiness waits by the remaining startup budget. Slow child bootstrap can return an ordinary timeout without unnecessarily cooling down the entire reachable node; genuine transport failures still trigger cooldown. - Enforce cumulative sticky-placement gates during request-driven remote placement so an existing server cannot move to a fallback node before that level unlocks. - Allow an expired restart attempt to be reclaimed immediately by another node at an allowed sticky-placement level, while preserving strict placement when no level matches. diff --git a/lib/durable_server/supervisor.ex b/lib/durable_server/supervisor.ex index 23e7642..31ed142 100644 --- a/lib/durable_server/supervisor.ex +++ b/lib/durable_server/supervisor.ex @@ -252,6 +252,7 @@ defmodule DurableServer.Supervisor do @placement_candidate_pool_multiplier 4 @placement_candidate_pool_min 10 @placement_node_timeout_cooldown_ms :timer.seconds(15) + @placement_erpc_response_headroom_ms 1_000 @placement_erpc_timeout_same_region_ms 3_000 @placement_erpc_timeout_cross_region_ms 8_000 @restart_claim_race_poll_ms 100 @@ -770,6 +771,10 @@ defmodule DurableServer.Supervisor do complete, including internal retries. Returns `{:error, :timeout}` on expiration. Set to `:infinity` to disable. Default: `#{@default_start_child_timeout}`ms. + Remote placement reserves part of each RPC budget for the reply, so a child + startup timeout can return without treating the reachable node as a transport + failure. Timing out a caller's wait does not cancel an already-supervised bootstrap. + ## Examples # Start with init args @@ -902,8 +907,14 @@ defmodule DurableServer.Supervisor do # When max_placement_retries is 0, this is a remote placement call from another node. # Wait for the supervisor tree to be ready before touching ETS/Group-backed state. if max_placement_retries == 0 do + ready_timeout = + case remaining_timeout_ms(caller_deadline_ms) do + :infinity -> @remote_placement_ready_timeout + remaining_ms -> min(@remote_placement_ready_timeout, remaining_ms) + end + case wait_until_ready(supervisor, - timeout: @remote_placement_ready_timeout, + timeout: ready_timeout, poll_interval: 50 ) do :ok -> @@ -911,7 +922,7 @@ defmodule DurableServer.Supervisor do {:error, :timeout} -> Logger.warning( - "DurableServer.Supervisor #{inspect(supervisor)} not ready after #{@remote_placement_ready_timeout}ms on remote placement" + "DurableServer.Supervisor #{inspect(supervisor)} not ready after #{ready_timeout}ms on remote placement" ) throw({:error, :not_ready}) @@ -1854,20 +1865,32 @@ defmodule DurableServer.Supervisor do [node | rest], placement_opts ) do - Logger.info("Attempting to place #{inspect(module)} on remote node #{inspect(node)}") - report_placement_diagnostic(supervisor, :remote_placement_erpc_attempt) shutdown_retries = Keyword.get(placement_opts, :shutdown_retries, 0) deadline = Keyword.get(placement_opts, :deadline) erpc_timeout_ms = __placement_erpc_timeout_ms__(supervisor, node, deadline) {remote_child_spec, remote_opts} = remote_start_child_args(child_spec) - remote_opts = Keyword.put(remote_opts, :timeout, erpc_timeout_ms) + + # Let a slow bootstrap return its ordinary timeout before the enclosing RPC + # expires and penalizes every key on this node with a transport cooldown. + # The outer timer starts before remote execution, so reserve time for transit + # in both directions and scheduling, not just the reply. One second is a + # conservative allowance, not a measured latency bound; cap it at half a + # short budget rounded up. If no startup budget remains, do not dispatch. + response_headroom_ms = + min(@placement_erpc_response_headroom_ms, div(erpc_timeout_ms + 1, 2)) + + remote_timeout_ms = erpc_timeout_ms - response_headroom_ms + remote_opts = Keyword.put(remote_opts, :timeout, remote_timeout_ms) # NOTE: we MUST pass max_placement_retries: 0 to prevent recursive retry on the other side try do - if erpc_timeout_ms == 0 do + if remote_timeout_ms == 0 do throw({:error, :placement_deadline_expired}) end + Logger.info("Attempting to place #{inspect(module)} on remote node #{inspect(node)}") + report_placement_diagnostic(supervisor, :remote_placement_erpc_attempt) + result = safe_erpc_call( node, diff --git a/test/durable_server/placement_deadline_test.exs b/test/durable_server/placement_deadline_test.exs new file mode 100644 index 0000000..dc091e9 --- /dev/null +++ b/test/durable_server/placement_deadline_test.exs @@ -0,0 +1,211 @@ +defmodule DurableServer.PlacementDeadlineTest do + use ExUnit.Case, async: false + + alias DurableServer.{LifecycleManager, PlacementTestServer, TestInMemoryBackend} + alias DurableServer.Supervisor, as: DurableSupervisor + + @moduletag :capture_log + + setup_all do + unless Node.alive?() do + {_, 0} = System.cmd("epmd", ["-daemon"]) + + {:ok, _} = + Node.start(:"placement_deadline_test_#{System.pid()}@127.0.0.1", :longnames) + + on_exit(fn -> Node.stop() end) + end + + :ok + end + + setup context do + suffix = "#{System.pid()}_#{System.unique_integer([:positive])}" + rpc_timeout = Map.get(context, :placement_rpc_timeout_ms, 500) + + {:ok, peer, remote_node} = + :peer.start_link(%{ + name: :"placement_deadline_peer_#{suffix}", + host: ~c"127.0.0.1", + args: [~c"+S", ~c"2:2", ~c"-pa" | :code.get_path()] + }) + + Process.unlink(peer) + + on_exit(fn -> + if Process.alive?(peer), do: :peer.stop(peer) + end) + + {:ok, _} = :erpc.call(remote_node, Application, :ensure_all_started, [:durable_server]) + supervisor = :"placement_deadline_sup_#{suffix}" + + opts = [ + name: supervisor, + prefix: "placement-deadline/#{suffix}/", + backend: {TestInMemoryBackend, []}, + initial_discovery_delay_ms: 60_000, + graceful_shutdown_timeout_ms: 500, + placement_erpc_timeout_same_region_ms: rpc_timeout, + placement_erpc_timeout_cross_region_ms: rpc_timeout + ] + + {:ok, _} = + :erpc.call(remote_node, PlacementTestServer, :start_supervisor, [ + Keyword.put(opts, :max_children, %{total: 10}) + ]) + + start_supervised!( + {DurableSupervisor, Keyword.put(opts, :max_children, %{PlacementTestServer => 0})} + ) + + :ok = DurableSupervisor.wait_until_ready(supervisor) + :ok = :erpc.call(remote_node, DurableSupervisor, :wait_until_ready, [supervisor]) + advertise_remote(supervisor, remote_node) + + %{supervisor: supervisor, remote_node: remote_node, peer: peer} + end + + for {rpc_timeout, start_timeout} <- [{500, 250}, {3_000, 2_000}, {8_000, 7_000}] do + @tag placement_rpc_timeout_ms: rpc_timeout + test "reserves the expected reply headroom for a #{rpc_timeout}ms RPC", context do + %{supervisor: supervisor, remote_node: remote_node} = context + :ok = :erpc.call(remote_node, PlacementTestServer, :trace_start_timeouts, [self()]) + + assert {:ok, {pid, _meta}} = + DurableSupervisor.start_child( + supervisor, + {PlacementTestServer, key: "budget", initial_state: %{}}, + timeout: 10_000, + max_placement_retries: 1 + ) + + assert node(pid) == remote_node + assert_receive {:placement_start_timeout, unquote(start_timeout)}, 1_000 + end + end + + test "slow bootstrap does not cool down its reachable node or block another key", context do + %{supervisor: supervisor, remote_node: remote_node} = context + observer = self() + + task = + Task.async(fn -> + DurableSupervisor.start_child( + supervisor, + {PlacementTestServer, key: "slow", initial_state: %{blocked: true, observer: observer}}, + timeout: 500, + max_placement_retries: 1 + ) + end) + + assert_receive {:bootstrap_started, "slow", bootstrap_pid}, 2_000 + assert node(bootstrap_pid) == remote_node + + try do + assert {:error, _reason} = Task.await(task, 2_000) + + diagnostics = LifecycleManager.get_discovery_diagnostics(supervisor) + assert Map.get(diagnostics, :remote_placement_node_cooldown_trip, 0) == 0 + assert Map.get(diagnostics, {:remote_placement_erpc_error, :timeout}, 0) == 0 + + assert {:ok, {fast_pid, _meta}} = + DurableSupervisor.start_child( + supervisor, + {PlacementTestServer, key: "fast", initial_state: %{}}, + timeout: 500, + max_placement_retries: 1 + ) + + assert node(fast_pid) == remote_node + assert :erpc.call(remote_node, Process, :alive?, [bootstrap_pid]) + after + send(bootstrap_pid, :finish_bootstrap) + end + + assert_eventually(fn -> + case :erpc.call(remote_node, DurableSupervisor, :lookup, [supervisor, "slow"]) do + {pid, _meta} -> pid == bootstrap_pid + nil -> false + end + end) + + assert {:ok, {^bootstrap_pid, _meta}} = + DurableSupervisor.ensure_started_child( + supervisor, + {PlacementTestServer, key: "slow", initial_state: %{}}, + timeout: 1_000 + ) + end + + test "a genuine transport failure still trips node cooldown", context do + %{supervisor: supervisor, peer: peer, remote_node: remote_node} = context + :ok = :peer.stop(peer) + assert Node.ping(remote_node) == :pang + advertise_remote(supervisor, remote_node) + + assert {:error, _reason} = + DurableSupervisor.start_child( + supervisor, + {PlacementTestServer, key: "disconnected", initial_state: %{}}, + timeout: 500, + max_placement_retries: 1 + ) + + diagnostics = LifecycleManager.get_discovery_diagnostics(supervisor) + assert diagnostics.remote_placement_node_cooldown_trip == 1 + assert diagnostics.remote_placement_erpc_error == 1 + end + + test "remote readiness waiting honors a short caller budget" do + supervisor = :"absent_placement_sup_#{System.unique_integer([:positive])}" + started_at = System.monotonic_time(:millisecond) + + assert catch_throw( + DurableSupervisor.__start_child__( + supervisor, + {PlacementTestServer, [key: "not-ready", initial_state: %{}], nil}, + max_placement_retries: 0, + timeout: 30 + ) + ) == {:error, :not_ready} + + assert System.monotonic_time(:millisecond) - started_at < 250 + end + + test "a one-millisecond budget does not dispatch a remote bootstrap", context do + %{supervisor: supervisor, remote_node: remote_node} = context + + assert {:error, :timeout} = + DurableSupervisor.start_child( + supervisor, + {PlacementTestServer, + key: "expired", initial_state: %{blocked: true, observer: self()}}, + timeout: 1, + max_placement_retries: 1 + ) + + refute_receive {:bootstrap_started, "expired", _pid}, 50 + assert :erpc.call(remote_node, DurableSupervisor, :lookup, [supervisor, "expired"]) == nil + diagnostics = LifecycleManager.get_discovery_diagnostics(supervisor) + assert Map.get(diagnostics, :remote_placement_node_cooldown_trip, 0) == 0 + assert Map.get(diagnostics, :remote_placement_erpc_attempt, 0) == 0 + end + + defp advertise_remote(supervisor, remote_node) do + :ets.insert( + :"durable_server_heartbeats_#{supervisor}", + {Atom.to_string(remote_node), 1, System.system_time(:millisecond), + %{total: %{current: 0, limit: 10}}, nil, %{}, %{}} + ) + end + + defp assert_eventually(fun, attempts \\ 100) + defp assert_eventually(fun, 0), do: assert(fun.()) + + defp assert_eventually(fun, attempts) do + unless fun.() do + Process.sleep(10) + assert_eventually(fun, attempts - 1) + end + end +end diff --git a/test/durable_server/supervisor_backend_spec_test.exs b/test/durable_server/supervisor_backend_spec_test.exs index 52d17fa..845f697 100644 --- a/test/durable_server/supervisor_backend_spec_test.exs +++ b/test/durable_server/supervisor_backend_spec_test.exs @@ -6,110 +6,12 @@ defmodule DurableServer.SupervisorBackendSpecTest do alias DurableServer.LifecycleManager alias DurableServer.Backends.EKVStore alias DurableServer.Backends.MirrorStore - alias DurableServer.StorageBackend + alias DurableServer.TestInMemoryBackend, as: InMemoryBackend def throw_not_ready do throw({:error, :not_ready}) end - defmodule InMemoryBackend do - @behaviour StorageBackend - - @impl true - def init_backend(raw_opts) do - opts = - case raw_opts do - %{} = map -> map - opts when is_list(opts) -> Map.new(opts) - other -> %{raw_opts: other} - end - - {:ok, - %{ - state: %{ - table: :ets.new(__MODULE__, [:set, :public]), - name: Map.get(opts, :name) - }, - defaults: %{ - heartbeat_tracking_mode: :poll, - discovery_interval_ms: 60_000, - heartbeat_interval_ms: 10_000, - heartbeat_reconcile_interval_ms: 10_000 - } - }} - end - - @impl true - def ensure_ready(_state), do: :ok - - @impl true - def get_object(%{table: table}, key, _opts) do - case :ets.lookup(table, key) do - [{^key, %{body: body, etag: etag}}] -> {:ok, %{body: body, etag: etag}} - [] -> {:error, :not_found} - end - end - - @impl true - def list_all_objects_stream(%{table: table}, prefix, _opts) do - table - |> :ets.tab2list() - |> Stream.filter(fn {key, _value} -> String.starts_with?(key, prefix) end) - |> Stream.map(fn {key, %{etag: etag}} -> %{key: key, etag: etag} end) - end - - @impl true - def put_object(%{table: table}, key, data, _opts) do - etag = next_etag() - :ets.insert(table, {key, %{body: data, etag: etag}}) - {:ok, %{body: data, etag: etag}} - end - - @impl true - def delete_object(%{table: table}, key) do - case :ets.lookup(table, key) do - [{^key, _value}] -> - :ets.delete(table, key) - :ok - - [] -> - {:error, :not_found} - end - end - - @impl true - def try_claim(%{table: table}, key, body) do - case :ets.lookup(table, key) do - [] -> - etag = next_etag() - :ets.insert(table, {key, %{body: body, etag: etag}}) - {:ok, {:claimed, etag}} - - [_existing] -> - {:error, :taken} - end - end - - @impl true - def update_object(%{table: table} = state, key, update_fn, _opts) do - with {:ok, %{body: body, etag: etag}} <- get_object(state, key, []), - {:ok, new_body} <- update_fn.(%{body: body, etag: etag}) do - put_object(%{table: table}, key, new_body, []) - end - end - - @impl true - def encode(_state, data), do: {:ok, data} - - @impl true - def decode(_state, data), do: {:ok, data} - - defp next_etag do - System.unique_integer([:positive, :monotonic]) - |> Integer.to_string() - end - end - test "rejects heartbeats beyond the configured future clock-skew tolerance" do supervisor_name = unique_supervisor_name("heartbeat_skew") prefix = unique_prefix("heartbeat_skew") diff --git a/test/support/in_memory_backend.ex b/test/support/in_memory_backend.ex new file mode 100644 index 0000000..c37f82c --- /dev/null +++ b/test/support/in_memory_backend.ex @@ -0,0 +1,93 @@ +defmodule DurableServer.TestInMemoryBackend do + @moduledoc """ + Node-local storage shared by backend configuration and placement tests. + + Each instance owns an independent ETS table. Claims are create-only, but + updates do not enforce compare-and-swap options. This is not distributed + storage and must not be used to test competing owners across nodes. + """ + @behaviour DurableServer.StorageBackend + + @impl true + def init_backend(raw_opts) do + opts = + case raw_opts do + %{} = map -> map + opts when is_list(opts) -> Map.new(opts) + other -> %{raw_opts: other} + end + + {:ok, + %{ + state: %{ + table: :ets.new(__MODULE__, [:set, :public]), + name: Map.get(opts, :name) + }, + defaults: %{ + heartbeat_tracking_mode: :poll, + discovery_interval_ms: 60_000, + heartbeat_interval_ms: 10_000, + heartbeat_reconcile_interval_ms: 10_000 + } + }} + end + + @impl true + def ensure_ready(_state), do: :ok + + @impl true + def get_object(%{table: table}, key, _opts) do + case :ets.lookup(table, key) do + [{^key, object}] -> {:ok, object} + [] -> {:error, :not_found} + end + end + + @impl true + def put_object(%{table: table}, key, body, _opts) do + object = %{body: body, etag: etag()} + true = :ets.insert(table, {key, object}) + {:ok, object} + end + + @impl true + def try_claim(%{table: table}, key, body) do + object = %{body: body, etag: etag()} + + if :ets.insert_new(table, {key, object}), + do: {:ok, {:claimed, object.etag}}, + else: {:error, :taken} + end + + @impl true + def update_object(state, key, fun, opts) do + with {:ok, object} <- get_object(state, key, opts), + {:ok, body} <- fun.(object) do + put_object(state, key, body, opts) + end + end + + @impl true + def delete_object(%{table: table}, key) do + case :ets.take(table, key) do + [] -> {:error, :not_found} + [_object] -> :ok + end + end + + @impl true + def list_all_objects_stream(%{table: table}, prefix, _opts) do + table + |> :ets.tab2list() + |> Stream.filter(fn {key, _object} -> String.starts_with?(key, prefix) end) + |> Stream.map(fn {key, object} -> %{key: key, etag: object.etag} end) + end + + @impl true + def encode(_state, value), do: {:ok, value} + + @impl true + def decode(_state, value), do: {:ok, value} + + defp etag, do: Integer.to_string(System.unique_integer([:positive, :monotonic])) +end diff --git a/test/support/placement_test_server.ex b/test/support/placement_test_server.ex new file mode 100644 index 0000000..2d25ab1 --- /dev/null +++ b/test/support/placement_test_server.ex @@ -0,0 +1,48 @@ +defmodule DurableServer.PlacementTestServer do + @moduledoc false + use DurableServer, vsn: 1 + + # Keep the remote supervisor alive after the short-lived RPC worker exits. + def start_supervisor(opts) do + {:ok, pid} = DurableServer.Supervisor.start_link(opts) + Process.unlink(pid) + {:ok, pid} + end + + # Observe the budget actually sent over RPC without sleeping for multi-second + # deadlines or exposing a production API solely for checking timeout arithmetic. + # The tracer and trace pattern live only on this test's disposable peer node. + def trace_start_timeouts(observer) do + tracer = spawn(fn -> forward_start_timeouts(observer) end) + :erlang.trace_pattern({DurableServer.Supervisor, :__start_child__, 3}, true, []) + :erlang.trace(:new, true, [:call, {:tracer, tracer}]) + :ok + end + + defp forward_start_timeouts(observer) do + receive do + {:trace, _pid, :call, {DurableServer.Supervisor, :__start_child__, [_sup, _spec, opts]}} -> + send(observer, {:placement_start_timeout, Keyword.fetch!(opts, :timeout)}) + forward_start_timeouts(observer) + end + end + + @impl true + def init(state, info) do + if state[:blocked] do + send(state.observer, {:bootstrap_started, info.key, self()}) + + receive do + :finish_bootstrap -> :ok + end + end + + {:ok, state} + end + + @impl true + def dump_state(state), do: state + + @impl true + def load_state(_vsn, state), do: state +end