From ab15afbf4c14e83adab17e61c328856ab0ce3b62 Mon Sep 17 00:00:00 2001 From: Peter Solnica Date: Thu, 10 Sep 2026 08:25:19 +0000 Subject: [PATCH] fix(telemetry_processor): wake scheduler for pending buffer deadlines Now TelemetryProcessor schedules the next wake-up from pending buffer deadlines, delivering partial batches without waiting for new items. It also continues processing ready batches across priority cycles and resumes buffered items when transport capacity becomes available. --- lib/sentry/telemetry/buffer.ex | 27 ++++-- lib/sentry/telemetry/scheduler.ex | 31 +++++-- test/sentry/telemetry/scheduler_test.exs | 101 +++++++++++++++++++++++ 3 files changed, 144 insertions(+), 15 deletions(-) diff --git a/lib/sentry/telemetry/buffer.ex b/lib/sentry/telemetry/buffer.ex index b281de72..cb94f394 100644 --- a/lib/sentry/telemetry/buffer.ex +++ b/lib/sentry/telemetry/buffer.ex @@ -120,6 +120,15 @@ defmodule Sentry.Telemetry.Buffer do GenServer.call(server, :category) end + @doc """ + Returns milliseconds until a pending batch is ready, or `:infinity` when + the buffer is empty or a partial batch has no timeout. + """ + @spec next_timeout(GenServer.server()) :: timeout() + def next_timeout(server) do + GenServer.call(server, :next_timeout) + end + ## GenServer Callbacks @impl true @@ -173,6 +182,10 @@ defmodule Sentry.Telemetry.Buffer do {:reply, ready_to_flush?(state), state} end + def handle_call(:next_timeout, _from, %Buffer{} = state) do + {:reply, time_until_ready(state), state} + end + def handle_call(:category, _from, %Buffer{} = state) do {:reply, state.category, state} end @@ -217,15 +230,13 @@ defmodule Sentry.Telemetry.Buffer do poll_batch(state, count - 1, [item | acc]) end - defp ready_to_flush?(%{size: 0}), do: false - - defp ready_to_flush?(%{size: size, batch_size: batch_size} = state) do - size >= batch_size or timeout_elapsed?(state) - end + defp ready_to_flush?(state), do: time_until_ready(state) == 0 - defp timeout_elapsed?(%{timeout: nil}), do: false + defp time_until_ready(%{size: 0}), do: :infinity + defp time_until_ready(%{size: size, batch_size: batch_size}) when size >= batch_size, do: 0 + defp time_until_ready(%{timeout: nil}), do: :infinity - defp timeout_elapsed?(%{timeout: timeout, last_flush_time: last_flush_time}) do - System.monotonic_time(:millisecond) - last_flush_time >= timeout + defp time_until_ready(%{timeout: timeout, last_flush_time: last_flush_time}) do + max(0, timeout - (System.monotonic_time(:millisecond) - last_flush_time)) end end diff --git a/lib/sentry/telemetry/scheduler.ex b/lib/sentry/telemetry/scheduler.ex index 29934a47..b57a45c0 100644 --- a/lib/sentry/telemetry/scheduler.ex +++ b/lib/sentry/telemetry/scheduler.ex @@ -14,9 +14,10 @@ defmodule Sentry.Telemetry.Scheduler do ## Signal-Based Wake - The scheduler sleeps until signaled via `signal/1`. When signaled, it wakes - and attempts to process items from the current buffer in the cycle. If the - buffer is not ready or the transport queue is full, it advances to the next position. + The scheduler wakes when signaled via `signal/1` or when a pending buffer's + timeout expires. It processes ready batches in priority order, sleeping until + the next buffer deadline when no batches are ready. When the transport queue + is full, it waits for a send to finish before checking buffers again. ## Transport Queue @@ -186,7 +187,7 @@ defmodule Sentry.Telemetry.Scheduler do @impl true def handle_cast(:signal, %Scheduler{} = state) do state = process_cycle(state) - {:noreply, state} + {:noreply, state, next_timeout(state)} end @impl true @@ -194,10 +195,15 @@ defmodule Sentry.Telemetry.Scheduler do state = flush_all_buffers(state) state = wait_for_active(state) state = flush_queue(state) - {:reply, :ok, state} + {:reply, :ok, state, next_timeout(state)} end @impl true + def handle_info(:timeout, %Scheduler{} = state) do + state = process_cycle(state) + {:noreply, state, next_timeout(state)} + end + def handle_info({:DOWN, ref, :process, _pid, reason}, %{active_ref: ref} = state) do if reason != :normal do LoggerUtils.log(fn -> @@ -214,11 +220,22 @@ defmodule Sentry.Telemetry.Scheduler do state = maybe_process_next(state) - {:noreply, state} + {:noreply, state, next_timeout(state)} end def handle_info({:DOWN, _ref, :process, _pid, _reason}, state) do - {:noreply, state} + {:noreply, state, next_timeout(state)} + end + + defp next_timeout(%Scheduler{} = state) do + if transport_has_space?(state) do + Enum.reduce(state.buffers, :infinity, fn {_category, buffer}, timeout -> + min(timeout, Buffer.next_timeout(buffer)) + end) + else + # A transport completion will wake us when capacity becomes available. + :infinity + end end defp process_cycle(%Scheduler{} = state) do diff --git a/test/sentry/telemetry/scheduler_test.exs b/test/sentry/telemetry/scheduler_test.exs index fca8516b..17d52aee 100644 --- a/test/sentry/telemetry/scheduler_test.exs +++ b/test/sentry/telemetry/scheduler_test.exs @@ -15,6 +15,12 @@ defmodule Sentry.Telemetry.SchedulerTest do } end + defp make_item(:log), do: make_log_event() + + defp make_item(:metric) do + %Sentry.Metric{name: "test.metric", type: :gauge, value: 42, timestamp: 1.0} + end + describe "build_priority_cycle/0" do test "builds cycle with correct weights for all categories" do cycle = Scheduler.build_priority_cycle() @@ -84,6 +90,101 @@ defmodule Sentry.Telemetry.SchedulerTest do end describe "signal/1" do + for category <- [:log, :metric] do + @category category + test "delivers a partial #{@category} batch after its timeout without another signal" do + category = @category + owner = self() + + buffer = + start_supervised!({Buffer, category: category, batch_size: 100, timeout: 200}) + + scheduler = + start_supervised!( + {Scheduler, + buffers: %{category => buffer}, + on_envelope: fn envelope -> send(owner, {:envelope, envelope}) end} + ) + + item = make_item(category) + + Buffer.add(buffer, item) + assert Buffer.size(buffer) == 1 + Scheduler.signal(scheduler) + + refute_receive {:envelope, _}, 50 + assert_receive {:envelope, _}, 1_000 + assert Buffer.size(buffer) == 0 + end + end + + test "continues processing ready batches after a full priority cycle" do + owner = self() + buffer = start_supervised!({Buffer, category: :log, batch_size: 1}) + + scheduler = + start_supervised!( + {Scheduler, + buffers: %{log: buffer}, + on_envelope: fn envelope -> send(owner, {:envelope, envelope}) end} + ) + + for i <- 1..5, do: Buffer.add(buffer, make_log_event("log_#{i}")) + assert Buffer.size(buffer) == 5 + Scheduler.signal(scheduler) + + for i <- 1..5 do + assert_receive {:envelope, envelope}, 1_000 + assert [%Sentry.LogBatch{log_events: [%LogEvent{body: body}]}] = envelope.items + assert body == "log_#{i}" + end + + assert Buffer.size(buffer) == 0 + end + + test "resumes buffered items when transport capacity becomes available" do + %{bypass: bypass, telemetry_processor: processor} = + Sentry.Test.setup_sentry( + collect_envelopes: true, + telemetry_processor: [ + transport_capacity: 1, + buffer_configs: %{metric: %{batch_size: 1}} + ] + ) + + owner = self() + + Bypass.stub(bypass, "POST", "/api/1/envelope/", fn conn -> + {:ok, body, conn} = Plug.Conn.read_body(conn) + send(owner, {:request_started, self(), body}) + + receive do + :release -> Plug.Conn.resp(conn, 200, "{}") + after + 5_000 -> Plug.Conn.resp(conn, 500, "response was not released") + end + end) + + Sentry.Metrics.gauge("first", 1) + assert_receive {:request_started, handler, _}, 1_000 + + try do + Sentry.Metrics.gauge("second", 2) + scheduler = Sentry.TelemetryProcessor.get_scheduler(processor) + assert :sys.get_state(scheduler).size == 1 + assert Sentry.TelemetryProcessor.buffer_size(processor, :metric) == 1 + refute_receive {:request_started, _, _}, 50 + after + send(handler, :release) + end + + assert_receive {:request_started, handler, body}, 1_000 + send(handler, :release) + + assert [%{"items" => [%{"name" => "second"}]}] = + extract_metric_items([decode_envelope!(body)]) + end + test "wakes scheduler to process log items" do buffers = start_test_buffers(batch_size: 1) test_pid = self()