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),