Fix OOM in Runner Node killing FLAME Pool

lib/flame/pool.ex

@@ -497,7 +497,7 @@ defmodule FLAME.Pool do
         {:noreply, replace_caller(state, ref, caller_pid, child_pids)}
 
       reason when reason in [:ok, :timeout, :catch] ->
-        {:noreply, checkin_runner(state, ref, caller_pid, reason)}
+        {:noreply, checkin_runner(state, ref, caller_pid)}
     end
   end
 
@@ -570,21 +570,19 @@ defmodule FLAME.Pool do
     %{state | callers: new_callers}
   end
 
-  defp checkin_runner(state, ref, caller_pid, reason)
+  defp checkin_runner(state, ref, caller_pid)
        when is_reference(ref) and is_pid(caller_pid) do
     case state.callers do
       %{^caller_pid => %Caller{checkout_ref: ^ref} = caller} ->
         Process.demonitor(caller.monitor_ref, [:flush])
         drop_caller(state, caller_pid, caller)
 
-      # the only way to race a checkin is if the caller has expired while still in the
-      # waiting state and checks in on the timeout before we lease it a runner.
-      %{} when reason == :timeout ->
-        maybe_drop_waiting(state, caller_pid)
-
+      # A checkin can race an already-dropped caller. This is benign and must not
+      # crash the pool, which owns the named ETS meta table and whose death would
+      # cascade `lookup_meta/1` and `:noproc` failures to every other caller
+      # as shown in https://github.com/phoenixframework/flame/issues/66.
       %{} ->
-        raise ArgumentError,
-              "expected to checkin runner for #{inspect(caller_pid)} that does not exist"
+        maybe_drop_waiting(state, caller_pid)
     end
   end
 

test/flame_test.exs

@@ -641,4 +641,36 @@ defmodule FLAME.FLAMETest do
     Supervisor.stop(pool_pid)
     refute File.exists?(artifact)
   end
+
+  # Reproduces the runner OOM race from https://github.com/phoenixframework/flame/issues/66.
+  @tag runner: [min: 1, max: 1, max_concurrency: 2]
+  test "runner OOM mid-call does not crash the pool", %{runner_sup: runner_sup} = config do
+    ExUnit.CaptureLog.capture_log(fn ->
+      pool = Process.whereis(config.test)
+      pool_ref = Process.monitor(pool)
+
+      {:ok, caller} = sim_long_running(config.test, 500)
+      Process.unlink(caller)
+      caller_ref = Process.monitor(caller)
+
+      assert [{:undefined, runner, :worker, [FLAME.Runner]}] =
+               Supervisor.which_children(runner_sup)
+
+      # simulate the OOM 100ms into processing
+      Process.sleep(100)
+      Process.exit(runner, :brutal_kill)
+
+      # the pool drops and kills the caller from the runner `:DOWN` first, which
+      # guarantees the caller is no longer tracked once the stray cancel arrives
+      assert_receive {:DOWN, ^caller_ref, :process, ^caller, :killed}, 1000
+
+      # ...then the caller's in-flight `:catch` cancel lands for the now-unknown caller
+      send(pool, {:cancel, make_ref(), caller, :catch})
+
+      # the pool must stay up (same pid) and keep serving work
+      refute_receive {:DOWN, ^pool_ref, :process, ^pool, _}, 500
+      assert Process.alive?(pool)
+      assert FLAME.call(config.test, fn -> :works end) == :works
+    end)
+  end
 end