From bca05de08c1ee9666c928c2c1720a0957800d4c9 Mon Sep 17 00:00:00 2001 From: Sai Asish Y Date: Wed, 17 Jun 2026 18:43:43 -0700 Subject: [PATCH 1/3] Handle :EXIT messages in ingestion WriteBuffer to prevent crash Signed-off-by: Sai Asish Y --- lib/plausible/ingestion/write_buffer.ex | 7 ++++++ .../plausible/ingestion/write_buffer_test.exs | 25 +++++++++++++++++++ 2 files changed, 32 insertions(+) create mode 100644 test/plausible/ingestion/write_buffer_test.exs diff --git a/lib/plausible/ingestion/write_buffer.ex b/lib/plausible/ingestion/write_buffer.ex index 8fc6b802f8c9..abb9bf7bf02a 100644 --- a/lib/plausible/ingestion/write_buffer.ex +++ b/lib/plausible/ingestion/write_buffer.ex @@ -66,6 +66,13 @@ defmodule Plausible.Ingestion.WriteBuffer do {:noreply, %{state | buffer: [], buffer_size: 0, timer: timer}} end + def handle_info({:EXIT, _from, _reason}, state) do + # We trap exits so terminate/2 can flush on shutdown. Linked ports and + # processes (e.g. the ClickHouse connection socket) deliver :EXIT messages + # here on disconnect, which we ignore to keep the buffer alive. + {:noreply, state} + end + @impl true def handle_call(:flush, _from, state) do %{timer: timer, flush_interval_ms: flush_interval_ms} = state diff --git a/test/plausible/ingestion/write_buffer_test.exs b/test/plausible/ingestion/write_buffer_test.exs new file mode 100644 index 000000000000..37362dcf83f3 --- /dev/null +++ b/test/plausible/ingestion/write_buffer_test.exs @@ -0,0 +1,25 @@ +defmodule Plausible.Ingestion.WriteBufferTest do + use ExUnit.Case, async: true + + alias Plausible.Ingestion.WriteBuffer + + @opts WriteBuffer.compile_time_prepare(Plausible.ClickhouseEventV2) + + test "survives :EXIT messages from linked processes while trapping exits" do + opts = + @opts + |> Map.take([:header, :insert_sql, :insert_opts]) + |> Map.to_list() + |> Keyword.put(:name, :"write_buffer_exit_#{System.unique_integer([:positive])}") + # large buffer so the empty buffer never flushes to ClickHouse during the test + |> Keyword.put(:flush_interval_ms, 60_000) + + {:ok, pid} = WriteBuffer.start_link(opts) + + send(pid, {:EXIT, self(), :normal}) + + # process is still alive and responsive after the :EXIT message + assert Process.alive?(pid) + assert WriteBuffer.flush(opts[:name]) == :ok + end +end From da23d412ee8f3f5297cbb62e8536b7e7c44d6e25 Mon Sep 17 00:00:00 2001 From: Sai Asish Y Date: Thu, 2 Jul 2026 02:09:41 -0700 Subject: [PATCH 2/3] test: trigger real linked-process :EXIT in write buffer test Signed-off-by: Sai Asish Y --- .../plausible/ingestion/write_buffer_test.exs | 19 +++++++++++++++---- 1 file changed, 15 insertions(+), 4 deletions(-) diff --git a/test/plausible/ingestion/write_buffer_test.exs b/test/plausible/ingestion/write_buffer_test.exs index 37362dcf83f3..673e100f3ba5 100644 --- a/test/plausible/ingestion/write_buffer_test.exs +++ b/test/plausible/ingestion/write_buffer_test.exs @@ -5,20 +5,31 @@ defmodule Plausible.Ingestion.WriteBufferTest do @opts WriteBuffer.compile_time_prepare(Plausible.ClickhouseEventV2) - test "survives :EXIT messages from linked processes while trapping exits" do + test "keeps running when a linked process exits while trapping exits" do opts = @opts |> Map.take([:header, :insert_sql, :insert_opts]) |> Map.to_list() |> Keyword.put(:name, :"write_buffer_exit_#{System.unique_integer([:positive])}") - # large buffer so the empty buffer never flushes to ClickHouse during the test + # large interval so the empty buffer never flushes to ClickHouse during the test |> Keyword.put(:flush_interval_ms, 60_000) {:ok, pid} = WriteBuffer.start_link(opts) - send(pid, {:EXIT, self(), :normal}) + # The buffer traps exits, so a linked process exiting delivers a real + # {:EXIT, from, reason} message to handle_info/2. Linking a throwaway + # process to the buffer and killing it reproduces the crash in #6382 + # without sending a synthetic message. + child = + spawn(fn -> + Process.link(pid) + Process.sleep(:infinity) + end) + + ref = Process.monitor(child) + Process.exit(child, :shutdown) + assert_receive {:DOWN, ^ref, :process, ^child, _}, 1_000 - # process is still alive and responsive after the :EXIT message assert Process.alive?(pid) assert WriteBuffer.flush(opts[:name]) == :ok end From 22dfd390fdf6571ea2298363506fb8dedd878b31 Mon Sep 17 00:00:00 2001 From: Sai Asish Y Date: Fri, 10 Jul 2026 13:01:20 -0700 Subject: [PATCH 3/3] Log EXIT reason in write buffer and assert it in test Signed-off-by: Sai Asish Y --- lib/plausible/ingestion/write_buffer.ex | 6 ++-- .../plausible/ingestion/write_buffer_test.exs | 36 ++++++++++++------- 2 files changed, 25 insertions(+), 17 deletions(-) diff --git a/lib/plausible/ingestion/write_buffer.ex b/lib/plausible/ingestion/write_buffer.ex index abb9bf7bf02a..26ace34dd0fe 100644 --- a/lib/plausible/ingestion/write_buffer.ex +++ b/lib/plausible/ingestion/write_buffer.ex @@ -66,10 +66,8 @@ defmodule Plausible.Ingestion.WriteBuffer do {:noreply, %{state | buffer: [], buffer_size: 0, timer: timer}} end - def handle_info({:EXIT, _from, _reason}, state) do - # We trap exits so terminate/2 can flush on shutdown. Linked ports and - # processes (e.g. the ClickHouse connection socket) deliver :EXIT messages - # here on disconnect, which we ignore to keep the buffer alive. + def handle_info({:EXIT, _from, reason}, state) do + Logger.warning("#{state.name} received EXIT, keeping buffer alive: #{inspect(reason)}") {:noreply, state} end diff --git a/test/plausible/ingestion/write_buffer_test.exs b/test/plausible/ingestion/write_buffer_test.exs index 673e100f3ba5..226a6c44b543 100644 --- a/test/plausible/ingestion/write_buffer_test.exs +++ b/test/plausible/ingestion/write_buffer_test.exs @@ -1,18 +1,22 @@ defmodule Plausible.Ingestion.WriteBufferTest do use ExUnit.Case, async: true + import ExUnit.CaptureLog + alias Plausible.Ingestion.WriteBuffer - @opts WriteBuffer.compile_time_prepare(Plausible.ClickhouseEventV2) + @compile_time_opts WriteBuffer.compile_time_prepare(Plausible.ClickhouseEventV2) - test "keeps running when a linked process exits while trapping exits" do - opts = - @opts - |> Map.take([:header, :insert_sql, :insert_opts]) - |> Map.to_list() - |> Keyword.put(:name, :"write_buffer_exit_#{System.unique_integer([:positive])}") - # large interval so the empty buffer never flushes to ClickHouse during the test - |> Keyword.put(:flush_interval_ms, 60_000) + test "keeps running and logs the reason when a linked process exits", %{test: test} do + opts = [ + header: @compile_time_opts.header, + insert_sql: @compile_time_opts.insert_sql, + insert_opts: @compile_time_opts.insert_opts, + # unique atom is at our disposal already + name: test, + # no need to ever flush automatically on slow CI + flush_interval_ms: 1_000_000_000 + ] {:ok, pid} = WriteBuffer.start_link(opts) @@ -27,10 +31,16 @@ defmodule Plausible.Ingestion.WriteBufferTest do end) ref = Process.monitor(child) - Process.exit(child, :shutdown) - assert_receive {:DOWN, ^ref, :process, ^child, _}, 1_000 - assert Process.alive?(pid) - assert WriteBuffer.flush(opts[:name]) == :ok + log = + capture_log(fn -> + Process.exit(child, :shutdown) + assert_receive {:DOWN, ^ref, :process, ^child, _}, 1_000 + assert Process.alive?(pid) + assert WriteBuffer.flush(test) == :ok + end) + + assert log =~ "received EXIT, keeping buffer alive" + assert log =~ ":shutdown" end end