defmodule Revenant do @moduledoc """ Durable GenServers backed by Postgres. A reply is a commit receipt. Revenant gives you process-per-entity GenServers whose state survives crashes, restarts, and deploys, using only the Postgres you already run. The callback module looks exactly like a GenServer: defmodule Account do use Revenant, repo: MyApp.Repo def initial_state(_id), do: %{balance: 0} def handle_call({:deposit, amount}, _from, state) do {:reply, :ok, %{state | balance: state.balance + amount}} end def handle_call(:balance, _from, state) do {:reply, state.balance, state} end end Revenant.call({Account, "acct_42"}, {:deposit, 100}) #=> :ok (the new state is committed to Postgres before you see this) Processes start lazily on the first message and are addressed by `{module, entity_id}` - a stable identity, not a pid. When a process is down, the next call revives it from its committed state. After `:idle_timeout` milliseconds without a message (default 5 minutes) a process passivates: it flushes any pending state and stops, and the next message revives it. Set `idle_timeout: :infinity` to keep processes resident. The `:timeout` info message is reserved for passivation on any server with a finite idle timeout. ## Topology The registry is node-local. Run Revenant on a single node, or route each entity's messages to one node yourself (consistent hashing, a fronting queue). If two nodes load the same entity, the version column keeps the data safe - the stale process exits with `{:revenant_conflict, key}` instead of overwriting - but under relaxed durability the losing node has already acked writes it can no longer commit. If you cannot guarantee routing, use `:strict`. ## Durability levels Set per server via the `:durability` option, escalate per call: * `:strict` (default) - every state change is committed before the caller gets its reply. Anything a caller ever observed is durable. * `{:interval, milliseconds}` - state changes mark the server dirty and the latest snapshot is flushed at most once per interval, plus always on graceful shutdown. Replies precede the commit, so a hard crash can lose up to one interval of acknowledged writes. * `:on_stop` - state is flushed only on graceful shutdown. The loss window on a hard crash is the whole session. Any single call on a relaxed server can demand strictness: Revenant.call({Account, id}, {:charge, amount}, durability: :strict) This flushes all pending state (it is one snapshot - flushing everything is the same write as flushing anything) and only then replies. ## Guarantees and non-guarantees Writes are guarded by a version column, so a stale process can never overwrite a newer commit - it exits with `{:revenant_conflict, key}` instead. A handler crash before commit rolls state back to the last commit; a poison message cannot persist a state no caller was acked on. State must be serializable: no pids, references, ports, or anonymous functions (captures of named functions are fine). Set `config :revenant, validate_state: true` in dev and test to deep-check every mutation. Side effects are not journaled - a handler that sends an email and then crashes before commit will send it again on retry. ## Setup Add the states table in a migration: defmodule MyApp.Repo.Migrations.AddRevenant do use Revenant.Migration end Start the supervisor after your repo: children = [MyApp.Repo, Revenant.Supervisor] ## State evolution Snapshots carry a schema version, set with the `:vsn` option (default 1). Bump it when the shape of your state changes and implement `upgrade/2` to migrate old snapshots as they load: use Revenant, repo: MyApp.Repo, vsn: 2 def upgrade(1, state), do: Map.put(state, :currency, :usd) ## Telemetry Revenant emits: * `[:revenant, :flush]` - a snapshot was committed; measurements `%{bytes: n}`, metadata `%{module, id, version}` * `[:revenant, :conflict]` - a stale process lost a version race and is exiting; metadata `%{module, id, version}` * `[:revenant, :load]` - a process revived from a committed snapshot (not first-ever starts); metadata `%{module, id, version}` """ import Ecto.Query alias Revenant.State @callback initial_state(id :: String.t()) :: term() @callback handle_call(message :: term(), from :: GenServer.from(), state :: term()) :: {:reply, term(), term()} | {:stop, term(), term(), term()} @callback handle_cast(message :: term(), state :: term()) :: {:noreply, term()} | {:stop, term(), term()} @callback handle_info(message :: term(), state :: term()) :: {:noreply, term()} | {:stop, term(), term()} @callback upgrade(old_vsn :: pos_integer(), state :: term()) :: term() @callback terminate(reason :: term(), state :: term()) :: term() @optional_callbacks handle_cast: 2, handle_info: 2, upgrade: 2, terminate: 2 defmacro __using__(opts) do quote bind_quoted: [opts: opts] do @behaviour Revenant @revenant_config Revenant.compile_config!(__MODULE__, opts) def __revenant__, do: @revenant_config def child_spec(id) do %{ id: {__MODULE__, id}, start: {Revenant.Server, :start_link, [__MODULE__, id]}, restart: :temporary } end def handle_cast(message, _state) do exit({:bad_cast, message}) end def handle_info(_message, state), do: {:noreply, state} def upgrade(old_vsn, _state) do raise "#{inspect(__MODULE__)} loaded a vsn #{old_vsn} snapshot but is at " <> "vsn #{@revenant_config.vsn} and does not implement upgrade/2" end def terminate(_reason, _state), do: :ok defoverridable child_spec: 1, handle_cast: 2, handle_info: 2, upgrade: 2, terminate: 2 end end @doc false def compile_config!(module, opts) do repo = Keyword.get(opts, :repo) || raise ArgumentError, "use Revenant requires a :repo option in #{inspect(module)}" vsn = Keyword.get(opts, :vsn, 1) unless is_integer(vsn) and vsn > 0 do raise ArgumentError, ":vsn must be a positive integer, got: #{inspect(vsn)}" end durability = Keyword.get(opts, :durability, :strict) case durability do :strict -> :ok :on_stop -> :ok {:interval, milliseconds} when is_integer(milliseconds) and milliseconds > 0 -> :ok other -> raise ArgumentError, ":durability must be :strict, {:interval, milliseconds}, or :on_stop, " <> "got: #{inspect(other)}" end idle_timeout = Keyword.get(opts, :idle_timeout, :timer.minutes(5)) unless idle_timeout == :infinity or (is_integer(idle_timeout) and idle_timeout > 0) do raise ArgumentError, ":idle_timeout must be a positive integer in milliseconds or :infinity, " <> "got: #{inspect(idle_timeout)}" end %{repo: repo, vsn: vsn, durability: durability, idle_timeout: idle_timeout} end @doc """ Calls the server identified by `{module, id}`, starting it if necessary. Options: * `:timeout` - the call timeout (default 5000) * `:durability` - pass `:strict` to commit before this reply even on a server running a relaxed durability level """ def call(server, message, opts \\ []) do server = normalize(server) timeout = Keyword.get(opts, :timeout, 5000) do_call(server, message, opts, timeout, 3) end defp do_call(server, message, opts, timeout, attempts_left) do pid = whereis(server) || start(server) try do GenServer.call(pid, {:"$revenant", message, opts}, timeout) catch # :noproc - the pid died between lookup and call; :normal - the server # passivated after the call was sent but before processing it. Both # mean the message was never handled, so a retry cannot double-apply. :exit, {reason, _} when reason in [:noproc, :normal] and attempts_left > 1 -> do_call(server, message, opts, timeout, attempts_left - 1) end end @doc """ Casts to the server identified by `{module, id}`, starting it if necessary. Fire-and-forget, exactly like `GenServer.cast/2`: a crash before the next commit loses the message. Use `call/3` when you need the durability receipt. """ def cast(server, message) do server = normalize(server) pid = whereis(server) || start(server) GenServer.cast(pid, {:"$revenant", message}) end @doc """ Returns the pid of a live server, or `nil`. A `nil` does not mean the entity has no state - only that no process is currently loaded for it. """ def whereis(server) do case Registry.lookup(Revenant.Registry, normalize(server)) do [{pid, _}] -> if Process.alive?(pid), do: pid, else: nil [] -> nil end end @doc """ Gracefully stops a live server, flushing any pending state first. A no-op if the server is not running. """ def stop(server, reason \\ :normal) do case whereis(server) do nil -> :ok pid -> try do GenServer.stop(pid, reason) catch # the server passivated between the lookup and the stop :exit, :noproc -> :ok end end end @doc """ Stops a live server and deletes its committed state. The next call addressing the entity starts fresh from `initial_state/1`. A concurrent call racing this function can revive the entity, so stop sending messages to it before deleting. """ def delete(server) do {module, id} = normalize(server) :ok = stop({module, id}) State |> where(module: ^inspect(module), id: ^id) |> module.__revenant__().repo.delete_all() :ok end defp normalize({module, id}) when is_atom(module), do: {module, to_string(id)} defp start({module, id} = server) do case DynamicSupervisor.start_child(Revenant.DynamicSupervisor, {module, id}) do {:ok, pid} -> pid {:error, {:already_started, pid}} -> pid {:error, reason} -> exit({:revenant_start_failed, server, reason}) end end end