From d974ea25d77c77c6e4e9b1d2bbe65cb3dcce9697 Mon Sep 17 00:00:00 2001 From: Neel Shah Date: Mon, 14 Sep 2026 15:15:35 +0200 Subject: [PATCH] fix(logs, metrics): Reduce mutex surface area for TelemetryEventBuffer --- .../lib/sentry/telemetry_event_buffer.rb | 44 +++++++++++-------- ...ed_examples_for_telemetry_event_buffers.rb | 38 ++++++++++++++-- 2 files changed, 60 insertions(+), 22 deletions(-) diff --git a/sentry-ruby/lib/sentry/telemetry_event_buffer.rb b/sentry-ruby/lib/sentry/telemetry_event_buffer.rb index 790702d0e..73dcf59f5 100644 --- a/sentry-ruby/lib/sentry/telemetry_event_buffer.rb +++ b/sentry-ruby/lib/sentry/telemetry_event_buffer.rb @@ -41,14 +41,18 @@ def initialize(configuration, client, event_class:, max_items:, max_items_before end def flush - @mutex.synchronize do - return if empty? + pending_items = @mutex.synchronize do + next if @pending_items.empty? - log_debug("[#{self.class}] flushing #{size} #{@event_class}") - - send_items + 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 alias_method :run, :flush @@ -57,23 +61,29 @@ def add_item(item) # Prevent ThreadError from re-entrant locking (e.g. transport instrumentation calling Sentry.metrics.*) return self if @mutex.owned? - @mutex.synchronize do + dropped = false + size_exceeded = @mutex.synchronize do return unless ensure_thread if size >= @max_items_before_drop - log_debug("[#{self.class}] exceeded max capacity, dropping event") - @client.transport.record_lost_event( - :queue_overflow, - @data_category, - num_bytes: JSON.generate(item.to_h).bytesize - ) + dropped = true else @pending_items << item end - send_items if size >= @max_items + size >= @max_items + end + + if dropped + log_debug("[#{self.class}] exceeded max capacity, dropping event") + @client.transport.record_lost_event( + :queue_overflow, + @data_category, + num_bytes: JSON.generate(item.to_h).bytesize + ) end + flush if size_exceeded self end @@ -91,7 +101,7 @@ def clear! private - def send_items + def send_items(pending_items) envelope = Envelope.new(sent_at: Sentry.utc_now.iso8601) discarded_count = 0 @@ -99,7 +109,7 @@ def send_items envelope_items = [] if callback = @configuration.send(@before_send) - @pending_items.each do |item| + pending_items.each do |item| processed_item = safe_dispatch_callback(@before_send.to_s, callback, [item]) if processed_item @@ -110,7 +120,7 @@ def send_items end end else - envelope_items = @pending_items.map(&:to_h) + envelope_items = pending_items.map(&:to_h) end unless discarded_count.zero? @@ -131,8 +141,6 @@ def send_items @client.send_envelope(envelope) rescue => e log_error("[#{self.class}] Failed to send #{@event_class}", e, debug: @debug) - ensure - clear! 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 e966f514b..3dd1dcd22 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 @@ -58,6 +58,39 @@ end end + describe "sending" do + let(:max_items) { 3 } + + it "does not hold the mutex while sending" do + send_started = Queue.new + continue_send = Queue.new + add_finished = Queue.new + + allow(client).to receive(:send_envelope) do + send_started << true + continue_send.pop + end + + # trigger a flush + sender = Thread.new do + max_items.times { subject.add_item(event) } + end + + # start sending, will block till continue_send + send_started.pop + + Thread.new do + subject.add_item(event) + add_finished << true + end + + # sending shouldn't block the add + expect(add_finished.pop).to be(true) + continue_send << true + sender.join + end + end + describe "multi-threaded access" do let(:max_items) { 30 } @@ -79,14 +112,11 @@ end describe "max capacity and dropping events" do - let(:max_items) { 3 } + let(:max_items) { max_items_before_drop + 1 } let(:max_items_before_drop) { 10 } before do subject.instance_variable_set(:@max_items_before_drop, max_items_before_drop) - - # don't clear pending items to allow buffer to grow - allow(subject).to receive(:clear!) end it "adds items up to max_items_before_drop capacity" do