From 2c72cfa3319b67e28f7b3a7ad405f2c0926f6bb8 Mon Sep 17 00:00:00 2001 From: Gary Rennie Date: Tue, 13 Oct 2026 14:36:42 +0100 Subject: [PATCH] Add durable state machines backed by native gen_statem Allow modules to select GenServer or :gen_statem through the :behaviour option, keeping GenServer as the default. State machines use the existing supervisor, persistence, placement, and lifecycle APIs. Run state machines through proc_lib startup and gen_statem.enter_loop, letting OTP manage transitions, state entry, postponed events, timers, and inserted events. Acknowledge process startup before acquiring the storage lock so slow initialization does not block other children. Support both handle_event_function and state_functions, with an explicit callback_mode/0 and optional state-entry callbacks. Persist the state name and data together as {state_name, data}, using application-defined dump_state/1 and load_state/2 callbacks. Translate {:durable, options} actions into durability operations. Explicit sync commits state before executing native actions, including replies and inserted events. Failed explicit writes terminate the owner without executing those actions. Preserve native initialization and termination semantics: - Track crashes and invoke termination callbacks when initial entry fails. - Preserve the initial state name throughout repeated entry callbacks. - Ignore init postponement actions without replaying initialization. - Release storage claims when initialization refuses to start. - Wait for termination and final persistence before acknowledging supervisor stop requests. - Retain native reply-before-termination ordering for stop_and_reply. Move callback defaults, startup, dispatch, termination, validation, and code-change handling into internal DurableServer.Behaviours adapters. Store adapter configuration in a behaviour tuple and retain compatibility with existing GenServer processes that lack that field. Support hot upgrades within an engine and reject engine changes requiring a restart. Reduce callback overhead by inlining common helpers, reusing unchanged clean state, avoiding redundant dirty-flag writes, and returning early when automatic synchronization has no work. --- README.md | 68 +- lib/durable_server.ex | 259 +++--- lib/durable_server/behaviours/gen_server.ex | 113 +++ .../behaviours/state_machine.ex | 336 +++++++ lib/durable_server/lifecycle_manager.ex | 7 + lib/durable_server/supervisor.ex | 50 +- test/durable_server/state_machine_test.exs | 844 ++++++++++++++++++ .../supervisor_backend_spec_test.exs | 91 ++ 8 files changed, 1638 insertions(+), 130 deletions(-) create mode 100644 lib/durable_server/behaviours/gen_server.ex create mode 100644 lib/durable_server/behaviours/state_machine.ex create mode 100644 test/durable_server/state_machine_test.exs diff --git a/README.md b/README.md index 39b37e9..c8ba080 100644 --- a/README.md +++ b/README.md @@ -1,6 +1,6 @@ # DurableServer -DurableServer provides durable, distributed GenServer processes backed by pluggable storage backends. +DurableServer provides durable, distributed GenServer and `gen_statem` processes backed by pluggable storage backends. It implements fault-tolerant, stateful processes that can survive node failures, restarts, and deployments by automatically persisting state and coordinating across a distributed cluster. @@ -94,11 +94,75 @@ GenServer.call(pid, :increment) # => 2 GenServer.call(pid, :get_count) # => 2 ``` -`:initial_state` is required and must be a map. On first boot, DurableServer +For GenServers, `:initial_state` is required and must be a map. On first boot, DurableServer passes it through `dump_state/1`, the configured backend's encode/decode path, and then `load_state/2` before `init/1` or `init/2`. The dumped initial state must therefore be encodable by your configured backend. +## State Machines + +Opt in with `use DurableServer, vsn: 1, behaviour: :gen_statem`. The process runs +on OTP's native [`gen_statem`](https://www.erlang.org/doc/apps/stdlib/gen_statem.html) +engine, with state entry callbacks, postponed events, inserted events, named +timeouts, and state timeouts. Implement `callback_mode/0` to select +`:handle_event_function` or `:state_functions`, optionally with `:state_enter`, +just as with native `gen_statem`. +The `:behaviour` option accepts `GenServer` (the default) or `:gen_statem`. + +```elixir +defmodule DurableSwitch do + use DurableServer, vsn: 1, behaviour: :gen_statem + + def callback_mode, do: :handle_event_function + + # The state name and data are persisted together, using your chosen encoding. + def dump_state({state, count}), do: %{"state" => Atom.to_string(state), "count" => count} + def load_state(_, %{"state" => "off", "count" => count}), do: {:off, count} + def load_state(_, %{"state" => "on", "count" => count}), do: {:on, count} + + def init({state, count}) do + {:ok, state, count, [{:durable, [permanent: true]}]} + end + + def handle_event({:call, from}, :toggle, state, count) do + next = if state == :off, do: :on, else: :off + + {:next_state, next, count + 1, + [{:durable, [sync: true, meta: %{state: next}]}, {:reply, from, next}]} + end +end + +{:ok, {pid, _meta}} = DurableServer.Supervisor.start_child( + MyDurableSup, + {DurableSwitch, key: "switch_123", initial_state: {:off, 0}} +) + +:gen_statem.call(pid, :toggle) # => :on +``` + +`:initial_state`, `load_state/2`, and `dump_state/1` use `{state_name, data}`. +The data can be any term your serialization callbacks can encode into a plain +map. Without a custom `init/1`, the loaded pair becomes the initial state and +data. `init/2` is also supported and receives the usual supervisor info map. + +Use `{:durable, options}` in a native action list for durability options: + +- In `init/1` or `init/2`: `:auto_sync`, `:sync_every_ms`, `:meta`, and `:permanent`. +- In event callbacks: `:sync` and `:meta`. + +`sync: true` persists the transition before any native actions execute, including +replies and `:next_event`. A failed explicit write terminates the owner without +running those actions. `handle_sync/3` receives the old and new state/data pairs +and has the same post-write semantics as with GenServer. Native timers and +postponed events remain in memory; reconstruct any required work on cold starts. + +Use the supervisor's existing lifecycle API for stopping, deleting, and cordoning. +Supervisor stop calls wait for `terminate/3` and final persistence. A user callback's +native `:stop_and_reply` retains OTP's reply-before-termination semantics. +`code_change/4` receives the state name and data separately for hot upgrades. +For inspection, `:sys.get_state/1` returns +`{state_name, %DurableServer{user_state: {state_name, data}}}`. + ## Administrative Cordon Use `terminate_and_cordon_child/3` when you need to stop a DurableServer and diff --git a/lib/durable_server.ex b/lib/durable_server.ex index 5154e38..58e37f6 100644 --- a/lib/durable_server.ex +++ b/lib/durable_server.ex @@ -1,6 +1,6 @@ defmodule DurableServer do @moduledoc ~S""" - DurableServer provides durable, distributed GenServer processes backed by pluggable storage. + DurableServer provides durable, distributed GenServer and `gen_statem` processes backed by pluggable storage. DurableServer implements fault-tolerant, stateful processes that can survive node failures, restarts, and deployments by automatically persisting state to storage and @@ -28,6 +28,32 @@ defmodule DurableServer do See `DurableServer.Supervisor` for supervisor setup and configuration options. + ## State machines + + `use DurableServer, vsn: 1, behaviour: :gen_statem` selects OTP's native state + machine engine. The `:behaviour` option accepts `GenServer` (the default) or + `:gen_statem`. The same supervisor, placement, storage, and lifecycle APIs + serve both engines. + + Implement `callback_mode/0` to select `:handle_event_function` or + `:state_functions`, optionally with `:state_enter`. Event callbacks, + `terminate/3`, and `code_change/4` use their native state-machine signatures. + + Durable user state is `{state_name, data}`. Both `:initial_state` and the + return from `load_state/2` use that pair; `dump_state/1` receives the pair and + must return a plain storage-compatible map. The application controls how the + state name and data are encoded. + + Add `{:durable, options}` to a native action list to use DurableServer options. + In `init/1` or `init/2`, the options are `:auto_sync`, `:sync_every_ms`, `:meta`, + and `:permanent`. In event callbacks, they are `:sync` and `:meta`. An explicit + `sync: true` commits the transition before any native actions execute, + including replies and inserted events. `handle_sync/3` retains its existing + contract and receives the old and new `{state_name, data}` pairs. + + Native postponed events and timers are process-local; reconstruct any + required work after a cold start. + ## Basic Usage defmodule MyCounterServer do @@ -864,6 +890,7 @@ defmodule DurableServer do sync_every_ms: nil, sync_timer_ref: nil, user_state: nil, + behaviour: {DurableServer.Behaviours.GenServer, %{}}, user_meta: nil, crash_history: [], module: nil, @@ -900,6 +927,8 @@ defmodule DurableServer do @lock_claim_retries 2 @bootstrap_continue {@durable, :bootstrap} + @compile {:inline, behaviour: 1, __dispatch__: 3, update_state: 2, auto_sync_to_storage: 2} + defguardp is_callback_action(action) when is_atom(action) or is_tuple(action) or (is_integer(action) and action >= 0) @@ -918,75 +947,30 @@ defmodule DurableServer do "the current :vsn must be provided:, ie: `use DurableServer, vsn: 1`" end - quote do - @behaviour DurableServer - @vsn unquote(vsn) - import unquote(__MODULE__) - - def __durable_server_config__() do - %{ - vsn: @vsn - } - end - - # Default implementations - def init(state) do - {:ok, state} - end - - def handle_call(_request, _from, state) do - {:reply, :ok, state} - end - - def handle_cast(_request, state) do - {:noreply, state} - end + behaviour = Macro.expand(Keyword.get(opts, :behaviour, GenServer), __CALLER__) - def handle_info(_msg, state) do - {:noreply, state} - end - - def handle_continue(_continue, state) do - {:noreply, state} - end + defaults = + case behaviour do + GenServer -> + DurableServer.Behaviours.GenServer.defaults() - def handle_sync(_info, _old_state, new_state) do - {:ok, new_state} - end + :gen_statem -> + DurableServer.Behaviours.StateMachine.defaults() - def terminate(_reason, _state) do - :ok + other -> + raise ArgumentError, + "expected :behaviour to be GenServer or :gen_statem, got: #{inspect(other)}" end - def after_terminate(_terminate_return, _info) do - :ok - end + quote do + @vsn unquote(vsn) + import unquote(__MODULE__) - # Default code_change for hot upgrades: no migration needed - # Override this to provide version-specific state migrations: - # - # def code_change(1, old_state, 2) do - # # Migrate from v1 to v2 - # {:ok, Map.put(old_state, :new_field, "default")} - # end - # - # def code_change(_, state, _) do - # # No migration for other versions - # {:ok, state} - # end - def code_change(_old_vsn, state, _new_vsn) do - {:ok, state} + def __durable_server_config__() do + %{vsn: @vsn, behaviour: unquote(behaviour)} end - defoverridable init: 1, - handle_call: 3, - handle_cast: 2, - handle_info: 2, - handle_continue: 2, - handle_sync: 3, - terminate: 2, - after_terminate: 2, - code_change: 3 + unquote(defaults) end end @@ -1010,7 +994,8 @@ defmodule DurableServer do config: config } = info ) do - GenServer.start_link(__MODULE__, info, start_opts(info.module, config)) + adapter = callback_adapter(info.module) + adapter.start_link(info, start_opts(info.module, config)) end defp start_opts(module, config) do @@ -1064,6 +1049,7 @@ defmodule DurableServer do key: key, prefix: prefix, module: module, + behaviour: {callback_adapter(module), %{}}, supervisor: supervisor_name, dynamic_supervisor: DurableServer.Supervisor.get_dynamic_supervisor(supervisor_name), task_supervisor: DurableServer.Supervisor.get_task_supervisor(supervisor_name), @@ -1317,6 +1303,7 @@ defmodule DurableServer do old_vsn: old_vsn, user_state: user_state, module: module, + behaviour: {callback_adapter(module), %{}}, supervisor: supervisor_name, dynamic_supervisor: DurableServer.Supervisor.get_dynamic_supervisor(supervisor_name), task_supervisor: DurableServer.Supervisor.get_task_supervisor(supervisor_name), @@ -2083,16 +2070,12 @@ defmodule DurableServer do end def handle_call(request, from, %DurableServer{} = state) do - state = maybe_migrate_on_callback(state) - result = state.module.handle_call(request, from, state.user_state) - process_callback_result(result, state) + __dispatch__({:call, from}, request, state) end @impl true def handle_cast(request, %__MODULE__{} = state) do - state = maybe_migrate_on_callback(state) - result = state.module.handle_cast(request, state.user_state) - process_callback_result(result, state) + __dispatch__(:cast, request, state) end @impl true @@ -2189,9 +2172,7 @@ defmodule DurableServer do end def handle_info(msg, %DurableServer{} = state) do - state = maybe_migrate_on_callback(state) - result = state.module.handle_info(msg, state.user_state) - process_callback_result(result, state) + __dispatch__(:info, msg, state) end @impl true @@ -2201,9 +2182,7 @@ defmodule DurableServer do @impl true def handle_continue(continue, %DurableServer{} = state) do - state = maybe_migrate_on_callback(state) - result = state.module.handle_continue(continue, state.user_state) - process_callback_result(result, state) + __dispatch__(:continue, continue, state) end @impl true @@ -2219,8 +2198,9 @@ defmodule DurableServer do end def terminate(reason, %DurableServer{} = state) do - # Ensure user callback terminate/2 finishes before we persist final status metadata. - terminate_return = state.module.terminate(reason, state.user_state) + # Ensure the user terminate callback finishes before persisting final status metadata. + {adapter, _adapter_state} = behaviour(state) + terminate_return = adapter.terminate(reason, state) {final_status, sync_result, delete_result} = case state.user_initiated_stop do @@ -2392,14 +2372,20 @@ defmodule DurableServer do init_info = Map.fetch!(config, :init_info) info = Map.merge(info, init_info) - init_result = - if function_exported?(module, :init, 2) do - module.init(loaded_init_state, info) - else - module.init(loaded_init_state) - end + init_args = + if function_exported?(module, :init, 2), + do: [loaded_init_state, info], + else: [loaded_init_state] + + {adapter, _adapter_state} = behaviour(locked_state) + {locked_state, init_result} = adapter.init_callback(init_args, locked_state) case init_result do + {:statem_init_stop, reason} -> + repair_failed_boot_storage(locked_state) + send(reply_to, {from_ref, {:error, reason}}) + {:stop, {:shutdown, {@durable, {:init_failed, reason}}}, locked_state} + :ignore -> handle_ignore(locked_state, init_from) @@ -2410,11 +2396,11 @@ defmodule DurableServer do handle_init(locked_state, user_state, opts, _continue_or_timeout = nil) {:ok, user_state, {tag, _} = continue_or_timeout} - when tag in [:continue, :timeout] -> + when tag in [:continue, :timeout, :statem] -> handle_init(locked_state, user_state, [], continue_or_timeout) {:ok, user_state, {tag, _} = continue_or_timeout, opts} - when tag in [:continue, :timeout] and is_list(opts) -> + when tag in [:continue, :timeout, :statem] and is_list(opts) -> handle_init(locked_state, user_state, opts, continue_or_timeout) other -> @@ -2824,13 +2810,20 @@ defmodule DurableServer do @impl true def code_change(old_vsn, %__MODULE__{} = state, extra) do - case state.module.code_change(old_vsn, state.user_state, extra) do - {:ok, new_user_state} -> - # The dumped shape can change across upgrades even if user state is unchanged. - {:ok, %{state | user_state: new_user_state, needs_sync_check: true}} + {adapter, _adapter_state} = behaviour(state) - {:error, reason} -> - {:error, reason} + with true <- adapter == callback_adapter(state.module), + {:ok, new_user_state, adapter_state} <- adapter.code_change(old_vsn, state, extra) do + state = + state + |> Map.put(:behaviour, {adapter, adapter_state}) + |> update_state(new_user_state) + + # The dumped shape can change even when the user state is unchanged. + {:ok, %{state | needs_sync_check: true}} + else + false -> {:error, :behaviour_change_requires_restart} + {:error, _reason} = error -> error end end @@ -2977,19 +2970,62 @@ defmodule DurableServer do "[DurableServer] Hot upgrade: #{state.key} migrating from v#{inspect(state.vsn)} to v#{inspect(current_vsn)}" ) - {:ok, new_user_state} = state.module.code_change(state.vsn, state.user_state, current_vsn) + {:ok, migrated} = code_change(state.vsn, state, current_vsn) + %{migrated | vsn: current_vsn, old_vsn: state.vsn} + end + end + + @doc false + def __callback_behaviour__(module), + do: Map.get(module.__durable_server_config__(), :behaviour, GenServer) - %{ - state - | vsn: current_vsn, - old_vsn: state.vsn, - user_state: new_user_state, - needs_sync_check: true - } + defp callback_adapter(module) do + case __callback_behaviour__(module) do + GenServer -> DurableServer.Behaviours.GenServer + :gen_statem -> DurableServer.Behaviours.StateMachine end end - defp process_callback_result(result, %__MODULE__{} = state) do + defp behaviour(state) do + case state do + %{behaviour: behaviour} -> behaviour + # Processes started before adapter support have no behaviour field. + _ -> {DurableServer.Behaviours.GenServer, %{}} + end + end + + @doc false + def __dispatch__(type, content, state) do + state = maybe_migrate_on_callback(state) + {adapter, _adapter_state} = behaviour(state) + adapter.dispatch(type, content, state) + end + + @doc false + def __statem_transition__(user_state, opts, state), + do: __process_callback_result__({:noreply, user_state, opts}, state) + + @doc false + def __statem_stop__(reason, user_state, state) + when reason in [ + :normal, + :delete, + :permanent, + {:shutdown, :normal}, + {:shutdown, :delete}, + {:shutdown, :permanent} + ] do + __process_callback_result__({:stop, reason, user_state}, state) + end + + def __statem_stop__({:error, _} = reason, user_state, state), + do: __process_callback_result__({:stop, reason, user_state}, state) + + def __statem_stop__(reason, user_state, state), + do: {:stop, reason, update_state(state, user_state)} + + @doc false + def __process_callback_result__(result, %__MODULE__{} = state) do case result do {:reply, reply, new_user_state} -> old_user_state = state.user_state @@ -3350,23 +3386,20 @@ defmodule DurableServer do end end - defp validate_user_state!(%{} = user_state), do: user_state - - defp validate_user_state!(user_state) do - raise ArgumentError, - "expected callback to return a map of user state, got: #{inspect(user_state)}" - end - defp update_state(%__MODULE__{} = state, new_user_state) do - new_user_state = validate_user_state!(new_user_state) + {adapter, _adapter_state} = behaviour(state) + new_user_state = adapter.validate_user_state!(new_user_state) - %{ - state - | user_state: new_user_state, - needs_sync_check: state.needs_sync_check or new_user_state !== state.user_state - } + cond do + state.needs_sync_check -> %{state | user_state: new_user_state} + new_user_state === state.user_state -> state + true -> %{state | user_state: new_user_state, needs_sync_check: true} + end end + defp auto_sync_to_storage(%{auto_sync: false} = state, _old_user_state), do: state + defp auto_sync_to_storage(%{needs_sync_check: false} = state, _old_user_state), do: state + defp auto_sync_to_storage(%DurableServer{module: module, key: key} = state, old_user_state) do if state.auto_sync && state.needs_sync_check do old_user_state = old_user_state || state.user_state diff --git a/lib/durable_server/behaviours/gen_server.ex b/lib/durable_server/behaviours/gen_server.ex new file mode 100644 index 0000000..d7009f3 --- /dev/null +++ b/lib/durable_server/behaviours/gen_server.ex @@ -0,0 +1,113 @@ +defmodule DurableServer.Behaviours.GenServer do + @moduledoc false + + def defaults do + quote do + @behaviour DurableServer + # Default implementations + def init(state) do + {:ok, state} + end + + def handle_call(_request, _from, state) do + {:reply, :ok, state} + end + + def handle_cast(_request, state) do + {:noreply, state} + end + + def handle_info(_msg, state) do + {:noreply, state} + end + + def handle_continue(_continue, state) do + {:noreply, state} + end + + def handle_sync(_info, _old_state, new_state) do + {:ok, new_state} + end + + def terminate(_reason, _state) do + :ok + end + + def after_terminate(_terminate_return, _info) do + :ok + end + + # Default code_change for hot upgrades: no migration needed + # Override this to provide version-specific state migrations: + # + # def code_change(1, old_state, 2) do + # # Migrate from v1 to v2 + # {:ok, Map.put(old_state, :new_field, "default")} + # end + # + # def code_change(_, state, _) do + # # No migration for other versions + # {:ok, state} + # end + def code_change(_old_vsn, state, _new_vsn) do + {:ok, state} + end + + defoverridable init: 1, + handle_call: 3, + handle_cast: 2, + handle_info: 2, + handle_continue: 2, + handle_sync: 3, + terminate: 2, + after_terminate: 2, + code_change: 3 + end + end + + def start_link(info, opts) do + GenServer.start_link(DurableServer, info, opts) + end + + def init_callback(args, durable) do + {durable, apply(durable.module, :init, args)} + end + + def dispatch({:call, from}, request, %{module: module, user_state: user_state} = durable) do + result = module.handle_call(request, from, user_state) + DurableServer.__process_callback_result__(result, durable) + end + + def dispatch(:cast, request, %{module: module, user_state: user_state} = durable) do + result = module.handle_cast(request, user_state) + DurableServer.__process_callback_result__(result, durable) + end + + def dispatch(:info, message, %{module: module, user_state: user_state} = durable) do + result = module.handle_info(message, user_state) + DurableServer.__process_callback_result__(result, durable) + end + + def dispatch(:continue, continue, %{module: module, user_state: user_state} = durable) do + result = module.handle_continue(continue, user_state) + DurableServer.__process_callback_result__(result, durable) + end + + def terminate(reason, durable) do + durable.module.terminate(reason, durable.user_state) + end + + def code_change(old_vsn, durable, extra) do + case durable.module.code_change(old_vsn, durable.user_state, extra) do + {:ok, user_state} -> {:ok, user_state, %{}} + {:error, _reason} = error -> error + end + end + + def validate_user_state!(%{} = user_state), do: user_state + + def validate_user_state!(user_state) do + raise ArgumentError, + "expected callback to return a map of user state, got: #{inspect(user_state)}" + end +end diff --git a/lib/durable_server/behaviours/state_machine.ex b/lib/durable_server/behaviours/state_machine.ex new file mode 100644 index 0000000..fac885b --- /dev/null +++ b/lib/durable_server/behaviours/state_machine.ex @@ -0,0 +1,336 @@ +defmodule DurableServer.Behaviours.StateMachine do + @moduledoc false + + @behaviour :gen_statem + + @type user_state :: {:gen_statem.state(), :gen_statem.data()} + @callback dump_state(user_state()) :: map() + @callback load_state(pos_integer() | nil, map()) :: user_state() + @callback init(user_state(), map()) :: :gen_statem.init_result(:gen_statem.state()) + @callback handle_sync(map(), user_state(), user_state()) :: {:ok, user_state()} + @callback after_terminate(term(), map()) :: term() + @optional_callbacks init: 2, handle_sync: 3, after_terminate: 2 + + @doc false + def defaults do + quote do + @behaviour :gen_statem + @behaviour DurableServer.Behaviours.StateMachine + + def init({name, data}), do: {:ok, name, data} + def handle_sync(_info, _old_state, new_state), do: {:ok, new_state} + def terminate(_reason, _name, _data), do: :ok + def after_terminate(_return, _info), do: :ok + def code_change(_old_vsn, name, data, _extra), do: {:ok, name, data} + + defoverridable init: 1, + handle_sync: 3, + terminate: 3, + after_terminate: 2, + code_change: 4 + end + end + + @doc false + def start_link(info, opts) do + :proc_lib.start_link(__MODULE__, :init, [{info, opts}]) + end + + @impl true + def callback_mode, do: [:handle_event_function, :state_enter] + + @impl true + def init({info, opts}) do + {:ok, durable, {:continue, bootstrap}} = DurableServer.init(info) + # Acknowledge before lock acquisition so storage cannot block other starts. + # DurableServer separately notifies the caller when bootstrap completes. + :proc_lib.init_ack({:ok, self()}) + + case DurableServer.handle_continue(bootstrap, durable) do + {:noreply, initialized, {:statem, actions}} -> + {name, _data} = initialized.user_state + :gen_statem.enter_loop(__MODULE__, opts, name, initialized, self(), actions) + + {:stop, reason, stopped} -> + DurableServer.terminate(reason, stopped) + exit(reason) + end + end + + @impl true + def handle_event(:enter, previous, name, durable) do + {__MODULE__, adapter_state} = durable.behaviour + + if adapter_state.state_enter? do + durable_event(:__dispatch__, [:enter, previous, durable], name) + else + {:keep_state, durable} + end + end + + def handle_event({:call, from}, request, name, durable) do + durable_event(:handle_call, [request, from, durable], name, from) + end + + def handle_event(:cast, request, name, durable) do + durable_event(:handle_cast, [request, durable], name) + end + + def handle_event(:info, message, name, durable) do + durable_event(:handle_info, [message, durable], name) + end + + def handle_event(type, content, name, durable) do + durable_event(:__dispatch__, [type, content, durable], name) + end + + @impl true + def terminate(reason, _name, durable) do + result = DurableServer.terminate(reason, durable) + {__MODULE__, adapter_state} = durable.behaviour + + # DurableServer's supervisor control calls acknowledge only after the + # terminate callback and final storage write. Native stop_and_reply sends + # replies before terminate, so it cannot implement these control calls. + if reply = adapter_state[:stop_reply] do + {from, response} = reply + :gen_statem.reply(from, response) + end + + result + end + + @impl true + def code_change(old_vsn, _name, durable, extra) do + case DurableServer.code_change(old_vsn, durable, extra) do + {:ok, updated} -> + {name, _data} = updated.user_state + {:ok, name, updated} + + {:error, _reason} = error -> + error + end + end + + @doc false + def init_callback(args, durable) do + init_result(durable, invoke_callback(durable.module, :init, args)) + end + + @doc false + def terminate(reason, durable) do + {name, data} = durable.user_state + invoke_callback(durable.module, :terminate, [reason, name, data]) + end + + @doc false + def code_change(old_vsn, durable, extra) do + {name, data} = durable.user_state + {__MODULE__, adapter_state} = durable.behaviour + + case invoke_callback(durable.module, :code_change, [old_vsn, name, data, extra]) do + {:ok, name, data} -> + adapter_state = Map.merge(adapter_state, callback_config(durable.module)) + {:ok, {name, data}, adapter_state} + + {:error, _reason} = error -> + error + end + end + + @doc false + def validate_user_state!({_name, _data} = user_state), do: user_state + + def validate_user_state!(user_state) do + raise ArgumentError, + "expected state-machine user state to be {state_name, data}, got: #{inspect(user_state)}" + end + + defp init_result(durable, result) do + case result do + {:ok, name, data} -> + init_result(durable, {:ok, name, data, []}) + + {:ok, name, data, actions} -> + {actions, opts} = split_actions(actions) + durable = %{durable | behaviour: {__MODULE__, callback_config(durable.module)}} + validate_state!(durable, name) + {durable, {:ok, {name, data}, {:statem, actions}, opts}} + + result when result == :ignore -> + {durable, result} + + {tag, reason} when tag in [:stop, :error] -> + {durable, {:statem_init_stop, reason}} + + other -> + {durable, {:bad_statem_init_return, other}} + end + end + + defp callback_config(module) do + modes = actions_list(invoke_callback(module, :callback_mode, [])) + {enters, modes} = Enum.split_with(modes, &(&1 == :state_enter)) + + case modes do + [mode] when mode in [:handle_event_function, :state_functions] -> + %{mode: mode, state_enter?: enters != []} + + _ -> + raise ArgumentError, + "invalid #{inspect(module)}.callback_mode/0: #{inspect(modes ++ enters)}" + end + end + + @doc false + def dispatch(type, content, durable) do + {name, data} = durable.user_state + {__MODULE__, adapter_state} = durable.behaviour + validate_state!(durable, name) + + result = + case adapter_state.mode do + :handle_event_function -> + invoke_callback(durable.module, :handle_event, [type, content, name, data]) + + :state_functions -> + invoke_callback(durable.module, name, [type, content, data]) + end + + callback_result(result, durable) + end + + # OTP accepts thrown values from native callbacks as ordinary returns. Catch + # them before returning to OTP, or they could skip durability and replace the + # wrapper's data with the application's data. + defp invoke_callback(module, function, args) do + apply(module, function, args) + catch + :throw, result -> result + end + + defp durable_event(function, args, name, from \\ nil) do + apply(DurableServer, function, args) |> native_result(name, from) + catch + # A throw from a storage or durability callback is not a native return. + :throw, value -> :erlang.raise(:error, {:nocatch, value}, __STACKTRACE__) + end + + defp callback_result({:next_state, name, data}, durable), + do: transition(:next_state, name, data, [], durable) + + defp callback_result({:next_state, name, data, actions}, durable), + do: transition(:next_state, name, data, actions, durable) + + defp callback_result({kind, data}, durable) when kind in [:keep_state, :repeat_state], + do: callback_result({kind, data, []}, durable) + + defp callback_result({kind, data, actions}, durable) + when kind in [:keep_state, :repeat_state] do + {name, _data} = durable.user_state + transition(kind, name, data, actions, durable) + end + + defp callback_result(kind, durable) when kind in [:keep_state_and_data, :repeat_state_and_data], + do: callback_result({kind, []}, durable) + + defp callback_result({kind, actions}, durable) + when kind in [:keep_state_and_data, :repeat_state_and_data] do + {name, data} = durable.user_state + kind = if kind == :repeat_state_and_data, do: :repeat_state, else: :keep_state + transition(kind, name, data, actions, durable) + end + + defp callback_result(:stop, durable), do: callback_result({:stop, :normal}, durable) + + defp callback_result({:stop, reason}, durable) do + {_name, data} = durable.user_state + callback_result({:stop, reason, data}, durable) + end + + defp callback_result({:stop, reason, data}, durable) do + {name, _data} = durable.user_state + DurableServer.__statem_stop__(reason, {name, data}, durable) + end + + defp callback_result({:stop_and_reply, reason, replies}, durable) do + {_name, data} = durable.user_state + callback_result({:stop_and_reply, reason, replies, data}, durable) + end + + defp callback_result({:stop_and_reply, reason, replies, data}, durable) do + {name, _data} = durable.user_state + {:stop, reason, durable} = DurableServer.__statem_stop__(reason, {name, data}, durable) + {:statem_stop_and_reply, reason, replies, durable} + end + + defp callback_result(other, _durable), + do: raise(ArgumentError, "invalid state-machine callback return: #{inspect(other)}") + + defp transition(kind, name, data, actions, durable) do + validate_state!(durable, name) + {actions, opts} = split_actions(actions) + {:noreply, durable} = DurableServer.__statem_transition__({name, data}, opts, durable) + {:noreply, durable, {:statem, kind, actions}} + end + + defp split_actions(actions) do + {native, opts} = + Enum.reduce(actions_list(actions), {[], []}, fn + {:durable, options}, {native, opts} when is_list(options) -> + {native, Keyword.merge(opts, options)} + + action, _acc + when action == :pop_callback_module or + (is_tuple(action) and tuple_size(action) == 2 and + elem(action, 0) in [:change_callback_module, :push_callback_module]) -> + raise ArgumentError, + "callback-module switching bypasses DurableServer: #{inspect(action)}" + + action, {native, opts} -> + {[action | native], opts} + end) + + {Enum.reverse(native), opts} + end + + defp actions_list(actions) when is_list(actions), do: actions + defp actions_list(action), do: [action] + + defp validate_state!(%{behaviour: {__MODULE__, %{mode: :state_functions}}}, name) + when not is_atom(name), + do: raise(ArgumentError, "state_functions requires an atom state, got: #{inspect(name)}") + + defp validate_state!(_durable, _name), do: :ok + + defp native_result({:noreply, durable}, _previous, _from) do + {name, _data} = durable.user_state + {:next_state, name, durable} + end + + defp native_result({:noreply, durable, {:statem, kind, actions}}, previous, _from) do + {name, _data} = durable.user_state + + if kind == :repeat_state and name === previous, + do: {:repeat_state, durable, actions}, + else: {:next_state, name, durable, actions} + end + + defp native_result({:reply, response, durable}, _previous, from) do + {name, _data} = durable.user_state + {:next_state, name, durable, [{:reply, from, response}]} + end + + defp native_result({:stop, reason, response, durable}, _previous, from) do + {__MODULE__, adapter_state} = durable.behaviour + adapter_state = Map.put(adapter_state, :stop_reply, {from, response}) + durable = %{durable | behaviour: {__MODULE__, adapter_state}} + {:stop, reason, durable} + end + + defp native_result({:stop, reason, durable}, _previous, _from), + do: {:stop, reason, durable} + + defp native_result({:statem_stop_and_reply, reason, replies, durable}, _previous, _from), + do: {:stop_and_reply, reason, replies, durable} +end diff --git a/lib/durable_server/lifecycle_manager.ex b/lib/durable_server/lifecycle_manager.ex index 5b96b53..9651116 100644 --- a/lib/durable_server/lifecycle_manager.ex +++ b/lib/durable_server/lifecycle_manager.ex @@ -2612,6 +2612,13 @@ defmodule DurableServer.LifecycleManager do defp try_restart_child(%LifecycleManager{} = state, {module, init_arg}, preloaded) do if Code.ensure_loaded?(module) do + # Recovery loads the real state from the preloaded object. Keep the unused + # initial-state placeholder compatible with the child's callback engine. + init_arg = + if DurableServer.__callback_behaviour__(module) == :gen_statem, + do: Keyword.put(init_arg, :initial_state, {nil, nil}), + else: init_arg + # note: we must pass max_placement_retries: 0 to prevent remote placement # because lifecycle manager is only concered with its own local node DurableServer.Supervisor.__start_child__( diff --git a/lib/durable_server/supervisor.ex b/lib/durable_server/supervisor.ex index 2553a92..94e470a 100644 --- a/lib/durable_server/supervisor.ex +++ b/lib/durable_server/supervisor.ex @@ -765,7 +765,8 @@ defmodule DurableServer.Supervisor do automatic restarts, and sticky placement retain their existing behavior. The child spec is `{Module, key: key, initial_state: initial_state}`. - `:initial_state` is required and must be a map. Before the first `init/1` or + `:initial_state` is required: a map for GenServers, or `{state_name, data}` for + modules using `behaviour: :gen_statem`. Before the first `init/1` or `init/2` call, DurableServer passes it through the module's `dump_state/1`, the configured backend's encode/decode path, and then `load_state/2`. This means the dumped initial state must be encodable by the configured backend, @@ -821,7 +822,7 @@ defmodule DurableServer.Supervisor do def start_child(supervisor, child_spec, opts \\ []) def start_child(supervisor, {module, init_arg}, opts) do - init_arg = validate_child_init_arg!(init_arg, "start_child") + init_arg = validate_child_init_arg!(module, init_arg, "start_child") __start_child__(supervisor, {module, init_arg, nil}, opts) end @@ -836,7 +837,7 @@ defmodule DurableServer.Supervisor do def __start_child__(supervisor, {module, init_arg, boot_info}, opts) when is_map(boot_info) or is_nil(boot_info) do - init_arg = validate_child_init_arg!(init_arg, "__start_child__") + init_arg = validate_child_init_arg!(module, init_arg, "__start_child__") do_start_child_with_init_arg(supervisor, {module, init_arg, boot_info}, opts) end @@ -846,7 +847,7 @@ defmodule DurableServer.Supervisor do "__start_child__ expects {Module, init_arg, boot_info}, got: #{inspect(child_spec)}" end - defp validate_child_init_arg!(args, function_name) when is_list(args) do + defp validate_child_init_arg!(module, args, function_name) when is_list(args) do args = Keyword.validate!(args, [:key, :initial_state]) key = @@ -865,12 +866,28 @@ defmodule DurableServer.Supervisor do initial_state = case Keyword.fetch(args, :initial_state) do - {:ok, initial_state} when is_map(initial_state) -> - initial_state - - {:ok, other} -> - raise ArgumentError, - "#{function_name} :initial_state must be a map, got: #{inspect(other)}" + {:ok, initial_state} -> + # Callers may route to workers with callback modules unavailable here. + # The destination validates the state when it receives the start request. + if Code.ensure_loaded?(module) do + case {DurableServer.__callback_behaviour__(module), initial_state} do + {GenServer, %{} = initial_state} -> + initial_state + + {:gen_statem, {_name, _data} = initial_state} -> + initial_state + + {GenServer, other} -> + raise ArgumentError, + "#{function_name} :initial_state must be a map, got: #{inspect(other)}" + + {:gen_statem, other} -> + raise ArgumentError, + "#{function_name} :initial_state must be {state_name, data}, got: #{inspect(other)}" + end + else + initial_state + end :error -> raise ArgumentError, "#{function_name} requires :initial_state" @@ -881,7 +898,7 @@ defmodule DurableServer.Supervisor do |> Keyword.put(:initial_state, initial_state) end - defp validate_child_init_arg!(init_arg, function_name) do + defp validate_child_init_arg!(_module, init_arg, function_name) do raise ArgumentError, "#{function_name} expects {Module, key: \"...\", initial_state: %{...}}, got: #{inspect(init_arg)}" end @@ -2106,7 +2123,8 @@ defmodule DurableServer.Supervisor do ensure a process exists but don't know if it's already running. The child spec is `{Module, key: key, initial_state: initial_state}`. - `:initial_state` is required and must be a map. If a new process is started + `:initial_state` is required: a map for GenServers, or `{state_name, data}` for + modules using `behaviour: :gen_statem`. If a new process is started and no persisted state exists yet, DurableServer passes `:initial_state` through the module's `dump_state/1`, the configured backend's encode/decode path, and then `load_state/2` before `init/1` or `init/2`. This means the @@ -2161,7 +2179,7 @@ defmodule DurableServer.Supervisor do def ensure_started_child(supervisor, child_spec, opts \\ []) def ensure_started_child(supervisor, {module, init_arg}, opts) do - init_arg = validate_child_init_arg!(init_arg, "ensure_started_child") + init_arg = validate_child_init_arg!(module, init_arg, "ensure_started_child") __ensure_started_child__(supervisor, {module, init_arg, nil}, opts) end @@ -2176,7 +2194,7 @@ defmodule DurableServer.Supervisor do def __ensure_started_child__(supervisor, {module, init_arg, boot_info}, opts) when is_map(boot_info) or is_nil(boot_info) do - init_arg = validate_child_init_arg!(init_arg, "__ensure_started_child__") + init_arg = validate_child_init_arg!(module, init_arg, "__ensure_started_child__") child_spec = {module, init_arg, boot_info} opts = @@ -2826,6 +2844,8 @@ defmodule DurableServer.Supervisor do """ def terminate_child(supervisor_name, pid) when is_atom(supervisor_name) and is_pid(pid) do Logger.debug(fn -> "terminating #{inspect(pid)} for #{inspect(supervisor_name)}" end) + + # A PID does not identify its behaviour; GenServer and :gen_statem share the call protocol. GenServer.call(pid, {@durable, {:stop_with_status, :stopped_graceful, :normal}}) end @@ -2965,7 +2985,7 @@ defmodule DurableServer.Supervisor do ) """ def rehome_child(supervisor, {module, init_arg}, opts \\ []) do - init_arg = validate_child_init_arg!(init_arg, "rehome_child") + init_arg = validate_child_init_arg!(module, init_arg, "rehome_child") opts = Keyword.validate!(opts, [:target_node, :force, :shutdown_timeout]) shutdown_timeout = Keyword.get(opts, :shutdown_timeout, 15_000) diff --git a/test/durable_server/state_machine_test.exs b/test/durable_server/state_machine_test.exs new file mode 100644 index 0000000..7913d91 --- /dev/null +++ b/test/durable_server/state_machine_test.exs @@ -0,0 +1,844 @@ +defmodule DurableServer.StateMachineTest do + use ExUnit.Case, async: true + import DurableServer.TestHelper + + @moduletag :capture_log + + defmodule Machine do + use DurableServer, vsn: 1, behaviour: :gen_statem + + def callback_mode, do: [:handle_event_function, :state_enter] + + def dump_state({name, data}) do + %{"name" => Atom.to_string(name), "count" => data.count} + end + + def load_state(_vsn, %{"name" => name, "count" => count}) do + name = + case name do + "idle" -> :idle + "running" -> :running + "done" -> :done + end + + {name, %{count: count}} + end + + def init({name, data}, info) do + data = Map.put(data, :observer, info.observer) + + {:ok, name, data, + [ + {:durable, + [ + meta: %{phase: name}, + auto_sync: info.auto_sync, + permanent: info.permanent, + sync_every_ms: 60_000 + ]} + ]} + end + + def handle_event(:enter, previous, name, data) do + send(data.observer, {:entered, self(), previous, name}) + :keep_state_and_data + end + + def handle_event({:call, from}, :get, name, data), + do: {:keep_state_and_data, [{:reply, from, {name, data.count}}]} + + def handle_event({:call, from}, :start, _name, data) do + {:next_state, :running, %{data | count: data.count + 1}, + [{:reply, from, :ok}, {:durable, [sync: true, meta: %{phase: :running}]}]} + end + + def handle_event({:call, from}, :increment, _name, data), + do: {:keep_state, %{data | count: data.count + 1}, [{:reply, from, :ok}]} + + def handle_event({:call, from}, :throw_transition, _name, data) do + throw( + {:next_state, :running, %{data | count: data.count + 1}, + [{:reply, from, :ok}, {:durable, [sync: true]}]} + ) + end + + def handle_event({:call, from}, :hold_terminate, _name, data), + do: {:keep_state, Map.put(data, :hold_terminate?, true), [{:reply, from, :ok}]} + + def handle_event({:call, from}, {:switch_callback, action}, _name, _data), + do: {:keep_state_and_data, [{:reply, from, :ok}, action]} + + def handle_event({:call, from}, :deferred_reply, _name, _data) do + send(self(), {:reply_later, from}) + :keep_state_and_data + end + + def handle_event(:info, {:reply_later, from}, _name, _data) do + :gen_statem.reply(from, :deferred) + :keep_state_and_data + end + + def handle_event(:cast, :needs_running, :idle, data) do + send(data.observer, {:postponed, self()}) + {:keep_state_and_data, [:postpone]} + end + + def handle_event(:cast, :needs_running, :running, data) do + send(data.observer, {:processed, self()}) + :keep_state_and_data + end + + def handle_event(:cast, :increment, _name, data), + do: {:keep_state, %{data | count: data.count + 1}} + + def handle_event({:call, from}, :repeat, _name, _data), + do: {:repeat_state_and_data, [{:reply, from, :ok}]} + + def handle_event({:call, from}, :arm_timeouts, _name, _data) do + {:keep_state_and_data, + [{:state_timeout, 50, :expired}, {{:timeout, :probe}, 100, :expired}, {:reply, from, :ok}]} + end + + def handle_event(type, :expired, name, data) do + send(data.observer, {:expired, self(), type, name}) + :keep_state_and_data + end + + def handle_event({:call, from}, :finish, _name, data) do + {:next_state, :done, %{data | count: data.count + 1}, + [{:durable, [sync: true]}, {:next_event, :internal, {:committed, from}}]} + end + + def handle_event(:internal, {:committed, from}, name, data) do + send(data.observer, {:committed_event, self(), name}) + {:keep_state_and_data, [{:reply, from, {name, data.count}}]} + end + + def handle_event({:call, from}, {:stop, reason}, _name, data), + do: {:stop_and_reply, reason, [{:reply, from, :bye}], data} + + def handle_sync(_info, old, {name, data} = current) do + if observer = Map.get(data, :observer), + do: send(observer, {:synced, self(), old, current}) + + {:ok, {name, data}} + end + + def code_change(0, _name, data, 1), + do: {:ok, :running, %{data | count: data.count + 100}} + + def code_change(_old, name, data, :upgrade), + do: {:ok, name, %{data | count: data.count + 10}} + + def code_change(_old, _name, _data, :reject), do: {:error, :upgrade_rejected} + + def code_change(_old, name, data, _extra), do: {:ok, name, data} + + def terminate(reason, name, data) do + if Map.get(data, :hold_terminate?, false) do + send(data.observer, {:terminate_waiting, self()}) + + receive do + :finish_terminate -> :ok + end + end + + send(data.observer, {:terminated, self(), reason, name, data.count}) + {:finished, data.observer} + end + + def after_terminate({:finished, observer}, info) do + send(observer, {:after_terminate, self(), :finished, info.final_status}) + :ok + end + end + + defmodule FaultBackend do + @behaviour DurableServer.StorageBackend + alias DurableServer.Backends.ObjectStore, as: Backend + + def init_backend(opts) do + {:ok, config} = Backend.init_backend(Keyword.fetch!(opts, :store)) + state = %{delegate: config.state, table: Keyword.fetch!(opts, :table)} + {:ok, %{config | state: state}} + end + + for {function, arity} <- [ + ensure_ready: 1, + get_object: 3, + list_all_objects_stream: 3, + delete_object: 2, + delete_object: 3, + update_object: 4, + encode: 2, + decode: 2 + ] do + args = Macro.generate_arguments(arity - 1, __MODULE__) + + def unquote(function)(%{delegate: delegate}, unquote_splicing(args)), + do: Backend.unquote(function)(delegate, unquote_splicing(args)) + end + + def try_claim(%{delegate: delegate, table: table}, key, body) do + case :ets.lookup(table, {:block_claim, key}) do + [{{:block_claim, ^key}, observer}] -> + send(observer, {:claim_blocked, self()}) + + receive do + :continue_claim -> :ok + end + + [] -> + :ok + end + + Backend.try_claim(delegate, key, body) + end + + def put_object(%{delegate: delegate, table: table}, key, body, opts) do + case :ets.lookup(table, :fail_key) do + [{:fail_key, ^key}] -> {:error, :unavailable} + _ -> Backend.put_object(delegate, key, body, opts) + end + end + end + + defmodule StateFunctions do + use DurableServer, vsn: 1, behaviour: :gen_statem + + def callback_mode, do: :state_functions + def dump_state({name, count}), do: %{"name" => Atom.to_string(name), "count" => count} + def load_state(_, %{"name" => "idle", "count" => count}), do: {:idle, count} + def load_state(_, %{"name" => "running", "count" => count}), do: {:running, count} + + def idle({:call, from}, :start, count), + do: {:next_state, :running, count + 1, [{:reply, from, :ok}, {:durable, [sync: true]}]} + + def running({:call, from}, :get, count), + do: {:keep_state_and_data, [{:reply, from, count}]} + end + + defmodule GenServerCallbacks do + use DurableServer, vsn: 1, behaviour: GenServer + + def dump_state(state), do: state + def load_state(_vsn, state), do: state + def handle_call(:get, _from, state), do: {:reply, state, state} + + def code_change(_old, _state, :reject), do: {:error, :upgrade_rejected} + def code_change(_old, state, _extra), do: {:ok, state} + end + + defmodule AliasedGenServerCallbacks do + alias GenServer, as: Server + use DurableServer, vsn: 1, behaviour: Server + + defdelegate dump_state(state), to: GenServerCallbacks + defdelegate load_state(vsn, state), to: GenServerCallbacks + defdelegate handle_call(request, from, state), to: GenServerCallbacks + end + + defmodule InitResult do + use DurableServer, vsn: 1, behaviour: :gen_statem + + def callback_mode, do: :handle_event_function + + defdelegate dump_state(state), to: StateFunctions + defdelegate load_state(vsn, state), to: StateFunctions + + def init({:idle, 0}), do: :ignore + def init({:idle, 1}), do: {:stop, :not_ready} + def init({:idle, 2}), do: {:error, :not_ready} + def init({:idle, 3}), do: raise("init failed") + + def handle_event(_, _, _, _), do: :keep_state_and_data + end + + defmodule InitialEntry do + use DurableServer, vsn: 1, behaviour: :gen_statem + + def callback_mode, do: [:handle_event_function, :state_enter] + + defdelegate dump_state(state), to: Machine + defdelegate load_state(vsn, state), to: Machine + defdelegate terminate(reason, name, data), to: Machine + defdelegate after_terminate(result, info), to: Machine + + def init({name, data}, info) do + data = Map.merge(data, %{observer: info.observer, actions: info.init_actions}) + __MODULE__.init({name, data}) + end + + def init({name, data}), do: {:ok, name, data, Map.get(data, :actions, [])} + + def handle_event(:enter, :idle, :idle, data) do + send(data.observer, {:initial_entry, self(), data}) + + receive do + :raise -> raise "initial entry failed" + {:return, result} -> result + end + end + + def handle_event(:enter, :idle, :running, data) do + send(data.observer, {:entered, self(), :idle, :running}) + :keep_state_and_data + end + + def handle_event(:internal, :advance, :idle, data), + do: {:next_state, :running, data} + + def handle_event({:call, from}, :get, name, data), + do: {:keep_state_and_data, [{:reply, from, {name, data.count}}]} + end + + setup context do + id = System.unique_integer([:positive]) + supervisor = :"statem_sup_#{id}" + prefix = "statem_#{id}/" + + table_owner = start_supervised!({Agent, fn -> :ets.new(:statem_faults, [:public]) end}) + table = Agent.get(table_owner, & &1) + + start_supervised!( + {DurableServer.Supervisor, + name: supervisor, + prefix: prefix, + backend: {FaultBackend, store: test_object_store_opts(), table: table}, + init_info: %{ + observer: self(), + auto_sync: context[:auto_sync] || false, + permanent: context[:permanent] || false, + init_actions: context[:init_actions] || [] + }, + initial_discovery_delay_ms: {60_000, 60_000}, + graceful_shutdown_timeout_ms: 1_000} + ) + + %{supervisor: supervisor, prefix: prefix, table: table} + end + + defp start_machine(context, module \\ Machine, initial \\ {:idle, %{count: 0}}) do + key = "machine-#{System.unique_integer([:positive])}" + + assert {:ok, {pid, _meta}} = + DurableServer.Supervisor.start_child( + context.supervisor, + {module, key: key, initial_state: initial} + ) + + {pid, key} + end + + defp stored!(context, key) do + assert {:ok, stored} = + DurableServer.fetch_stored_state( + context.supervisor, + %{key: key, prefix: context.prefix}, + consistent: true + ) + + stored + end + + test "validates initial state against the callback behaviour on the destination node", + context do + for {module, initial_state, expected} <- [ + {GenServerCallbacks, {:idle, %{}}, "must be a map"}, + {Machine, %{}, "must be {state_name, data}"} + ] do + error = + assert_raise ArgumentError, fn -> + DurableServer.Supervisor.__start_child__( + context.supervisor, + {module, [key: "invalid-state", initial_state: initial_state], nil}, + max_placement_retries: 0 + ) + end + + assert error.message =~ expected + end + end + + test "retains compatibility with GenServer call and cast clients", + context do + {pid, _key} = start_machine(context) + assert GenServer.call(pid, :get) == {:idle, 0} + + assert {:idle, + %DurableServer{ + behaviour: + {DurableServer.Behaviours.StateMachine, + %{mode: :handle_event_function, state_enter?: true}}, + user_state: {:idle, %{count: 0}} + }} = :sys.get_state(pid) + + assert_receive {:entered, ^pid, :idle, :idle} + :gen_statem.cast(pid, :increment) + GenServer.cast(pid, :increment) + assert :gen_statem.call(pid, :get) == {:idle, 2} + assert GenServer.call(pid, :deferred_reply) == :deferred + assert :gen_statem.call(pid, :deferred_reply) == :deferred + end + + test "blocked lock acquisition does not block the supervisor from starting other children", + context do + key = "blocked-#{System.unique_integer([:positive])}" + :ets.insert(context.table, {{:block_claim, context.prefix <> key}, self()}) + + blocked = + Task.async(fn -> + DurableServer.Supervisor.start_child( + context.supervisor, + {Machine, key: key, initial_state: {:idle, %{count: 0}}} + ) + end) + + assert_receive {:claim_blocked, pid}, 1_000 + + try do + fast = Task.async(fn -> start_machine(context) end) + assert {fast_pid, _key} = Task.await(fast, 2_000) + assert :gen_statem.call(fast_pid, :get) == {:idle, 0} + assert Task.yield(blocked, 0) == nil + after + send(pid, :continue_claim) + end + + assert {:ok, {^pid, _meta}} = Task.await(blocked, 2_000) + assert :gen_statem.call(pid, :get) == {:idle, 0} + end + + test "native parent shutdown terminates the state machine and persists its state", context do + {pid, key} = start_machine(context) + assert :gen_statem.call(pid, :increment) == :ok + assert :proc_lib.translate_initial_call(pid) == {Machine, :init, 1} + ref = Process.monitor(pid) + supervisor = DurableServer.Supervisor.get_dynamic_supervisor(context.supervisor) + + assert :ok = DynamicSupervisor.terminate_child(supervisor, pid) + assert_receive {:terminated, ^pid, :shutdown, :idle, 1} + assert_receive {:DOWN, ^ref, :process, ^pid, :shutdown} + assert stored!(context, key).meta.status == :stopped_graceful + assert stored!(context, key).state == %{"name" => "idle", "count" => 1} + end + + test "syncs the state name and data before replies and updates registry metadata", context do + {pid, key} = start_machine(context) + assert :gen_statem.call(pid, :start) == :ok + assert stored!(context, key).state == %{"name" => "running", "count" => 1} + assert {^pid, %{phase: :running}} = DurableServer.Supervisor.lookup(context.supervisor, key) + assert_receive {:synced, ^pid, {:idle, %{count: 0}}, {:running, %{count: 1}}} + assert_receive {:entered, ^pid, :idle, :running} + assert :gen_statem.call(pid, :finish) == {:done, 2} + assert stored!(context, key).state == %{"name" => "done", "count" => 2} + end + + test "postpones events until a state change, not a data update or durable sync", context do + {pid, _key} = start_machine(context) + :gen_statem.cast(pid, :needs_running) + assert_receive {:postponed, ^pid} + assert :gen_statem.call(pid, :increment) == :ok + send(pid, {:durable, :sync}) + assert :gen_statem.call(pid, :get) == {:idle, 1} + refute_receive {:postponed, ^pid}, 20 + refute_receive {:processed, ^pid}, 20 + assert :gen_statem.call(pid, :start) == :ok + assert_receive {:processed, ^pid} + end + + test "native state timeouts cancel on transitions while named timeouts survive", context do + {pid, _key} = start_machine(context) + assert :gen_statem.call(pid, :arm_timeouts) == :ok + assert :gen_statem.call(pid, :start) == :ok + assert_receive {:expired, ^pid, {:timeout, :probe}, :running}, 1_000 + refute_receive {:expired, ^pid, :state_timeout, _}, 20 + assert :gen_statem.call(pid, :arm_timeouts) == :ok + assert_receive {:expired, ^pid, :state_timeout, :running}, 1_000 + end + + test "repeat_state invokes state entry without changing the state name", context do + {pid, _key} = start_machine(context) + assert_receive {:entered, ^pid, :idle, :idle} + assert :gen_statem.call(pid, :repeat) == :ok + assert_receive {:entered, ^pid, :idle, :idle} + end + + @tag auto_sync: true + test "auto sync also applies to callbacks returning native actions", context do + {pid, key} = start_machine(context) + assert :gen_statem.call(pid, :increment) == :ok + assert stored!(context, key).state == %{"name" => "idle", "count" => 1} + end + + test "recovers persisted state through the existing supervisor API", context do + {pid, key} = start_machine(context) + assert :gen_statem.call(pid, :start) == :ok + assert :ok = DurableServer.Supervisor.terminate_child(context.supervisor, pid) + assert_receive {:terminated, ^pid, :normal, :running, 1} + assert stored!(context, key).meta.status == :stopped_graceful + + assert {:ok, {restarted, _}} = + DurableServer.Supervisor.ensure_started_child( + context.supervisor, + {Machine, key: key, initial_state: {:idle, %{count: -1}}} + ) + + assert :gen_statem.call(restarted, :get) == {:running, 1} + end + + test "supports state_functions and non-map native data", context do + {pid, key} = start_machine(context, StateFunctions, {:idle, 0}) + assert :gen_statem.call(pid, :start) == :ok + assert :gen_statem.call(pid, :get) == 1 + + assert {:running, + %DurableServer{ + behaviour: + {DurableServer.Behaviours.StateMachine, + %{mode: :state_functions, state_enter?: false}} + }} = :sys.get_state(pid) + + assert stored!(context, key).state == %{"name" => "running", "count" => 1} + end + + test "accepts GenServer and aliases as the behaviour option", context do + for module <- [GenServerCallbacks, AliasedGenServerCallbacks] do + {pid, _key} = start_machine(context, module, %{"count" => 0}) + assert DurableServer.__callback_behaviour__(module) == GenServer + assert GenServer.call(pid, :get) == %{"count" => 0} + + assert %DurableServer{behaviour: {DurableServer.Behaviours.GenServer, %{}}} = + :sys.get_state(pid) + end + end + + test "GenServer processes without a behaviour field still dispatch and upgrade", context do + {pid, _key} = start_machine(context, GenServerCallbacks, %{"count" => 0}) + :sys.replace_state(pid, &Map.delete(&1, :behaviour)) + assert GenServer.call(pid, :get) == %{"count" => 0} + :ok = :sys.suspend(pid) + + try do + assert :ok = :sys.change_code(pid, GenServerCallbacks, 1, :upgrade) + + assert %DurableServer{behaviour: {DurableServer.Behaviours.GenServer, %{}}} = + :sys.get_state(pid) + after + :ok = :sys.resume(pid) + end + + assert GenServer.call(pid, :get) == %{"count" => 0} + end + + test "native init refusals release the storage claim and return to the starter", context do + for {count, expected} <- [{0, :ignore}, {1, {:error, :not_ready}}, {2, {:error, :not_ready}}] do + key = "init-result-#{System.unique_integer([:positive])}" + + assert DurableServer.Supervisor.start_child( + context.supervisor, + {InitResult, key: key, initial_state: {:idle, count}} + ) == expected + + assert stored!(context, key).meta.status == :stopped_graceful + assert DurableServer.Supervisor.lookup(context.supervisor, key) == nil + end + end + + test "an exception during init returns the failure to the starter", context do + key = "init-crash-#{System.unique_integer([:positive])}" + + assert {:error, {%RuntimeError{message: "init failed"}, _stacktrace}} = + DurableServer.Supervisor.start_child( + context.supervisor, + {InitResult, key: key, initial_state: {:idle, 3}} + ) + end + + test "initial entry failures invoke termination and persist crash history", context do + {pid, key} = start_machine(context, InitialEntry) + assert_receive {:initial_entry, ^pid, %{count: 0}} + ref = Process.monitor(pid) + send(pid, :raise) + + assert_receive {:terminated, ^pid, %RuntimeError{message: "initial entry failed"}, :idle, 0} + + assert_receive {:DOWN, ^ref, :process, ^pid, {%RuntimeError{}, _}} + stored = stored!(context, key) + assert stored.meta.status == :crashed + assert [%{reason: reason, timestamp: timestamp}] = stored.meta.crash_history + assert reason =~ "initial entry failed" + assert is_integer(timestamp) + end + + for kind <- [ + :repeat_state, + :repeat_state_with_actions, + :repeat_state_and_data, + :repeat_state_and_data_with_actions + ] do + @tag repeat_kind: kind, init_actions: [{:next_event, :internal, :advance}] + test "initial entry preserves the previous state through #{kind}", context do + {pid, _key} = start_machine(context, InitialEntry) + + for _ <- 1..2 do + assert_receive {:initial_entry, ^pid, data} + + result = + case context.repeat_kind do + :repeat_state -> {:repeat_state, %{data | count: data.count + 1}} + :repeat_state_with_actions -> {:repeat_state, %{data | count: data.count + 1}, []} + :repeat_state_and_data -> :repeat_state_and_data + :repeat_state_and_data_with_actions -> {:repeat_state_and_data, []} + end + + send(pid, {:return, result}) + end + + assert_receive {:initial_entry, ^pid, data} + send(pid, {:return, :keep_state_and_data}) + assert_receive {:entered, ^pid, :idle, :running} + assert :gen_statem.call(pid, :get) == {:running, data.count} + end + end + + for action <- [:postpone, {:postpone, true}, {:postpone, false}] do + @tag init_actions: [action, {:next_event, :internal, :advance}] + test "init ignores #{inspect(action)} without replaying the bootstrap event", context do + {pid, _key} = start_machine(context, InitialEntry) + + assert_receive {:initial_entry, ^pid, _data} + send(pid, {:return, :keep_state_and_data}) + assert_receive {:entered, ^pid, :idle, :running} + assert :gen_statem.call(pid, :get) == {:running, 0} + end + end + + @tag permanent: true + test "the lifecycle manager restarts a native state machine from its persisted state", + context do + {pid, key} = start_machine(context) + assert :gen_statem.call(pid, :start) == :ok + ref = Process.monitor(pid) + Process.exit(pid, :kill) + assert_receive {:DOWN, ^ref, :process, ^pid, :killed} + + # Make the orphan eligible immediately without waiting for a real dead + # node's heartbeat timeout. Recovery still takes the normal claim path. + stored = stored!(context, key) + + meta = %{ + stored.meta + | status: :crashed, + node_str: "statem-dead@host", + node_ref: "dead", + last_heartbeat_at: System.system_time(:millisecond) - 600_000 + } + + %{storage_backend: backend} = DurableServer.Supervisor.__get_config__(context.supervisor) + + assert {:ok, _} = + DurableServer.StorageBackend.put_object(backend, context.prefix <> key, %{ + stored + | meta: meta + }) + + send(DurableServer.LifecycleManager.name(context.supervisor), :discover_and_restart) + assert_receive {:entered, restarted, :running, :running} when restarted != pid, 5_000 + assert :gen_statem.call(restarted, :get) == {:running, 1} + assert {:running, %DurableServer{}} = :sys.get_state(restarted) + assert stored!(context, key).meta.status == :running + end + + test "hot upgrades invoke code_change/4 and retain durable state and metadata", context do + {pid, key} = start_machine(context) + :ok = :sys.suspend(pid) + + try do + assert :ok = :sys.change_code(pid, Machine, 1, :upgrade) + + assert {:idle, + %DurableServer{ + behaviour: + {DurableServer.Behaviours.StateMachine, + %{mode: :handle_event_function, state_enter?: true}}, + user_state: {:idle, %{count: 10}}, + vsn: 1 + }} = + :sys.get_state(pid) + after + :ok = :sys.resume(pid) + end + + send(pid, {:durable, :sync}) + assert :gen_statem.call(pid, :get) == {:idle, 10} + assert stored!(context, key).state == %{"name" => "idle", "count" => 10} + end + + test "schema-version migrations update the native state name before dispatch", context do + {pid, key} = start_machine(context) + :sys.replace_state(pid, fn {name, durable} -> {name, %{durable | vsn: 0}} end) + assert :gen_statem.call(pid, :get) == {:running, 100} + assert {:running, %{vsn: 1, old_vsn: 0}} = :sys.get_state(pid) + send(pid, {:durable, :sync}) + assert :gen_statem.call(pid, :get) == {:running, 100} + assert stored!(context, key).state == %{"name" => "running", "count" => 100} + end + + test "rejected hot upgrades preserve state for both behaviours", context do + for {module, api, initial_state, reply} <- [ + {GenServerCallbacks, GenServer, %{"count" => 3}, %{"count" => 3}}, + {Machine, :gen_statem, {:idle, %{count: 3}}, {:idle, 3}} + ] do + {pid, _key} = start_machine(context, module, initial_state) + assert api.call(pid, :get) == reply + :ok = :sys.suspend(pid) + + try do + before = :sys.get_state(pid) + + assert {:error, {:error, :upgrade_rejected}} = + :sys.change_code(pid, module, 1, :reject) + + assert :sys.get_state(pid) == before + after + :ok = :sys.resume(pid) + end + + assert api.call(pid, :get) == reply + end + end + + test "native thrown callback returns still pass through durability", context do + {pid, key} = start_machine(context) + assert :gen_statem.call(pid, :throw_transition) == :ok + assert stored!(context, key).state == %{"name" => "running", "count" => 1} + assert {:running, %DurableServer{}} = :sys.get_state(pid) + end + + test "an explicit sync failure cannot acknowledge or execute inserted events", context do + {pid, key} = start_machine(context) + :ets.insert(context.table, {:fail_key, context.prefix <> key}) + ref = Process.monitor(pid) + assert catch_exit(:gen_statem.call(pid, :finish)) + + assert_receive {:DOWN, ^ref, :process, ^pid, + {:shutdown, {:durable, {:fatal_exit, {:sync_failed, :unavailable}}}}}, + 1_000 + + refute_receive {:committed_event, ^pid, _}, 20 + assert stored!(context, key).state == %{"name" => "idle", "count" => 0} + end + + test "supervisor stop waits for user termination and final persistence", context do + {pid, key} = start_machine(context) + assert :gen_statem.call(pid, :increment) == :ok + assert :gen_statem.call(pid, :hold_terminate) == :ok + task = Task.async(fn -> DurableServer.Supervisor.terminate_child(context.supervisor, pid) end) + assert_receive {:terminate_waiting, ^pid} + assert Task.yield(task, 20) == nil + assert stored!(context, key).meta.status == :running + send(pid, :finish_terminate) + assert Task.await(task) == :ok + assert_receive {:after_terminate, ^pid, :finished, :stopped_graceful} + assert stored!(context, key).meta.status == :stopped_graceful + assert stored!(context, key).state == %{"name" => "idle", "count" => 1} + end + + test "native stop_and_reply replies before termination while still persisting final state", + context do + {pid, key} = start_machine(context) + assert :gen_statem.call(pid, :increment) == :ok + assert :gen_statem.call(pid, :hold_terminate) == :ok + ref = Process.monitor(pid) + + assert :gen_statem.call(pid, {:stop, :normal}) == :bye + assert_receive {:terminate_waiting, ^pid} + assert stored!(context, key).meta.status == :running + send(pid, :finish_terminate) + assert_receive {:DOWN, ^ref, :process, ^pid, :normal} + assert_receive {:after_terminate, ^pid, :finished, :stopped_graceful} + assert stored!(context, key).state == %{"name" => "idle", "count" => 1} + end + + test "supervisor cordon persists final state and blocks a subsequent start", context do + {pid, key} = start_machine(context) + assert :gen_statem.call(pid, :increment) == :ok + + assert :ok = DurableServer.Supervisor.terminate_and_cordon_child(context.supervisor, pid) + assert stored!(context, key).meta.status == :cordoned + assert stored!(context, key).state == %{"name" => "idle", "count" => 1} + + assert {:error, :cordoned} = + DurableServer.Supervisor.start_child( + context.supervisor, + {Machine, key: key, initial_state: {:idle, %{count: 0}}} + ) + end + + test "OTP shutdown drains a native child and persists its final state", context do + {pid, key} = start_machine(context) + assert :gen_statem.call(pid, :increment) == :ok + ref = Process.monitor(pid) + %{storage_backend: backend} = DurableServer.Supervisor.__get_config__(context.supervisor) + + assert :ok = stop_supervised(context.supervisor) + assert_receive {:DOWN, ^ref, :process, ^pid, :normal} + assert_receive {:terminated, ^pid, :normal, :idle, 1} + assert_receive {:after_terminate, ^pid, :finished, :stopped_graceful} + + assert {:ok, stored} = + DurableServer.fetch_stored_state( + backend, + %{key: key, prefix: context.prefix, supervisor: context.supervisor}, + consistent: true + ) + + assert stored.meta.status == :stopped_graceful + assert stored.state == %{"name" => "idle", "count" => 1} + end + + test "callback-module switching cannot bypass persistence", context do + for action <- [ + {:change_callback_module, StateFunctions}, + {:push_callback_module, StateFunctions}, + :pop_callback_module + ] do + {pid, _key} = start_machine(context) + assert catch_exit(:gen_statem.call(pid, {:switch_callback, action})) + end + end + + test "switching the process engine requires a restart" do + old_genserver = %DurableServer{module: Machine, user_state: %{count: 0}} + + assert {:error, :behaviour_change_requires_restart} = + DurableServer.code_change(1, old_genserver, []) + + old_statem = %DurableServer{ + module: GenServerCallbacks, + user_state: {:idle, %{count: 0}}, + behaviour: + {DurableServer.Behaviours.StateMachine, + %{mode: :handle_event_function, state_enter?: true}} + } + + assert {:error, :behaviour_change_requires_restart} = + DurableServer.code_change(1, old_statem, []) + end + + test "stop_and_reply retains durable deletion semantics", context do + {pid, key} = start_machine(context) + ref = Process.monitor(pid) + assert :gen_statem.call(pid, {:stop, :delete}) == :bye + assert_receive {:DOWN, ^ref, :process, ^pid, :normal}, 2_000 + assert_receive {:terminated, ^pid, :normal, :idle, 0} + + assert {:error, :not_found} = + DurableServer.fetch_stored_state( + context.supervisor, + %{key: key, prefix: context.prefix}, + consistent: true + ) + end +end diff --git a/test/durable_server/supervisor_backend_spec_test.exs b/test/durable_server/supervisor_backend_spec_test.exs index 435406a..d9e7ee6 100644 --- a/test/durable_server/supervisor_backend_spec_test.exs +++ b/test/durable_server/supervisor_backend_spec_test.exs @@ -7,6 +7,7 @@ defmodule DurableServer.SupervisorBackendSpecTest do alias DurableServer.Backends.EKVStore alias DurableServer.Backends.MirrorStore alias DurableServer.StorageBackend + alias __MODULE__.RemoteOnlyServer def throw_not_ready do throw({:error, :not_ready}) @@ -204,6 +205,96 @@ defmodule DurableServer.SupervisorBackendSpecTest do assert object_store == nil end + describe "callback modules available only on worker nodes" do + setup do + supervisor = unique_supervisor_name("remote_module") + + start_supervised!( + {DurableServer.Supervisor, + name: supervisor, + prefix: unique_prefix("remote_module"), + backend: {InMemoryBackend, []}, + max_children: %{RemoteOnlyServer => 0}, + graceful_shutdown_timeout_ms: 500} + ) + + refute Code.ensure_loaded?(RemoteOnlyServer) + %{supervisor: supervisor} + end + + for {behaviour, initial_state} <- [ + {GenServer, %{count: 0}}, + {:gen_statem, {:idle, %{count: 0}}} + ] do + test "returns an existing owner for an unavailable #{behaviour} module", %{ + supervisor: supervisor + } do + key = "existing-owner" + pid = self() + meta = %{count: 1} + :ok = Group.register(supervisor, key, %{user_meta: meta}) + args = [key: key, initial_state: unquote(Macro.escape(initial_state))] + + try do + assert {:ok, {^pid, ^meta}} = + DurableServer.Supervisor.ensure_started_child( + supervisor, + {RemoteOnlyServer, args} + ) + + assert {:ok, {^pid, ^meta}} = + DurableServer.Supervisor.__ensure_started_child__( + supervisor, + {RemoteOnlyServer, args, nil} + ) + + assert {:error, {:already_started, {^pid, ^meta}}} = + DurableServer.Supervisor.start_child(supervisor, {RemoteOnlyServer, args}) + + assert {:error, {:already_started, {^pid, ^meta}}} = + DurableServer.Supervisor.__start_child__( + supervisor, + {RemoteOnlyServer, args, nil} + ) + after + Group.unregister(supervisor, key) + end + end + + for start <- [:start_child, :ensure_started_child] do + test "#{start} attempts remote placement for an unavailable #{behaviour} module", %{ + supervisor: supervisor + } do + remote = :remote_only_worker@test + heartbeat_table = :"durable_server_heartbeats_#{supervisor}" + + :ets.insert( + heartbeat_table, + {to_string(remote), 1, System.system_time(:millisecond), + %{RemoteOnlyServer => %{current: 0, limit: 1}}, nil, %{}, %{}} + ) + + log = + capture_log(fn -> + assert {:error, {:capacity_limit, :all_placement_attempts_failed}} = + apply(DurableServer.Supervisor, unquote(start), [ + supervisor, + {RemoteOnlyServer, + key: "remote-placement", + initial_state: unquote(Macro.escape(initial_state))}, + [placement_timeout: 0, timeout: 1_000] + ]) + end) + + # The worker is deliberately unreachable; reaching ERPC proves that + # local callback inspection did not prevent routing to it. + assert log =~ "Attempting to place #{inspect(RemoteOnlyServer)} on remote node" + assert log =~ to_string(remote) + end + end + end + end + test "caps placement ERPC timeout by the caller deadline" do supervisor_name = unique_supervisor_name("placement_deadline") prefix = unique_prefix("placement_deadline")