defmodule Snakepit.Bridge.SessionStore do @moduledoc """ Centralized session store using ETS for high-performance session management. This GenServer manages a centralized ETS table for storing session data, providing CRUD operations, TTL-based expiration, and automatic cleanup. The store is designed for high concurrency with optimized ETS settings. """ use GenServer alias Snakepit.Bridge.Session alias Snakepit.Defaults alias Snakepit.Logger, as: SLog @log_category :bridge @default_table_name :snakepit_sessions ## Client API @doc """ Starts the SessionStore GenServer. ## Options - `:name` - The name to register the GenServer (default: __MODULE__) - `:table_name` - The ETS table name (default: :snakepit_sessions) - `:cleanup_interval` - Cleanup interval in milliseconds (default: 60_000) - `:default_ttl` - Default TTL for sessions in seconds (default: 3600) """ @spec start_link(keyword()) :: GenServer.on_start() def start_link(opts \\ []) do name = Keyword.get(opts, :name, __MODULE__) GenServer.start_link(__MODULE__, opts, name: name) end @doc """ Creates a new session with the given ID and options. ## Parameters - `session_id` - Unique session identifier - `opts` - Keyword list of options passed to Session.new/2 ## Returns `{:ok, session}` if successful, `{:error, reason}` if failed. ## Examples {:ok, session} = SessionStore.create_session("session_123") {:ok, session} = SessionStore.create_session("session_456", ttl: 7200) """ @spec create_session(String.t(), keyword()) :: {:ok, Session.t()} | {:error, term()} def create_session(session_id, opts \\ []) when is_binary(session_id) do GenServer.call(__MODULE__, {:create_session, session_id, opts}) end @spec create_session(GenServer.server(), String.t(), keyword()) :: {:ok, Session.t()} | {:error, term()} def create_session(server, session_id, opts) when is_binary(session_id) do GenServer.call(server, {:create_session, session_id, opts}) end @doc """ Gets a session by ID, automatically updating the last_accessed timestamp. ## Parameters - `session_id` - The session identifier ## Returns `{:ok, session}` if found, `{:error, :not_found}` if not found. """ @spec get_session(String.t()) :: {:ok, Session.t()} | {:error, :not_found} def get_session(session_id) when is_binary(session_id) do get_session(__MODULE__, session_id) end @spec get_session(GenServer.server(), String.t()) :: {:ok, Session.t()} | {:error, :not_found} def get_session(server, session_id) when is_binary(session_id) do GenServer.call(server, {:get_session, session_id}) end @doc """ Updates a session using the provided update function. The update function receives the current session and should return the updated session. The operation is atomic. ## Parameters - `session_id` - The session identifier - `update_fn` - Function that takes a session and returns an updated session ## Returns `{:ok, updated_session}` if successful, `{:error, reason}` if failed. ## Examples {:ok, session} = SessionStore.update_session("session_123", fn session -> Map.put(session, :data, %{key: "value"}) end) """ @spec update_session(String.t(), (Session.t() -> Session.t())) :: {:ok, Session.t()} | {:error, term()} def update_session(session_id, update_fn) when is_binary(session_id) and is_function(update_fn, 1) do update_session(__MODULE__, session_id, update_fn) end @spec update_session(GenServer.server(), String.t(), (Session.t() -> Session.t())) :: {:ok, Session.t()} | {:error, term()} def update_session(server, session_id, update_fn) when is_binary(session_id) and is_function(update_fn, 1) do GenServer.call(server, {:update_session, session_id, update_fn}) end @doc """ Deletes a session by ID. ## Parameters - `session_id` - The session identifier ## Returns `:ok` always (idempotent operation). """ @spec delete_session(String.t()) :: :ok def delete_session(session_id) when is_binary(session_id) do delete_session(__MODULE__, session_id) end @spec delete_session(GenServer.server(), String.t()) :: :ok def delete_session(server, session_id) when is_binary(session_id) do GenServer.call(server, {:delete_session, session_id}) end @doc """ Manually triggers cleanup of expired sessions. ## Returns The number of sessions that were cleaned up. """ @spec cleanup_expired_sessions() :: non_neg_integer() def cleanup_expired_sessions do cleanup_expired_sessions(__MODULE__) end @spec cleanup_expired_sessions(GenServer.server()) :: non_neg_integer() def cleanup_expired_sessions(server) do GenServer.call(server, :cleanup_expired_sessions) end @doc """ Gets statistics about the session store. ## Returns A map containing various statistics about the session store. """ @spec get_stats() :: map() def get_stats do get_stats(__MODULE__) end @spec get_stats(GenServer.server()) :: map() def get_stats(server) do GenServer.call(server, :get_stats) end @doc """ Lists all active session IDs. ## Returns A list of all active session IDs. """ @spec list_sessions() :: [String.t()] def list_sessions do list_sessions(__MODULE__) end @spec list_sessions(GenServer.server()) :: [String.t()] def list_sessions(server) do GenServer.call(server, :list_sessions) end @doc """ Checks if a session exists. ## Parameters - `session_id` - The session identifier ## Returns `true` if the session exists, `false` otherwise. """ @spec session_exists?(String.t()) :: boolean() def session_exists?(session_id) when is_binary(session_id) do session_exists?(__MODULE__, session_id) end @spec session_exists?(GenServer.server(), String.t()) :: boolean() def session_exists?(server, session_id) when is_binary(session_id) do GenServer.call(server, {:session_exists, session_id}) end @doc """ Stores worker-session affinity mapping. """ @spec store_worker_session(String.t(), String.t()) :: :ok def store_worker_session(session_id, worker_id) do GenServer.call(__MODULE__, {:upsert_worker_session, session_id, worker_id}) end ## GenServer Callbacks @impl true def init(opts) do table_name = Keyword.get(opts, :table_name, @default_table_name) table = :ets.new(table_name, [ :set, :protected, :named_table, {:read_concurrency, true}, {:write_concurrency, true}, {:decentralized_counters, true} ]) cleanup_interval = Keyword.get(opts, :cleanup_interval, Defaults.session_cleanup_interval()) default_ttl = Keyword.get(opts, :default_ttl, Defaults.session_default_ttl()) quota_config = Application.get_env(:snakepit, :session_store, %{}) max_sessions = resolve_quota(opts, quota_config, :max_sessions, Defaults.session_max_sessions()) # Strict mode for dev/test - warns loudly on session accumulation strict_mode = Keyword.get(opts, :strict_mode, Map.get(quota_config, :strict_mode, false)) Process.send_after(self(), :cleanup_expired_sessions, cleanup_interval) state = %{ table: table, table_name: table_name, cleanup_interval: cleanup_interval, default_ttl: default_ttl, max_sessions: max_sessions, strict_mode: strict_mode, stats: %{ sessions_created: 0, sessions_deleted: 0, sessions_expired: 0, cleanup_runs: 0 } } SLog.info(@log_category, "SessionStore started with table #{table}") {:ok, state} end @impl true def handle_call({:create_session, session_id, opts}, _from, state) do opts = Keyword.put_new(opts, :ttl, state.default_ttl) session = Session.new(session_id, opts) with :ok <- Session.validate(session), :ok <- check_session_quota(state) do insert_new_session(session_id, session, state) else {:error, reason} -> {:reply, {:error, reason}, state} end end @impl true def handle_call({:update_session, session_id, update_fn}, _from, state) do case :ets.lookup(state.table, session_id) do [{^session_id, {_last_accessed, _ttl, session}}] -> try do updated_session = update_fn.(session) case Session.validate(updated_session) do :ok -> touched_session = Session.touch(updated_session) ets_record = {session_id, {touched_session.last_accessed, touched_session.ttl, touched_session}} :ets.insert(state.table, ets_record) {:reply, {:ok, touched_session}, state} {:error, reason} -> {:reply, {:error, reason}, state} end rescue error -> SLog.error(@log_category, "Error updating session #{session_id}: #{inspect(error)}") {:reply, {:error, {:update_failed, error}}, state} end [] -> {:reply, {:error, :not_found}, state} end end @impl true def handle_call(:cleanup_expired_sessions, _from, state) do {expired_count, new_stats} = do_cleanup_expired_sessions(state.table, state.stats) {:reply, expired_count, %{state | stats: new_stats}} end @impl true def handle_call(:get_stats, _from, state) do current_sessions = :ets.info(state.table, :size) memory_usage = :ets.info(state.table, :memory) * :erlang.system_info(:wordsize) stats = Map.merge(state.stats, %{ current_sessions: current_sessions, memory_usage_bytes: memory_usage, table_info: :ets.info(state.table) }) {:reply, stats, state} end @impl true def handle_call({:get_session, session_id}, _from, state) do case :ets.lookup(state.table, session_id) do [{^session_id, {_last_accessed, _ttl, session}}] -> touched_session = Session.touch(session) ets_record = {session_id, {touched_session.last_accessed, touched_session.ttl, touched_session}} :ets.insert(state.table, ets_record) {:reply, {:ok, touched_session}, state} [] -> {:reply, {:error, :not_found}, state} end end @impl true def handle_call({:delete_session, session_id}, _from, state) do :ets.delete(state.table, session_id) new_stats = Map.update(state.stats, :sessions_deleted, 1, &(&1 + 1)) {:reply, :ok, %{state | stats: new_stats}} end @impl true def handle_call(:list_sessions, _from, state) do session_ids = :ets.select(state.table, [{{:"$1", :_}, [], [:"$1"]}]) {:reply, session_ids, state} end @impl true def handle_call({:session_exists, session_id}, _from, state) do exists = case :ets.lookup(state.table, session_id) do [{^session_id, _}] -> true [] -> false end {:reply, exists, state} end @impl true def handle_call({:upsert_worker_session, session_id, worker_id}, _from, state) do case :ets.lookup(state.table, session_id) do [{^session_id, {_last_accessed, _ttl, session}}] -> updated_session = session |> Map.put(:last_worker_id, worker_id) |> Session.touch() ets_record = {session_id, {updated_session.last_accessed, updated_session.ttl, updated_session}} :ets.insert(state.table, ets_record) {:reply, :ok, state} [] -> opts = [ttl: state.default_ttl] session = Session.new(session_id, opts) |> Map.put(:last_worker_id, worker_id) case Session.validate(session) do :ok -> ets_record = {session_id, {session.last_accessed, session.ttl, session}} :ets.insert(state.table, ets_record) new_stats = Map.update(state.stats, :sessions_created, 1, &(&1 + 1)) {:reply, :ok, %{state | stats: new_stats}} {:error, reason} -> SLog.warning( @log_category, "Failed to validate session for worker affinity: #{inspect(reason)}" ) {:reply, :ok, state} end end end @impl true def handle_info(:cleanup_expired_sessions, state) do {expired_count, new_stats} = do_cleanup_expired_sessions(state.table, state.stats) # Emit telemetry for session pruning if expired_count > 0 do emit_cleanup_telemetry(state, expired_count, :ttl) end # Check session accumulation thresholds and warn if needed maybe_warn_session_accumulation(state) Process.send_after(self(), :cleanup_expired_sessions, state.cleanup_interval) {:noreply, %{state | stats: new_stats}} end @impl true def handle_info(msg, state) do SLog.warning(@log_category, "SessionStore received unexpected message: #{inspect(msg)}") {:noreply, state} end ## Private Functions defp do_cleanup_expired_sessions(table, stats) do current_time = System.monotonic_time(:second) match_spec = [ {{:_, {:"$1", :"$2", :_}}, [ {:<, {:+, :"$1", :"$2"}, current_time} ], [true]} ] expired_count = :ets.select_delete(table, match_spec) if expired_count > 0 do SLog.debug( @log_category, "Cleaned up #{expired_count} expired sessions using high-performance select_delete" ) end new_stats = stats |> Map.update(:sessions_expired, expired_count, &(&1 + expired_count)) |> Map.update(:cleanup_runs, 1, &(&1 + 1)) {expired_count, new_stats} end defp resolve_quota(opts, config, key, default) do value = Keyword.get(opts, key, Map.get(config, key, default)) normalize_quota(value, default) end defp normalize_quota(:infinity, _default), do: :infinity defp normalize_quota(value, _default) when is_integer(value) and value > 0, do: value defp normalize_quota(value, default) when is_binary(value) do case Integer.parse(value) do {int, _} when int > 0 -> int _ -> default end end defp normalize_quota(_value, default), do: default defp session_quota_reached?(%{max_sessions: :infinity}), do: false defp session_quota_reached?(state) do :ets.info(state.table, :size) >= state.max_sessions end defp check_session_quota(state) do if session_quota_reached?(state) do {:error, :session_quota_exceeded} else :ok end end defp insert_new_session(session_id, session, state) do ets_record = {session_id, {session.last_accessed, session.ttl, session}} case :ets.insert_new(state.table, ets_record) do true -> SLog.debug(@log_category, "Created new session: #{session_id}") new_stats = Map.update(state.stats, :sessions_created, 1, &(&1 + 1)) {:reply, {:ok, session}, %{state | stats: new_stats}} false -> SLog.debug( @log_category, "Session #{session_id} already exists - reusing (concurrent init)" ) [{^session_id, {_last_accessed, _ttl, existing_session}}] = :ets.lookup(state.table, session_id) {:reply, {:ok, existing_session}, state} end end # Telemetry emission for session pruning events defp emit_cleanup_telemetry(state, count, reason) do current_sessions = :ets.info(state.table, :size) :telemetry.execute( [:snakepit, :bridge, :session, :pruned], %{ count: count, remaining_sessions: current_sessions, system_time: System.system_time() }, %{ reason: reason, table_name: state.table_name, max_sessions: state.max_sessions } ) end # Warn on session accumulation in strict mode or when approaching limits defp maybe_warn_session_accumulation(state) do current_sessions = :ets.info(state.table, :size) max_sessions = state.max_sessions cond do max_sessions == :infinity -> :ok current_sessions >= max_sessions -> emit_accumulation_warning(state, current_sessions, :quota_exceeded) if state.strict_mode do SLog.warning( @log_category, """ [STRICT MODE] Session quota exceeded! Current: #{current_sessions}, Max: #{max_sessions} Sessions are being rejected. Check for session leaks. """ ) end current_sessions >= trunc(max_sessions * Defaults.session_warning_threshold()) -> emit_accumulation_warning(state, current_sessions, :threshold_warning) if state.strict_mode do utilization = Float.round(current_sessions / max_sessions * 100, 1) SLog.warning( @log_category, """ [STRICT MODE] High session count warning! Current: #{current_sessions}, Max: #{max_sessions} (#{utilization}% utilization) Consider checking for session leaks or increasing max_sessions. """ ) end true -> :ok end end defp emit_accumulation_warning(state, current_sessions, reason) do :telemetry.execute( [:snakepit, :bridge, :session, :accumulation_warning], %{ current_sessions: current_sessions, max_sessions: state.max_sessions, utilization: if(state.max_sessions != :infinity, do: current_sessions / state.max_sessions, else: 0.0 ), system_time: System.system_time() }, %{ reason: reason, table_name: state.table_name, strict_mode: state.strict_mode } ) end end