Fix memory leak by preventing stale SSE unregister dropping active heandlers (#109)

## Problem

After adding Anubis 0.17.1 I noticed that memory usage of my Phoenix
application slowly climbed over the hours until it ran out of memory and
died. It's a pre-production app with little usage and when that happened
there were just logs to connect to the MCP server.

## Solution

After adding some tracking, GPT-5.3-Codex diagnosed this as follows:

- In streamable HTTP transport, SSE handlers are keyed by session_id.
- Reconnects on the same `session_id` can overwrite map entries, and old
stream processes can become orphaned (no longer in `sse_handlers`).
- Orphaned stream loops sit in receive forever.
- `expiry_timers` map in Anubis server is not cleaned on
`:session_expired` path.

With this fix:

- adding pid-aware `unregister_sse_handler/3` so only the expected
handler can remove itself
- replacing existing handlers safely on register (demonitor old ref,
close old handler, monitor/store the new handler)
- updating plug on-close and stale-handler cleanup paths to pass the
handler pid

## Rationale

I've tested this and deployed it to my app, and memory usage has been
stable the last few hours.

<!-- This is an auto-generated comment: release notes by coderabbit.ai
-->

## Summary by CodeRabbit

## Release Notes

* **Bug Fixes**
* Enhanced Server-Sent Events (SSE) handler registration logic to
properly manage concurrent handler instances for the same session,
preventing race conditions and ensuring consistent handler state.

* **Tests**
* Added test case to verify SSE handler lifecycle behavior when multiple
handlers are registered for the same session.

<!-- end of auto-generated comment: release notes by coderabbit.ai -->

Author: nighthawk

lib/anubis/server/transport/streamable_http.ex

@@ -192,9 +192,9 @@ defmodule Anubis.Server.Transport.StreamableHTTP do
   @doc """
   Unregisters the SSE handler for a session. Called when the SSE connection closes.
   """
-  @spec unregister_sse_handler(GenServer.server(), String.t()) :: :ok
-  def unregister_sse_handler(transport, session_id) do
-    GenServer.cast(transport, {:unregister_sse_handler, session_id})
+  @spec unregister_sse_handler(GenServer.server(), String.t(), pid() | nil) :: :ok
+  def unregister_sse_handler(transport, session_id, expected_pid \\ nil) do
+    GenServer.cast(transport, {:unregister_sse_handler, session_id, expected_pid})
   end
 
   @doc """
@@ -251,9 +251,23 @@ defmodule Anubis.Server.Transport.StreamableHTTP do
 
   @impl GenServer
   def handle_call({:register_sse_handler, session_id, pid}, _from, state) do
-    ref = Process.monitor(pid)
+    sse_handlers =
+      case Map.get(state.sse_handlers, session_id) do
+        {^pid, old_ref} ->
+          Process.demonitor(old_ref, [:flush])
+          state.sse_handlers
+
+        {old_pid, old_ref} ->
+          Process.demonitor(old_ref, [:flush])
+          send(old_pid, :close_sse)
+          state.sse_handlers
+
+        nil ->
+          state.sse_handlers
+      end
 
-    sse_handlers = Map.put(state.sse_handlers, session_id, {pid, ref})
+    ref = Process.monitor(pid)
+    sse_handlers = Map.put(sse_handlers, session_id, {pid, ref})
 
     Logging.transport_event("sse_handler_registered", %{
       session_id: session_id,
@@ -299,8 +313,16 @@ defmodule Anubis.Server.Transport.StreamableHTTP do
 
   @impl GenServer
   def handle_cast({:unregister_sse_handler, session_id}, state) do
+    handle_cast({:unregister_sse_handler, session_id, nil}, state)
+  end
+
+  @impl GenServer
+  def handle_cast({:unregister_sse_handler, session_id, expected_pid}, state) do
     sse_handlers =
       case Map.get(state.sse_handlers, session_id) do
+        {pid, _ref} when is_pid(expected_pid) and pid != expected_pid ->
+          state.sse_handlers
+
         {_pid, ref} ->
           Process.demonitor(ref, [:flush])
           Map.delete(state.sse_handlers, session_id)

lib/anubis/server/transport/streamable_http/plug.ex

@@ -275,7 +275,7 @@ if Code.ensure_loaded?(Plug) do
           |> send_resp(202, "{}")
 
         handler_pid ->
-          StreamableHTTP.unregister_sse_handler(transport, session_id)
+          StreamableHTTP.unregister_sse_handler(transport, session_id, handler_pid)
           establish_sse_for_request(conn, response, session_id, opts)
 
         true ->
@@ -288,6 +288,7 @@ if Code.ensure_loaded?(Plug) do
 
       case StreamableHTTP.register_sse_handler(transport, session_id) do
         :ok ->
+          handler_pid = self()
           self_pid = self()
           Task.start(fn -> send(self_pid, {:sse_message, response}) end)
 
@@ -296,7 +297,7 @@ if Code.ensure_loaded?(Plug) do
           |> Streaming.prepare_connection()
           |> Streaming.start(transport, session_id,
             on_close: fn ->
-              StreamableHTTP.unregister_sse_handler(transport, session_id)
+              StreamableHTTP.unregister_sse_handler(transport, session_id, handler_pid)
             end
           )
 
@@ -525,13 +526,14 @@ if Code.ensure_loaded?(Plug) do
 
     defp start_sse_streaming(conn, params) do
       %{transport: transport, session_id: session_id, session_header: session_header} = params
+      handler_pid = self()
 
       conn
       |> put_resp_header(session_header, session_id)
       |> Streaming.prepare_connection()
       |> Streaming.start(transport, session_id,
         on_close: fn ->
-          StreamableHTTP.unregister_sse_handler(transport, session_id)
+          StreamableHTTP.unregister_sse_handler(transport, session_id, handler_pid)
         end
       )
     end

test/anubis/server/transport/streamable_http_test.exs

@@ -47,6 +47,46 @@ defmodule Anubis.Server.Transport.StreamableHTTPTest do
       refute StreamableHTTP.get_sse_handler(transport, session_id)
     end
 
+    test "stale unregister cannot remove a newer handler", %{transport: transport} do
+      session_id = "test-session-race"
+      test_pid = self()
+
+      old_handler =
+        spawn(fn ->
+          :ok = StreamableHTTP.register_sse_handler(transport, session_id)
+          send(test_pid, {:registered, self()})
+
+          receive do
+            :stop -> :ok
+          end
+        end)
+
+      assert_receive {:registered, ^old_handler}
+
+      new_handler =
+        spawn(fn ->
+          :ok = StreamableHTTP.register_sse_handler(transport, session_id)
+          send(test_pid, {:registered, self()})
+
+          receive do
+            :stop -> :ok
+          end
+        end)
+
+      assert_receive {:registered, ^new_handler}
+      assert ^new_handler = StreamableHTTP.get_sse_handler(transport, session_id)
+
+      # Simulate delayed close from old SSE connection.
+      assert :ok = StreamableHTTP.unregister_sse_handler(transport, session_id, old_handler)
+      assert ^new_handler = StreamableHTTP.get_sse_handler(transport, session_id)
+
+      assert :ok = StreamableHTTP.unregister_sse_handler(transport, session_id, new_handler)
+      refute StreamableHTTP.get_sse_handler(transport, session_id)
+
+      send(old_handler, :stop)
+      send(new_handler, :stop)
+    end
+
     test "routes messages to sessions", %{transport: transport} do
       session_id = "test-session-789"