diff --git a/lib/flame/pool.ex b/lib/flame/pool.ex index 02ee7ca..3c04bfc 100644 --- a/lib/flame/pool.ex +++ b/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 diff --git a/test/flame_test.exs b/test/flame_test.exs index 458d4db..8954121 100644 --- a/test/flame_test.exs +++ b/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