fix(logs, metrics): Reduce mutex surface area for TelemetryEventBuffer

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,15 +101,15 @@ def clear!
 
     private
 
-    def send_items
+    def send_items(pending_items)
       envelope = Envelope.new(sent_at: Sentry.utc_now.iso8601)
 
       discarded_count = 0
       discarded_bytes = 0
       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

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