From 68733c9fbb6688d1d5c98d7ed2854c63bfcbf1fc Mon Sep 17 00:00:00 2001 From: Faisal Nazir Date: Thu, 23 Jul 2026 19:14:43 +0530 Subject: [PATCH 1/7] fix(session_store): make Redis store start idempotent under :one_for_all restart --- lib/anubis/server/session/store/redis.ex | 33 +++-- .../server/session/store/redis_test.exs | 117 ++++++++++++++++++ 2 files changed, 143 insertions(+), 7 deletions(-) create mode 100644 test/anubis/server/session/store/redis_test.exs diff --git a/lib/anubis/server/session/store/redis.ex b/lib/anubis/server/session/store/redis.ex index 2c6db2e2..962f5ea2 100644 --- a/lib/anubis/server/session/store/redis.ex +++ b/lib/anubis/server/session/store/redis.ex @@ -61,7 +61,13 @@ if Code.ensure_loaded?(Redix) do @impl Store def start_link(opts) do - GenServer.start_link(__MODULE__, opts, name: __MODULE__) + # Idempotent under a :one_for_all supervisor restart: if a prior instance + # is still registered under our name, adopt it instead of crashing with + # {:error, {:already_started, pid}}. + case GenServer.start_link(__MODULE__, opts, name: __MODULE__) do + {:error, {:already_started, pid}} -> {:ok, pid} + other -> other + end end @impl Store @@ -137,12 +143,7 @@ if Code.ensure_loaded?(Redix) do # 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 - } + state = build_state(conn_name, namespace, ttl, pool_size) Logging.log(:info, "Redis session store started successfully", namespace: namespace, @@ -159,6 +160,15 @@ if Code.ensure_loaded?(Redix) do {:ok, state} + {:error, {:already_started, _pid}} -> + # A racing :one_for_all restart re-ran init/1 before the prior pool + # supervisor released its fixed name. The live pool is still usable + # (connections are name-addressed via get_connection/1), so adopt it + # instead of crashing with {:stop, {:already_started, _}}. + Logging.log(:debug, "Reusing live Redis session store pool", namespace: namespace) + + {:ok, build_state(conn_name, namespace, ttl, pool_size)} + {:error, reason} = error -> Logging.log(:error, "Failed to start Redis session store", reason: inspect(reason)) @@ -300,6 +310,15 @@ if Code.ensure_loaded?(Redix) do # Private functions + defp build_state(conn_name, namespace, ttl, pool_size) do + %State{ + conn_name: conn_name, + namespace: namespace, + ttl: ttl, + pool_size: pool_size + } + end + defp validate_redix_opts(nil), do: [] defp validate_redix_opts(opts) do 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..d6b16fc0 --- /dev/null +++ b/test/anubis/server/session/store/redis_test.exs @@ -0,0 +1,117 @@ +if Code.ensure_loaded?(Redix) do + defmodule Anubis.Server.Session.Store.RedisTest do + # Regression coverage for the {:already_started} crash observed when the + # store restarts under the server's :one_for_all supervisor. Redix starts + # with sync_connect: false, so the store starts without a live Redis, which + # lets 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 "start_link/1 idempotency" do + test "adopts the live instance instead of crashing on {:already_started}", %{config: config} do + pid1 = start_supervised!({Redis, config}) + assert Process.alive?(pid1) + + # A second start under the same global name must not crash; it adopts + # the already-registered process. + assert {:ok, ^pid1} = Redis.start_link(config) + assert Process.alive?(pid1) + end + end + + describe "init/1 pool supervisor collision" do + test "reuses a pre-registered pool supervisor instead of stopping", %{ + conn_name: conn_name, + config: config + } do + supervisor_name = :"anubis_#{conn_name}_supervisor" + + # Simulate the racing restart: the prior pool supervisor's fixed name is + # still taken when the new store's init/1 runs. + {:ok, sup} = Supervisor.start_link([], strategy: :one_for_one, name: supervisor_name) + + on_exit(fn -> + if Process.alive?(sup), do: Supervisor.stop(sup) + end) + + # Before the fix this returned {:stop, {:error, {:already_started, _}}} + # and the child failed to start. Now it starts and reuses the pool. + pid = start_supervised!({Redis, config}) + assert Process.alive?(pid) + + # A Redis-free call confirms the GenServer is live and answering. + assert {:ok, 0} = Redis.cleanup_expired([]) + end + end + + describe ":one_for_all restart cascade" do + test "store survives a sibling crash without an {:already_started} failure", %{config: config} do + {:ok, sup} = + Supervisor.start_link( + [ + %{id: :sibling, start: {Agent, :start_link, [fn -> 0 end]}}, + {Redis, config} + ], + strategy: :one_for_all + ) + + on_exit(fn -> + if Process.alive?(sup), do: Supervisor.stop(sup) + end) + + store_before = Process.whereis(Redis) + assert is_pid(store_before) + + [{_, sibling_pid, _, _}] = + Enum.filter(Supervisor.which_children(sup), fn {id, _, _, _} -> id == :sibling end) + + ref = Process.monitor(sibling_pid) + Process.exit(sibling_pid, :kill) + assert_receive {:DOWN, ^ref, :process, ^sibling_pid, _}, 1_000 + + # :one_for_all restarts every child, including the store. Its restart + # must succeed (fresh pid, re-registered name) rather than flapping on + # {:already_started}. + assert eventually(fn -> + case Process.whereis(Redis) do + pid when is_pid(pid) -> Process.alive?(pid) + _ -> false + end + end) + + assert Process.alive?(sup) + end + end + + defp eventually(fun, retries \\ 50) do + cond do + fun.() -> + true + + retries <= 0 -> + false + + true -> + Process.sleep(20) + eventually(fun, retries - 1) + end + end + end +end From cc2cabf31818231e75db9c681e2039dc1b6401b7 Mon Sep 17 00:00:00 2001 From: Faisal Nazir Date: Thu, 23 Jul 2026 21:36:45 +0530 Subject: [PATCH 2/7] refactor(session_store): supervise Redis store subtree to fix restart race --- lib/anubis/server/session/store/redis.ex | 523 +++++++++--------- .../server/session/store/redis_test.exs | 185 +++++-- 2 files changed, 390 insertions(+), 318 deletions(-) diff --git a/lib/anubis/server/session/store/redis.ex b/lib/anubis/server/session/store/redis.ex index 962f5ea2..8a6ee576 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 (`#{__MODULE__}.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 `Anubis.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,62 +62,59 @@ 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 @impl Store def start_link(opts) do - # Idempotent under a :one_for_all supervisor restart: if a prior instance - # is still registered under our name, adopt it instead of crashing with - # {:error, {:already_started, pid}}. - case GenServer.start_link(__MODULE__, opts, name: __MODULE__) do - {:error, {:already_started, pid}} -> {:ok, pid} - other -> other - end + Supervisor.start_link(__MODULE__, opts, name: __MODULE__) + end + + def child_spec(opts) do + %{ + id: __MODULE__, + start: {__MODULE__, :start_link, [opts]}, + type: :supervisor + } end @impl Store 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 @impl Store def load(session_id, opts \\ []) do - GenServer.call(__MODULE__, {:load, session_id, opts}) + GenServer.call(__MODULE__.Server, {:load, session_id, opts}) end @impl Store def delete(session_id, opts \\ []) do - GenServer.call(__MODULE__, {:delete, session_id, opts}) + GenServer.call(__MODULE__.Server, {:delete, session_id, opts}) end @impl Store def list_active(opts \\ []) do - GenServer.call(__MODULE__, {:list_active, opts}) + GenServer.call(__MODULE__.Server, {:list_active, opts}) end @impl Store 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 @impl Store 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 @impl Store 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) @@ -121,8 +128,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}" @@ -137,312 +144,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 = build_state(conn_name, namespace, ttl, 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.server_event("redis_store_started", %{ - namespace: namespace, - pool_size: pool_size, - ttl: ttl - }) - - {:ok, state} + Logging.log(:info, "Redis session store started successfully", + namespace: namespace, + pool_size: pool_size, + ttl: ttl, + redis_url: redis_url + ) - {:error, {:already_started, _pid}} -> - # A racing :one_for_all restart re-ran init/1 before the prior pool - # supervisor released its fixed name. The live pool is still usable - # (connections are name-addressed via get_connection/1), so adopt it - # instead of crashing with {:stop, {:already_started, _}}. - Logging.log(:debug, "Reusing live Redis session store pool", namespace: namespace) + Logging.server_event("redis_store_started", %{ + namespace: namespace, + pool_size: pool_size, + ttl: ttl + }) - {:ok, build_state(conn_name, namespace, ttl, pool_size)} + 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) + } - case load_and_decode(state, key) do - {:ok, data} -> - {:reply, {:ok, data}, state} + {:ok, state, :hibernate} + 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) - {: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 build_state(conn_name, namespace, ttl, pool_size) do - %State{ - conn_name: conn_name, - namespace: namespace, - ttl: ttl, - pool_size: pool_size - } - end + {:error, :not_found} = error -> + {:reply, error, state} - defp validate_redix_opts(nil), do: [] + {:error, reason} = error -> + Logging.log(:error, "Failed to update session", + session_id: session_id, + error: reason + ) - defp validate_redix_opts(opts) do - if Keyword.keyword?(opts) do - opts - else - raise ArgumentError, ":redix_opts must be a keyword list" + {: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 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) + 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 index d6b16fc0..f546c0d1 100644 --- a/test/anubis/server/session/store/redis_test.exs +++ b/test/anubis/server/session/store/redis_test.exs @@ -1,9 +1,12 @@ if Code.ensure_loaded?(Redix) do defmodule Anubis.Server.Session.Store.RedisTest do # Regression coverage for the {:already_started} crash observed when the - # store restarts under the server's :one_for_all supervisor. Redix starts - # with sync_connect: false, so the store starts without a live Redis, which - # lets these run deterministically outside the :integration suite. + # 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 @@ -24,46 +27,83 @@ if Code.ensure_loaded?(Redix) do %{conn_name: conn_name, config: config} end - describe "start_link/1 idempotency" do - test "adopts the live instance instead of crashing on {:already_started}", %{config: config} do - pid1 = start_supervised!({Redis, config}) - assert Process.alive?(pid1) + 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 - # A second start under the same global name must not crash; it adopts - # the already-registered process. - assert {:ok, ^pid1} = Redis.start_link(config) - assert Process.alive?(pid1) + 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 "init/1 pool supervisor collision" do - test "reuses a pre-registered pool supervisor instead of stopping", %{ - conn_name: conn_name, - config: config - } do - supervisor_name = :"anubis_#{conn_name}_supervisor" + describe ":one_for_all restart cascade" do + test "store subtree is replaced by a fresh instance when a sibling crashes", %{config: 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) end) - # Simulate the racing restart: the prior pool supervisor's fixed name is - # still taken when the new store's init/1 runs. - {:ok, sup} = Supervisor.start_link([], strategy: :one_for_one, name: supervisor_name) + store_before = wait_for_pid(Redis) + ref = Process.monitor(store_before) - on_exit(fn -> - if Process.alive?(sup), do: Supervisor.stop(sup) - end) + sibling = child_pid(parent, :sibling) + Process.exit(sibling, :kill) - # Before the fix this returned {:stop, {:error, {:already_started, _}}} - # and the child failed to start. Now it starts and reuses the pool. - pid = start_supervised!({Redis, config}) - assert Process.alive?(pid) + # :one_for_all restarts every child, so the store subtree is torn down... + assert_receive {:DOWN, ^ref, :process, ^store_before, _}, 2_000 - # A Redis-free call confirms the GenServer is live and answering. + # ...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 ":one_for_all restart cascade" do - test "store survives a sibling crash without an {:already_started} failure", %{config: config} do - {:ok, sup} = + 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) + + {:ok, parent} = Supervisor.start_link( [ %{id: :sibling, start: {Agent, :start_link, [fn -> 0 end]}}, @@ -72,45 +112,70 @@ if Code.ensure_loaded?(Redix) do strategy: :one_for_all ) - on_exit(fn -> - if Process.alive?(sup), do: Supervisor.stop(sup) - end) + on_exit(fn -> stop_supervisor(parent) end) - store_before = Process.whereis(Redis) - assert is_pid(store_before) + wait_for_pid(Redis) + session_id = "roundtrip_#{System.unique_integer([:positive])}" + assert :ok = eventually_ok(fn -> Redis.save(session_id, %{id: session_id}) end) - [{_, sibling_pid, _, _}] = - Enum.filter(Supervisor.which_children(sup), fn {id, _, _, _} -> id == :sibling end) + store_before = wait_for_pid(Redis) + Process.exit(child_pid(parent, :sibling), :kill) + wait_for_new_pid(Redis, store_before) - ref = Process.monitor(sibling_pid) - Process.exit(sibling_pid, :kill) - assert_receive {:DOWN, ^ref, :process, ^sibling_pid, _}, 1_000 + # 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 - # :one_for_all restarts every child, including the store. Its restart - # must succeed (fresh pid, re-registered name) rather than flapping on - # {:already_started}. - assert eventually(fn -> - case Process.whereis(Redis) do - pid when is_pid(pid) -> Process.alive?(pid) - _ -> false - end - end) + defp stop_supervisor(pid) do + Supervisor.stop(pid) + catch + :exit, _ -> :ok + end - assert Process.alive?(sup) - 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 eventually(fun, retries \\ 50) do - cond do - fun.() -> - true + 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 - retries <= 0 -> - false + 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 - true -> + _ when retries > 0 -> Process.sleep(20) - eventually(fun, retries - 1) + 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 From 88dc5be9351c36e44200beb549e586539ccf91d7 Mon Sep 17 00:00:00 2001 From: Faisal Nazir Date: Thu, 23 Jul 2026 21:56:48 +0530 Subject: [PATCH 3/7] refactor(session_store): add public @specs, drop redundant child_spec, dedupe cascade test setup --- lib/anubis/server/session/store/redis.ex | 16 ++++----- .../server/session/store/redis_test.exs | 36 +++++++++---------- 2 files changed, 24 insertions(+), 28 deletions(-) diff --git a/lib/anubis/server/session/store/redis.ex b/lib/anubis/server/session/store/redis.ex index 8a6ee576..4325b6dc 100644 --- a/lib/anubis/server/session/store/redis.ex +++ b/lib/anubis/server/session/store/redis.ex @@ -65,49 +65,49 @@ if Code.ensure_loaded?(Redix) do # Client API @impl Store + @spec start_link(keyword()) :: Supervisor.on_start() def start_link(opts) do Supervisor.start_link(__MODULE__, opts, name: __MODULE__) end - def child_spec(opts) do - %{ - id: __MODULE__, - start: {__MODULE__, :start_link, [opts]}, - type: :supervisor - } - end - @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__.Server, {:save, session_id, state, opts}) end @impl Store + @spec load(Store.session_id(), Store.opts()) :: {:ok, Store.session_state()} | Store.error() def load(session_id, opts \\ []) do GenServer.call(__MODULE__.Server, {:load, session_id, opts}) end @impl Store + @spec delete(Store.session_id(), Store.opts()) :: :ok | Store.error() def delete(session_id, opts \\ []) do GenServer.call(__MODULE__.Server, {:delete, session_id, opts}) end @impl Store + @spec list_active(Store.opts()) :: {:ok, [Store.session_id()]} | Store.error() def list_active(opts \\ []) do GenServer.call(__MODULE__.Server, {:list_active, opts}) end @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__.Server, {:update_ttl, session_id, ttl_ms, opts}) end @impl Store + @spec update(Store.session_id(), map(), Store.opts()) :: :ok | Store.error() def update(session_id, updates, opts \\ []) do GenServer.call(__MODULE__.Server, {:update, session_id, updates, opts}) end @impl Store + @spec cleanup_expired(Store.opts()) :: {:ok, non_neg_integer()} | Store.error() def cleanup_expired(opts \\ []) do GenServer.call(__MODULE__.Server, {:cleanup_expired, opts}) end diff --git a/test/anubis/server/session/store/redis_test.exs b/test/anubis/server/session/store/redis_test.exs index f546c0d1..6db968b0 100644 --- a/test/anubis/server/session/store/redis_test.exs +++ b/test/anubis/server/session/store/redis_test.exs @@ -48,16 +48,7 @@ if Code.ensure_loaded?(Redix) do describe ":one_for_all restart cascade" do test "store subtree is replaced by a fresh instance when a sibling crashes", %{config: 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) end) + parent = start_one_for_all!(config) store_before = wait_for_pid(Redis) ref = Process.monitor(store_before) @@ -103,16 +94,7 @@ if Code.ensure_loaded?(Redix) do end end) - {: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) end) + parent = start_one_for_all!(config) wait_for_pid(Redis) session_id = "roundtrip_#{System.unique_integer([:positive])}" @@ -127,6 +109,20 @@ if Code.ensure_loaded?(Redix) do 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) end) + parent + end + defp stop_supervisor(pid) do Supervisor.stop(pid) catch From 4cb3d0f4385bcdf90979b63d91663739ad2d0a24 Mon Sep 17 00:00:00 2001 From: Faisal Nazir Date: Thu, 23 Jul 2026 22:12:41 +0530 Subject: [PATCH 4/7] docs(session_store): document Redis store start_link/1 public API --- lib/anubis/server/session/store/redis.ex | 29 ++++++++++++++++++++---- 1 file changed, 25 insertions(+), 4 deletions(-) diff --git a/lib/anubis/server/session/store/redis.ex b/lib/anubis/server/session/store/redis.ex index 4325b6dc..b9d32f9b 100644 --- a/lib/anubis/server/session/store/redis.ex +++ b/lib/anubis/server/session/store/redis.ex @@ -43,12 +43,12 @@ if Code.ensure_loaded?(Redix) do ## Architecture This module is a `Supervisor` whose children are the Redix connection pool - plus an internal state server (`#{__MODULE__}.Server`) that answers the + 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 `Anubis.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}`. + 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 @@ -64,6 +64,27 @@ if Code.ensure_loaded?(Redix) do # 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 From eade9b4afb83f77cb25409f690ffaf13d66af5a6 Mon Sep 17 00:00:00 2001 From: Faisal Nazir Date: Fri, 24 Jul 2026 10:34:56 +0530 Subject: [PATCH 5/7] docs(session_store): document each Redis store operation with examples --- lib/anubis/server/session/store/redis.ex | 72 ++++++++++++++++++++++++ 1 file changed, 72 insertions(+) diff --git a/lib/anubis/server/session/store/redis.ex b/lib/anubis/server/session/store/redis.ex index b9d32f9b..6028c938 100644 --- a/lib/anubis/server/session/store/redis.ex +++ b/lib/anubis/server/session/store/redis.ex @@ -91,42 +91,114 @@ if Code.ensure_loaded?(Redix) do 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__.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__.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__.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__.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__.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__.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 From 478635c3fee659f3842675c1bb133fc0d7d299f1 Mon Sep 17 00:00:00 2001 From: Faisal Nazir Date: Fri, 24 Jul 2026 10:35:29 +0530 Subject: [PATCH 6/7] test(session_store): make cascade-test teardown deterministic --- .../server/session/store/redis_test.exs | 23 ++++++++++++++++++- 1 file changed, 22 insertions(+), 1 deletion(-) diff --git a/test/anubis/server/session/store/redis_test.exs b/test/anubis/server/session/store/redis_test.exs index 6db968b0..988b1cf1 100644 --- a/test/anubis/server/session/store/redis_test.exs +++ b/test/anubis/server/session/store/redis_test.exs @@ -119,7 +119,14 @@ if Code.ensure_loaded?(Redix) do strategy: :one_for_all ) - on_exit(fn -> stop_supervisor(parent) end) + 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 @@ -129,6 +136,20 @@ if Code.ensure_loaded?(Redix) do :exit, _ -> :ok end + defp await_unregistered(name, retries \\ 50) do + cond do + is_nil(Process.whereis(name)) -> + :ok + + retries <= 0 -> + :ok + + 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) From 41313751c41b1c230098488503cd645fa97922ae Mon Sep 17 00:00:00 2001 From: Faisal Nazir Date: Fri, 24 Jul 2026 10:50:57 +0530 Subject: [PATCH 7/7] test(session_store): fail teardown wait when a name never clears --- test/anubis/server/session/store/redis_test.exs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/test/anubis/server/session/store/redis_test.exs b/test/anubis/server/session/store/redis_test.exs index 988b1cf1..ab2f1fb3 100644 --- a/test/anubis/server/session/store/redis_test.exs +++ b/test/anubis/server/session/store/redis_test.exs @@ -142,7 +142,7 @@ if Code.ensure_loaded?(Redix) do :ok retries <= 0 -> - :ok + flunk("#{inspect(name)} remained registered after teardown") true -> Process.sleep(20)