diff --git a/CHANGELOG.md b/CHANGELOG.md index d25bd1c..b922cf4 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,3 +1,23 @@ +## 0.1.5 (unreleased) +- Add `:heartbeat_future_skew_tolerance_ms` supervisor option (default: `5_000`). Node heartbeats stamped further than this into the future are ignored for liveness decisions instead of being treated as always-fresh, and local heartbeat/watchdog deadlines now use monotonic time so wall-clock (NTP) adjustments no longer stretch or shrink safety windows. +- Reject child keys in the reserved internal `__nodes/` namespace: `start_child/3`, `ensure_started_child/3`, and `rehome_child/3` now raise `ArgumentError` for keys that would collide with node heartbeat storage. +- Enforce `max_children` limits atomically with local capacity reservations so concurrent starts can no longer exceed total or per-module limits. Integer `max_children` is enforced through the same path and returns `{:error, {:capacity_limit, :max_children_total}}` instead of leaking DynamicSupervisor's raw `{:error, :max_children}`; integer values must now be positive (`0` or negative raises `ArgumentError`). +- Terminate servers using `auto_sync: true` when a storage CAS conflict shows the object was replaced by another owner, matching explicit and periodic sync behavior. Previously the stale process logged the conflict and kept serving requests. +- Accept documented callback returns that previously crashed the process: integer timeout actions (e.g. `{:reply, reply, state, 5_000}`) and `{:stop, {:shutdown, :normal}, ...}` stops, which persist `:stopped_graceful` while preserving the exit reason. +- Return `{:error, reason}` from `terminate_and_delete_child/2,3` when the storage delete fails instead of `:ok`, and only delete storage after an owned `:deleting` tombstone is CAS-written so a stale process cannot delete an object claimed by a newer owner. +- Fence crash-time metadata writes to the dying owner so a stale crashing process cannot overwrite a newer owner's status or crash history. +- Release the storage prefix claim when a supervisor stops or fails to start. Supervisors can now be restarted with the same `:prefix` in the same VM; previously the claim leaked until VM restart. +- Bound graceful shutdown with one global `graceful_shutdown_timeout_ms` deadline shared by discovery shutdown and all children (previously the timeout applied per child and multiplied across concurrency batches), and warn when children fail final persistence during shutdown. +- Cap remote placement ERPC calls and remote child timeouts by the caller's remaining deadline instead of issuing fresh fixed per-node timeouts after the budget expired. +- Retry transient Req heartbeat failures (retryable HTTP statuses, transport timeouts/refusals/closures, HTTP/2 unprocessed) within the heartbeat deadline instead of crashing the supervisor tree on the first error. +- Recover from discovery task crashes by logging, clearing discovery state, and rescheduling instead of restarting the whole supervisor tree. +- Make mirror fallback promotion create-only (`try_claim`) so promotion can no longer overwrite an object concurrently created in the preferred backend. +- Retry waiting `ensure_started_child/3` callers immediately when the singleflight leader finishes before the waiter registers instead of sleeping until the full timeout, and never run the guarded operation twice when it raises `ArgumentError`. +- Harden persisted metadata decoding with bounded sizes, `:safe` external-term decoding, and per-field validation. Corrupt or unsafe metadata is rejected with `{:error, %ArgumentError{}}` from storage reads instead of being admitted as partially typed data; legacy binary node references remain accepted. +- Parse IAM XML responses with DTDs disabled and a 1 MiB limit to prevent external-entity and entity-expansion attacks from a malicious endpoint. +- Redact secrets from logging: IAM CreateAccessKey bodies (which contain `SecretAccessKey`) are no longer logged, supervisor startup logs omit backend state and `init_info`, and backend errors no longer echo raw options. +- Update Mint to 1.9.3 for the HTTP/1 chunk-size parser security fix (CVE-2026-59249). + ## 0.1.4 (2026-06-04) - Reject explicit local `start_child/3` and `ensure_started_child/3` attempts before spawning a child when the local supervisor is draining. diff --git a/lib/durable_server.ex b/lib/durable_server.ex index 34eddd1..6aad1b0 100644 --- a/lib/durable_server.ex +++ b/lib/durable_server.ex @@ -817,6 +817,8 @@ defmodule DurableServer do permanent: false, was_permanently_crashed: false, user_initiated_stop: nil, + delete_request_ref: nil, + delete_requester_pid: nil, start_time: nil, restart_attempt_node: nil, restart_attempt_time: nil, @@ -839,6 +841,10 @@ defmodule DurableServer do @max_sync_retries 5 @bootstrap_continue {@durable, :bootstrap} + defguardp is_callback_action(action) + when is_atom(action) or is_tuple(action) or + (is_integer(action) and action >= 0) + defmacro __using__(opts) do vsn = case Keyword.fetch(opts, :vsn) do @@ -1376,6 +1382,9 @@ defmodule DurableServer do {:error, :timeout} -> delete_with_lock_attempt(key, timeout, config) + + {:error, _reason} = error -> + error end nil -> @@ -1419,11 +1428,6 @@ defmodule DurableServer do delete_by_pid(pid, timeout) - {:error, :not_found} -> - # object doesn't exist, consider this success - Logger.info("Object #{storage_key} already deleted") - :ok - {:error, reason} -> Logger.error("Failed to acquire delete lock for #{storage_key}: #{inspect(reason)}") {:error, reason} @@ -1431,32 +1435,51 @@ defmodule DurableServer do end defp delete_by_pid(pid, timeout) when is_pid(pid) and is_integer(timeout) do - start_time = System.system_time(:millisecond) + deadline = System.monotonic_time(:millisecond) + timeout ref = make_ref() monitor_ref = Process.monitor(pid) send(pid, {@durable, {:delete_request, ref, self()}}) - receive do - # process is shutting down and attempting to delete itself - {:delete_in_progress, ^ref} -> - Logger.info("Process #{inspect(pid)} completed self-deletion") - remaining_timeout = timeout - (System.system_time(:millisecond) - start_time) + try do + receive do + {:delete_in_progress, ^ref} -> + await_delete_completion(pid, ref, monitor_ref, deadline) - # await shutdown + # process is dead and did not process our delete request + {:DOWN, ^monitor_ref, :process, ^pid, _reason} -> + {:error, :noproc} + after + remaining_delete_timeout(deadline) -> {:error, :timeout} + end + after + Process.demonitor(monitor_ref, [:flush]) + end + end + + defp await_delete_completion(pid, ref, monitor_ref, deadline) do + receive do + {:delete_complete, ^ref, result} -> receive do - {:DOWN, ^monitor_ref, :process, ^pid, _} -> :ok + {:DOWN, ^monitor_ref, :process, ^pid, _reason} -> result after - remaining_timeout -> {:error, :timeout} + remaining_delete_timeout(deadline) -> {:error, :timeout} end - # process is dead and did not process our delete request - {:DOWN, ^monitor_ref, :process, ^pid, _} -> - {:error, :noproc} + {:DOWN, ^monitor_ref, :process, ^pid, reason} -> + receive do + {:delete_complete, ^ref, result} -> result + after + 0 -> {:error, {:delete_not_completed, reason}} + end after - timeout -> {:error, :timeout} + remaining_delete_timeout(deadline) -> {:error, :timeout} end end + defp remaining_delete_timeout(deadline) do + max(deadline - System.monotonic_time(:millisecond), 0) + end + @doc false def claim_restart_attempt(%ObjectStore{} = store, %StoredState{} = stored_state, opts) do backend = StorageBackend.new(DurableServer.Backends.ObjectStore, store) @@ -1659,9 +1682,7 @@ defmodule DurableServer do {:noreply, schedule_sync(new_state)} {:error, :conflict} -> - fatal_exit!( - "#{state.key} object updated out from underneath: #{inspect(node: node(), pid: self())}" - ) + fatal_sync_conflict!(state) {:error, reason} -> # continue without stopping for transient errors (ie timeouts), but log the error @@ -1694,7 +1715,13 @@ defmodule DurableServer do # Defer status persistence to terminate/2 after user callback terminate/2 has run. updated_state = - %{state | user_initiated_stop: {:shutdown, :delete}, final_status_set: :deleting} + %{ + state + | user_initiated_stop: {:shutdown, :delete}, + final_status_set: :deleting, + delete_request_ref: ref, + delete_requester_pid: requester_pid + } # stop with delete reason to trigger deletion in terminate/2 {:stop, {:shutdown, :delete}, updated_state} @@ -1743,9 +1770,21 @@ defmodule DurableServer do handle_user_initiated_terminate(user_stop, reason, state) end + notify_delete_result(state, sync_result) maybe_invoke_after_terminate(state, terminate_return, reason, final_status, sync_result) end + defp notify_delete_result( + %DurableServer{delete_request_ref: ref, delete_requester_pid: requester_pid}, + result + ) + when is_reference(ref) and is_pid(requester_pid) do + send(requester_pid, {:delete_complete, ref, result}) + :ok + end + + defp notify_delete_result(%DurableServer{}, _result), do: :ok + defp bootstrap_init( %DurableServer{ module: module, @@ -1769,7 +1808,12 @@ defmodule DurableServer do } ) do with :ok <- maybe_check_global_lock_circuit_breaker(circuit_breaker, preloaded_boot), - :ok <- LifecycleManager.check_capacity(supervisor_name, module, capacity_opts) do + :ok <- + LifecycleManager.check_capacity( + supervisor_name, + module, + Keyword.put(capacity_opts, :skip_count_limits, true) + ) do current_node_str = to_string(Node.self()) load_result = @@ -1948,18 +1992,30 @@ defmodule DurableServer do Logger.info("DurableServer #{state.key} terminating for deletion - removing from storage") final_status = state.final_status_set || :deleting - sync_result = maybe_sync_final_status(state, final_status) - case StorageBackend.delete_object(state.object_store, storage_key(state)) do - :ok -> - Logger.info("Successfully deleted storage for #{state.key}") - - {:error, :not_found} -> - Logger.info("Storage already deleted for #{state.key}") + sync_result = + case persist_final_status(state, final_status) do + {:ok, %DurableServer{} = synced_state} -> + case StorageBackend.delete_object( + synced_state.object_store, + storage_key(synced_state) + ) do + :ok -> + Logger.info("Successfully deleted storage for #{state.key}") + :ok + + {:error, :not_found} -> + Logger.info("Storage already deleted for #{state.key}") + :ok + + {:error, reason} -> + Logger.error("Failed to delete storage for #{state.key}: #{inspect(reason)}") + {:error, reason} + end - {:error, reason} -> - Logger.error("Failed to delete storage for #{state.key}: #{inspect(reason)}") - end + {:error, _reason} = error -> + error + end {final_status, sync_result} @@ -2085,15 +2141,22 @@ defmodule DurableServer do end defp maybe_sync_final_status(%DurableServer{} = state, status) when is_atom(status) do + case persist_final_status(state, status) do + {:ok, %DurableServer{}} -> :ok + {:error, _reason} = error -> error + end + end + + defp persist_final_status(%DurableServer{} = state, status) when is_atom(status) do report_sync_and_stop = state.terminator_handled and status == :stopped_graceful case sync_to_storage(state, meta: %{status: status}) do - {:ok, %DurableServer{} = _new_state} -> + {:ok, %DurableServer{} = new_state} -> if report_sync_and_stop do LifecycleManager.report_diagnostic(state.supervisor, :sync_and_stop_ok) end - :ok + {:ok, new_state} {:error, sync_reason} -> if report_sync_and_stop do @@ -2321,7 +2384,7 @@ defmodule DurableServer do # Handle action + options tuple. {:reply, reply, new_user_state, action, opts} - when (is_atom(action) or is_tuple(action)) and is_list(opts) -> + when is_callback_action(action) and is_list(opts) -> {updated_state, sync?} = apply_callback_options(state, opts) {final_state, final_action} = handle_action(updated_state, new_user_state, action) @@ -2343,7 +2406,7 @@ defmodule DurableServer do {:reply, reply, new_state} - {:reply, reply, new_user_state, action} when is_atom(action) or is_tuple(action) -> + {:reply, reply, new_user_state, action} when is_callback_action(action) -> {final_state, final_action} = handle_action(state, new_user_state, action) if final_action do @@ -2359,7 +2422,7 @@ defmodule DurableServer do |> auto_sync_to_storage()} {:noreply, new_user_state, action, opts} - when (is_atom(action) or is_tuple(action)) and is_list(opts) -> + when is_callback_action(action) and is_list(opts) -> {updated_state, sync?} = apply_callback_options(state, opts) {final_state, final_action} = handle_action(updated_state, new_user_state, action) @@ -2381,7 +2444,7 @@ defmodule DurableServer do {:noreply, new_state} - {:noreply, new_user_state, action} when is_atom(action) or is_tuple(action) -> + {:noreply, new_user_state, action} when is_callback_action(action) -> {final_state, final_action} = handle_action(state, new_user_state, action) if final_action do @@ -2430,6 +2493,32 @@ defmodule DurableServer do {:stop, :normal, stopped_state} + {:stop, {:shutdown, :normal}, reply, new_user_state} -> + stopped_state = + update_state( + %{ + state + | user_initiated_stop: {:shutdown, :normal}, + final_status_set: :stopped_graceful + }, + new_user_state + ) + + {:stop, {:shutdown, :normal}, reply, stopped_state} + + {:stop, {:shutdown, :normal}, new_user_state} -> + stopped_state = + update_state( + %{ + state + | user_initiated_stop: {:shutdown, :normal}, + final_status_set: :stopped_graceful + }, + new_user_state + ) + + {:stop, {:shutdown, :normal}, stopped_state} + {:stop, {:shutdown, :permanent}, reply, new_user_state} -> # shutdown-wrapped permanent stop stopped_state = @@ -2609,9 +2698,7 @@ defmodule DurableServer do synced_state {:error, :conflict} -> - fatal_exit!( - "#{state.key} object updated out from underneath: #{inspect(node: node(), pid: self())}" - ) + fatal_sync_conflict!(state) {:error, reason} -> if is_map(metadata) do @@ -2644,6 +2731,9 @@ defmodule DurableServer do {:ok, %DurableServer{} = new_state} -> new_state + {:error, :conflict} -> + fatal_sync_conflict!(state) + {:error, reason} -> Logger.error(fn -> "#{inspect(module)} (key=#{key}) unable to auto_sync: #{inspect(reason)}" @@ -2656,6 +2746,12 @@ defmodule DurableServer do end end + defp fatal_sync_conflict!(%DurableServer{} = state) do + fatal_exit!( + "#{state.key} object updated out from underneath: #{inspect(node: node(), pid: self())}" + ) + end + defp sync_to_storage(%DurableServer{} = state, opts \\ []) do opts = Keyword.validate!(opts, [:meta]) # if meta overrides are provided, we always force sync @@ -3179,15 +3275,19 @@ defmodule DurableServer do store = state.object_store case StorageBackend.get_object(store, storage_key, consistent: true) do - {:ok, %{body: %StoredState{} = stored_state, etag: etag}} -> - updated_data = - stored_state - |> attach_stored_state_context(%{key: state.key, prefix: state.prefix}) - |> Map.put(:meta, updated_meta) - - case put_object(%{state | etag: etag}, storage_key, updated_data) do - {:ok, %DurableServer{} = new_state} -> {:ok, new_state} - {:error, reason} -> {:error, reason} + {:ok, %{body: %StoredState{meta: %Meta{} = current_meta} = stored_state, etag: etag}} -> + if same_boot_owner?(state, current_meta) do + updated_data = + stored_state + |> attach_stored_state_context(%{key: state.key, prefix: state.prefix}) + |> Map.put(:meta, updated_meta) + + case put_object(%{state | etag: etag}, storage_key, updated_data) do + {:ok, %DurableServer{} = new_state} -> {:ok, new_state} + {:error, reason} -> {:error, reason} + end + else + {:error, :ownership_mismatch} end {:ok, %{body: other}} -> diff --git a/lib/durable_server/application.ex b/lib/durable_server/application.ex index cfdfaa3..acb8cf0 100644 --- a/lib/durable_server/application.ex +++ b/lib/durable_server/application.ex @@ -6,6 +6,7 @@ defmodule DurableServer.Application do @impl true def start(_type, _args) do children = [ + {Registry, keys: :unique, name: DurableServer.PrefixRegistry}, {Finch, name: DurableServer.Finch}, {Task.Supervisor, name: DurableServer.TaskSupervisor} ] diff --git a/lib/durable_server/backends/mirror_store.ex b/lib/durable_server/backends/mirror_store.ex index b0a1d47..2d1c4d4 100644 --- a/lib/durable_server/backends/mirror_store.ex +++ b/lib/durable_server/backends/mirror_store.ex @@ -49,9 +49,9 @@ defmodule DurableServer.Backends.MirrorStore do `-- promote_on_fallback=true | v - put into read_preference - |-- success/conflict-resolved -> return read_preference object - `-- transient failure -> return {:error, {:promotion_failed, reason}} + claim absent key in read_preference + |-- success/concurrent-create -> return read_preference object + `-- transient failure -> return {:error, {:promotion_failed, reason}} ``` ### Why promotion matters @@ -391,12 +391,13 @@ defmodule DurableServer.Backends.MirrorStore do def unsubscribe(%{} = _state, _subscription_ref), do: :ok defp promote_fallback_object(read_backend, key, body, read_opts) do - # Promote to the read backend so returned etag matches subsequent CAS writes. - case StorageBackend.put_object(read_backend, key, body, max_retries: 3) do - {:ok, promoted_obj} -> - {:ok, promoted_obj} + # Promotion is create-if-absent. A writer may create the preferred object + # between its initial miss and this fallback read; never overwrite that value. + case StorageBackend.try_claim(read_backend, key, body) do + {:ok, {:claimed, etag}} -> + {:ok, %{body: body, etag: etag}} - {:error, :conflict} -> + {:error, reason} when reason in [:already_claimed, :taken, :conflict] -> StorageBackend.get_object(read_backend, key, read_opts) {:error, reason} -> diff --git a/lib/durable_server/heartbeat_watchdog.ex b/lib/durable_server/heartbeat_watchdog.ex index ee51ca3..5101438 100644 --- a/lib/durable_server/heartbeat_watchdog.ex +++ b/lib/durable_server/heartbeat_watchdog.ex @@ -122,7 +122,7 @@ defmodule DurableServer.HeartbeatWatchdog do {:heartbeat_deadline, timer_token}, %__MODULE__{timer_token: timer_token} = state ) do - elapsed_since_last = System.system_time(:millisecond) - state.last_heartbeat_at + elapsed_since_last = System.monotonic_time(:millisecond) - state.last_heartbeat_at Logger.error(fn -> "#{inspect(state.supervisor_name)}: heartbeat watchdog deadline exceeded " <> @@ -149,7 +149,7 @@ defmodule DurableServer.HeartbeatWatchdog do defp schedule_deadline(%__MODULE__{} = state) do deadline_at = state.last_heartbeat_at + state.deadline_ms - delay_ms = max(deadline_at - System.system_time(:millisecond), 0) + delay_ms = max(deadline_at - System.monotonic_time(:millisecond), 0) timer_token = make_ref() timer_ref = Process.send_after(self(), {:heartbeat_deadline, timer_token}, delay_ms) diff --git a/lib/durable_server/lifecycle_manager.ex b/lib/durable_server/lifecycle_manager.ex index 1105e03..72b03c0 100644 --- a/lib/durable_server/lifecycle_manager.ex +++ b/lib/durable_server/lifecycle_manager.ex @@ -125,6 +125,7 @@ defmodule DurableServer.LifecycleManager do capacity_limits: %{}, heartbeat_meta: nil, last_successful_heartbeat_at: nil, + last_successful_heartbeat_monotonic_at: nil, heartbeat_watchdog: nil, # Last successful heartbeat timing for diagnostics last_heartbeat_timing: nil, @@ -160,6 +161,7 @@ defmodule DurableServer.LifecycleManager do @default_parallel_restart_batch_size 50 # Default threshold for considering a node's heartbeat stale @default_heartbeat_staleness_threshold_ms :timer.seconds(30) + @default_heartbeat_future_skew_tolerance_ms :timer.seconds(5) # Threshold for considering a node unhealthy when finding eligible placement nodes @node_health_staleness_threshold_ms :timer.seconds(50) @resource_check_interval_ms :timer.seconds(60) @@ -285,8 +287,8 @@ defmodule DurableServer.LifecycleManager do defp high_cardinality_diag_key?(_), do: false - def stop_discovery(supervisor_name) do - GenServer.call(name(supervisor_name), :stop_discovery) + def stop_discovery(supervisor_name, timeout \\ 5_000) do + GenServer.call(name(supervisor_name), :stop_discovery, timeout) end @impl true @@ -347,6 +349,7 @@ defmodule DurableServer.LifecycleManager do restart_claim_gate_disable_after_ms: @restart_claim_gate_disable_after_ms, heartbeat_interval_ms: 10_000, heartbeat_staleness_threshold_ms: @default_heartbeat_staleness_threshold_ms, + heartbeat_future_skew_tolerance_ms: @default_heartbeat_future_skew_tolerance_ms, heartbeat_tracking_mode: :poll, heartbeat_reconcile_interval_ms: 10_000, prefix: "test/" @@ -360,6 +363,10 @@ defmodule DurableServer.LifecycleManager do |> Map.put_new(:restart_claim_gate_expand_after_ms, @restart_claim_gate_expand_after_ms) |> Map.put_new(:restart_claim_gate_disable_after_ms, @restart_claim_gate_disable_after_ms) |> Map.put_new(:heartbeat_staleness_threshold_ms, @default_heartbeat_staleness_threshold_ms) + |> Map.put_new( + :heartbeat_future_skew_tolerance_ms, + @default_heartbeat_future_skew_tolerance_ms + ) |> Map.put_new(:heartbeat_tracking_mode, :poll) |> Map.put_new(:heartbeat_reconcile_interval_ms, 10_000) @@ -456,7 +463,8 @@ defmodule DurableServer.LifecycleManager do # we MUST start with a populated node heartbeat cache # perform_heartbeat writes our heartbeat and refreshes the node health cache - {timing, heartbeat_entry} = perform_heartbeat(state, refresh_cache?: true) + {timing, heartbeat_entry, heartbeat_monotonic_at} = + perform_heartbeat(state, refresh_cache?: true) # Join Group with heartbeat data so other nodes see us instantly via peer_connect. # S3 is the source of truth for liveness; Group is the fast path for discovery. @@ -466,7 +474,7 @@ defmodule DurableServer.LifecycleManager do HeartbeatWatchdog.arm( state.heartbeat_watchdog, self(), - heartbeat_entry_timestamp(heartbeat_entry), + heartbeat_monotonic_at, heartbeat_hard_deadline_ms(state) ) @@ -474,15 +482,16 @@ defmodule DurableServer.LifecycleManager do %{ state | last_successful_heartbeat_at: heartbeat_entry_timestamp(heartbeat_entry), + last_successful_heartbeat_monotonic_at: heartbeat_monotonic_at, last_heartbeat_timing: timing }} end @impl true def handle_info(:heartbeat, %LifecycleManager{} = state) do - now = System.system_time(:millisecond) + now = System.monotonic_time(:millisecond) deadline_ms = heartbeat_hard_deadline_ms(state) - deadline_at = state.last_successful_heartbeat_at + deadline_ms + deadline_at = state.last_successful_heartbeat_monotonic_at + deadline_ms # Check if we've already exceeded the deadline if now >= deadline_at do @@ -498,15 +507,11 @@ defmodule DurableServer.LifecycleManager do task = Task.Supervisor.async(state.task_sup, fn -> :ok = HeartbeatWatchdog.track_heartbeat_task(heartbeat_watchdog, owner, self()) - {timing, heartbeat_entry} = perform_heartbeat(state) + {timing, heartbeat_entry, heartbeat_monotonic_at} = perform_heartbeat(state) - HeartbeatWatchdog.renew( - heartbeat_watchdog, - owner, - heartbeat_entry_timestamp(heartbeat_entry) - ) + HeartbeatWatchdog.renew(heartbeat_watchdog, owner, heartbeat_monotonic_at) - {:heartbeat, {timing, heartbeat_entry}} + {:heartbeat, {timing, heartbeat_entry, heartbeat_monotonic_at}} end) {:noreply, %{state | current_heartbeat_task: task}} @@ -589,7 +594,7 @@ defmodule DurableServer.LifecycleManager do end task = - Task.Supervisor.async(state.task_sup, fn -> + Task.Supervisor.async_nolink(state.task_sup, fn -> {:discover, discover_and_restart_servers(state)} end) @@ -623,7 +628,7 @@ defmodule DurableServer.LifecycleManager do end def handle_info( - {ref, {:heartbeat, {timing, heartbeat_entry}}}, + {ref, {:heartbeat, {timing, heartbeat_entry, heartbeat_monotonic_at}}}, %LifecycleManager{current_heartbeat_task: %Task{ref: ref}} = state ) do # heartbeat task completed successfully and already renewed the watchdog directly. @@ -646,6 +651,7 @@ defmodule DurableServer.LifecycleManager do state | current_heartbeat_task: nil, last_successful_heartbeat_at: heartbeat_entry_timestamp(heartbeat_entry), + last_successful_heartbeat_monotonic_at: heartbeat_monotonic_at, last_heartbeat_timing: timing }} end @@ -703,15 +709,20 @@ defmodule DurableServer.LifecycleManager do # without waiting for the next periodic heartbeat tick. state = case write_node_heartbeat(state) do - {:ok, heartbeat_entry} -> + {:ok, {heartbeat_entry, heartbeat_monotonic_at}} -> HeartbeatWatchdog.renew( state.heartbeat_watchdog, self(), - heartbeat_entry_timestamp(heartbeat_entry) + heartbeat_monotonic_at ) join_group_heartbeat(state, heartbeat_entry) - %{state | last_successful_heartbeat_at: heartbeat_entry_timestamp(heartbeat_entry)} + + %{ + state + | last_successful_heartbeat_at: heartbeat_entry_timestamp(heartbeat_entry), + last_successful_heartbeat_monotonic_at: heartbeat_monotonic_at + } {:error, reason} -> log(state, :warning, fn -> @@ -760,15 +771,15 @@ defmodule DurableServer.LifecycleManager do # Do the critical heartbeat PUT inline put_start = System.monotonic_time(:millisecond) - heartbeat_entry = + {heartbeat_entry, heartbeat_monotonic_at} = case write_node_heartbeat(state) do - {:ok, entry} -> + {:ok, {entry, monotonic_at}} -> log(state, :debug, fn -> put_duration = System.monotonic_time(:millisecond) - put_start "Node heartbeat written successfully in #{put_duration}ms" end) - entry + {entry, monotonic_at} {:error, reason} -> put_duration = System.monotonic_time(:millisecond) - put_start @@ -792,7 +803,11 @@ defmodule DurableServer.LifecycleManager do total_duration = System.monotonic_time(:millisecond) - start_time - {%{put_ms: put_duration, cache_ms: cache_duration, total_ms: total_duration}, heartbeat_entry} + { + %{put_ms: put_duration, cache_ms: cache_duration, total_ms: total_duration}, + heartbeat_entry, + heartbeat_monotonic_at + } end defp refresh_heartbeat_cache_with_timing!(%LifecycleManager{} = state) do @@ -947,6 +962,7 @@ defmodule DurableServer.LifecycleManager do node_str = to_string(Node.self()) node_ref = DurableServer.Supervisor.node_ref(state.supervisor_name) current_time = System.system_time(:millisecond) + current_monotonic_time = System.monotonic_time(:millisecond) # calculate capacity and resource info capacity = DurableServer.Supervisor.current_capacity(state.supervisor_name) @@ -983,14 +999,15 @@ defmodule DurableServer.LifecycleManager do key = "#{state.prefix}__nodes/#{node_str}" entry = {node_str, node_ref, current_time, capacity, resources, env_vars, heartbeat_meta} - deadline_at = heartbeat_deadline_at(state, state.last_successful_heartbeat_at) + deadline_at = + heartbeat_deadline_at(state, state.last_successful_heartbeat_monotonic_at) case put_heartbeat_until_deadline(state, key, heartbeat_data, deadline_at) do {:ok, _} -> # update local ets cache with full capacity info :ets.insert(state.heartbeat_table, entry) - {:ok, entry} + {:ok, {entry, current_monotonic_time}} {:error, reason} -> {:error, reason} @@ -1008,7 +1025,7 @@ defmodule DurableServer.LifecycleManager do deadline_at, attempt ) do - remaining_ms = max(deadline_at - System.system_time(:millisecond), 0) + remaining_ms = max(deadline_at - System.monotonic_time(:millisecond), 0) if remaining_ms <= 0 do {:error, :heartbeat_deadline_exceeded} @@ -1039,11 +1056,11 @@ defmodule DurableServer.LifecycleManager do end defp heartbeat_deadline_at(%LifecycleManager{} = state, nil), - do: System.system_time(:millisecond) + heartbeat_hard_deadline_ms(state) + do: System.monotonic_time(:millisecond) + heartbeat_hard_deadline_ms(state) - defp heartbeat_deadline_at(%LifecycleManager{} = state, last_successful_heartbeat_at) - when is_integer(last_successful_heartbeat_at) do - last_successful_heartbeat_at + heartbeat_hard_deadline_ms(state) + defp heartbeat_deadline_at(%LifecycleManager{} = state, last_successful_heartbeat_monotonic_at) + when is_integer(last_successful_heartbeat_monotonic_at) do + last_successful_heartbeat_monotonic_at + heartbeat_hard_deadline_ms(state) end defp heartbeat_entry_timestamp( @@ -1056,6 +1073,15 @@ defmodule DurableServer.LifecycleManager do defp heartbeat_write_retryable?({:mirror_failed, reason}), do: heartbeat_write_retryable?(reason) + defp heartbeat_write_retryable?(%Req.Response{status: status}), + do: status in [408, 429, 500, 502, 503, 504] + + defp heartbeat_write_retryable?(%Req.TransportError{reason: reason}), + do: reason in [:timeout, :econnrefused, :closed] + + defp heartbeat_write_retryable?(%Req.HTTPError{protocol: :http2, reason: :unprocessed}), + do: true + defp heartbeat_write_retryable?(reason) do reason in [ :no_quorum, @@ -1167,8 +1193,14 @@ defmodule DurableServer.LifecycleManager do _ -> false end) - {dead_nodes, errors} = + {future_skewed_nodes, non_future_skewed_rest} = Enum.split_with(non_missing_rest, fn + {:future_skewed, _key} -> true + _ -> false + end) + + {dead_nodes, errors} = + Enum.split_with(non_future_skewed_rest, fn {:dead, _key, _node, _node_ref, _timestamp} -> true _ -> false end) @@ -1192,6 +1224,12 @@ defmodule DurableServer.LifecycleManager do end) end + if future_skewed_nodes != [] do + log(state, :warning, fn -> + "Ignored #{length(future_skewed_nodes)} heartbeat(s) beyond the future clock-skew tolerance" + end) + end + error_count = length(errors) # extract heartbeat data for alive nodes @@ -1230,17 +1268,35 @@ defmodule DurableServer.LifecycleManager do defp process_heartbeat_list_entry( %{key: key, body: body}, - _state, + %LifecycleManager{} = state, current_time, dead_node_threshold_ms ) when is_binary(key) and is_integer(current_time) and is_integer(dead_node_threshold_ms) do case parse_heartbeat_data(body) do {:ok, {node, node_ref, timestamp, capacity, resources, env_vars, heartbeat_meta}} -> - if current_time - timestamp > dead_node_threshold_ms do - {:dead, key, node, node_ref, timestamp} - else - {:alive, node, node_ref, timestamp, capacity, resources, env_vars, heartbeat_meta} + future_skew_tolerance_ms = state.config.heartbeat_future_skew_tolerance_ms + + cond do + not heartbeat_timestamp_acceptable?( + timestamp, + current_time, + future_skew_tolerance_ms + ) -> + # Do not delete a future-skewed record: its writer may still be alive, + # but never admit it into the local liveness cache. + {:future_skewed, key} + + heartbeat_fresh?( + timestamp, + current_time, + dead_node_threshold_ms, + future_skew_tolerance_ms + ) -> + {:alive, node, node_ref, timestamp, capacity, resources, env_vars, heartbeat_meta} + + true -> + {:dead, key, node, node_ref, timestamp} end {:error, :invalid_format} -> @@ -1380,17 +1436,20 @@ defmodule DurableServer.LifecycleManager do table_name = heartbeat_table_name(supervisor_name) %{ - heartbeat_staleness_threshold_ms: heartbeat_staleness_threshold_ms - } = - DurableServer.Supervisor.__get_config__(supervisor_name) + heartbeat_staleness_threshold_ms: heartbeat_staleness_threshold_ms, + heartbeat_future_skew_tolerance_ms: future_skew_tolerance_ms + } = DurableServer.Supervisor.__get_config__(supervisor_name) case :ets.lookup(table_name, node_str) do [{^node_str, node_ref, timestamp, capacity, resources, env_vars, heartbeat_meta}] -> current_time = System.system_time(:millisecond) - if current_time - timestamp > heartbeat_staleness_threshold_ms do - :stale - else + if heartbeat_fresh?( + timestamp, + current_time, + heartbeat_staleness_threshold_ms, + future_skew_tolerance_ms + ) do {:healthy, %{ node_ref: node_ref, @@ -1399,6 +1458,8 @@ defmodule DurableServer.LifecycleManager do env_vars: env_vars, heartbeat_meta: heartbeat_meta }} + else + :stale end # Node not found in heartbeat table @@ -1433,6 +1494,7 @@ defmodule DurableServer.LifecycleManager do config = DurableServer.Supervisor.__get_config__(supervisor_name) prefix = config.prefix heartbeat_staleness_threshold_ms = config.heartbeat_staleness_threshold_ms + future_skew_tolerance_ms = config.heartbeat_future_skew_tolerance_ms storage_backend = config.storage_backend heartbeat_store = Map.get(config, :heartbeat_backend, storage_backend) @@ -1445,12 +1507,17 @@ defmodule DurableServer.LifecycleManager do {_node_str, node_ref, timestamp, _capacity, _resources, _env_vars, _heartbeat_meta}} -> current_time = System.system_time(:millisecond) - if current_time - timestamp > heartbeat_staleness_threshold_ms do - :stale - else + if heartbeat_fresh?( + timestamp, + current_time, + heartbeat_staleness_threshold_ms, + future_skew_tolerance_ms + ) do # Cache the fetched heartbeat so subsequent lookups are fast cache_fetched_heartbeat(supervisor_name, body) {:healthy, %{node_ref: node_ref}} + else + :stale end {:error, :invalid_format} -> @@ -1589,6 +1656,15 @@ defmodule DurableServer.LifecycleManager do "heartbeat_staleness_threshold_ms must be an integer greater than #{@heartbeat_deadline_buffer_ms}, got: #{inspect(other)}" end + case Map.fetch!(config, :heartbeat_future_skew_tolerance_ms) do + value when is_integer(value) and value >= 0 -> + :ok + + other -> + raise ArgumentError, + "heartbeat_future_skew_tolerance_ms must be a non-negative integer, got: #{inspect(other)}" + end + max_heartbeat_interval = div(Map.fetch!(config, :heartbeat_staleness_threshold_ms), 2) case Map.fetch!(config, :heartbeat_interval_ms) do @@ -1623,6 +1699,33 @@ defmodule DurableServer.LifecycleManager do Map.fetch!(state.config, :heartbeat_staleness_threshold_ms) end + defp heartbeat_fresh?(timestamp, now, staleness_threshold_ms, future_skew_tolerance_ms) + when is_integer(timestamp) and is_integer(now) do + age_ms = now - timestamp + age_ms >= -future_skew_tolerance_ms and age_ms <= staleness_threshold_ms + end + + defp heartbeat_fresh?(_timestamp, _now, _staleness_threshold_ms, _future_skew_tolerance_ms), + do: false + + defp heartbeat_timestamp_acceptable?(timestamp, now, future_skew_tolerance_ms) + when is_integer(timestamp), + do: timestamp <= now + future_skew_tolerance_ms + + defp heartbeat_timestamp_acceptable?(_timestamp, _now, _future_skew_tolerance_ms), do: false + + defp configured_future_skew_tolerance_ms(supervisor_name) do + supervisor_name + |> DurableServer.Supervisor.__get_config__() + |> Map.get( + :heartbeat_future_skew_tolerance_ms, + @default_heartbeat_future_skew_tolerance_ms + ) + rescue + RuntimeError -> @default_heartbeat_future_skew_tolerance_ms + ArgumentError -> @default_heartbeat_future_skew_tolerance_ms + end + defp preferred_restart_claimer?( supervisor_name, %Meta{} = meta, @@ -1729,6 +1832,7 @@ defmodule DurableServer.LifecycleManager do ) delays = get_sticky_placement_delays(supervisor_name, meta.module) + future_skew_tolerance_ms = configured_future_skew_tolerance_ms(supervisor_name) node_health = lookup_node_health(meta) @@ -1748,7 +1852,12 @@ defmodule DurableServer.LifecycleManager do try do node = String.to_existing_atom(node_str) - if now - timestamp <= @node_health_staleness_threshold_ms do + if heartbeat_fresh?( + timestamp, + now, + @node_health_staleness_threshold_ms, + future_skew_tolerance_ms + ) do candidate_health = {:healthy, %{ @@ -2723,15 +2832,53 @@ defmodule DurableServer.LifecycleManager do the local disk, so rejecting based on disk usage would be counterproductive. """ def check_capacity(supervisor_name, module, opts \\ []) do - opts = Keyword.validate!(opts, [:bypass_disk_check]) + opts = Keyword.validate!(opts, [:bypass_disk_check, :skip_count_limits]) + + count_result = + if Keyword.get(opts, :skip_count_limits, false) do + :ok + else + check_count_limits(supervisor_name, module) + end with :ok <- check_shutting_down(supervisor_name), - :ok <- check_count_limits(supervisor_name, module), - :ok <- check_resource_limits(supervisor_name, opts) do + :ok <- count_result, + :ok <- check_resource_limits(supervisor_name, Keyword.take(opts, [:bypass_disk_check])) do :ok end end + @doc false + def reserve_capacity(supervisor_name, module, opts \\ []) do + opts = Keyword.validate!(opts, [:bypass_disk_check]) + + %{ets_table: table_name} = + DurableServer.Supervisor.__get_config__(supervisor_name) + + with :ok <- check_shutting_down(supervisor_name), + :ok <- check_resource_limits(supervisor_name, opts), + {:ok, counter_keys} <- reserve_count_slots(table_name, module) do + {:ok, start_capacity_reservation_keeper(table_name, counter_keys, self())} + end + end + + @doc false + def commit_capacity_reservation(nil, _pid), do: :ok + + def commit_capacity_reservation(reservation, pid) + when is_pid(reservation) and is_pid(pid) do + send(reservation, {:commit, pid}) + :ok + end + + @doc false + def cancel_capacity_reservation(nil), do: :ok + + def cancel_capacity_reservation(reservation) when is_pid(reservation) do + send(reservation, :cancel) + :ok + end + defp check_shutting_down(supervisor_name) do %{ets_table: table_name} = DurableServer.Supervisor.__get_config__(supervisor_name) @@ -2753,13 +2900,8 @@ defmodule DurableServer.LifecycleManager do :ok max_children_limits -> - global_count = Group.local_registry_count(supervisor_name) - - module_count = - Group.local_member_count( - supervisor_name, - DurableServer.Supervisor.__module_group_prefix__(module) - ) + global_count = capacity_counter(table_name, :capacity_count_total) + module_count = capacity_counter(table_name, capacity_module_counter_key(module)) total_limit = max_children_limits[:total] module_limit = max_children_limits[module] @@ -2780,6 +2922,96 @@ defmodule DurableServer.LifecycleManager do end end + defp reserve_count_slots(table_name, module) do + [{:capacity_limits, limits}] = :ets.lookup(table_name, :capacity_limits) + max_children_limits = Map.get(limits, :max_children, %{}) + total_limit = max_children_limits[:total] + module_limit = max_children_limits[module] + + case reserve_capacity_counter( + table_name, + :capacity_count_total, + total_limit, + :max_children_total, + %{limit: total_limit} + ) do + {:ok, total_keys} -> + case reserve_capacity_counter( + table_name, + capacity_module_counter_key(module), + module_limit, + :max_children_module, + %{module: module, limit: module_limit} + ) do + {:ok, module_keys} -> + {:ok, total_keys ++ module_keys} + + {:error, reason} -> + release_capacity_counters(table_name, total_keys) + {:error, reason} + end + + {:error, reason} -> + {:error, reason} + end + end + + defp reserve_capacity_counter(_table_name, _key, nil, _reason, _details), do: {:ok, []} + + defp reserve_capacity_counter(table_name, key, limit, reason, details) do + current = :ets.update_counter(table_name, key, {2, 1}, {key, 0}) + + if current <= limit do + {:ok, [key]} + else + :ets.update_counter(table_name, key, {2, -1}) + + {:error, {:limit_reached, reason, Map.put(details, :current, current - 1)}} + end + end + + defp start_capacity_reservation_keeper(_table_name, [], _owner), do: nil + + defp start_capacity_reservation_keeper(table_name, counter_keys, owner) do + spawn(fn -> + owner_ref = Process.monitor(owner) + + receive do + {:commit, child_pid} when is_pid(child_pid) -> + Process.demonitor(owner_ref, [:flush]) + child_ref = Process.monitor(child_pid) + + receive do + {:DOWN, ^child_ref, :process, ^child_pid, _reason} -> + release_capacity_counters(table_name, counter_keys) + end + + :cancel -> + Process.demonitor(owner_ref, [:flush]) + release_capacity_counters(table_name, counter_keys) + + {:DOWN, ^owner_ref, :process, ^owner, _reason} -> + release_capacity_counters(table_name, counter_keys) + end + end) + end + + defp release_capacity_counters(table_name, counter_keys) do + Enum.each(counter_keys, &:ets.update_counter(table_name, &1, {2, -1})) + :ok + rescue + ArgumentError -> :ok + end + + defp capacity_counter(table_name, key) do + case :ets.lookup(table_name, key) do + [{^key, count}] -> count + [] -> 0 + end + end + + defp capacity_module_counter_key(module), do: {:capacity_count_module, module} + defp check_resource_limits(supervisor_name, opts) do opts = Keyword.validate!(opts, [:bypass_disk_check]) bypass_disk_check = Keyword.get(opts, :bypass_disk_check, false) @@ -3095,6 +3327,7 @@ defmodule DurableServer.LifecycleManager do end now = System.system_time(:millisecond) + future_skew_tolerance_ms = configured_future_skew_tolerance_ms(supervisor_name) # Merge two data sources per node, picking whichever has the more recent timestamp: # 1. S3 heartbeat ETS cache — source of truth for "can this node reach S3?" @@ -3108,10 +3341,13 @@ defmodule DurableServer.LifecycleManager do try do node = String.to_existing_atom(node_str) - heartbeat_age_ms = now - timestamp - health = - if heartbeat_age_ms <= @node_health_staleness_threshold_ms do + if heartbeat_fresh?( + timestamp, + now, + @node_health_staleness_threshold_ms, + future_skew_tolerance_ms + ) do {:healthy, %{ node_ref: node_ref, @@ -3152,9 +3388,17 @@ defmodule DurableServer.LifecycleManager do # Merge heartbeat data from S3 ETS cache and Group PG members. # For each node present in both sources, pick the entry with the more recent timestamp. # Returns a list of heartbeat tuples in the same shape as the ETS entries. - defp merge_heartbeat_sources(supervisor_name, heartbeat_table, _now) do - # Start with ETS cache as the base (keyed by node_str) - ets_entries = :ets.tab2list(heartbeat_table) + defp merge_heartbeat_sources(supervisor_name, heartbeat_table, now) do + future_skew_tolerance_ms = configured_future_skew_tolerance_ms(supervisor_name) + + # Start with ETS cache as the base (keyed by node_str). Ignore timestamps + # too far in the future before choosing the newest source for a node. + ets_entries = + heartbeat_table + |> :ets.tab2list() + |> Enum.filter(fn {_node, _ref, timestamp, _capacity, _resources, _env, _meta} -> + heartbeat_timestamp_acceptable?(timestamp, now, future_skew_tolerance_ms) + end) ets_map = Map.new(ets_entries, fn {node_str, _node_ref, _ts, _cap, _res, _env, _meta} = entry -> @@ -3179,18 +3423,22 @@ defmodule DurableServer.LifecycleManager do {node_str, meta.node_ref, group_ts, meta.capacity, meta.resources, meta.env_vars, meta.heartbeat_meta} - case Map.get(acc, node_str) do - nil -> - # Node only in Group (new node, not yet in S3 cache) — use Group data - Map.put(acc, node_str, group_entry) + if heartbeat_timestamp_acceptable?(group_ts, now, future_skew_tolerance_ms) do + case Map.get(acc, node_str) do + nil -> + # Node only in Group (new node, not yet in S3 cache) — use Group data + Map.put(acc, node_str, group_entry) - {_ns, _nr, ets_ts, _c, _r, _e, _m} when group_ts > ets_ts -> - # Group data is more recent — use it - Map.put(acc, node_str, group_entry) + {_ns, _nr, ets_ts, _c, _r, _e, _m} when group_ts > ets_ts -> + # Group data is more recent — use it + Map.put(acc, node_str, group_entry) - _ets_entry -> - # ETS data is same age or more recent — keep it - acc + _ets_entry -> + # ETS data is same age or more recent — keep it + acc + end + else + acc end end) diff --git a/lib/durable_server/meta.ex b/lib/durable_server/meta.ex index 20737bc..a6fc5ab 100644 --- a/lib/durable_server/meta.ex +++ b/lib/durable_server/meta.ex @@ -31,6 +31,9 @@ defmodule DurableServer.Meta do @permanently_crashed :permanently_crashed @deleting :deleting + @max_metadata_binary_bytes 65_536 + @max_metadata_base64_bytes div(@max_metadata_binary_bytes + 2, 3) * 4 + @statuses [ @stopped_graceful, @stopped_permanent, @@ -40,13 +43,55 @@ defmodule DurableServer.Meta do @deleting ] - # note we are using struct! so keys cannot be removed (but can be added) - # if we remove keys we need to change the decode function to pluck out valid keys def decode_from_binary(meta_str, %{key: key, prefix: prefix}) when is_binary(meta_str) do - meta_str - |> Base.decode64!() - |> :erlang.binary_to_term() - |> from_storage_term(%{key: key, prefix: prefix}) + with :ok <- validate_encoded_size(meta_str), + {:ok, binary} <- decode_base64(meta_str), + :ok <- validate_binary_format(binary), + {:ok, term} <- decode_safe_term(binary) do + from_storage_term(term, %{key: key, prefix: prefix}) + else + {:error, reason} -> + raise ArgumentError, to_string(reason) + end + rescue + error in ArgumentError -> + raise ArgumentError, "invalid persisted metadata: #{Exception.message(error)}" + end + + defp validate_encoded_size(meta_str) do + if byte_size(meta_str) <= @max_metadata_base64_bytes do + :ok + else + {:error, "encoded value exceeds #{@max_metadata_binary_bytes} byte limit"} + end + end + + defp decode_base64(meta_str) do + case Base.decode64(meta_str) do + {:ok, binary} -> {:ok, binary} + :error -> {:error, "invalid base64"} + end + end + + # This encoder has never emitted compressed ETF. Rejecting it prevents a tiny + # persisted value from expanding without a useful pre-decode size bound. + defp validate_binary_format(<<131, 80, _compressed_size::32, _rest::binary>>), + do: {:error, "compressed external terms are not accepted"} + + defp validate_binary_format(<<131, _rest::binary>> = binary) do + if byte_size(binary) <= @max_metadata_binary_bytes do + :ok + else + {:error, "decoded value exceeds #{@max_metadata_binary_bytes} byte limit"} + end + end + + defp validate_binary_format(_binary), do: {:error, "invalid external term"} + + defp decode_safe_term(binary) do + {:ok, :erlang.binary_to_term(binary, [:safe])} + rescue + _error -> {:error, "unsafe or malformed external term"} end def encode_to_binary(%Meta{} = meta) do @@ -56,21 +101,126 @@ defmodule DurableServer.Meta do |> Base.encode64() end - def from_storage_term(%Meta{} = meta, %{key: key, prefix: prefix}) do - %{meta | key: key, prefix: prefix} + def from_storage_term(%Meta{} = meta, context) do + meta + |> Map.from_struct() + |> from_storage_term(context) end def from_storage_term(meta_map, %{key: key, prefix: prefix}) when is_map(meta_map) do - valid_keys = Map.keys(%__MODULE__{}) + unless safe_metadata_term?(meta_map) do + raise ArgumentError, "metadata contains unsupported external terms" + end + + valid_keys = Map.keys(Map.from_struct(%Meta{})) meta_map = Map.take(meta_map, valid_keys) + validate_storage_term!(meta_map) + %{struct!(Meta, meta_map) | key: key, prefix: prefix} + end + + def from_storage_term(_term, _context) do + raise ArgumentError, "invalid meta storage term" + end - unless Map.has_key?(meta_map, :status) do - raise ArgumentError, "invalid meta storage term: #{inspect(meta_map)}" + defp validate_storage_term!(meta) do + unless Map.has_key?(meta, :status) do + raise ArgumentError, "metadata is missing required field :status" end - %{struct!(Meta, meta_map) | key: key, prefix: prefix} + validate_field!(meta, :vsn, &(is_integer(&1) and &1 > 0), "a positive integer") + validate_field!(meta, :module, &is_atom/1, "an atom") + validate_field!(meta, :permanent, &is_boolean/1, "a boolean") + validate_field!(meta, :pid, &(is_nil(&1) or is_pid(&1)), "a pid or nil") + validate_field!(meta, :status, &(&1 in @statuses), "a supported status") + validate_field!(meta, :sticky_placement, &(is_nil(&1) or is_list(&1)), "a list or nil") + validate_field!(meta, :sticky_placement_history, &is_list/1, "a list") + validate_field!(meta, :supervisor, &(is_nil(&1) or is_atom(&1)), "an atom or nil") + validate_field!(meta, :task_supervisor, &(is_nil(&1) or is_atom(&1)), "an atom or nil") + validate_field!(meta, :dynamic_supervisor, &(is_nil(&1) or is_atom(&1)), "an atom or nil") + + validate_field!( + meta, + :node_ref, + &(is_nil(&1) or is_integer(&1) or is_binary(&1)), + "an integer, binary, or nil" + ) + + validate_field!(meta, :node_str, &(is_nil(&1) or is_binary(&1)), "a binary or nil") + + validate_field!( + meta, + :last_heartbeat_at, + &(is_nil(&1) or is_integer(&1)), + "an integer or nil" + ) + + validate_field!(meta, :crash_history, &is_list/1, "a list") + + validate_field!( + meta, + :restart_attempt_node, + &(is_nil(&1) or is_binary(&1)), + "a binary or nil" + ) + + validate_field!( + meta, + :restart_attempt_time, + &(is_nil(&1) or is_integer(&1)), + "an integer or nil" + ) + + validate_field!( + meta, + :restart_attempt_ttl, + &(is_nil(&1) or is_integer(&1)), + "an integer or nil" + ) + + validate_field!( + meta, + :init_from_ref, + &(is_nil(&1) or is_reference(&1)), + "a reference or nil" + ) + + validate_field!( + meta, + :init_from_pid, + &(is_nil(&1) or is_pid(&1)), + "a pid or nil" + ) + + :ok end + defp validate_field!(meta, key, predicate, expected) do + value = Map.get(meta, key, Map.fetch!(Map.from_struct(%Meta{}), key)) + + unless predicate.(value) do + raise ArgumentError, "invalid metadata field #{inspect(key)}: expected #{expected}" + end + end + + defp safe_metadata_term?(term) + when is_atom(term) or is_binary(term) or is_number(term) or is_pid(term) or + is_reference(term), + do: true + + defp safe_metadata_term?(list) when is_list(list), do: Enum.all?(list, &safe_metadata_term?/1) + + defp safe_metadata_term?(tuple) when is_tuple(tuple) do + tuple |> Tuple.to_list() |> Enum.all?(&safe_metadata_term?/1) + end + + defp safe_metadata_term?(map) when is_map(map) do + Enum.all?(map, fn {key, value} -> + safe_metadata_term?(key) and safe_metadata_term?(value) + end) + end + + defp safe_metadata_term?(_term), do: false + def to_storage_term(%Meta{} = meta) do meta |> Map.from_struct() diff --git a/lib/durable_server/object_store.ex b/lib/durable_server/object_store.ex index 6540987..be82e33 100644 --- a/lib/durable_server/object_store.ex +++ b/lib/durable_server/object_store.ex @@ -15,6 +15,7 @@ defmodule DurableServer.ObjectStore do """ @default_timeout 30_000 + @max_iam_xml_bytes 1_048_576 @derive {Inspect, only: []} defstruct access_key_id: nil, @@ -34,7 +35,6 @@ defmodule DurableServer.ObjectStore do task_supervisor: nil require Logger - require ReqS3 # Import SweetXml for XML parsing and sigils import SweetXml @@ -258,8 +258,8 @@ defmodule DurableServer.ObjectStore do :ok -> Logger.info("Successfully created IAM user: #{user_name}") - {:error, reason} -> - Logger.warning("Failed to create IAM user (may already exist): #{inspect(reason)}") + {:error, _reason} -> + Logger.warning("Failed to create IAM user (it may already exist)") end end @@ -362,6 +362,41 @@ defmodule DurableServer.ObjectStore do end end + @doc false + def __parse_iam_xml__(xml_body) + when is_binary(xml_body) and byte_size(xml_body) <= @max_iam_xml_bytes do + {:ok, SweetXml.parse(xml_body, dtd: :none, quiet: true)} + rescue + _error -> {:error, :invalid_xml} + catch + _kind, _reason -> {:error, :invalid_xml} + end + + def __parse_iam_xml__(_xml_body), do: {:error, :invalid_xml} + + @doc false + def __parse_create_access_key_response__(xml_body, user_name) + when is_binary(xml_body) and is_binary(user_name) do + with {:ok, xml_document} <- __parse_iam_xml__(xml_body), + access_key_id when access_key_id != "" <- + xpath(xml_document, ~x"//AccessKeyId/text()"s), + secret_access_key when secret_access_key != "" <- + xpath(xml_document, ~x"//SecretAccessKey/text()"s) do + {:ok, + %{ + access_key_id: access_key_id, + secret_access_key: secret_access_key, + user_name: user_name + }} + else + {:error, :invalid_xml} -> + {:error, {:xml_parse_error, :invalid_xml}} + + _missing_value -> + {:error, {:invalid_response, "Missing access key information in response"}} + end + end + # Creates an IAM user (for LocalStack only) defp create_iam_user(client, user_name) do Logger.info("Creating IAM user: #{user_name}") @@ -401,34 +436,15 @@ defmodule DurableServer.ObjectStore do } case iam_request(client, params) do - {:ok, %{body: xml_body} = result} -> - try do - # Parse the XML response manually - Logger.debug(fn -> "Received CreateAccessKey response: #{inspect(xml_body)}" end) - - # Extract the access key ID and secret from the XML - access_key_id = xml_body |> xpath(~x"//AccessKeyId/text()"s) - secret_access_key = xml_body |> xpath(~x"//SecretAccessKey/text()"s) - - if access_key_id != "" and secret_access_key != "" do - Logger.info("Successfully created access key with ID: #{access_key_id}") - - {:ok, - %{ - access_key_id: access_key_id, - secret_access_key: secret_access_key, - user_name: user_name - }} - else - {:error, {:invalid_response, "Missing access key information in response"}} - end - rescue - error -> - Logger.error( - "Failed to parse CreateAccessKey XML response: #{inspect(error)} #{inspect(result)}" - ) - - {:error, {:xml_parse_error, error}} + {:ok, %{body: xml_body}} -> + case __parse_create_access_key_response__(xml_body, user_name) do + {:ok, _credentials} = ok -> + Logger.info("Successfully created access key") + ok + + {:error, reason} = error -> + Logger.error("Failed to parse CreateAccessKey XML response: #{inspect(reason)}") + error end {:error, reason} -> @@ -450,16 +466,20 @@ defmodule DurableServer.ObjectStore do case iam_request(client, params) do {:ok, %{body: xml_body}} when is_binary(xml_body) and xml_body != "" -> # Parse the XML response manually - Logger.debug( - "Received CreatePolicy response: #{inspect(String.slice(xml_body, 0, 100))}..." - ) + Logger.debug("Received CreatePolicy response") try do + xml_document = + case __parse_iam_xml__(xml_body) do + {:ok, document} -> document + {:error, :invalid_xml} -> raise ArgumentError, "invalid IAM XML response" + end + # Extract the policy ARN from the XML - policy_arn = xml_body |> xpath(~x"//Arn/text()"s) + policy_arn = xml_document |> xpath(~x"//Arn/text()"s) if policy_arn != "" do - Logger.info("Successfully created policy with ARN: #{policy_arn}") + Logger.info("Successfully created IAM policy") {:ok, %{policy_arn: policy_arn}} else {:error, {:invalid_response, "Missing policy ARN in response"}} @@ -467,14 +487,14 @@ defmodule DurableServer.ObjectStore do catch kind, reason -> Logger.error( - "Failed to parse CreatePolicy XML response: #{inspect(kind: kind, reason: reason, body: xml_body)}" + "Failed to parse CreatePolicy XML response: #{inspect(kind: kind, reason: reason)}" ) {:error, {:xml_parse_error, {kind, reason}}} end - {:ok, %{body: xml_body}} -> - Logger.error("CreatePolicy returned empty or invalid body: #{inspect(xml_body)}") + {:ok, %{body: _xml_body}} -> + Logger.error("CreatePolicy returned an empty or invalid body") {:error, {:invalid_response, "Empty or invalid response body"}} {:error, reason} -> @@ -484,7 +504,7 @@ defmodule DurableServer.ObjectStore do # Attaches a policy to a user (access key) using the AttachUserPolicy API operation via Req defp attach_user_policy(client, user_name, policy_arn) do - Logger.info("Attaching policy #{policy_arn} to user: #{user_name}") + Logger.info("Attaching IAM policy to user") params = %{ "Action" => "AttachUserPolicy", @@ -501,7 +521,7 @@ defmodule DurableServer.ObjectStore do # Deletes an access key using the DeleteAccessKey API operation defp delete_access_key(client, access_key_id) do - Logger.info("Deleting access key: #{access_key_id}") + Logger.info("Deleting access key") params = %{ "Action" => "DeleteAccessKey", @@ -519,7 +539,7 @@ defmodule DurableServer.ObjectStore do # Deletes a policy using the DeletePolicy API operation defp delete_policy(client, policy_arn) do - Logger.info("Deleting policy: #{policy_arn}") + Logger.info("Deleting IAM policy") # First detach the policy from the user detach_result = detach_user_policy(client, policy_arn) @@ -538,8 +558,8 @@ defmodule DurableServer.ObjectStore do end # Detaches a policy from all users that it's attached to - defp detach_user_policy(_config, policy_arn) do - Logger.info("Detaching policy: #{policy_arn} from users") + defp detach_user_policy(_config, _policy_arn) do + Logger.info("Detaching IAM policy from users") # We would need to list all users the policy is attached to # For simplicity, we'll just return success for now @@ -574,20 +594,26 @@ defmodule DurableServer.ObjectStore do case iam_request(client, params) do {:ok, %{body: xml_body}} -> - policies = - xml_body - |> xpath(~x"//Policies/member"l) - |> Enum.map(fn policy -> - %{ - arn: xpath(policy, ~x"./Arn/text()"s), - name: xpath(policy, ~x"./PolicyName/text()"s) - } - end) - |> Enum.filter(fn policy -> - String.contains?(policy.name, policy_prefix) - end) - - {:ok, policies} + case __parse_iam_xml__(xml_body) do + {:ok, xml_document} -> + policies = + xml_document + |> xpath(~x"//Policies/member"l) + |> Enum.map(fn policy -> + %{ + arn: xpath(policy, ~x"./Arn/text()"s), + name: xpath(policy, ~x"./PolicyName/text()"s) + } + end) + |> Enum.filter(fn policy -> + String.contains?(policy.name, policy_prefix) + end) + + {:ok, policies} + + {:error, :invalid_xml} -> + {:error, {:xml_parse_error, :invalid_xml}} + end {:error, reason} -> {:error, reason} @@ -618,16 +644,18 @@ defmodule DurableServer.ObjectStore do } with {:ok, %{body: policy_xml}} <- iam_request(client, get_policy_params), - policy_name = policy_xml |> xpath(~x"//PolicyName/text()"s), - default_version = policy_xml |> xpath(~x"//DefaultVersionId/text()"s), + {:ok, policy_document} <- __parse_iam_xml__(policy_xml), + policy_name = policy_document |> xpath(~x"//PolicyName/text()"s), + default_version = policy_document |> xpath(~x"//DefaultVersionId/text()"s), get_version_params = %{ "Action" => "GetPolicyVersion", "Version" => "2010-05-08", "PolicyArn" => policy_arn, "VersionId" => default_version }, - {:ok, %{body: version_xml}} <- iam_request(client, get_version_params) do - encoded_document = version_xml |> xpath(~x"//Document/text()"s) + {:ok, %{body: version_xml}} <- iam_request(client, get_version_params), + {:ok, version_document} <- __parse_iam_xml__(version_xml) do + encoded_document = version_document |> xpath(~x"//Document/text()"s) document = URI.decode(encoded_document) {:ok, @@ -861,9 +889,6 @@ defmodule DurableServer.ObjectStore do {:ok, %{keys: keys, next_continuation_token: nil}} -> {keys, :done} - {:ok, %{keys: keys}} -> - {keys, :done} - {:error, reason} -> case error_handler.(reason) do :halt -> nil diff --git a/lib/durable_server/supervisor.ex b/lib/durable_server/supervisor.ex index 6b75cd1..580a6cd 100644 --- a/lib/durable_server/supervisor.ex +++ b/lib/durable_server/supervisor.ex @@ -57,7 +57,7 @@ defmodule DurableServer.Supervisor do ### Node Heartbeats The LifecycleManager maintains node-level heartbeats in object storage at - `{prefix}nodes/{node_name}` and caches them locally for efficient health + `{prefix}__nodes/{node_name}` and caches them locally for efficient health checking during restart decisions. ## Configuration Options @@ -88,6 +88,8 @@ defmodule DurableServer.Supervisor do - `:heartbeat_interval_ms` - How often to write node heartbeats (default: 10_000) - `:heartbeat_staleness_threshold_ms` - How long a node heartbeat may go without success before the node is considered stale/orphan-claimable (default: 30_000) + - `:heartbeat_future_skew_tolerance_ms` - Maximum accepted clock lead in a persisted + cross-node heartbeat (default: 5_000) - `:heartbeat_tracking_mode` - Heartbeat cache strategy: `:poll` or `:subscribe`. Defaults from backend capabilities. - `:heartbeat_reconcile_interval_ms` - Full heartbeat cache reconcile interval used @@ -229,6 +231,7 @@ defmodule DurableServer.Supervisor do @default_parallel_restart_batch_size 50 @default_restart_start_timeout_ms 30_000 @default_heartbeat_staleness_threshold_ms 30_000 + @default_heartbeat_future_skew_tolerance_ms 5_000 @default_restart_claim_preferred_fanout 2 @default_restart_claim_expanded_fanout 4 @default_restart_claim_gate_expand_after_ms :timer.seconds(30) @@ -597,7 +600,7 @@ defmodule DurableServer.Supervisor do # add total capacity if configured capacity_map = if total_limit = max_children[:total] do - current = Group.local_registry_count(supervisor_name) + current = capacity_count(table_name, :capacity_count_total) Map.put(capacity_map, :total, %{current: current, limit: total_limit}) else capacity_map @@ -609,10 +612,7 @@ defmodule DurableServer.Supervisor do |> Enum.reject(fn {k, _v} -> k == :total end) |> Enum.reduce(capacity_map, fn {module, limit}, acc -> current = - Group.local_member_count( - supervisor_name, - __module_group_prefix__(module) - ) + capacity_count(table_name, {:capacity_count_module, module}) Map.put(acc, module, %{current: current, limit: limit}) end) @@ -623,6 +623,13 @@ defmodule DurableServer.Supervisor do _ -> nil end + defp capacity_count(table_name, key) do + case :ets.lookup(table_name, key) do + [{^key, count}] -> count + [] -> 0 + end + end + @doc """ Starts a DurableServer.Supervisor with the given options. @@ -673,23 +680,51 @@ defmodule DurableServer.Supervisor do raise ArgumentError, "prefix must end with '/', got: #{inspect(prefix)}" end - # claim the prefix to prevent conflicts between supervisors - prefix_key = {__MODULE__, :prefix, prefix} + ensure_prefix_unclaimed!(prefix) + Supervisor.start_link(__MODULE__, opts, name: name) + end - case :persistent_term.get(prefix_key, nil) do - nil -> - :persistent_term.put(prefix_key, name) + defp ensure_prefix_unclaimed!(prefix) do + deadline = System.monotonic_time(:millisecond) + 1_000 + await_prefix_release!(prefix, deadline) + end - existing_name when existing_name != name -> - raise ArgumentError, - "prefix #{inspect(prefix)} is already claimed by supervisor #{inspect(existing_name)}" + defp await_prefix_release!(prefix, deadline) do + case Registry.lookup(DurableServer.PrefixRegistry, prefix) do + [] -> + :ok - ^name -> - raise ArgumentError, - "the prefix #{inspect(prefix)} has already been claimed by another process" + [{pid, existing_name}] -> + if Process.alive?(pid) do + raise ArgumentError, + "prefix #{inspect(prefix)} is already claimed by supervisor #{inspect(existing_name)}" + else + if System.monotonic_time(:millisecond) < deadline do + Process.sleep(1) + await_prefix_release!(prefix, deadline) + else + raise ArgumentError, + "timed out releasing prefix #{inspect(prefix)} from a stopped owner" + end + end end + end - Supervisor.start_link(__MODULE__, opts, name: name) + defp claim_prefix!(prefix, name) do + case Registry.register(DurableServer.PrefixRegistry, prefix, name) do + {:ok, _owner} -> + :ok + + {:error, {:already_registered, owner_pid}} -> + existing_name = + case Registry.lookup(DurableServer.PrefixRegistry, prefix) do + [{^owner_pid, owner_name}] -> owner_name + _ -> owner_pid + end + + raise ArgumentError, + "prefix #{inspect(prefix)} is already claimed by supervisor #{inspect(existing_name)}" + end end @doc false @@ -802,6 +837,8 @@ defmodule DurableServer.Supervisor do raise ArgumentError, "#{function_name} requires :key" end + validate_durable_key!(key, function_name) + initial_state = case Keyword.fetch(args, :initial_state) do {:ok, initial_state} when is_map(initial_state) -> @@ -825,6 +862,15 @@ defmodule DurableServer.Supervisor do "#{function_name} expects {Module, key: \"...\", initial_state: %{...}}, got: #{inspect(init_arg)}" end + defp validate_durable_key!(key, function_name) do + if key == "__nodes" or String.starts_with?(key, "__nodes/") do + raise ArgumentError, + "#{function_name} :key uses the reserved internal namespace __nodes/: #{inspect(key)}" + end + + :ok + end + defp do_start_child_with_init_arg(supervisor, {module, init_arg, boot_info}, opts) do opts = Keyword.validate!(opts, [ @@ -1083,27 +1129,36 @@ defmodule DurableServer.Supervisor do reply_to ) do # Reject before spawning a local child when this node is draining. - case check_local_start_capacity(supervisor, module, boot_info) do - :ok -> - do_start_child_inner( - supervisor, - module, - init_arg, - boot_info, - key, - retries, - deadline_ms, - reply_to - ) + case reserve_local_start_capacity(supervisor, module, boot_info) do + {:ok, capacity_reservation} -> + try do + do_start_child_inner( + supervisor, + module, + init_arg, + boot_info, + key, + retries, + deadline_ms, + reply_to, + capacity_reservation + ) + after + LifecycleManager.cancel_capacity_reservation(capacity_reservation) + end {:error, {:capacity_limit, _reason}} = error -> error end end - defp check_local_start_capacity(supervisor, module, boot_info) do - case LifecycleManager.check_capacity(supervisor, module, local_start_capacity_opts(boot_info)) do - :ok -> :ok + defp reserve_local_start_capacity(supervisor, module, boot_info) do + case LifecycleManager.reserve_capacity( + supervisor, + module, + local_start_capacity_opts(boot_info) + ) do + {:ok, reservation} -> {:ok, reservation} {:error, {:limit_reached, reason, _details}} -> {:error, {:capacity_limit, reason}} end end @@ -1124,7 +1179,8 @@ defmodule DurableServer.Supervisor do key, retries, deadline_ms, - reply_to + reply_to, + capacity_reservation ) do dynamic_sup = get_dynamic_supervisor(supervisor) config = __get_config__(supervisor) @@ -1147,6 +1203,7 @@ defmodule DurableServer.Supervisor do case DynamicSupervisor.start_child(dynamic_sup, child_spec) do {:ok, pid} -> + LifecycleManager.commit_capacity_reservation(capacity_reservation, pid) monitor_ref = Process.monitor(pid) timeout_ms = remaining_timeout_ms(deadline_ms) @@ -1321,6 +1378,9 @@ defmodule DurableServer.Supervisor do end end + {:error, :max_children} -> + {:error, {:capacity_limit, :max_children_total}} + {:error, reason} -> {:error, reason} end @@ -1516,7 +1576,12 @@ defmodule DurableServer.Supervisor do defp restart_claim_race_final_retry?(_retries, _remaining_ms), do: false - defp try_remote_placement(supervisor, {module, init_arg, boot_info} = child_spec, max_retries) do + defp try_remote_placement( + supervisor, + {module, init_arg, boot_info} = child_spec, + max_retries, + deadline + ) do key = Keyword.fetch!(init_arg, :key) sticky_placement = @@ -1551,7 +1616,8 @@ defmodule DurableServer.Supervisor do nodes -> try_nodes(supervisor, child_spec, nodes, key: key, - sticky_placement: sticky_placement + sticky_placement: sticky_placement, + deadline: deadline ) end end @@ -1632,7 +1698,7 @@ defmodule DurableServer.Supervisor do end defp try_remote_placement_with_retry(supervisor, child_spec, max_retries, deadline) do - case try_remote_placement(supervisor, child_spec, max_retries) do + case try_remote_placement(supervisor, child_spec, max_retries, deadline) do {:ok, result} -> {:ok, result} @@ -1744,11 +1810,17 @@ defmodule DurableServer.Supervisor 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) - erpc_timeout_ms = placement_erpc_timeout_ms(supervisor, node) + 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) # NOTE: we MUST pass max_placement_retries: 0 to prevent recursive retry on the other side try do + if erpc_timeout_ms == 0 do + throw({:error, :placement_deadline_expired}) + end + result = safe_erpc_call( node, @@ -1784,6 +1856,9 @@ defmodule DurableServer.Supervisor do try_nodes(supervisor, child_spec, rest, placement_opts) end catch + :throw, {:error, :placement_deadline_expired} -> + {:error, :timeout} + :throw, {:error, :not_ready} -> report_placement_diagnostic(supervisor, :remote_placement_not_ready) Logger.warning("Node #{inspect(node)} not ready (still starting up), trying next node") @@ -1850,6 +1925,17 @@ defmodule DurableServer.Supervisor do end end + @doc false + def __placement_erpc_timeout_ms__(supervisor, node, deadline) + when is_atom(supervisor) and is_atom(node) do + configured_timeout = placement_erpc_timeout_ms(supervisor, node) + + case remaining_timeout_ms(deadline) do + :infinity -> configured_timeout + remaining_ms -> min(configured_timeout, remaining_ms) + end + end + defp placement_erpc_timeout_ms(supervisor, node) when is_atom(supervisor) and is_atom(node) do local_region = lookup_local_region(supervisor) @@ -2270,7 +2356,7 @@ defmodule DurableServer.Supervisor do owner_registry = ensure_started_singleflight_registry_name(supervisor) waiters_registry = ensure_started_singleflight_waiters_registry_name(supervisor) - case Registry.register(owner_registry, singleflight_key, :singleflight_owner) do + case safe_registry_register(owner_registry, singleflight_key, :singleflight_owner) do {:ok, _owner_pid} -> report_singleflight_diagnostic(supervisor, diagnostics, :leader) @@ -2300,11 +2386,17 @@ defmodule DurableServer.Supervisor do wait_timeout_ms, diagnostics ) + + {:error, :registry_unavailable} -> + # Supervisor is shutting down or the registry is already gone. + {:result, fun.()} end + end + + defp safe_registry_register(registry, key, value) do + Registry.register(registry, key, value) rescue - ArgumentError -> - # Supervisor is shutting down or registry already gone; fall back to direct call. - {:result, fun.()} + ArgumentError -> {:error, :registry_unavailable} end defp wait_for_supervisor_singleflight_owner( @@ -2335,12 +2427,14 @@ defmodule DurableServer.Supervisor do result = try do - case Registry.register( + case __register_singleflight_waiter__( + owner_registry, waiters_registry, singleflight_key, + owner_pid, {waiter_ref, reply_alias} ) do - {:ok, _} -> + :wait -> receive do {:singleflight_done, ^singleflight_key, ^waiter_ref, singleflight_result} -> {:result, singleflight_result} @@ -2366,8 +2460,11 @@ defmodule DurableServer.Supervisor do end end - {:error, {:already_registered, _}} -> + :retry -> :retry + + {:follow_owner, new_owner_pid} -> + {:follow_owner, new_owner_pid} end after :erlang.unalias(reply_alias) @@ -2398,6 +2495,30 @@ defmodule DurableServer.Supervisor do ArgumentError -> :retry end + @doc false + def __register_singleflight_waiter__( + owner_registry, + waiters_registry, + singleflight_key, + owner_pid, + waiter_value + ) + when is_atom(owner_registry) and is_atom(waiters_registry) and is_pid(owner_pid) do + case Registry.register(waiters_registry, singleflight_key, waiter_value) do + {:ok, _} -> + case Registry.lookup(owner_registry, singleflight_key) do + [{^owner_pid, _value}] -> :wait + [{new_owner_pid, _value}] when is_pid(new_owner_pid) -> {:follow_owner, new_owner_pid} + [] -> :retry + end + + {:error, {:already_registered, _pid}} -> + :retry + end + rescue + ArgumentError -> :retry + end + defp report_singleflight_diagnostic(supervisor, diagnostics, key) when is_atom(supervisor) and is_map(diagnostics) and is_atom(key) do case Map.fetch(diagnostics, key) do @@ -2454,7 +2575,7 @@ defmodule DurableServer.Supervisor do deadline ) do # First attempt remote placement (single round, no retry loop — we handle retry here) - case try_remote_placement(supervisor, child_spec, 3) do + case try_remote_placement(supervisor, child_spec, 3, deadline) do {:ok, result} -> {:ok, result} @@ -3110,6 +3231,7 @@ defmodule DurableServer.Supervisor do :restart_claim_gate_disable_after_ms, :heartbeat_interval_ms, :heartbeat_staleness_threshold_ms, + :heartbeat_future_skew_tolerance_ms, :heartbeat_tracking_mode, :heartbeat_reconcile_interval_ms, :graceful_shutdown_timeout_ms, @@ -3137,6 +3259,7 @@ defmodule DurableServer.Supervisor do name = Keyword.fetch!(opts, :name) prefix = Keyword.fetch!(opts, :prefix) + :ok = claim_prefix!(prefix, name) # Extract infrastructure options with defaults finch = Keyword.get(opts, :finch, DurableServer.Finch) @@ -3240,6 +3363,13 @@ defmodule DurableServer.Supervisor do @default_heartbeat_staleness_threshold_ms ) + heartbeat_future_skew_tolerance_ms = + extract_non_negative_integer!( + opts, + :heartbeat_future_skew_tolerance_ms, + @default_heartbeat_future_skew_tolerance_ms + ) + max_heartbeat_interval = div(heartbeat_staleness_threshold_ms, 2) if heartbeat_interval_ms > max_heartbeat_interval do @@ -3349,6 +3479,7 @@ defmodule DurableServer.Supervisor do restart_claim_gate_disable_after_ms: restart_claim_gate_disable_after_ms, heartbeat_interval_ms: heartbeat_interval_ms, heartbeat_staleness_threshold_ms: heartbeat_staleness_threshold_ms, + heartbeat_future_skew_tolerance_ms: heartbeat_future_skew_tolerance_ms, heartbeat_tracking_mode: heartbeat_tracking_mode, heartbeat_reconcile_interval_ms: heartbeat_reconcile_interval_ms, graceful_shutdown_timeout_ms: Keyword.get(opts, :graceful_shutdown_timeout_ms, 30_000), @@ -3367,10 +3498,23 @@ defmodule DurableServer.Supervisor do warn_on_shutdown_timeout_mismatch(config, backend_resources) - Logger.info("starting #{inspect(name)}: #{inspect(config)}") + logged_config = + config + |> Map.drop([ + :storage_backend, + :heartbeat_backend, + :object_store, + :circuit_breaker, + :init_info + ]) + |> Map.put(:storage_backend, config.storage_backend.adapter) + |> Map.put(:heartbeat_backend, config.heartbeat_backend.adapter) + + Logger.info("starting #{inspect(name)}: #{inspect(logged_config)}") :ets.insert(table_name, {:config, config}) :ets.insert(table_name, {:capacity_limits, capacity_limits}) + :ets.insert(table_name, {:capacity_count_total, 0}) :ets.insert(table_name, {:sticky_placement_config, sticky_placement_config}) :ets.insert( @@ -3736,9 +3880,9 @@ defmodule DurableServer.Supervisor do init_backend!(adapter, raw_opts) end - defp init_backend_spec(spec, _finch, _task_sup) do + defp init_backend_spec(_spec, _finch, _task_sup) do raise ArgumentError, - "invalid :backend option #{inspect(spec)}. Expected {BackendModule, opts} or %DurableServer.StorageBackend{}" + "invalid :backend option. Expected {BackendModule, opts} or %DurableServer.StorageBackend{}" end defp prepare_backend_init_opts( @@ -3784,17 +3928,16 @@ defmodule DurableServer.Supervisor do {:ok, %StorageBackend{} = backend} -> backend - {:error, reason} -> - raise ArgumentError, - "failed to initialize backend #{inspect(adapter)} with #{inspect(raw_opts)}: #{inspect(reason)}" + {:error, _reason} -> + raise ArgumentError, "failed to initialize backend #{inspect(adapter)}" end end defp normalize_backend_opts(opts) when is_list(opts), do: opts defp normalize_backend_opts(opts) when is_map(opts), do: Map.to_list(opts) - defp normalize_backend_opts(other) do - raise ArgumentError, "expected backend options as keyword or map, got: #{inspect(other)}" + defp normalize_backend_opts(_other) do + raise ArgumentError, "expected backend options as keyword or map" end defp maybe_extract_object_store(%StorageBackend{ @@ -3825,8 +3968,11 @@ defmodule DurableServer.Supervisor do :infinity -> limits + limit when is_integer(limit) and limit > 0 -> + Map.put(limits, :max_children, %{total: limit}) + limit when is_integer(limit) -> - %{:total => limit} + raise ArgumentError, "max_children must be a positive integer, got: #{inspect(limit)}" limit_map when is_map(limit_map) -> validate_max_children_map!(limit_map) diff --git a/lib/durable_server/terminator.ex b/lib/durable_server/terminator.ex index 1c365c6..87a3638 100644 --- a/lib/durable_server/terminator.ex +++ b/lib/durable_server/terminator.ex @@ -8,7 +8,7 @@ defmodule DurableServer.Terminator do 1. Sends sync_and_stop messages to DurableServer children (with limited concurrency) 2. Monitors each child process for DOWN messages - 3. Waits up to a configurable timeout for each child to sync and terminate + 3. Waits within one configurable deadline for children to sync and terminate 4. Returns to continue the shutdown process This ensures that DurableServer processes have an opportunity to persist their @@ -18,7 +18,7 @@ defmodule DurableServer.Terminator do ## Configuration The Terminator uses the same configuration as its parent DurableServer.Supervisor: - - `:graceful_shutdown_timeout_ms` - Maximum time to wait for each child to shutdown + - `:graceful_shutdown_timeout_ms` - Maximum total time for coordinated child shutdown (default: 30_000ms) - `:graceful_shutdown_concurrency` - Maximum concurrent shutdown operations (default: 50, should match Finch pool size to avoid connection exhaustion) @@ -29,7 +29,7 @@ defmodule DurableServer.Terminator do 2. Terminator's terminate/2 is called with reason and state 3. Terminator uses Task.async_stream with limited concurrency to: a. Send {:durable, {:sync_and_stop, reason}} to each DurableServer - b. Wait for each child to terminate (up to timeout) + b. Wait for children within the shared shutdown deadline 4. Each DurableServer calls sync_state/1 then stops normally 5. After all children stop or timeout is reached, terminate/2 returns 6. Supervisor continues shutdown process @@ -75,16 +75,25 @@ defmodule DurableServer.Terminator do "Terminator initiating graceful shutdown for #{state.supervisor_name}: #{inspect(reason)}" ) - try do - DurableServer.LifecycleManager.stop_discovery(state.supervisor_name) - catch - :exit, _ -> :ok + deadline = + System.monotonic_time(:millisecond) + state.graceful_shutdown_timeout_ms + + case remaining_timeout(deadline) do + 0 -> + :ok + + timeout -> + try do + DurableServer.LifecycleManager.stop_discovery(state.supervisor_name, timeout) + catch + :exit, _ -> :ok + end end - perform_graceful_shutdown(state) + perform_graceful_shutdown(state, deadline) end - defp perform_graceful_shutdown(state) do + defp perform_graceful_shutdown(state, deadline) do case get_durable_server_children(state.supervisor_name) do [_ | _] = children -> child_count = length(children) @@ -95,25 +104,36 @@ defmodule DurableServer.Terminator do ) # Use Task.async_stream to limit concurrent shutdown operations. - # This prevents overwhelming the Finch connection pool when many - # DurableServers try to persist their state simultaneously. - per_child_timeout = state.graceful_shutdown_timeout_ms + # Every task shares the same deadline so later batches cannot multiply + # the configured graceful shutdown timeout. start_time = System.monotonic_time(:millisecond) + diagnostics_before = + DurableServer.LifecycleManager.get_discovery_diagnostics(state.supervisor_name) + killed_count = children |> Task.async_stream( fn {_id, pid, _type, _modules} -> - shutdown_child(pid, per_child_timeout) + shutdown_child(pid, deadline) end, max_concurrency: state.graceful_shutdown_concurrency, timeout: :infinity, ordered: false ) - |> Enum.reduce(0, fn {:ok, result}, acc -> - if result == :killed, do: acc + 1, else: acc + |> Enum.reduce(0, fn + {:ok, :killed}, acc -> acc + 1 + {:ok, :ok}, acc -> acc + {:exit, _reason}, acc -> acc + 1 end) + diagnostics_after = + DurableServer.LifecycleManager.get_discovery_diagnostics(state.supervisor_name) + + sync_error_count = + Map.get(diagnostics_after, :sync_and_stop_error, 0) - + Map.get(diagnostics_before, :sync_and_stop_error, 0) + elapsed_ms = System.monotonic_time(:millisecond) - start_time Logger.info( @@ -121,6 +141,12 @@ defmodule DurableServer.Terminator do "(#{child_count} children, #{killed_count} killed due to timeout)" ) + if sync_error_count > 0 do + Logger.warning( + "#{sync_error_count} DurableServer children failed final persistence during shutdown" + ) + end + :ok [] -> @@ -129,7 +155,7 @@ defmodule DurableServer.Terminator do end end - defp shutdown_child(pid, timeout) do + defp shutdown_child(pid, deadline) do ref = Process.monitor(pid) send(pid, {:durable, {:sync_and_stop, :shutdown}}) @@ -137,21 +163,19 @@ defmodule DurableServer.Terminator do {:DOWN, ^ref, :process, ^pid, _reason} -> :ok after - timeout -> - # Didn't finish in time - kill to avoid blocking DynamicSupervisor shutdown + remaining_timeout(deadline) -> + # The shared budget is exhausted. Kill without adding another per-child wait. Process.exit(pid, :kill) - - receive do - {:DOWN, ^ref, :process, ^pid, _reason} -> :ok - after - 1000 -> Process.demonitor(ref, [:flush]) - end - - Logger.warning("Child #{inspect(pid)} did not terminate within #{timeout}ms, killed") + Process.demonitor(ref, [:flush]) + Logger.warning("Child #{inspect(pid)} did not terminate before shutdown deadline, killed") :killed end end + defp remaining_timeout(deadline) do + max(deadline - System.monotonic_time(:millisecond), 0) + end + defp get_durable_server_children(supervisor_name) do try do # Get the DynamicSupervisor child name diff --git a/mix.exs b/mix.exs index 7f8048d..c347d4f 100644 --- a/mix.exs +++ b/mix.exs @@ -41,7 +41,6 @@ defmodule DurableServer.MixProject do defp deps do [ {:group, "~> 0.2.0"}, - {:jason, "~> 1.4"}, {:req, "~> 0.5"}, {:req_s3, "~> 0.2"}, {:finch, "~> 0.18"}, diff --git a/mix.lock b/mix.lock index a08b172..1baebe6 100644 --- a/mix.lock +++ b/mix.lock @@ -1,23 +1,23 @@ %{ "cc_precompiler": {:hex, :cc_precompiler, "0.1.11", "8c844d0b9fb98a3edea067f94f616b3f6b29b959b6b3bf25fee94ffe34364768", [:mix], [{:elixir_make, "~> 0.7", [hex: :elixir_make, repo: "hexpm", optional: false]}], "hexpm", "3427232caf0835f94680e5bcf082408a70b48ad68a5f5c0b02a3bea9f3a075b9"}, - "earmark_parser": {:hex, :earmark_parser, "1.4.44", "f20830dd6b5c77afe2b063777ddbbff09f9759396500cdbe7523efd58d7a339c", [:mix], [], "hexpm", "4778ac752b4701a5599215f7030989c989ffdc4f6df457c5f36938cc2d2a2750"}, - "ekv": {:hex, :ekv, "0.4.0", "cf5030ee660f3d62516498afd2dfa3593b9ee038084b18bf4c840a1812aa9ff1", [:make, :mix], [{:cc_precompiler, "~> 0.1", [hex: :cc_precompiler, repo: "hexpm", optional: false]}, {:elixir_make, "~> 0.9", [hex: :elixir_make, repo: "hexpm", optional: false]}], "hexpm", "0dd2d78f999e343a171ec141af2fb1e7fc0f311f608702d4a5f0fef82cf1264f"}, - "elixir_make": {:hex, :elixir_make, "0.9.0", "6484b3cd8c0cee58f09f05ecaf1a140a8c97670671a6a0e7ab4dc326c3109726", [:mix], [], "hexpm", "db23d4fd8b757462ad02f8aa73431a426fe6671c80b200d9710caf3d1dd0ffdb"}, - "ex_doc": {:hex, :ex_doc, "0.39.3", "519c6bc7e84a2918b737aec7ef48b96aa4698342927d080437f61395d361dcee", [:mix], [{:earmark_parser, "~> 1.4.44", [hex: :earmark_parser, repo: "hexpm", optional: false]}, {:makeup_c, ">= 0.1.0", [hex: :makeup_c, repo: "hexpm", optional: true]}, {:makeup_elixir, "~> 0.14 or ~> 1.0", [hex: :makeup_elixir, repo: "hexpm", optional: false]}, {:makeup_erlang, "~> 0.1 or ~> 1.0", [hex: :makeup_erlang, repo: "hexpm", optional: false]}, {:makeup_html, ">= 0.1.0", [hex: :makeup_html, repo: "hexpm", optional: true]}], "hexpm", "0590955cf7ad3b625780ee1c1ea627c28a78948c6c0a9b0322bd976a079996e1"}, - "finch": {:hex, :finch, "0.20.0", "5330aefb6b010f424dcbbc4615d914e9e3deae40095e73ab0c1bb0968933cadf", [:mix], [{:mime, "~> 1.0 or ~> 2.0", [hex: :mime, repo: "hexpm", optional: false]}, {:mint, "~> 1.6.2 or ~> 1.7", [hex: :mint, repo: "hexpm", optional: false]}, {:nimble_options, "~> 0.4 or ~> 1.0", [hex: :nimble_options, repo: "hexpm", optional: false]}, {:nimble_pool, "~> 1.1", [hex: :nimble_pool, repo: "hexpm", optional: false]}, {:telemetry, "~> 0.4 or ~> 1.0", [hex: :telemetry, repo: "hexpm", optional: false]}], "hexpm", "2658131a74d051aabfcba936093c903b8e89da9a1b63e430bee62045fa9b2ee2"}, + "earmark_parser": {:hex, :earmark_parser, "1.4.45", "cba8369ab2a1342e419bc2760eec731b17be828941dcf494045d44766227e1d5", [:mix], [], "hexpm", "d3ec045bf122965db20c0bdb420e19ee1415843135327124918473feb4b328e8"}, + "ekv": {:hex, :ekv, "0.4.3", "a4aa948d84068572806c8ecfebaf63fdcdae04e2c6e6589e2314e412c26e4b6c", [:make, :mix], [{:cc_precompiler, "~> 0.1", [hex: :cc_precompiler, repo: "hexpm", optional: false]}, {:elixir_make, "~> 0.9", [hex: :elixir_make, repo: "hexpm", optional: false]}], "hexpm", "4005bdf00f3e1c1cd0958088005501cedd5babc77cd3495e9d0dbbd409a51102"}, + "elixir_make": {:hex, :elixir_make, "0.10.0", "16577e2583a79bb79237bbff349619ef5d80afffc07eac6e4faf0d00e2ddaf7d", [:mix], [], "hexpm", "dc1f09fb7fa68866b886abd5f0f3c83553b1a19a52359a899e92af1bb3b31982"}, + "ex_doc": {:hex, :ex_doc, "0.40.3", "4a972ffe64bc07dc605af487e98fc19b72a4185f55ca031b94c0552d6071c1d9", [:mix], [{:earmark_parser, "~> 1.4.44", [hex: :earmark_parser, repo: "hexpm", optional: false]}, {:makeup_c, ">= 0.1.0", [hex: :makeup_c, repo: "hexpm", optional: true]}, {:makeup_elixir, "~> 0.14 or ~> 1.0", [hex: :makeup_elixir, repo: "hexpm", optional: false]}, {:makeup_erlang, "~> 0.1 or ~> 1.0", [hex: :makeup_erlang, repo: "hexpm", optional: false]}, {:makeup_html, ">= 0.1.0", [hex: :makeup_html, repo: "hexpm", optional: true]}], "hexpm", "2756e357742fecd9749b489b85d67c9ce99c465f2e75728d9e6dc8d704b973de"}, + "finch": {:hex, :finch, "0.23.0", "e3f9287ac25a8832f848b144c2b57346aac65b205e2e0629a52adfe6507fd837", [:mix], [{:mime, "~> 1.0 or ~> 2.0", [hex: :mime, repo: "hexpm", optional: false]}, {:mint, "~> 1.8", [hex: :mint, repo: "hexpm", optional: false]}, {:nimble_options, "~> 0.4 or ~> 1.0", [hex: :nimble_options, repo: "hexpm", optional: false]}, {:nimble_pool, "~> 1.1", [hex: :nimble_pool, repo: "hexpm", optional: false]}, {:telemetry, "~> 0.4 or ~> 1.0", [hex: :telemetry, repo: "hexpm", optional: false]}], "hexpm", "80e58d3f936f57e3fdf404f83a3642897ae6d9fb642934e46da4d8fe761b99d5"}, "group": {:hex, :group, "0.2.0", "1f40d9b49bd0496ceb49ddf273b000d9291711a235631caeafa8607ce159d5ea", [:mix], [], "hexpm", "a58c2fb99165cd5b10756a7a51f0162d18e5e6d75b64b3253c004d2937e73d78"}, - "hpax": {:hex, :hpax, "1.0.3", "ed67ef51ad4df91e75cc6a1494f851850c0bd98ebc0be6e81b026e765ee535aa", [:mix], [], "hexpm", "8eab6e1cfa8d5918c2ce4ba43588e894af35dbd8e91e6e55c817bca5847df34a"}, - "jason": {:hex, :jason, "1.4.4", "b9226785a9aa77b6857ca22832cffa5d5011a667207eb2a0ad56adb5db443b8a", [:mix], [{:decimal, "~> 1.0 or ~> 2.0", [hex: :decimal, repo: "hexpm", optional: true]}], "hexpm", "c5eb0cab91f094599f94d55bc63409236a8ec69a21a67814529e8d5f6cc90b3b"}, - "makeup": {:hex, :makeup, "1.2.1", "e90ac1c65589ef354378def3ba19d401e739ee7ee06fb47f94c687016e3713d1", [:mix], [{:nimble_parsec, "~> 1.4", [hex: :nimble_parsec, repo: "hexpm", optional: false]}], "hexpm", "d36484867b0bae0fea568d10131197a4c2e47056a6fbe84922bf6ba71c8d17ce"}, + "hpax": {:hex, :hpax, "1.0.4", "777de5d433b0fbdc7c418159c8055910faa8047ffdb3d6b31098d2a46cd7685c", [:mix], [], "hexpm", "afc7cb142ebcc2d01ce7816190b98ce5dd49e799111b24249f3443d730f377ca"}, + "jason": {:hex, :jason, "1.4.5", "2e3a008590b0b8d7388c20293e9dcc9cf3e5d642fd2a114e4cbbb52e595d940a", [:mix], [{:decimal, "~> 1.0 or ~> 2.0 or ~> 3.0", [hex: :decimal, repo: "hexpm", optional: true]}], "hexpm", "b0c823996102bcd0239b3c2444eb00409b72f6a140c1950bc8b457d836b30684"}, + "makeup": {:hex, :makeup, "1.2.2", "882d46dc0905e9ff7abf2aab61a7e6b3dcc555533977d8a23b06019e6c89ac94", [:mix], [{:nimble_parsec, "~> 1.4", [hex: :nimble_parsec, repo: "hexpm", optional: false]}], "hexpm", "9a1a24e5b343b8ae16abea0822c10a6f75da27af7fa802ada5251f7579bfccfa"}, "makeup_elixir": {:hex, :makeup_elixir, "1.0.1", "e928a4f984e795e41e3abd27bfc09f51db16ab8ba1aebdba2b3a575437efafc2", [:mix], [{:makeup, "~> 1.0", [hex: :makeup, repo: "hexpm", optional: false]}, {:nimble_parsec, "~> 1.2.3 or ~> 1.3", [hex: :nimble_parsec, repo: "hexpm", optional: false]}], "hexpm", "7284900d412a3e5cfd97fdaed4f5ed389b8f2b4cb49efc0eb3bd10e2febf9507"}, - "makeup_erlang": {:hex, :makeup_erlang, "1.0.2", "03e1804074b3aa64d5fad7aa64601ed0fb395337b982d9bcf04029d68d51b6a7", [:mix], [{:makeup, "~> 1.0", [hex: :makeup, repo: "hexpm", optional: false]}], "hexpm", "af33ff7ef368d5893e4a267933e7744e46ce3cf1f61e2dccf53a111ed3aa3727"}, + "makeup_erlang": {:hex, :makeup_erlang, "1.1.0", "835f7e60792e08824cda445639555d7bf1bbbddb1b60b306e33cb6f6db24dc74", [:mix], [{:makeup, "~> 1.0", [hex: :makeup, repo: "hexpm", optional: false]}], "hexpm", "1cd6780fb1dd1a03979abaed0fe82712b0625118fd5257d3ebbf73f960c73c3c"}, "mime": {:hex, :mime, "2.0.7", "b8d739037be7cd402aee1ba0306edfdef982687ee7e9859bee6198c1e7e2f128", [:mix], [], "hexpm", "6171188e399ee16023ffc5b76ce445eb6d9672e2e241d2df6050f3c771e80ccd"}, - "mint": {:hex, :mint, "1.7.1", "113fdb2b2f3b59e47c7955971854641c61f378549d73e829e1768de90fc1abf1", [:mix], [{:castore, "~> 0.1.0 or ~> 1.0", [hex: :castore, repo: "hexpm", optional: true]}, {:hpax, "~> 0.1.1 or ~> 0.2.0 or ~> 1.0", [hex: :hpax, repo: "hexpm", optional: false]}], "hexpm", "fceba0a4d0f24301ddee3024ae116df1c3f4bb7a563a731f45fdfeb9d39a231b"}, + "mint": {:hex, :mint, "1.9.3", "3337184d69179695c7a9f1714d92c11e629d36c8c037a21cf490131d3d150554", [:mix], [{:castore, "~> 0.1.0 or ~> 1.0", [hex: :castore, repo: "hexpm", optional: true]}, {:hpax, "~> 0.1.1 or ~> 0.2.0 or ~> 1.0", [hex: :hpax, repo: "hexpm", optional: false]}], "hexpm", "5f7c9342480c069dbbc4eeac3490303c9e01870ff01a7f1d29b6107054fc1e74"}, "nimble_options": {:hex, :nimble_options, "1.1.1", "e3a492d54d85fc3fd7c5baf411d9d2852922f66e69476317787a7b2bb000a61b", [:mix], [], "hexpm", "821b2470ca9442c4b6984882fe9bb0389371b8ddec4d45a9504f00a66f650b44"}, "nimble_parsec": {:hex, :nimble_parsec, "1.4.2", "8efba0122db06df95bfaa78f791344a89352ba04baedd3849593bfce4d0dc1c6", [:mix], [], "hexpm", "4b21398942dda052b403bbe1da991ccd03a053668d147d53fb8c4e0efe09c973"}, "nimble_pool": {:hex, :nimble_pool, "1.1.0", "bf9c29fbdcba3564a8b800d1eeb5a3c58f36e1e11d7b7fb2e084a643f645f06b", [:mix], [], "hexpm", "af2e4e6b34197db81f7aad230c1118eac993acc0dae6bc83bac0126d4ae0813a"}, - "req": {:hex, :req, "0.5.16", "99ba6a36b014458e52a8b9a0543bfa752cb0344b2a9d756651db1281d4ba4450", [:mix], [{:brotli, "~> 0.3.1", [hex: :brotli, repo: "hexpm", optional: true]}, {:ezstd, "~> 1.0", [hex: :ezstd, repo: "hexpm", optional: true]}, {:finch, "~> 0.17", [hex: :finch, repo: "hexpm", optional: false]}, {:jason, "~> 1.0", [hex: :jason, repo: "hexpm", optional: false]}, {:mime, "~> 2.0.6 or ~> 2.1", [hex: :mime, repo: "hexpm", optional: false]}, {:nimble_csv, "~> 1.0", [hex: :nimble_csv, repo: "hexpm", optional: true]}, {:plug, "~> 1.0", [hex: :plug, repo: "hexpm", optional: true]}], "hexpm", "974a7a27982b9b791df84e8f6687d21483795882a7840e8309abdbe08bb06f09"}, + "req": {:hex, :req, "0.5.17", "0096ddd5b0ed6f576a03dde4b158a0c727215b15d2795e59e0916c6971066ede", [:mix], [{:brotli, "~> 0.3.1", [hex: :brotli, repo: "hexpm", optional: true]}, {:ezstd, "~> 1.0", [hex: :ezstd, repo: "hexpm", optional: true]}, {:finch, "~> 0.17", [hex: :finch, repo: "hexpm", optional: false]}, {:jason, "~> 1.0", [hex: :jason, repo: "hexpm", optional: false]}, {:mime, "~> 2.0.6 or ~> 2.1", [hex: :mime, repo: "hexpm", optional: false]}, {:nimble_csv, "~> 1.0", [hex: :nimble_csv, repo: "hexpm", optional: true]}, {:plug, "~> 1.0", [hex: :plug, repo: "hexpm", optional: true]}], "hexpm", "0b8bc6ffdfebbc07968e59d3ff96d52f2202d0536f10fef4dc11dc02a2a43e39"}, "req_s3": {:hex, :req_s3, "0.2.3", "ede5f4c792cf39995379307733ff4593032a876f38da29d9d7ea03881b498b51", [:mix], [{:req, "~> 0.5.6", [hex: :req, repo: "hexpm", optional: false]}], "hexpm", "31b5d52490495c8aeea7e3c5cbcec82f49035e11bdaf41f0e58ab716fefe44ca"}, "sweet_xml": {:hex, :sweet_xml, "0.7.5", "803a563113981aaac202a1dbd39771562d0ad31004ddbfc9b5090bdcd5605277", [:mix], [], "hexpm", "193b28a9b12891cae351d81a0cead165ffe67df1b73fe5866d10629f4faefb12"}, - "telemetry": {:hex, :telemetry, "1.3.0", "fedebbae410d715cf8e7062c96a1ef32ec22e764197f70cda73d82778d61e7a2", [:rebar3], [], "hexpm", "7015fc8919dbe63764f4b4b87a95b7c0996bd539e0d499be6ec9d7f3875b79e6"}, + "telemetry": {:hex, :telemetry, "1.4.2", "a0cb522801dffb1c49fe6e30561badffc7b6d0e180db1300df759faa22062855", [:rebar3], [], "hexpm", "928f6495066506077862c0d1646609eed891a4326bee3126ba54b60af61febb1"}, } diff --git a/test/durable_server/lifecycle_test.exs b/test/durable_server/lifecycle_test.exs index 71674aa..13b6f59 100644 --- a/test/durable_server/lifecycle_test.exs +++ b/test/durable_server/lifecycle_test.exs @@ -139,6 +139,83 @@ defmodule DurableServer.LifecycleTest do def decode(_state, data), do: {:ok, data} end + defmodule DiscoveryCrashOnceBackend do + @behaviour DurableServer.StorageBackend + + alias DurableServer.StorageBackend + + @impl true + def init_backend(opts) do + delegate = Keyword.fetch!(opts, :delegate) + + {:ok, + %{ + state: %{ + delegate: delegate, + table: Keyword.fetch!(opts, :table), + notify_pid: Keyword.fetch!(opts, :notify_pid) + }, + 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}, key, opts), + do: StorageBackend.get_object(delegate, key, opts) + + @impl true + def list_all_objects_stream( + %{delegate: delegate, table: table, notify_pid: notify_pid}, + prefix, + opts + ) do + if String.ends_with?(prefix, "__nodes/") do + StorageBackend.list_all_objects_stream(delegate, prefix, opts) + else + attempt = + :ets.update_counter( + table, + :discovery_list_attempts, + {2, 1}, + {:discovery_list_attempts, 0} + ) + + send(notify_pid, {:discovery_list_attempt, attempt}) + + if attempt == 1 do + raise "injected discovery failure" + else + StorageBackend.list_all_objects_stream(delegate, prefix, opts) + end + end + 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 + defmodule HeartbeatDelayBackend do @behaviour DurableServer.StorageBackend @@ -1514,20 +1591,11 @@ defmodule DurableServer.LifecycleTest do # Manager should still be alive despite corrupted data assert_process_alive(manager_pid) - # Should have attempted to process or clean up corruption - {:ok, data} = - DurableServer.fetch_stored_state(config.object_store, %{key: key, prefix: prefix}) - - # The restart attempt fields should be handled gracefully - # Since we started with corrupted string values, they might remain as strings - # or be cleaned up entirely. Both are acceptable for corruption recovery. - restart_time = Map.get(data.meta, :restart_attempt_time) + # Corrupt typed fields are rejected as data errors rather than admitted into Meta. + assert {:error, %ArgumentError{message: message}} = + DurableServer.fetch_stored_state(config.object_store, %{key: key, prefix: prefix}) - if restart_time do - # Either cleaned up and replaced with proper integer, or left as original corrupt string - assert is_integer(restart_time) or is_binary(restart_time), - "Restart time should be integer (if fixed) or string (if original corrupt data)" - end + assert message =~ "invalid metadata field :restart_attempt_time" GenServer.stop(manager_pid) end @@ -2553,6 +2621,40 @@ defmodule DurableServer.LifecycleTest do GenServer.stop(manager_pid) end + test "an abnormal discovery task exit is handled and retried", %{ + supervisor_name: supervisor_name, + config: config + } do + table = :ets.new(__MODULE__.DiscoveryCrashOnceBackend, [:set, :public]) + delegate = DurableServer.Supervisor.__get_config__(supervisor_name).storage_backend + + {:ok, storage_backend} = + DurableServer.StorageBackend.init_backend(DiscoveryCrashOnceBackend, + delegate: delegate, + table: table, + notify_pid: self() + ) + + 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, 10) + + {:ok, manager_pid} = start_standalone_lifecycle_manager(supervisor_name, test_config) + manager_ref = Process.monitor(manager_pid) + + send(manager_pid, :discover_and_restart) + + assert_receive {:discovery_list_attempt, 1}, 1_000 + assert_receive {:discovery_list_attempt, 2}, 1_000 + refute_receive {:DOWN, ^manager_ref, :process, ^manager_pid, _reason}, 50 + assert Process.alive?(manager_pid) + + GenServer.stop(manager_pid) + end + test "task supervision completes discovery cycle properly", %{ supervisor_name: supervisor_name, prefix: _prefix, @@ -2597,9 +2699,13 @@ defmodule DurableServer.LifecycleTest do assert [{:heartbeat_write, %{"last_heartbeat_at" => stored_heartbeat_at}}] = :ets.lookup(table, :heartbeat_write) - # Remote stale checks use this stored timestamp, so local self-kill math must too. + # The persisted wall timestamp remains available for cross-node freshness checks, + # while local deadline arithmetic tracks the equivalent monotonic instant. assert abs(manager_state.last_successful_heartbeat_at - stored_heartbeat_at) < 500 + assert System.monotonic_time(:millisecond) - + manager_state.last_successful_heartbeat_monotonic_at < 3_000 + GenServer.stop(manager_pid) end @@ -2709,7 +2815,10 @@ defmodule DurableServer.LifecycleTest do {us, result} = :timer.tc(fn -> LifecycleManager.handle_info( - {ref, {:heartbeat, {%{total_ms: 10, put_ms: 5, cache_ms: 5}, heartbeat_entry}}}, + {ref, + {:heartbeat, + {%{total_ms: 10, put_ms: 5, cache_ms: 5}, heartbeat_entry, + System.monotonic_time(:millisecond)}}}, state ) end) diff --git a/test/durable_server/meta_test.exs b/test/durable_server/meta_test.exs new file mode 100644 index 0000000..30d7bac --- /dev/null +++ b/test/durable_server/meta_test.exs @@ -0,0 +1,69 @@ +defmodule DurableServer.MetaTest do + use ExUnit.Case, async: true + + alias DurableServer.Meta + alias DurableServer.StoredState + + @context %{key: "server-key", prefix: "tenant/"} + + test "round-trips valid metadata through the hardened decoder" do + meta = %Meta{ + module: __MODULE__, + permanent: true, + pid: self(), + status: :running, + supervisor: :meta_test_supervisor, + task_supervisor: :meta_test_tasks, + dynamic_supervisor: :meta_test_dynamic, + node_ref: 123, + node_str: "node@example", + last_heartbeat_at: 456, + crash_history: [100, 200], + init_from_ref: make_ref(), + init_from_pid: self() + } + + decoded = meta |> Meta.encode_to_binary() |> Meta.decode_from_binary(@context) + + assert decoded == %{meta | key: @context.key, prefix: @context.prefix} + end + + test "rejects an ETF atom that does not already exist without interning it" do + atom_name = "durable_meta_untrusted_#{System.unique_integer([:positive, :monotonic])}" + assert_raise ArgumentError, fn -> String.to_existing_atom(atom_name) end + + encoded = + <<131, 119, byte_size(atom_name), atom_name::binary>> + |> Base.encode64() + + assert_raise ArgumentError, ~r/unsafe or malformed external term/, fn -> + Meta.decode_from_binary(encoded, @context) + end + + assert_raise ArgumentError, fn -> String.to_existing_atom(atom_name) end + end + + test "rejects executable and malformed external terms with a controlled error" do + executable = :erlang.term_to_binary(%{status: :running, payload: fn -> :unsafe end}) + + for encoded <- [ + Base.encode64(executable), + "not base64!", + Base.encode64(<<131, 80, 0, 0, 0, 1, 0>>) + ] do + term = %{"vsn" => 1, "state" => %{}, "meta" => encoded} + assert {:error, %ArgumentError{}} = StoredState.from_object_store_term(term) + end + end + + test "rejects invalid metadata field types" do + encoded = + %{status: :running, permanent: "yes"} + |> :erlang.term_to_binary() + |> Base.encode64() + + assert_raise ArgumentError, ~r/invalid metadata field :permanent/, fn -> + Meta.decode_from_binary(encoded, @context) + end + end +end diff --git a/test/durable_server/mirror_store_test.exs b/test/durable_server/mirror_store_test.exs new file mode 100644 index 0000000..8eaff2d --- /dev/null +++ b/test/durable_server/mirror_store_test.exs @@ -0,0 +1,125 @@ +defmodule DurableServer.MirrorStoreTest do + use ExUnit.Case, async: true + + alias DurableServer.Backends.MirrorStore + alias DurableServer.StorageBackend + + defmodule RaceBackend do + @behaviour StorageBackend + + @impl true + def init_backend(opts), do: {:ok, %{state: Map.new(opts)}} + + @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 list_all_objects_stream(_state, _prefix, _opts), do: [] + + @impl true + def put_object(state, key, body, _opts) do + maybe_pause(state, key, :put) + object = %{body: body, etag: next_etag()} + :ets.insert(state.table, {key, object}) + {:ok, object} + end + + @impl true + def delete_object(%{table: table}, key) do + :ets.delete(table, key) + :ok + end + + @impl true + def try_claim(state, key, body) do + maybe_pause(state, key, :try_claim) + object = %{body: body, etag: next_etag()} + + if :ets.insert_new(state.table, {key, object}) do + {:ok, {:claimed, object.etag}} + else + {:error, :already_claimed} + end + end + + @impl true + def update_object(_state, _key, _update_fn, _opts), do: {:error, :unsupported} + + @impl true + def encode(_state, data), do: {:ok, data} + + @impl true + def decode(_state, data), do: {:ok, data} + + defp maybe_pause(%{pause_owner: owner, pause_key: key}, key, operation) + when is_pid(owner) do + ref = make_ref() + send(owner, {:promotion_paused, self(), ref, operation}) + + receive do + {:continue_promotion, ^ref} -> :ok + end + end + + defp maybe_pause(_state, _key, _operation), do: :ok + + defp next_etag do + System.unique_integer([:positive, :monotonic]) + |> Integer.to_string() + end + end + + test "fallback promotion cannot overwrite a concurrent preferred-backend create" do + key = "promotion-race" + primary_table = :ets.new(:mirror_primary, [:set, :public]) + secondary_table = :ets.new(:mirror_secondary, [:set, :public]) + + primary = + StorageBackend.new(RaceBackend, %{ + table: primary_table, + pause_owner: self(), + pause_key: key + }) + + direct_primary = StorageBackend.new(RaceBackend, %{table: primary_table}) + secondary = StorageBackend.new(RaceBackend, %{table: secondary_table}) + + assert {:ok, _object} = StorageBackend.put_object(secondary, key, "fallback-value") + + mirror = + StorageBackend.new(MirrorStore, %{ + primary: primary, + secondary: secondary, + read_preference: :primary, + write_target: :primary, + fallback_reads: true, + promote_on_fallback: true, + mirror_writes: false, + mirror_mode: :best_effort, + secondary_required: false + }) + + read_task = Task.async(fn -> StorageBackend.get_object(mirror, key) end) + + assert_receive {:promotion_paused, promotion_pid, ref, operation} + assert operation in [:put, :try_claim] + + assert {:ok, %{body: "newer-preferred-value"}} = + StorageBackend.put_object(direct_primary, key, "newer-preferred-value") + + send(promotion_pid, {:continue_promotion, ref}) + + assert {:ok, %{body: "newer-preferred-value"}} = Task.await(read_task) + + assert {:ok, %{body: "newer-preferred-value"}} = + StorageBackend.get_object(direct_primary, key) + end +end diff --git a/test/durable_server/object_store_security_test.exs b/test/durable_server/object_store_security_test.exs new file mode 100644 index 0000000..18d33c5 --- /dev/null +++ b/test/durable_server/object_store_security_test.exs @@ -0,0 +1,150 @@ +defmodule DurableServer.ObjectStoreSecurityTest do + use ExUnit.Case, async: true + + import ExUnit.CaptureLog + import SweetXml + + alias DurableServer.ObjectStore + alias DurableServer.StorageBackend + + defmodule SensitiveBackend do + @behaviour StorageBackend + + @impl true + def init_backend(opts) do + {:ok, + %{ + state: %{ + table: :ets.new(__MODULE__, [:set, :public]), + secret: Keyword.fetch!(opts, :secret) + }, + defaults: %{ + discovery_interval_ms: 60_000, + heartbeat_interval_ms: 10_000, + heartbeat_tracking_mode: :poll, + 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 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} -> Map.put(object, :key, key) end) + end + + @impl true + def put_object(%{table: table}, key, body, _opts) do + object = %{body: body, etag: Integer.to_string(System.unique_integer([:positive]))} + :ets.insert(table, {key, object}) + {:ok, object} + end + + @impl true + def delete_object(%{table: table}, key) do + :ets.delete(table, key) + :ok + end + + @impl true + def try_claim(%{table: table}, key, body) do + object = %{body: body, etag: Integer.to_string(System.unique_integer([:positive]))} + + if :ets.insert_new(table, {key, object}) do + {:ok, {:claimed, object.etag}} + else + {:error, :already_claimed} + end + end + + @impl true + def update_object(_state, _key, _update_fn, _opts), do: {:error, :unsupported} + + @impl true + def encode(_state, data), do: {:ok, data} + + @impl true + def decode(_state, data), do: {:ok, data} + end + + test "supervisor startup logs omit backend state and init_info" do + backend_secret = "backend-secret-#{System.unique_integer([:positive])}" + init_secret = "init-secret-#{System.unique_integer([:positive])}" + unique = System.unique_integer([:positive, :monotonic]) + supervisor_name = :"redacted_supervisor_#{unique}" + + log = + capture_log([level: :debug], fn -> + start_supervised!( + {DurableServer.Supervisor, + name: supervisor_name, + prefix: "redacted/#{unique}/", + backend: {SensitiveBackend, secret: backend_secret}, + init_info: %{credential: init_secret}, + graceful_shutdown_timeout_ms: 100} + ) + end) + + refute log =~ backend_secret + refute log =~ init_secret + end + + test "CreateAccessKey parsing never logs returned credentials" do + access_key_id = "access-id-#{System.unique_integer([:positive])}" + secret = "secret-key-#{System.unique_integer([:positive])}" + + xml = """ + + #{access_key_id}#{secret} + + """ + + log = + capture_log([level: :debug], fn -> + assert {:ok, + %{access_key_id: ^access_key_id, secret_access_key: ^secret, user_name: "user"}} = + ObjectStore.__parse_create_access_key_response__(xml, "user") + end) + + refute log =~ access_key_id + refute log =~ secret + end + + test "IAM XML parsing rejects DTDs and never expands an external entity" do + secret = "iam-xxe-secret-#{System.unique_integer([:positive, :monotonic])}" + + path = + Path.join(System.tmp_dir!(), "durable-server-xxe-#{System.unique_integer([:positive])}") + + File.write!(path, secret) + on_exit(fn -> File.rm(path) end) + + xml = """ + + ]> + &xxe; + """ + + assert {:error, :invalid_xml} = ObjectStore.__parse_iam_xml__(xml) + + assert {:ok, document} = + ObjectStore.__parse_iam_xml__( + "safe-id" + ) + + assert xpath(document, ~x"//AccessKeyId/text()"s) == "safe-id" + end +end diff --git a/test/durable_server/supervisor_backend_spec_test.exs b/test/durable_server/supervisor_backend_spec_test.exs index 58095d9..5134a0b 100644 --- a/test/durable_server/supervisor_backend_spec_test.exs +++ b/test/durable_server/supervisor_backend_spec_test.exs @@ -28,7 +28,9 @@ defmodule DurableServer.SupervisorBackendSpecTest do %{ state: %{ table: :ets.new(__MODULE__, [:set, :public]), - name: Map.get(opts, :name) + control_table: init_control_table(opts), + name: Map.get(opts, :name), + owner: Map.get(opts, :owner) }, defaults: %{ heartbeat_tracking_mode: :poll, @@ -59,10 +61,17 @@ defmodule DurableServer.SupervisorBackendSpecTest do 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}} + def put_object(%{table: table} = state, key, data, _opts) do + case pop_heartbeat_failure(state, key) do + {:error, reason} -> + if state.owner, do: send(state.owner, {:heartbeat_put_failed, reason}) + {:error, reason} + + :ok -> + etag = next_etag() + :ets.insert(table, {key, %{body: data, etag: etag}}) + {:ok, %{body: data, etag: etag}} + end end @impl true @@ -91,10 +100,10 @@ defmodule DurableServer.SupervisorBackendSpecTest do end @impl true - def update_object(%{table: table} = state, key, update_fn, _opts) do + def update_object(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, []) + put_object(state, key, new_body, []) end end @@ -104,12 +113,92 @@ defmodule DurableServer.SupervisorBackendSpecTest do @impl true def decode(_state, data), do: {:ok, data} + defp init_control_table(opts) do + table = :ets.new(__MODULE__, [:set, :public]) + :ets.insert(table, {:heartbeat_failures, Map.get(opts, :heartbeat_failures, [])}) + table + end + + defp pop_heartbeat_failure(%{control_table: table}, key) do + if String.contains?(key, "__nodes/") do + case :ets.lookup(table, :heartbeat_failures) do + [{:heartbeat_failures, [reason | rest]}] -> + :ets.insert(table, {:heartbeat_failures, rest}) + {:error, reason} + + _ -> + :ok + end + else + :ok + end + end + 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") + + start_supervised!( + {DurableServer.Supervisor, + [ + name: supervisor_name, + prefix: prefix, + backend: {InMemoryBackend, name: :heartbeat_skew}, + heartbeat_future_skew_tolerance_ms: 50, + graceful_shutdown_timeout_ms: 500 + ]} + ) + + table = :"durable_server_heartbeats_#{supervisor_name}" + node_str = "future-heartbeat@host" + now = System.system_time(:millisecond) + + :ets.insert(table, {node_str, 1, now + 5_000, nil, nil, %{}, %{}}) + + assert :stale = + LifecycleManager.lookup_node_health(%{ + supervisor: supervisor_name, + node_str: node_str + }) + + :ets.insert(table, {node_str, 1, now + 25, nil, nil, %{}, %{}}) + + assert {:healthy, %{node_ref: 1}} = + LifecycleManager.lookup_node_health(%{ + supervisor: supervisor_name, + node_str: node_str + }) + end + + test "retries transient Req heartbeat failures during startup" do + supervisor_name = unique_supervisor_name("heartbeat_retry") + prefix = unique_prefix("heartbeat_retry") + response = %Req.Response{status: 503} + transport_error = %Req.TransportError{reason: :timeout} + + start_supervised!( + {DurableServer.Supervisor, + [ + name: supervisor_name, + prefix: prefix, + backend: + {InMemoryBackend, + name: :heartbeat_retry, owner: self(), heartbeat_failures: [response, transport_error]}, + graceful_shutdown_timeout_ms: 500 + ]} + ) + + assert_receive {:heartbeat_put_failed, ^response} + assert_receive {:heartbeat_put_failed, ^transport_error} + assert Process.alive?(Process.whereis(LifecycleManager.name(supervisor_name))) + end + test "accepts backend module spec directly" do supervisor_name = unique_supervisor_name("custom") prefix = unique_prefix("custom") @@ -132,6 +221,40 @@ defmodule DurableServer.SupervisorBackendSpecTest do assert object_store == nil end + test "caps placement ERPC timeout by the caller deadline" do + supervisor_name = unique_supervisor_name("placement_deadline") + prefix = unique_prefix("placement_deadline") + + start_supervised!( + {DurableServer.Supervisor, + [ + name: supervisor_name, + prefix: prefix, + backend: {InMemoryBackend, name: :placement_deadline}, + placement_erpc_timeout_cross_region_ms: 8_000 + ]} + ) + + deadline = System.monotonic_time(:millisecond) + 25 + + timeout = + DurableServer.Supervisor.__placement_erpc_timeout_ms__( + supervisor_name, + :remote@host, + deadline + ) + + assert timeout > 0 + assert timeout <= 25 + + assert 0 == + DurableServer.Supervisor.__placement_erpc_timeout_ms__( + supervisor_name, + :remote@host, + System.monotonic_time(:millisecond) - 1 + ) + end + test "child_spec uses configured supervisor shutdown timeout" do shutdown_timeout = 12_345 supervisor_name = unique_supervisor_name("child_spec") diff --git a/test/durable_server/watermark_test.exs b/test/durable_server/watermark_test.exs index 9f674e8..a8a6f7a 100644 --- a/test/durable_server/watermark_test.exs +++ b/test/durable_server/watermark_test.exs @@ -3,6 +3,22 @@ defmodule DurableServer.WatermarkTest do import DurableServer.TestHelper alias DurableServer + defmodule SlowInitWatermarkTestServer do + use DurableServer, vsn: 1 + + @impl true + def init(state, _info) do + Process.sleep(Map.fetch!(state, "init_delay_ms")) + {:ok, state} + end + + @impl true + def dump_state(state), do: state + + @impl true + def load_state(_vsn, state), do: state + end + defmodule WatermarkTestServer do use DurableServer, vsn: 1 @@ -196,6 +212,43 @@ defmodule DurableServer.WatermarkTest do {WatermarkTestServer, key: "key3", initial_state: %{}} ) end + + test "enforces a map limit atomically across concurrent starts", %{ + supervisor_name: supervisor_name, + prefix: prefix + } do + start_supervised!( + {DurableServer.Supervisor, + name: supervisor_name, + prefix: prefix, + object_store: test_object_store_opts(), + max_children: %{SlowInitWatermarkTestServer => 1}} + ) + + results = + 1..12 + |> Task.async_stream( + fn index -> + DurableServer.Supervisor.start_child( + supervisor_name, + {SlowInitWatermarkTestServer, + key: "concurrent-#{index}", initial_state: %{init_delay_ms: 300}}, + max_placement_retries: 0 + ) + end, + max_concurrency: 12, + timeout: 5_000 + ) + |> Enum.map(fn {:ok, result} -> result end) + + assert Enum.count(results, &match?({:ok, _}, &1)) == 1 + + assert Enum.all?(results, fn + {:ok, _} -> true + {:error, {:capacity_limit, :max_children_module}} -> true + _ -> false + end) + end end describe "resource limits" do @@ -428,18 +481,30 @@ defmodule DurableServer.WatermarkTest do ) end - test "accepts integer max_children for DynamicSupervisor (legacy)", %{ + test "enforces integer max_children with normalized errors", %{ supervisor_name: supervisor_name, prefix: prefix } do - # Integer max_children should be accepted but not used for capacity limiting assert {:ok, _pid} = start_supervised( {DurableServer.Supervisor, name: supervisor_name, prefix: prefix, object_store: test_object_store_opts(), - max_children: 100} + max_children: 1} + ) + + assert {:ok, {_pid, _meta}} = + DurableServer.Supervisor.start_child( + supervisor_name, + {WatermarkTestServer, key: "integer-1", initial_state: %{}} + ) + + assert {:error, {:capacity_limit, :max_children_total}} = + DurableServer.Supervisor.start_child( + supervisor_name, + {WatermarkTestServer, key: "integer-2", initial_state: %{}}, + max_placement_retries: 0 ) end end diff --git a/test/durable_server_test.exs b/test/durable_server_test.exs index 0e3253b..1a37ad5 100644 --- a/test/durable_server_test.exs +++ b/test/durable_server_test.exs @@ -107,6 +107,10 @@ defmodule DurableServerTest do :ets.insert(table, {{:override, key, consistent}, response}) end + defp put_backend_delete_override(table, key, response) do + :ets.insert(table, {{:delete_override, key}, response}) + end + defp insert_running_lock(table, storage_key, supervisor_name, prefix, key, pid) do node_ref = DurableServer.Supervisor.node_ref(supervisor_name) @@ -323,13 +327,19 @@ defmodule DurableServerTest do @impl true def delete_object(%{table: table}, key) do - case :ets.lookup(table, {:data, key}) do - [{{:data, ^key}, _value}] -> - :ets.delete(table, {:data, key}) - :ok + case :ets.lookup(table, {:delete_override, key}) do + [{{:delete_override, ^key}, response}] -> + response [] -> - {:error, :not_found} + case :ets.lookup(table, {:data, key}) do + [{{:data, ^key}, _value}] -> + :ets.delete(table, {:data, key}) + :ok + + [] -> + {:error, :not_found} + end end end @@ -500,6 +510,11 @@ defmodule DurableServerTest do {:noreply, new_state} end + def handle_cast(:increment_with_timeout, %{count: count} = state) do + new_state = %{state | count: count + 1} + {:noreply, new_state, 1_000} + end + def handle_cast(:increment_and_sync, %{count: count} = state) do new_state = %{state | count: count + 1} {:noreply, new_state, :sync} @@ -599,6 +614,18 @@ defmodule DurableServerTest do end end + defmodule SlowTerminateServer do + use DurableServer, vsn: 1 + + def dump_state(state), do: state + def load_state(_old_vsn, state), do: DurableServerTest.atomify_keys(state) + + def terminate(_reason, %{terminate_sleep_ms: sleep_ms}) do + Process.sleep(sleep_ms) + :ok + end + end + defmodule InitInfoServer do use DurableServer, vsn: 1 @@ -709,6 +736,52 @@ defmodule DurableServerTest do end describe "init/1" do + test "releases its prefix claim when the supervisor stops", %{prefix: _prefix} do + unique_id = DurableServer.UUID.uuid4() + supervisor_name = :"restartable_supervisor_#{unique_id}" + prefix = "restartable_#{unique_id}/" + opts = [name: supervisor_name, prefix: prefix, object_store: test_object_store_opts()] + + assert {:ok, first_pid} = DurableServer.Supervisor.start_link(opts) + + assert_raise ArgumentError, ~r/already claimed by supervisor/, fn -> + DurableServer.Supervisor.start_link( + Keyword.put(opts, :name, :"competing_supervisor_#{unique_id}") + ) + end + + assert :ok = Supervisor.stop(first_pid) + assert {:ok, second_pid} = DurableServer.Supervisor.start_link(opts) + assert :ok = Supervisor.stop(second_pid) + end + + test "releases its prefix claim when startup fails", %{prefix: _prefix} do + unique_id = DurableServer.UUID.uuid4() + supervisor_name = :"failed_restartable_supervisor_#{unique_id}" + prefix = "failed_restartable_#{unique_id}/" + + invalid_opts = [ + name: supervisor_name, + prefix: prefix, + backend: {ConsistencyProbeBackend, owner: self()}, + heartbeat_interval_ms: 0 + ] + + failed_start = + Task.async(fn -> + Process.flag(:trap_exit, true) + DurableServer.Supervisor.start_link(invalid_opts) + end) + + assert {:error, _reason} = Task.await(failed_start) + + valid_opts = + Keyword.delete(invalid_opts, :heartbeat_interval_ms) + + assert {:ok, supervisor_pid} = DurableServer.Supervisor.start_link(valid_opts) + assert :ok = Supervisor.stop(supervisor_pid) + end + test "initializes successfully with valid options", %{ supervisor_name: supervisor_name, prefix: _prefix @@ -849,6 +922,18 @@ defmodule DurableServerTest do end end + test "rejects child keys in the internal heartbeat namespace", %{ + supervisor_name: supervisor_name, + prefix: _prefix + } do + assert_raise ArgumentError, ~r/reserved internal namespace/, fn -> + DurableServer.Supervisor.start_child( + supervisor_name, + {TestServer, key: "__nodes/future@host", initial_state: %{}} + ) + end + end + test "validates child args require map :initial_state", %{ supervisor_name: supervisor_name, prefix: _prefix @@ -1318,6 +1403,14 @@ defmodule DurableServerTest do assert GenServer.call(pid, :get_count) == 1 end + test "supports integer timeout actions from call and noreply callbacks", %{pid: pid} do + assert GenServer.call(pid, :increment_with_timeout) == 1 + GenServer.cast(pid, :increment_with_timeout) + :sys.get_state(pid) + assert GenServer.call(pid, :get_count) == 2 + refute_process_down(pid) + end + test "handles cast messages", %{pid: pid} do GenServer.cast(pid, :increment) # Allow cast to process @@ -1486,9 +1579,7 @@ defmodule DurableServerTest do ) # Test that invalid meta types cause the GenServer call to exit with FunctionClauseError - catch_exit do - GenServer.call(pid, {:invalid_meta_test, "not_a_map"}) - end + catch_exit(GenServer.call(pid, {:invalid_meta_test, "not_a_map"})) end test "handles stop with reply", %{pid: pid} do @@ -1676,6 +1767,30 @@ defmodule DurableServerTest do } = persisted_data end + test "stops when auto sync loses the storage CAS", %{prefix: _prefix} do + {supervisor_name, _supervisor_pid, prefix} = + start_test_supervisor(backend: {ConsistencyProbeBackend, owner: self()}) + + key = "auto-sync-conflict-#{DurableServer.UUID.uuid4()}" + + {:ok, {pid, _meta}} = + DurableServer.Supervisor.start_child( + supervisor_name, + {AutoSyncServer, key: key, initial_state: %{}} + ) + + %{storage_backend: backend} = DurableServer.Supervisor.__get_config__(supervisor_name) + storage_key = prefix <> key + {:ok, %{body: stored_state, etag: etag}} = StorageBackend.get_object(backend, storage_key) + + assert {:ok, _obj} = + StorageBackend.put_object(backend, storage_key, stored_state, etag: etag) + + ref = Process.monitor(pid) + assert catch_exit(GenServer.call(pid, :increment)) + assert_receive {:DOWN, ^ref, :process, ^pid, _reason} + end + test "does not auto sync when disabled", %{supervisor_name: supervisor_name, prefix: prefix} do key = "no-auto-sync-test-#{DurableServer.UUID.uuid4()}" @@ -2114,6 +2229,76 @@ defmodule DurableServerTest do end end + describe "graceful shutdown" do + test "reports final persistence failures", %{prefix: _prefix} do + unique_id = DurableServer.UUID.uuid4() + supervisor_name = :"shutdown_failure_#{unique_id}" + prefix = "shutdown_failure_#{unique_id}/" + + assert {:ok, supervisor_pid} = + DurableServer.Supervisor.start_link( + name: supervisor_name, + prefix: prefix, + backend: {ConsistencyProbeBackend, owner: self()} + ) + + key = "shutdown-sync-failure" + + assert {:ok, {_server_pid, _meta}} = + DurableServer.Supervisor.start_child( + supervisor_name, + {TestServer, key: key, initial_state: %{}} + ) + + %{storage_backend: backend} = DurableServer.Supervisor.__get_config__(supervisor_name) + storage_key = prefix <> key + + assert {:ok, %{body: stored_state, etag: etag}} = + StorageBackend.get_object(backend, storage_key) + + assert {:ok, _obj} = + StorageBackend.put_object(backend, storage_key, stored_state, etag: etag) + + log = + capture_log(fn -> assert :ok = Supervisor.stop(supervisor_pid, :normal, :infinity) end) + + assert log =~ "1 DurableServer children failed final persistence during shutdown" + end + + test "uses one timeout budget across all concurrency batches", %{prefix: _prefix} do + unique_id = DurableServer.UUID.uuid4() + supervisor_name = :"shutdown_budget_#{unique_id}" + + assert {:ok, supervisor_pid} = + DurableServer.Supervisor.start_link( + name: supervisor_name, + prefix: "shutdown_budget_#{unique_id}/", + backend: {ConsistencyProbeBackend, owner: self()}, + graceful_shutdown_concurrency: 1, + graceful_shutdown_timeout_ms: 75 + ) + + pids = + for index <- 1..4 do + key = "slow-shutdown-#{index}-#{DurableServer.UUID.uuid4()}" + + assert {:ok, {pid, _meta}} = + DurableServer.Supervisor.start_child( + supervisor_name, + {SlowTerminateServer, key: key, initial_state: %{terminate_sleep_ms: 1_000}} + ) + + pid + end + + {elapsed_us, :ok} = :timer.tc(fn -> Supervisor.stop(supervisor_pid, :normal, :infinity) end) + elapsed_ms = div(elapsed_us, 1_000) + + assert elapsed_ms < 225 + assert Enum.all?(pids, &(not Process.alive?(&1))) + end + end + describe "error handling" do test "continues operation when sync operations encounter errors", %{ supervisor_name: supervisor_name, @@ -2472,6 +2657,56 @@ defmodule DurableServerTest do assert_receive {:DOWN, ^ref, :process, ^blocked_pid, _reason}, 2_000 end + test "singleflight waiter retries when the owner finishes before waiter registration", %{ + supervisor_name: supervisor_name, + prefix: _prefix + } do + singleflight_key = + {:ensure_started_child, "registration-race", DurableServerTest.BlockingInitServer} + + owner_registry = :"durable_sf_owner_#{supervisor_name}" + waiters_registry = :"durable_sf_waiters_#{supervisor_name}" + parent = self() + + owner_pid = + spawn(fn -> + {:ok, _} = Registry.register(owner_registry, singleflight_key, :singleflight_owner) + send(parent, {:singleflight_owner_ready, self()}) + + receive do + :finish -> + Registry.dispatch(waiters_registry, singleflight_key, fn _entries -> :ok end) + Registry.unregister(owner_registry, singleflight_key) + send(parent, {:singleflight_owner_finished, self()}) + + receive do + :stop -> :ok + end + end + end) + + assert_receive {:singleflight_owner_ready, ^owner_pid} + send(owner_pid, :finish) + assert_receive {:singleflight_owner_finished, ^owner_pid} + assert Process.alive?(owner_pid) + + waiter_ref = make_ref() + reply_alias = :erlang.alias() + + assert :retry = + DurableServer.Supervisor.__register_singleflight_waiter__( + owner_registry, + waiters_registry, + singleflight_key, + owner_pid, + {waiter_ref, reply_alias} + ) + + :erlang.unalias(reply_alias) + Registry.unregister(waiters_registry, singleflight_key) + send(owner_pid, :stop) + end + test "timed out singleflight waiter does not receive late singleflight_done", %{ supervisor_name: supervisor_name, prefix: _prefix @@ -2664,6 +2899,50 @@ defmodule DurableServerTest do assert data.meta.status == :crashed end + test "a stale crashing process does not mark a newer owner as crashed", %{ + prefix: _prefix + } do + {supervisor_name, _supervisor_pid, prefix} = + start_test_supervisor(backend: {ConsistencyProbeBackend, owner: self()}) + + key = "stale-crash-owner-#{DurableServer.UUID.uuid4()}" + + {:ok, {pid, _meta}} = + DurableServer.Supervisor.start_child( + supervisor_name, + {EdgeCaseTestServer, key: key, initial_state: %{count: 0}} + ) + + %{storage_backend: backend} = DurableServer.Supervisor.__get_config__(supervisor_name) + storage_key = prefix <> key + + assert {:ok, %{body: stored_state, etag: etag}} = + StorageBackend.get_object(backend, storage_key) + + newer_owner = + stored_state.meta + |> Map.put(:pid, self()) + |> Map.update!(:node_ref, &(&1 + 1)) + |> Meta.put_status(:running) + + assert {:ok, _obj} = + StorageBackend.put_object( + backend, + storage_key, + %{stored_state | meta: newer_owner}, + etag: etag + ) + + ref = Process.monitor(pid) + assert catch_exit(GenServer.call(pid, :crash)) + assert_receive {:DOWN, ^ref, :process, ^pid, _reason} + + assert {:ok, %{body: final_state}} = StorageBackend.get_object(backend, storage_key) + assert final_state.meta.pid == self() + assert final_state.meta.node_ref == newer_owner.node_ref + assert final_state.meta.status == :running + end + test "abnormal stop reasons mark status as crashed", %{ supervisor_name: supervisor_name, prefix: prefix @@ -3382,6 +3661,10 @@ defmodule DurableServerTest do {:stop, {:shutdown, :permanent}, :ok, state} end + def handle_call(:stop_shutdown_normal, _from, state) do + {:stop, {:shutdown, :normal}, :ok, state} + end + def handle_call(:stop_permanent_non_shutdown, _from, state) do {:stop, :permanent, :ok, state} end @@ -3438,6 +3721,77 @@ defmodule DurableServerTest do assert_receive {:terminate, _reason, ^key} end + test "a stale process cannot delete a newer owner's object", %{prefix: _prefix} do + {supervisor_name, _supervisor_pid, prefix} = + start_test_supervisor(backend: {ConsistencyProbeBackend, owner: self()}) + + key = "stale-delete-owner-#{DurableServer.UUID.uuid4()}" + + {:ok, {server_pid, _meta}} = + DurableServer.Supervisor.start_child( + supervisor_name, + {DeleteTestServer, key: key, initial_state: %{}} + ) + + %{storage_backend: backend} = DurableServer.Supervisor.__get_config__(supervisor_name) + storage_key = prefix <> key + + assert {:ok, %{body: stored_state, etag: etag}} = + StorageBackend.get_object(backend, storage_key) + + newer_owner = + stored_state.meta + |> Map.put(:pid, self()) + |> Map.update!(:node_ref, &(&1 + 1)) + |> Meta.put_status(:running) + + assert {:ok, _obj} = + StorageBackend.put_object( + backend, + storage_key, + %{stored_state | meta: newer_owner}, + etag: etag + ) + + ref = Process.monitor(server_pid) + assert :ok = GenServer.call(server_pid, :delete_self) + assert_receive {:DOWN, ^ref, :process, ^server_pid, {:shutdown, :delete}} + + assert {:ok, %{body: final_state}} = StorageBackend.get_object(backend, storage_key) + assert final_state.meta.pid == self() + assert final_state.meta.node_ref == newer_owner.node_ref + assert final_state.meta.status == :running + end + + test "terminate_and_delete_child returns a storage deletion failure", %{prefix: _prefix} do + {supervisor_name, _supervisor_pid, prefix} = + start_test_supervisor(backend: {ConsistencyProbeBackend, owner: self()}) + + %{storage_backend: backend} = DurableServer.Supervisor.__get_config__(supervisor_name) + + for target <- [:pid, :key] do + key = "delete-failure-#{target}-#{DurableServer.UUID.uuid4()}" + + {:ok, {server_pid, _meta}} = + DurableServer.Supervisor.start_child( + supervisor_name, + {DeleteTestServer, key: key, initial_state: %{}} + ) + + storage_key = prefix <> key + put_backend_delete_override(backend_table(backend), storage_key, {:error, :unavailable}) + delete_target = if target == :pid, do: server_pid, else: key + + assert {:error, :unavailable} = + DurableServer.Supervisor.terminate_and_delete_child( + supervisor_name, + delete_target + ) + + assert {:ok, _object} = StorageBackend.get_object(backend, storage_key) + end + end + test "terminate_and_delete_child/2 with key deletes running process and storage", %{ supervisor_name: supervisor_name, prefix: prefix @@ -3751,6 +4105,31 @@ defmodule DurableServerTest do assert_receive {:EXIT, ^server_pid, :normal}, 100 end + test "{:stop, {:shutdown, :normal}, state} stops gracefully with the wrapped reason", %{ + supervisor_name: supervisor_name, + prefix: prefix + } do + key = "normal_shutdown_#{DurableServer.UUID.uuid4()}" + + {:ok, {server_pid, _meta}} = + DurableServer.Supervisor.start_child( + supervisor_name, + {DeleteTestServer, key: key, initial_state: %{}} + ) + + ref = Process.monitor(server_pid) + assert :ok = GenServer.call(server_pid, :stop_shutdown_normal) + assert_receive {:DOWN, ^ref, :process, ^server_pid, {:shutdown, :normal}} + + assert {:ok, stored_state} = + DurableServer.fetch_stored_state( + DurableServer.Supervisor.__get_config__(supervisor_name).storage_backend, + %{key: key, prefix: prefix} + ) + + assert stored_state.meta.status == :stopped_graceful + end + test "{:stop, {:shutdown, :permanent}, state} stops with permanent status and propagates exit", %{ supervisor_name: supervisor_name, diff --git a/test/mirror_backend_e2e_test.exs b/test/mirror_backend_e2e_test.exs index 26b95ef..4eb56a1 100644 --- a/test/mirror_backend_e2e_test.exs +++ b/test/mirror_backend_e2e_test.exs @@ -376,7 +376,6 @@ defmodule DurableServer.MirrorBackendE2ETest do :ok = Supervisor.stop(pid) assert_receive {:DOWN, ^monitor_ref, :process, ^pid, _reason}, 5_000 assert_eventually(fn -> Process.whereis(supervisor_name) == nil end) - :persistent_term.erase({DurableServer.Supervisor, :prefix, prefix}) :ok end diff --git a/test/mirror_backend_integration_test.exs b/test/mirror_backend_integration_test.exs index 7993222..d55683f 100644 --- a/test/mirror_backend_integration_test.exs +++ b/test/mirror_backend_integration_test.exs @@ -34,8 +34,7 @@ defmodule DurableServer.MirrorBackendIntegrationTest do 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) + def try_claim(_state, _key, _body), do: {:error, :promotion_write_rejected} @impl true def update_object(%{delegate: delegate}, key, update_fn, opts),