From 9c5eca4a528c74d861a91afc06e4b64ab583d663 Mon Sep 17 00:00:00 2001
From: Neel Shah <neel.shah@sentry.io>
Date: Mon, 14 Sep 2026 16:42:02 +0200
Subject: [PATCH] fix(metrics, logs): Add a ConditionVariable to
 TelemetryEventBuffer so that both periodic and size based flushing happen on
 the buffer thread

---
 .../lib/sentry/telemetry_event_buffer.rb      |  44 +++--
 sentry-ruby/lib/sentry/test_helper.rb         |   4 +
 .../lib/sentry/threaded_periodic_worker.rb    |  76 ++++++--
 .../spec/isolated/sentry_logger_spec.rb       |  19 --
 .../spec/isolated/std_lib_logger_spec.rb      |  20 ++-
 .../spec/sentry/backpressure_monitor_spec.rb  |   6 +-
 .../spec/sentry/session_flusher_spec.rb       |   4 +-
 .../sentry/threaded_periodic_worker_spec.rb   | 162 ++++++++++++++++++
 ...ed_examples_for_telemetry_event_buffers.rb |  30 +++-
 9 files changed, 308 insertions(+), 57 deletions(-)
 create mode 100644 sentry-ruby/spec/sentry/threaded_periodic_worker_spec.rb

diff --git a/sentry-ruby/lib/sentry/telemetry_event_buffer.rb b/sentry-ruby/lib/sentry/telemetry_event_buffer.rb
index 73dcf59f5..0ab1f1355 100644
--- a/sentry-ruby/lib/sentry/telemetry_event_buffer.rb
+++ b/sentry-ruby/lib/sentry/telemetry_event_buffer.rb
@@ -41,30 +41,26 @@ def initialize(configuration, client, event_class:, max_items:, max_items_before
     end
 
     def flush
-      pending_items = @mutex.synchronize do
-        next if @pending_items.empty?
-
-        items = @pending_items
-        @pending_items = []
-        items
-      end
-
-      return unless pending_items
+      wait_until_idle
+      result = flush_pending_items
+      wait_until_idle
+      result
+    end
 
-      log_debug("[#{self.class}] flushing #{pending_items.size} #{@event_class}")
-      send_items(pending_items)
-      self
+    def run
+      flush_pending_items
     end
-    alias_method :run, :flush
 
     def add_item(item)
+      # the buffer thread can never add telemetry itself to prevent recursion
+      return self if Thread.current == thread
       # Prevent ThreadError from re-entrant locking (e.g. transport instrumentation calling Sentry.metrics.*)
       return self if @mutex.owned?
 
+      return unless ensure_thread
+
       dropped = false
       size_exceeded = @mutex.synchronize do
-        return unless ensure_thread
-
         if size >= @max_items_before_drop
           dropped = true
         else
@@ -83,7 +79,7 @@ def add_item(item)
         )
       end
 
-      flush if size_exceeded
+      wake if size_exceeded
       self
     end
 
@@ -101,6 +97,22 @@ def clear!
 
     private
 
+    def flush_pending_items
+      pending_items = @mutex.synchronize do
+        next if @pending_items.empty?
+
+        items = @pending_items
+        @pending_items = []
+        items
+      end
+
+      return unless pending_items
+
+      log_debug("[#{self.class}] flushing #{pending_items.size} #{@event_class}")
+      send_items(pending_items)
+      self
+    end
+
     def send_items(pending_items)
       envelope = Envelope.new(sent_at: Sentry.utc_now.iso8601)
 
diff --git a/sentry-ruby/lib/sentry/test_helper.rb b/sentry-ruby/lib/sentry/test_helper.rb
index d3bfbf8b6..0273ade81 100644
--- a/sentry-ruby/lib/sentry/test_helper.rb
+++ b/sentry-ruby/lib/sentry/test_helper.rb
@@ -127,6 +127,8 @@ def sentry_envelopes
     end
 
     def sentry_logs
+      Sentry.get_current_client&.log_event_buffer&.flush
+
       sentry_envelopes
         .flat_map(&:items)
         .select { |item| item.headers[:type] == "log" }
@@ -134,6 +136,8 @@ def sentry_logs
     end
 
     def sentry_metrics
+      Sentry.get_current_client&.metric_event_buffer&.flush
+
       sentry_envelopes
         .flat_map(&:items)
         .select { |item| item.headers[:type] == "trace_metric" }
diff --git a/sentry-ruby/lib/sentry/threaded_periodic_worker.rb b/sentry-ruby/lib/sentry/threaded_periodic_worker.rb
index 2ec35c108..8144e9213 100644
--- a/sentry-ruby/lib/sentry/threaded_periodic_worker.rb
+++ b/sentry-ruby/lib/sentry/threaded_periodic_worker.rb
@@ -4,41 +4,91 @@ module Sentry
   class ThreadedPeriodicWorker
     include LoggingHelper
 
+    attr_reader :thread
+
     def initialize(sdk_logger, interval)
       @thread = nil
       @exited = false
       @interval = interval
       @sdk_logger = sdk_logger
+
+      @thread_mutex = Mutex.new
+      @wake_condition = ConditionVariable.new
+      @idle_condition = ConditionVariable.new
+
+      @woken = false
+      @running = false
     end
 
     def ensure_thread
-      return false if @exited
-      return true if @thread&.alive?
+      @thread_mutex.synchronize do
+        return false if @exited
+        return true if @thread&.alive?
 
-      @thread = Thread.new do
-        loop do
-          sleep(@interval)
-          run
-        end
-      end
+        @thread = Thread.new { worker_loop }
 
-      true
+        true
+      end
     rescue ThreadError
+      @thread_mutex.synchronize { @exited = true }
       log_debug("[#{self.class.name}] thread creation failed")
-      @exited = true
       false
     end
 
+    def wake
+      @thread_mutex.synchronize do
+        next false if @exited
+
+        @woken = true
+        @wake_condition.signal
+        true
+      end
+    end
+
+    def wait_until_idle
+      @thread_mutex.synchronize do
+        @idle_condition.wait(@thread_mutex) while !@exited && (@running || @woken)
+      end
+    end
+
     def kill
-      @exited = true
+      thread = @thread_mutex.synchronize do
+        @exited = true
+        @woken = false
+        @idle_condition.broadcast
+        @thread
+      end
 
       # Only a started worker has a thread to kill (and to log about).
       # Guarding here keeps a never-started worker's teardown silent, so
       # killing one during test reset can't emit a stray debug line.
-      return unless @thread
+      return unless thread
 
       log_debug("[#{self.class.name}] thread killed")
-      @thread.kill
+      thread.kill
+    end
+
+    private
+
+    def worker_loop
+      loop do
+        @thread_mutex.synchronize do
+          @wake_condition.wait(@thread_mutex, @interval) unless @woken
+          @woken = false
+          @running = true
+        end
+
+        begin
+          run
+        rescue Exception => e
+          log_error("[#{self.class.name}] run failed", e)
+        ensure
+          @thread_mutex.synchronize do
+            @running = false
+            @idle_condition.broadcast
+          end
+        end
+      end
     end
   end
 end
diff --git a/sentry-ruby/spec/isolated/sentry_logger_spec.rb b/sentry-ruby/spec/isolated/sentry_logger_spec.rb
index 78a92c535..e9877018e 100644
--- a/sentry-ruby/spec/isolated/sentry_logger_spec.rb
+++ b/sentry-ruby/spec/isolated/sentry_logger_spec.rb
@@ -7,7 +7,6 @@
     perform_basic_setup do |config|
       config.breadcrumbs_logger = [:sentry_logger]
       config.max_log_events = 1
-      config.enabled_patches = [:logger]
     end
   end
 
@@ -104,22 +103,4 @@
       end
     end
   end
-
-  it "does not conflict with :logger patch" do
-    logger = ::Logger.new(nil)
-
-    logger.info("Hello World")
-
-    expect(sentry_logs).to_not be_empty
-
-    log_event = sentry_logs.last
-
-    expect(log_event[:level]).to eql("info")
-    expect(log_event[:body]).to eql("Hello World")
-
-    breadcrumb = breadcrumbs.peek
-
-    expect(breadcrumb.level).to eq("info")
-    expect(breadcrumb.message).to eq("Hello World")
-  end
 end
diff --git a/sentry-ruby/spec/isolated/std_lib_logger_spec.rb b/sentry-ruby/spec/isolated/std_lib_logger_spec.rb
index 5e43b7537..28de1283c 100644
--- a/sentry-ruby/spec/isolated/std_lib_logger_spec.rb
+++ b/sentry-ruby/spec/isolated/std_lib_logger_spec.rb
@@ -2,7 +2,7 @@
 
 SimpleCov.command_name "StdLibLogger"
 
-RSpec.describe Sentry::StdLibLogger do
+RSpec.describe Sentry::StdLibLogger, order: :defined do
   let(:logger) { ::Logger.new($stdout) }
 
   context "when logger patch is not enabled" do
@@ -224,5 +224,23 @@
         end
       end
     end
+
+    it "does not conflict with the Sentry logger patch" do
+      Sentry.configuration.breadcrumbs_logger = [:sentry_logger]
+      logger = ::Logger.new(nil)
+      logger.info("Hello World")
+
+      expect(sentry_logs).to_not be_empty
+
+      log_event = sentry_logs.last
+
+      expect(log_event[:level]).to eql("info")
+      expect(log_event[:body]).to eql("Hello World")
+
+      breadcrumb = Sentry.get_current_scope.breadcrumbs.peek
+
+      expect(breadcrumb.level).to eq("info")
+      expect(breadcrumb.message).to eq("Hello World")
+    end
   end
 end
diff --git a/sentry-ruby/spec/sentry/backpressure_monitor_spec.rb b/sentry-ruby/spec/sentry/backpressure_monitor_spec.rb
index 77040c2ed..97b425030 100644
--- a/sentry-ruby/spec/sentry/backpressure_monitor_spec.rb
+++ b/sentry-ruby/spec/sentry/backpressure_monitor_spec.rb
@@ -30,12 +30,12 @@
 
     it 'spawns new thread' do
       expect { subject.healthy? }.to change { Thread.list.count }.by(1)
-      expect(subject.instance_variable_get(:@thread)).to be_a(Thread)
+      expect(subject.thread).to be_a(Thread)
     end
 
     it 'spawns only one thread' do
       expect { subject.healthy? }.to change { Thread.list.count }.by(1)
-      thread = subject.instance_variable_get(:@thread)
+      thread = subject.thread
       expect(thread).to receive(:alive?).and_return(true)
       expect { subject.healthy? }.to change { Thread.list.count }.by(0)
     end
@@ -109,7 +109,7 @@
   describe '#kill' do
     it 'kills the thread and logs a message' do
       subject.healthy?
-      expect(subject.instance_variable_get(:@thread)).to receive(:kill)
+      expect(subject.thread).to receive(:kill)
       subject.kill
       expect(string_io.string).to include("[#{described_class.name}] thread killed")
     end
diff --git a/sentry-ruby/spec/sentry/session_flusher_spec.rb b/sentry-ruby/spec/sentry/session_flusher_spec.rb
index af88dbd79..aa7091f6c 100644
--- a/sentry-ruby/spec/sentry/session_flusher_spec.rb
+++ b/sentry-ruby/spec/sentry/session_flusher_spec.rb
@@ -105,7 +105,7 @@
         subject.add_session(session)
       end.to change { Thread.list.count }.by(1)
 
-      expect(subject.instance_variable_get(:@thread)).to be_a(Thread)
+      expect(subject.thread).to be_a(Thread)
     end
 
     it "spawns only one thread" do
@@ -113,7 +113,7 @@
         subject.add_session(session)
       end.to change { Thread.list.count }.by(1)
 
-      thread = subject.instance_variable_get(:@thread)
+      thread = subject.thread
       expect(thread).to receive(:alive?).and_return(true)
 
       expect do
diff --git a/sentry-ruby/spec/sentry/threaded_periodic_worker_spec.rb b/sentry-ruby/spec/sentry/threaded_periodic_worker_spec.rb
new file mode 100644
index 000000000..ad89947df
--- /dev/null
+++ b/sentry-ruby/spec/sentry/threaded_periodic_worker_spec.rb
@@ -0,0 +1,162 @@
+# frozen_string_literal: true
+
+RSpec.describe Sentry::ThreadedPeriodicWorker do
+  let(:worker_class) do
+    Class.new(described_class) do
+      attr_reader :runs
+      attr_writer :run_block
+
+      def initialize(*args)
+        super
+        @runs = Queue.new
+      end
+
+      def run
+        @run_block ? @run_block.call : @runs << Thread.current
+      end
+    end
+  end
+
+  let(:interval) { 60 }
+  let(:logger_output) { StringIO.new }
+  let(:worker) { worker_class.new(Logger.new(logger_output), interval) }
+
+  after do
+    worker.kill
+  end
+
+  describe "#initialize" do
+    it "does not start a thread just by initialization" do
+      expect(worker.thread).to be_nil
+    end
+  end
+
+  describe "#ensure_thread" do
+    context "when the interval expires" do
+      let(:interval) { 0 }
+
+      it "runs on the worker thread" do
+        worker.ensure_thread
+
+        expect(worker.runs.pop).to eq(worker.thread)
+      end
+    end
+
+    it "creates only one thread when ensured concurrently" do
+      barrier = Queue.new
+      callers = 2.times.map do
+        Thread.new do
+          barrier.pop
+          worker.ensure_thread
+        end
+      end
+
+      expect(Thread).to receive(:new).once.and_call_original
+      callers.each { barrier << true }
+      callers.each(&:join)
+    end
+
+    it "logs run errors and continues running" do
+      started = Queue.new
+      run_count = 0
+      worker.run_block = -> {
+        run_count += 1
+        started << run_count
+        raise Exception, "boom" if run_count == 1
+      }
+
+      worker.ensure_thread
+      worker.wake
+      expect(started.pop).to eq(1)
+
+      worker.wake
+      expect(started.pop).to eq(2)
+      expect(logger_output.string).to include("run failed: boom")
+    end
+  end
+
+  describe "#wake" do
+    it "runs before the interval expires" do
+      worker.ensure_thread
+      thread = worker.thread
+
+      worker.wake
+
+      expect(worker.runs.pop).to eq(thread)
+    end
+
+    it "does nothing after the worker exits" do
+      worker.kill
+
+      expect(worker.wake).to be(false)
+    end
+  end
+
+  describe "#wait_until_idle" do
+    it "waits for a running task to finish" do
+      started = Queue.new
+      continue = Queue.new
+      completed = Queue.new
+      worker.run_block = -> {
+        started << true
+        continue.pop
+        completed << :run
+      }
+
+      worker.ensure_thread
+      worker.wake
+      started.pop
+
+      waiter_started = Queue.new
+      waiter = Thread.new do
+        waiter_started << true
+        worker.wait_until_idle
+        completed << :wait
+      end
+      waiter_started.pop
+
+      continue << true
+
+      expect(completed.pop).to eq(:run)
+      expect(completed.pop).to eq(:wait)
+      waiter.join
+    end
+
+    it "does nothing after the worker exits" do
+      started = Queue.new
+      continue = Queue.new
+      worker.run_block = -> {
+        started << true
+        continue.pop
+      }
+
+      worker.ensure_thread
+      worker.wake
+      started.pop
+
+      waiter = Thread.new { worker.wait_until_idle }
+      worker.kill
+
+      expect(waiter.join(1)).to eq(waiter)
+
+      continue << true
+      expect(worker.thread.join(1)).to eq(worker.thread)
+    end
+  end
+
+  describe "#kill" do
+    it "stops the worker thread" do
+      worker.ensure_thread
+      thread = worker.thread
+
+      worker.kill
+      thread.join
+
+      expect(thread).not_to be_alive
+
+      result = nil
+      expect { result = worker.ensure_thread }.not_to change { Thread.list.count }
+      expect(result).to be(false)
+    end
+  end
+end
diff --git a/sentry-ruby/spec/support/shared_examples_for_telemetry_event_buffers.rb b/sentry-ruby/spec/support/shared_examples_for_telemetry_event_buffers.rb
index 3dd1dcd22..2ecc479f1 100644
--- a/sentry-ruby/spec/support/shared_examples_for_telemetry_event_buffers.rb
+++ b/sentry-ruby/spec/support/shared_examples_for_telemetry_event_buffers.rb
@@ -49,11 +49,16 @@
       2.times { subject.add_item(event) }
     end
 
-    it "auto-flushes pending items to the client when the number of items reaches max_items" do
-      expect(client).to receive(:send_envelope)
+    it "auto-flushes pending items to the client using the buffer thread when the number of items reaches max_items" do
+      thread = nil
+      expect(client).to receive(:send_envelope) do
+        thread = Thread.current
+      end
 
       3.times { subject.add_item(event) }
 
+      subject.flush
+      expect(thread).to eq(subject.thread)
       expect(subject).to be_empty
     end
   end
@@ -95,7 +100,7 @@
     let(:max_items) { 30 }
 
     it "thread-safely handles concurrent access" do
-      expect(client).to receive(:send_envelope).exactly(3).times
+      expect(client).to receive(:send_envelope).at_least(:once)
 
       threads = 3.times.map do
         Thread.new do
@@ -169,6 +174,7 @@
         3.times { subject.add_item(event) }
       }.not_to raise_error
 
+      subject.flush
       expect(reentrant_calls).to be >= 1
     end
 
@@ -182,9 +188,25 @@
 
       3.times { subject.add_item(event) }
 
+      subject.flush
       expect(items_sent).not_to be_empty
       expect(string_io.string).not_to include("deadlock")
     end
+
+    it "does not add items from the buffer thread" do
+      worker_thread = Queue.new
+
+      allow(client).to receive(:send_envelope) do
+        worker_thread << Thread.current
+        subject.add_item(event)
+      end
+
+      3.times { subject.add_item(event) }
+
+      expect(worker_thread.pop).to eq(subject.thread)
+      subject.wait_until_idle
+      expect(subject).to be_empty
+    end
   end
 
   describe "error handling" do
@@ -214,12 +236,14 @@
       it "logs the error to sdk_logger" do
         3.times { subject.add_item(event) }
 
+        subject.flush
         expect(string_io.string).to include("Failed to send #{event.class}")
       end
 
       it "clears the buffer after a failed send to avoid memory buildup" do
         3.times { subject.add_item(event) }
 
+        subject.flush
         expect(subject).to be_empty
       end
     end