diff --git a/lib/anubis/server/session/store/redis.ex b/lib/anubis/server/session/store/redis.ex index 2c6db2e2..6028c938 100644 --- a/lib/anubis/server/session/store/redis.ex +++ b/lib/anubis/server/session/store/redis.ex @@ -39,11 +39,21 @@ if Code.ensure_loaded?(Redix) do - Last-write-wins semantics for session updates - Connection pooling for high concurrency - Namespace support for multi-tenant deployments + + ## Architecture + + This module is a `Supervisor` whose children are the Redix connection pool + plus an internal state server that answers the + `Anubis.Server.Session.Store` behaviour calls. Because the pool lives *inside* + the supervision tree (rather than being started from `init/1`), a restart of + the enclosing server supervisor (which runs `:one_for_all`) tears the whole + subtree down synchronously — releasing every registered name before the store + is restarted. Restarts are therefore race-free: no `{:already_started}`. """ @behaviour Anubis.Server.Session.Store - use GenServer + use Supervisor use Anubis.Logging alias Anubis.Server.Session.Store @@ -52,56 +62,152 @@ if Code.ensure_loaded?(Redix) do @default_ttl 1_800_000 @default_namespace "anubis:sessions" - defmodule State do - @moduledoc false - defstruct [:conn_name, :namespace, :ttl, :pool_size] - end - # Client API + @doc """ + Starts the Redis session store supervisor. + + Supervises the Redix connection pool and the internal state server. `opts` + is the `:session_store` keyword config documented in the moduledoc + (`:redis_url`, `:pool_size`, `:ttl`, `:namespace`, `:connection_name`, + `:redix_opts`). + + ## Examples + + {:ok, _pid} = + Anubis.Server.Session.Store.Redis.start_link( + redis_url: "redis://localhost:6379/0", + namespace: "anubis:sessions" + ) + + The store operations (`save/3`, `load/2`, `delete/2`, `list_active/1`, + `update_ttl/3`, `update/3`, `cleanup_expired/1`) follow the + `Anubis.Server.Session.Store` behaviour; see its callback docs for the + request/response contract. + """ @impl Store + @spec start_link(keyword()) :: Supervisor.on_start() def start_link(opts) do - GenServer.start_link(__MODULE__, opts, name: __MODULE__) + Supervisor.start_link(__MODULE__, opts, name: __MODULE__) end + @doc """ + Persists `state` for `session_id`, expiring after the store TTL. + + Pass `opts[:ttl]` (milliseconds) to override the configured TTL for this + write. Implements `c:Anubis.Server.Session.Store.save/3`. + + ## Examples + + :ok = Anubis.Server.Session.Store.Redis.save("sess-1", %{initialized: true}) + :ok = Anubis.Server.Session.Store.Redis.save("sess-1", %{}, ttl: 60_000) + """ @impl Store + @spec save(Store.session_id(), Store.session_state(), Store.opts()) :: :ok | Store.error() def save(session_id, state, opts \\ []) do - GenServer.call(__MODULE__, {:save, session_id, state, opts}) + GenServer.call(__MODULE__.Server, {:save, session_id, state, opts}) end + @doc """ + Loads the persisted state for `session_id`. + + Returns `{:error, :not_found}` when the key is absent or has expired. + Implements `c:Anubis.Server.Session.Store.load/2`. + + ## Examples + + {:ok, state} = Anubis.Server.Session.Store.Redis.load("sess-1") + {:error, :not_found} = Anubis.Server.Session.Store.Redis.load("missing") + """ @impl Store + @spec load(Store.session_id(), Store.opts()) :: {:ok, Store.session_state()} | Store.error() def load(session_id, opts \\ []) do - GenServer.call(__MODULE__, {:load, session_id, opts}) + GenServer.call(__MODULE__.Server, {:load, session_id, opts}) end + @doc """ + Deletes any persisted state for `session_id`. + + Idempotent — returns `:ok` even when nothing was stored. Implements + `c:Anubis.Server.Session.Store.delete/2`. + + ## Examples + + :ok = Anubis.Server.Session.Store.Redis.delete("sess-1") + """ @impl Store + @spec delete(Store.session_id(), Store.opts()) :: :ok | Store.error() def delete(session_id, opts \\ []) do - GenServer.call(__MODULE__, {:delete, session_id, opts}) + GenServer.call(__MODULE__.Server, {:delete, session_id, opts}) end + @doc """ + Lists the session ids currently held in the store's namespace. + + Implements `c:Anubis.Server.Session.Store.list_active/1`. + + ## Examples + + {:ok, ids} = Anubis.Server.Session.Store.Redis.list_active() + """ @impl Store + @spec list_active(Store.opts()) :: {:ok, [Store.session_id()]} | Store.error() def list_active(opts \\ []) do - GenServer.call(__MODULE__, {:list_active, opts}) + GenServer.call(__MODULE__.Server, {:list_active, opts}) end + @doc """ + Refreshes the expiry of `session_id` to `ttl_ms` milliseconds from now. + + Returns `{:error, :not_found}` when the session is absent. Implements + `c:Anubis.Server.Session.Store.update_ttl/3`. + + ## Examples + + :ok = Anubis.Server.Session.Store.Redis.update_ttl("sess-1", 1_800_000) + """ @impl Store + @spec update_ttl(Store.session_id(), pos_integer(), Store.opts()) :: :ok | Store.error() def update_ttl(session_id, ttl_ms, opts \\ []) do - GenServer.call(__MODULE__, {:update_ttl, session_id, ttl_ms, opts}) + GenServer.call(__MODULE__.Server, {:update_ttl, session_id, ttl_ms, opts}) end + @doc """ + Merges `updates` into the persisted state for `session_id`. + + Read-modify-write with last-write-wins semantics; returns + `{:error, :not_found}` when the session is absent. Implements + `c:Anubis.Server.Session.Store.update/3`. + + ## Examples + + :ok = Anubis.Server.Session.Store.Redis.update("sess-1", %{log_level: "debug"}) + """ @impl Store + @spec update(Store.session_id(), map(), Store.opts()) :: :ok | Store.error() def update(session_id, updates, opts \\ []) do - GenServer.call(__MODULE__, {:update, session_id, updates, opts}) + GenServer.call(__MODULE__.Server, {:update, session_id, updates, opts}) end + @doc """ + No-op for Redis — expiry is handled natively by Redis TTLs. + + Always returns `{:ok, 0}`. Implements + `c:Anubis.Server.Session.Store.cleanup_expired/1`. + + ## Examples + + {:ok, 0} = Anubis.Server.Session.Store.Redis.cleanup_expired() + """ @impl Store + @spec cleanup_expired(Store.opts()) :: {:ok, non_neg_integer()} | Store.error() def cleanup_expired(opts \\ []) do - GenServer.call(__MODULE__, {:cleanup_expired, opts}) + GenServer.call(__MODULE__.Server, {:cleanup_expired, opts}) end - # GenServer callbacks + # Supervisor callback - @impl GenServer + @impl Supervisor def init(opts) do redis_url = Keyword.get(opts, :redis_url, "redis://localhost:6379/0") conn_name = Keyword.get(opts, :connection_name, :anubis_redis) @@ -115,8 +221,8 @@ if Code.ensure_loaded?(Redix) do |> validate_redix_opts() |> Keyword.delete(:name) - # Start Redix connection pool with anubis_ prefix to avoid conflicts - children = + # Redix connection pool with anubis_ prefix to avoid conflicts + pool_children = for i <- 1..pool_size do child_id = :"anubis_#{conn_name}_#{i}" @@ -131,299 +237,312 @@ if Code.ensure_loaded?(Redix) do } end - # Use anubis_ prefix for supervisor name - supervisor_name = :"anubis_#{conn_name}_supervisor" - - # Start connections under a supervisor - case Supervisor.start_link(children, strategy: :one_for_one, name: supervisor_name) do - {:ok, _pid} -> - state = %State{ - conn_name: conn_name, - namespace: namespace, - ttl: ttl, - pool_size: pool_size - } + server_child = + {__MODULE__.Server, conn_name: conn_name, namespace: namespace, ttl: ttl, pool_size: pool_size} - Logging.log(:info, "Redis session store started successfully", - namespace: namespace, - pool_size: pool_size, - ttl: ttl, - redis_url: redis_url - ) + Logging.log(:info, "Redis session store started successfully", + namespace: namespace, + pool_size: pool_size, + ttl: ttl, + redis_url: redis_url + ) - Logging.server_event("redis_store_started", %{ - namespace: namespace, - pool_size: pool_size, - ttl: ttl - }) + Logging.server_event("redis_store_started", %{ + namespace: namespace, + pool_size: pool_size, + ttl: ttl + }) - {:ok, state} + Supervisor.init(pool_children ++ [server_child], strategy: :one_for_one) + end - {:error, reason} = error -> - Logging.log(:error, "Failed to start Redis session store", reason: inspect(reason)) + defp validate_redix_opts(nil), do: [] - {:stop, error} + defp validate_redix_opts(opts) do + if Keyword.keyword?(opts) do + opts + else + raise ArgumentError, ":redix_opts must be a keyword list" end end - @impl GenServer - def handle_call({:save, session_id, session_state, opts}, _from, state) do - ttl = Keyword.get(opts, :ttl, state.ttl) - key = make_key(state.namespace, session_id) + defmodule Server do + @moduledoc false - case encode_and_save(state, key, session_state, ttl) do - :ok -> - Logging.server_event("session_saved", %{session_id: session_id, ttl: ttl}) - {:reply, :ok, state} + use GenServer + use Anubis.Logging - {:error, reason} = error -> - Logging.log(:error, "Failed to persist session", - session_id: session_id, - error: reason - ) + defmodule State do + @moduledoc false + defstruct [:conn_name, :namespace, :ttl, :pool_size] + end - {:reply, error, state} + @spec start_link(keyword()) :: GenServer.on_start() + def start_link(opts) do + GenServer.start_link(__MODULE__, opts, name: __MODULE__) end - end - @impl GenServer - def handle_call({:load, session_id, _opts}, _from, state) do - key = make_key(state.namespace, session_id) + @impl GenServer + def init(opts) do + state = %State{ + conn_name: Keyword.fetch!(opts, :conn_name), + namespace: Keyword.fetch!(opts, :namespace), + ttl: Keyword.fetch!(opts, :ttl), + pool_size: Keyword.fetch!(opts, :pool_size) + } + + {:ok, state, :hibernate} + end - case load_and_decode(state, key) do - {:ok, data} -> - {:reply, {:ok, data}, state} + @impl GenServer + def handle_call({:save, session_id, session_state, opts}, _from, state) do + ttl = Keyword.get(opts, :ttl, state.ttl) + key = make_key(state.namespace, session_id) - {:error, :not_found} = error -> - {:reply, error, state} + case encode_and_save(state, key, session_state, ttl) do + :ok -> + Logging.server_event("session_saved", %{session_id: session_id, ttl: ttl}) + {:reply, :ok, state} - {:error, reason} = error -> - Logging.log(:error, "Failed to load session #{session_id}", error: reason) + {:error, reason} = error -> + Logging.log(:error, "Failed to persist session", + session_id: session_id, + error: reason + ) - {:reply, error, state} + {:reply, error, state} + end end - end - @impl GenServer - def handle_call({:delete, session_id, _opts}, _from, state) do - key = make_key(state.namespace, session_id) - conn = get_connection(state) + @impl GenServer + def handle_call({:load, session_id, _opts}, _from, state) do + key = make_key(state.namespace, session_id) - case Redix.command(conn, ["DEL", key]) do - {:ok, _} -> - Logging.server_event("session_deleted", %{session_id: session_id}) - {:reply, :ok, state} + case load_and_decode(state, key) do + {:ok, data} -> + {:reply, {:ok, data}, state} - {:error, reason} -> - Logging.log(:error, "Failed to delete session", - session_id: session_id, - error: reason - ) + {:error, :not_found} = error -> + {:reply, error, state} - {:reply, {:error, reason}, state} - end - end + {:error, reason} = error -> + Logging.log(:error, "Failed to load session #{session_id}", error: reason) - @impl GenServer - def handle_call({:list_active, opts}, _from, state) do - pattern = make_key(state.namespace, "*") - server_filter = Keyword.get(opts, :server) - conn = get_connection(state) + {:reply, error, state} + end + end - case scan_keys(conn, pattern) do - {:ok, keys} -> - session_ids = - keys - |> Enum.map(&extract_session_id(state.namespace, &1)) - |> filter_by_server(server_filter) + @impl GenServer + def handle_call({:delete, session_id, _opts}, _from, state) do + key = make_key(state.namespace, session_id) + conn = get_connection(state) - {:reply, {:ok, session_ids}, state} + case Redix.command(conn, ["DEL", key]) do + {:ok, _} -> + Logging.server_event("session_deleted", %{session_id: session_id}) + {:reply, :ok, state} - {:error, reason} = error -> - Logging.log(:error, "Failed to list sessions from store", error: reason) + {:error, reason} -> + Logging.log(:error, "Failed to delete session", + session_id: session_id, + error: reason + ) - {:reply, error, state} + {:reply, {:error, reason}, state} + end end - end - @impl GenServer - def handle_call({:update_ttl, session_id, ttl_ms, _opts}, _from, state) do - key = make_key(state.namespace, session_id) - conn = get_connection(state) - ttl_seconds = ms_to_seconds(ttl_ms) + @impl GenServer + def handle_call({:list_active, opts}, _from, state) do + pattern = make_key(state.namespace, "*") + server_filter = Keyword.get(opts, :server) + conn = get_connection(state) - case Redix.command(conn, ["EXPIRE", key, ttl_seconds]) do - {:ok, 1} -> - {:reply, :ok, state} + case scan_keys(conn, pattern) do + {:ok, keys} -> + session_ids = + keys + |> Enum.map(&extract_session_id(state.namespace, &1)) + |> filter_by_server(server_filter) - {:ok, 0} -> - {:reply, {:error, :not_found}, state} + {:reply, {:ok, session_ids}, state} - {:error, reason} -> - Logging.log(:error, "Failed to update TTL for session", - session_id: session_id, - error: reason - ) + {:error, reason} = error -> + Logging.log(:error, "Failed to list sessions from store", error: reason) - {:reply, {:error, reason}, state} + {:reply, error, state} + end end - end - @impl GenServer - def handle_call({:update, session_id, updates, opts}, _from, state) do - key = make_key(state.namespace, session_id) - ttl = Keyword.get(opts, :ttl, state.ttl) + @impl GenServer + def handle_call({:update_ttl, session_id, ttl_ms, _opts}, _from, state) do + key = make_key(state.namespace, session_id) + conn = get_connection(state) + ttl_seconds = ms_to_seconds(ttl_ms) - case atomic_update(state, key, updates, ttl) do - :ok -> - {:reply, :ok, state} + case Redix.command(conn, ["EXPIRE", key, ttl_seconds]) do + {:ok, 1} -> + {:reply, :ok, state} - {:error, :not_found} = error -> - {:reply, error, state} + {:ok, 0} -> + {:reply, {:error, :not_found}, state} - {:error, reason} = error -> - Logging.log(:error, "Failed to update session", - session_id: session_id, - error: reason - ) + {:error, reason} -> + Logging.log(:error, "Failed to update TTL for session", + session_id: session_id, + error: reason + ) - {:reply, error, state} + {:reply, {:error, reason}, state} + end end - end - @impl GenServer - def handle_call({:cleanup_expired, _opts}, _from, state) do - # Redis handles expiration automatically via TTL - # This is a no-op but we could scan and count expired keys if needed - {:reply, {:ok, 0}, state} - end + @impl GenServer + def handle_call({:update, session_id, updates, opts}, _from, state) do + key = make_key(state.namespace, session_id) + ttl = Keyword.get(opts, :ttl, state.ttl) - # Private functions + case atomic_update(state, key, updates, ttl) do + :ok -> + {:reply, :ok, state} - defp validate_redix_opts(nil), do: [] + {:error, :not_found} = error -> + {:reply, error, state} - defp validate_redix_opts(opts) do - if Keyword.keyword?(opts) do - opts - else - raise ArgumentError, ":redix_opts must be a keyword list" + {:error, reason} = error -> + Logging.log(:error, "Failed to update session", + session_id: session_id, + error: reason + ) + + {:reply, error, state} + end end - end - defp make_key(namespace, session_id) do - "#{namespace}:#{session_id}" - end + @impl GenServer + def handle_call({:cleanup_expired, _opts}, _from, state) do + # Redis handles expiration automatically via TTL + # This is a no-op but we could scan and count expired keys if needed + {:reply, {:ok, 0}, state} + end - defp extract_session_id(namespace, key) do - prefix = "#{namespace}:" - String.replace_prefix(key, prefix, "") - end + # Private functions - defp get_connection(state) when is_struct(state, State) do - # Use cheap monotonic counter for pool selection instead of random - index = rem(:erlang.unique_integer([:positive]), state.pool_size) + 1 - :"anubis_#{state.conn_name}_#{index}" - end + defp make_key(namespace, session_id) do + "#{namespace}:#{session_id}" + end - defp json_encode(data) do - {:ok, JSON.encode!(data)} - rescue - error -> {:error, error} - end + defp extract_session_id(namespace, key) do + prefix = "#{namespace}:" + String.replace_prefix(key, prefix, "") + end - defp ms_to_seconds(milliseconds) do - div(milliseconds, 1000) - end + defp get_connection(state) when is_struct(state, State) do + # Use cheap monotonic counter for pool selection instead of random + index = rem(:erlang.unique_integer([:positive]), state.pool_size) + 1 + :"anubis_#{state.conn_name}_#{index}" + end + + defp json_encode(data) do + {:ok, JSON.encode!(data)} + rescue + error -> {:error, error} + end - defp encode_and_save(state, key, data, ttl) do - conn = get_connection(state) - ttl_seconds = ms_to_seconds(ttl) + defp ms_to_seconds(milliseconds) do + div(milliseconds, 1000) + end + + defp encode_and_save(state, key, data, ttl) do + conn = get_connection(state) + ttl_seconds = ms_to_seconds(ttl) - case json_encode(data) do - {:ok, json} -> - case Redix.command(conn, ["SETEX", key, ttl_seconds, json]) do - {:ok, "OK"} -> :ok - {:error, reason} -> {:error, reason} - end + case json_encode(data) do + {:ok, json} -> + case Redix.command(conn, ["SETEX", key, ttl_seconds, json]) do + {:ok, "OK"} -> :ok + {:error, reason} -> {:error, reason} + end - {:error, reason} -> - {:error, {:encoding_failed, reason}} + {:error, reason} -> + {:error, {:encoding_failed, reason}} + end end - end - defp load_and_decode(state, key) do - conn = get_connection(state) + defp load_and_decode(state, key) do + conn = get_connection(state) - case Redix.command(conn, ["GET", key]) do - {:ok, nil} -> - {:error, :not_found} + case Redix.command(conn, ["GET", key]) do + {:ok, nil} -> + {:error, :not_found} - {:ok, json} -> - case JSON.decode(json) do - {:ok, data} -> {:ok, data} - {:error, reason} -> {:error, {:decoding_failed, reason}} - end + {:ok, json} -> + case JSON.decode(json) do + {:ok, data} -> {:ok, data} + {:error, reason} -> {:error, {:decoding_failed, reason}} + end - {:error, reason} -> - {:error, reason} + {:error, reason} -> + {:error, reason} + end end - end - defp atomic_update(state, key, updates, ttl) do - conn = get_connection(state) - ttl_seconds = ms_to_seconds(ttl) - - # Simple read-modify-write (last-write-wins semantics) - # Good enough for session storage - sessions are single-writer in practice - with {:ok, current_data} <- fetch_current_data(conn, key), - updated_data = Map.merge(current_data, updates), - {:ok, new_json} <- json_encode(updated_data), - {:ok, "OK"} <- Redix.command(conn, ["SETEX", key, ttl_seconds, new_json]) do - :ok + defp atomic_update(state, key, updates, ttl) do + conn = get_connection(state) + ttl_seconds = ms_to_seconds(ttl) + + # Simple read-modify-write (last-write-wins semantics) + # Good enough for session storage - sessions are single-writer in practice + with {:ok, current_data} <- fetch_current_data(conn, key), + updated_data = Map.merge(current_data, updates), + {:ok, new_json} <- json_encode(updated_data), + {:ok, "OK"} <- Redix.command(conn, ["SETEX", key, ttl_seconds, new_json]) do + :ok + end end - end - defp fetch_current_data(conn, key) do - with {:ok, json} <- fetch_existing_key(conn, key), - {:ok, data} <- JSON.decode(json) do - {:ok, data} - else - {:error, :not_found} = err -> err - {:error, reason} -> {:error, {:decoding_failed, reason}} + defp fetch_current_data(conn, key) do + with {:ok, json} <- fetch_existing_key(conn, key), + {:ok, data} <- JSON.decode(json) do + {:ok, data} + else + {:error, :not_found} = err -> err + {:error, reason} -> {:error, {:decoding_failed, reason}} + end end - end - defp fetch_existing_key(conn, key) do - case Redix.command(conn, ["GET", key]) do - {:ok, nil} -> {:error, :not_found} - {:ok, json} -> {:ok, json} - {:error, reason} -> {:error, reason} + defp fetch_existing_key(conn, key) do + case Redix.command(conn, ["GET", key]) do + {:ok, nil} -> {:error, :not_found} + {:ok, json} -> {:ok, json} + {:error, reason} -> {:error, reason} + end end - end - defp scan_keys(conn, pattern, cursor \\ "0", acc \\ []) do - case Redix.command(conn, ["SCAN", cursor, "MATCH", pattern, "COUNT", "100"]) do - {:ok, [new_cursor, keys]} -> - new_acc = acc ++ keys + defp scan_keys(conn, pattern, cursor \\ "0", acc \\ []) do + case Redix.command(conn, ["SCAN", cursor, "MATCH", pattern, "COUNT", "100"]) do + {:ok, [new_cursor, keys]} -> + new_acc = acc ++ keys - if new_cursor == "0" do - {:ok, new_acc} - else - scan_keys(conn, pattern, new_cursor, new_acc) - end + if new_cursor == "0" do + {:ok, new_acc} + else + scan_keys(conn, pattern, new_cursor, new_acc) + end - {:error, reason} -> - {:error, reason} + {:error, reason} -> + {:error, reason} + end end - end - defp filter_by_server(session_ids, nil), do: session_ids + defp filter_by_server(session_ids, nil), do: session_ids - defp filter_by_server(session_ids, server) do - # If we need server-specific filtering, we'd need to load each session - # and check its server field. For now, return all. - _ = server - session_ids + defp filter_by_server(session_ids, server) do + # If we need server-specific filtering, we'd need to load each session + # and check its server field. For now, return all. + _ = server + session_ids + end end end end diff --git a/test/anubis/server/session/store/redis_test.exs b/test/anubis/server/session/store/redis_test.exs new file mode 100644 index 00000000..ab2f1fb3 --- /dev/null +++ b/test/anubis/server/session/store/redis_test.exs @@ -0,0 +1,199 @@ +if Code.ensure_loaded?(Redix) do + defmodule Anubis.Server.Session.Store.RedisTest do + # Regression coverage for the {:already_started} crash observed when the + # store restarted under the server's :one_for_all supervisor. The store is + # now a supervised subtree (pool + state server), so a restart tears the + # whole tree down synchronously and no fixed name is ever double-registered. + # + # Redix starts with sync_connect: false, so the subtree starts without a live + # Redis — these run deterministically outside the :integration suite. + use ExUnit.Case, async: false + + alias Anubis.Server.Session.Store.Redis + + @moduletag capture_log: true + + setup do + conn_name = :"redis_test_#{System.unique_integer([:positive])}" + + config = [ + redis_url: "redis://localhost:6379", + pool_size: 2, + ttl: 10_000, + namespace: "anubis:test", + connection_name: conn_name + ] + + %{conn_name: conn_name, config: config} + end + + describe "supervision structure" do + test "starts the pool and state server as supervised children", %{config: config} do + sup = start_supervised!({Redis, config}) + assert Process.alive?(sup) + + children = Supervisor.which_children(sup) + + # pool_size connections + one state server + assert length(children) == config[:pool_size] + 1 + + assert Enum.all?(children, fn {_id, pid, _type, _mods} -> + is_pid(pid) and Process.alive?(pid) + end) + + assert Enum.any?(children, fn {id, _pid, _type, _mods} -> id == Redis.Server end) + assert is_pid(Process.whereis(Redis.Server)) + end + end + + describe ":one_for_all restart cascade" do + test "store subtree is replaced by a fresh instance when a sibling crashes", %{config: config} do + parent = start_one_for_all!(config) + + store_before = wait_for_pid(Redis) + ref = Process.monitor(store_before) + + sibling = child_pid(parent, :sibling) + Process.exit(sibling, :kill) + + # :one_for_all restarts every child, so the store subtree is torn down... + assert_receive {:DOWN, ^ref, :process, ^store_before, _}, 2_000 + + # ...and replaced by a distinct, live instance — no {:already_started}. + store_after = wait_for_new_pid(Redis, store_before) + assert store_after != store_before + assert Process.alive?(store_after) + assert Process.alive?(parent) + end + end + + describe "behaviour delegation" do + test "cleanup_expired/1 reaches the state server without Redis", %{config: config} do + start_supervised!({Redis, config}) + assert {:ok, 0} = Redis.cleanup_expired([]) + end + end + + describe "post-restart round-trip" do + @describetag :integration + + @tag :integration + test "persisted sessions are served by the restarted pool", %{config: config} do + namespace = config[:namespace] + + on_exit(fn -> + {:ok, conn} = Redix.start_link(config[:redis_url]) + + try do + case Redix.command(conn, ["KEYS", "#{namespace}:*"]) do + {:ok, [_ | _] = keys} -> Redix.command(conn, ["DEL" | keys]) + _ -> :ok + end + after + Redix.stop(conn) + end + end) + + parent = start_one_for_all!(config) + + wait_for_pid(Redis) + session_id = "roundtrip_#{System.unique_integer([:positive])}" + assert :ok = eventually_ok(fn -> Redis.save(session_id, %{id: session_id}) end) + + store_before = wait_for_pid(Redis) + Process.exit(child_pid(parent, :sibling), :kill) + wait_for_new_pid(Redis, store_before) + + # The data lives in Redis; the freshly restarted pool must still serve it. + assert {:ok, %{"id" => ^session_id}} = eventually_ok(fn -> Redis.load(session_id) end) + end + end + + defp start_one_for_all!(config) do + {:ok, parent} = + Supervisor.start_link( + [ + %{id: :sibling, start: {Agent, :start_link, [fn -> 0 end]}}, + {Redis, config} + ], + strategy: :one_for_all + ) + + on_exit(fn -> + stop_supervisor(parent) + # The parent is linked to the (now-dead) test process, so teardown may be + # racing; wait for the global names to clear before the next test starts. + await_unregistered(Redis) + await_unregistered(Redis.Server) + end) + + parent + end + + defp stop_supervisor(pid) do + Supervisor.stop(pid) + catch + :exit, _ -> :ok + end + + defp await_unregistered(name, retries \\ 50) do + cond do + is_nil(Process.whereis(name)) -> + :ok + + retries <= 0 -> + flunk("#{inspect(name)} remained registered after teardown") + + true -> + Process.sleep(20) + await_unregistered(name, retries - 1) + end + end + + defp child_pid(sup, id) do + {^id, pid, _type, _mods} = + Enum.find(Supervisor.which_children(sup), fn {child_id, _, _, _} -> child_id == id end) + + pid + end + + defp wait_for_pid(name, retries \\ 50) do + case Process.whereis(name) do + pid when is_pid(pid) -> + pid + + nil when retries > 0 -> + Process.sleep(20) + wait_for_pid(name, retries - 1) + + nil -> + flunk("#{inspect(name)} was never registered") + end + end + + defp wait_for_new_pid(name, old_pid, retries \\ 50) do + case Process.whereis(name) do + pid when is_pid(pid) and pid != old_pid -> + pid + + _ when retries > 0 -> + Process.sleep(20) + wait_for_new_pid(name, old_pid, retries - 1) + + _ -> + flunk("#{inspect(name)} was not replaced with a new pid") + end + end + + defp eventually_ok(fun, retries \\ 25) do + case fun.() do + {:error, _} when retries > 0 -> + Process.sleep(40) + eventually_ok(fun, retries - 1) + + other -> + other + end + end + end +end