Fix cascade failure in PhoenixSignaling.Registry (#40)

* Fix cascade failure in PhoenixSignaling.Registry

Signaling processes created by get_or_create were linked to the Registry
via Signaling.new() (which calls GenServer.start_link). When any signaling
process exited with a non-normal reason (e.g. a viewer's WebRTC connection
failing), the link took down the Registry, which sent EXIT signals to
every other linked signaling process, killing all active streams.

Changes:
- Unlink signaling processes from the Registry and monitor them instead
- Clean up dead entries from the signaling map via :DOWN handler
- Use get/1 instead of get!/1 in Channel.handle_in to gracefully close
  the channel when a signaling process has already exited
- Change supervisor strategy from one_for_all to one_for_one since the
  WhipRegistry and PhoenixSignaling.Registry are independent

---------

Co-authored-by: George Alexander Day <georgealexanderday@proton.me>

Author: georgeday-firefly

lib/membrane_webrtc/app.ex

@@ -8,6 +8,6 @@ defmodule Membrane.WebRTC.App do
       [{Registry, name: Membrane.WebRTC.WhipRegistry, keys: :unique}] ++
         if Code.ensure_loaded?(Phoenix), do: [Membrane.WebRTC.PhoenixSignaling.Registry], else: []
 
-    Supervisor.start_link(children, strategy: :one_for_all, name: __MODULE__.Supervisor)
+    Supervisor.start_link(children, strategy: :one_for_one, name: __MODULE__.Supervisor)
   end
 end

lib/membrane_webrtc/phoenix_signaling/channel.ex

@@ -3,18 +3,21 @@ if Code.ensure_loaded?(Phoenix) do
     @moduledoc false
     use Phoenix.Channel
     alias Membrane.WebRTC.PhoenixSignaling
+    alias Membrane.WebRTC.Signaling
 
     @impl true
     def join(signaling_id, _payload, socket) do
-      PhoenixSignaling.register_channel(signaling_id)
-      socket = assign(socket, :signaling_id, signaling_id)
+      signaling = PhoenixSignaling.Registry.get_or_create(signaling_id)
+      Process.monitor(signaling.pid)
+      Signaling.register_peer(signaling, message_format: :json_data, pid: self())
+      socket = assign(socket, signaling_id: signaling_id, signaling: signaling)
       {:ok, socket}
     end
 
     @impl true
-    def handle_in(signaling_id, msg, socket) do
+    def handle_in(_signaling_id, msg, socket) do
       msg = Jason.decode!(msg)
-      PhoenixSignaling.signal(signaling_id, msg)
+      Signaling.signal(socket.assigns.signaling, msg)
       {:noreply, socket}
     end
 
@@ -23,5 +26,10 @@ if Code.ensure_loaded?(Phoenix) do
       push(socket, socket.assigns.signaling_id, msg)
       {:noreply, socket}
     end
+
+    @impl true
+    def handle_info({:DOWN, _ref, :process, _pid, _reason}, socket) do
+      {:stop, :normal, socket}
+    end
   end
 end

lib/membrane_webrtc/phoenix_signaling/registry.ex

@@ -38,15 +38,17 @@ if Code.ensure_loaded?(Phoenix) do
 
     @impl true
     def init(_args) do
-      {:ok, %{signaling_map: %{}}}
+      {:ok, %{signaling_map: %{}, refs: %{}}}
     end
 
     @impl true
     def handle_call({:get_or_create, signaling_id}, _from, state) do
       case Map.get(state.signaling_map, signaling_id) do
         nil ->
-          signaling = Signaling.new()
+          signaling = Signaling.start()
+          ref = Process.monitor(signaling.pid)
           state = put_in(state, [:signaling_map, signaling_id], signaling)
+          state = put_in(state, [:refs, ref], signaling_id)
           {:reply, signaling, state}
 
         signaling ->
@@ -58,5 +60,17 @@ if Code.ensure_loaded?(Phoenix) do
     def handle_call({:get, signaling_id}, _from, state) do
       {:reply, Map.get(state.signaling_map, signaling_id), state}
     end
+
+    @impl true
+    def handle_info({:DOWN, ref, :process, _pid, _reason}, state) do
+      case Map.pop(state.refs, ref) do
+        {nil, _refs} ->
+          {:noreply, state}
+
+        {signaling_id, refs} ->
+          {:noreply,
+           %{state | signaling_map: Map.delete(state.signaling_map, signaling_id), refs: refs}}
+      end
+    end
   end
 end

lib/membrane_webrtc/signaling.ex

@@ -82,6 +82,18 @@ defmodule Membrane.WebRTC.Signaling do
     %__MODULE__{pid: pid}
   end
 
+  @doc """
+  Spawns Signaling GenServer and wraps it in a struct, without linking to the caller.
+
+  Be aware that the GenServer is not started under any supervision tree, so it needs to be manually
+  stopped or monitored.
+  """
+  @spec start() :: t()
+  def start() do
+    {:ok, pid} = GenServer.start(__MODULE__, [])
+    %__MODULE__{pid: pid}
+  end
+
   @doc """
   Starts and links a Signaling GenServer.