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