fix(telemetry_processor): wake scheduler for pending buffer deadlines

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

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,18 +187,23 @@ 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
   def handle_call(:flush, _from, %Scheduler{} = state) 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

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()