fix(logs, metrics): Make ThreadedPeriodicWorker and TelemetryEventBuffer fork friendly

sentry-rails/spec/sentry/rails_spec.rb

@@ -146,40 +146,34 @@
       Rails.application.load_runner
     end
 
-    def capture_in_separate_process(exit_code:)
-      pipe_in, pipe_out = IO.pipe
-
-      fork do
-        pipe_in.close
-
+    it "captures exception if exit code is non-zero" do
+      skip('fork not supported in jruby') if RUBY_PLATFORM == 'java'
+      captured_message = capture_in_separate_process(exit_code: 1) do |pipe_out|
         allow(Sentry::Rails).to receive(:capture_exception) do |event|
           pipe_out.puts event
         end
 
         # silence process
         $stderr.reopen('/dev/null', 'w')
         $stdout.reopen('/dev/null', 'w')
-
-        exit exit_code
       end
-
-      pipe_out.close
-      captured_messages = pipe_in.read
-      pipe_in.close
-      # sometimes the at_exit hook was registered multiple times
-      captured_messages.split("\n").last
-    end
-
-    it "captures exception if exit code is non-zero" do
-      skip('fork not supported in jruby') if RUBY_PLATFORM == 'java'
-      captured_message = capture_in_separate_process(exit_code: 1)
+      captured_message = captured_message.split("\n").last
 
       expect(captured_message).to eq('exit')
     end
 
     it "does not capture exception if exit code is zero" do
       skip('fork not supported in jruby') if RUBY_PLATFORM == 'java'
-      captured_message = capture_in_separate_process(exit_code: 0)
+      captured_message = capture_in_separate_process(exit_code: 0) do |pipe_out|
+        allow(Sentry::Rails).to receive(:capture_exception) do |event|
+          pipe_out.puts event
+        end
+
+        # silence process
+        $stderr.reopen('/dev/null', 'w')
+        $stdout.reopen('/dev/null', 'w')
+      end
+      captured_message = captured_message.split("\n").last
 
       expect(captured_message).to be_nil
     end

sentry-rails/spec/spec_helper.rb

@@ -11,6 +11,7 @@
 
 require "sentry-ruby"
 require "sentry-rails"
+require_relative "../../spec/support/fork_helper"
 
 require "simplecov"
 
@@ -49,6 +50,8 @@
   # Disable RSpec exposing methods globally on `Module` and `main`
   config.disable_monkey_patching!
 
+  config.include(Test::ForkHelper)
+
   config.expect_with :rspec do |c|
     c.syntax = :expect
   end

sentry-ruby/lib/sentry/telemetry_event_buffer.rb

@@ -54,11 +54,11 @@ def run
     def add_item(item)
       # the buffer thread can never add telemetry itself to prevent recursion
       return self if Thread.current == thread
+      return unless ensure_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
         if size >= @max_items_before_drop
@@ -97,6 +97,14 @@ def clear!
 
     private
 
+    def reset_if_forked
+      return unless super
+
+      # Discard items inherited from the parent to avoid duplicate delivery.
+      @mutex = Mutex.new
+      @pending_items = []
+    end
+
     def flush_pending_items
       pending_items = @mutex.synchronize do
         next if @pending_items.empty?

sentry-ruby/lib/sentry/threaded_periodic_worker.rb

@@ -8,6 +8,7 @@ class ThreadedPeriodicWorker
     attr_reader :thread, :thread_mutex, :idle_condition
 
     def initialize(sdk_logger, interval)
+      @process_id = Process.pid
       @thread = nil
       @exited = false
       @interval = interval
@@ -22,6 +23,8 @@ def initialize(sdk_logger, interval)
     end
 
     def ensure_thread
+      reset_if_forked
+
       @thread_mutex.synchronize do
         return false if @exited
         return true if @thread&.alive?
@@ -80,6 +83,20 @@ def kill
 
     private
 
+    def reset_if_forked
+      return false if @process_id == Process.pid
+
+      @process_id = Process.pid
+      @thread = nil
+      @exited = false
+      @woken = false
+      @running = false
+      @thread_mutex = Mutex.new
+      @wake_condition = ConditionVariable.new
+      @idle_condition = ConditionVariable.new
+      true
+    end
+
     def worker_loop
       loop do
         @thread_mutex.synchronize do

sentry-ruby/spec/spec_helper.rb

@@ -30,6 +30,7 @@
 
 require "sentry-ruby"
 require "sentry/test_helper"
+require_relative "../../spec/support/fork_helper"
 
 require "webmock/rspec"
 require_relative "support/profiler"
@@ -43,6 +44,7 @@
   config.disable_monkey_patching!
 
   config.include(Sentry::TestHelper)
+  config.include(Test::ForkHelper)
 
   config.expect_with :rspec do |c|
     c.syntax = :expect

sentry-ruby/spec/support/shared_examples_for_telemetry_event_buffers.rb

@@ -324,6 +324,37 @@
     end
   end
 
+  context "when a child adds telemetry after a fork", when: { ruby_engine?: "ruby" } do
+    let(:max_items) { 3 }
+
+    it "resets inherited worker state before adding telemetry" do
+      subject.add_item(event)
+      subject.wait_until_idle
+      expect(subject.size).to eq(1)
+
+      result = capture_in_separate_process do |writer|
+        subject.add_item(event)
+        subject.flush
+        item_count = sentry_envelopes.first.items.first.headers[:item_count]
+        writer.puts [subject.size, sentry_envelopes.size, item_count].join(",")
+      end
+
+      child_size, child_envelopes, child_item_count = result.split(",").map(&:to_i)
+      expect(child_size).to eq(0)
+      expect(child_envelopes).to eq(1)
+      expect(child_item_count).to eq(1)
+      expect(subject.size).to eq(1)
+      expect(sentry_envelopes).to be_empty
+
+      subject.flush
+      expect(sentry_envelopes.size).to eq(1)
+
+      subject.add_item(event)
+      subject.flush
+      expect(sentry_envelopes.size).to eq(2)
+    end
+  end
+
   describe "error handling" do
     let(:max_items) { 3 }
 

spec/support/fork_helper.rb

@@ -0,0 +1,69 @@
+# frozen_string_literal: true
+
+module Test
+  module ForkHelper
+    TIMEOUT = 1
+
+    def capture_in_separate_process(exit_code: 0)
+      reader, writer = IO.pipe
+      pid = fork do
+        reader.close
+        yield(writer)
+        exit(exit_code)
+      end
+
+      writer.close
+      output = read_from_child(reader)
+      wait_for_child(pid)
+      output
+    ensure
+      reader&.close
+      writer&.close
+      terminate_child(pid)
+    end
+
+    private
+
+    def read_from_child(reader)
+      output = +""
+      deadline = Process.clock_gettime(Process::CLOCK_MONOTONIC) + TIMEOUT
+
+      loop do
+        remaining = deadline - Process.clock_gettime(Process::CLOCK_MONOTONIC)
+        raise "child did not respond" if remaining <= 0
+        raise "child did not respond" unless IO.select([reader], nil, nil, remaining)
+
+        begin
+          output << reader.read_nonblock(4096)
+        rescue IO::WaitReadable
+          next
+        rescue EOFError
+          return output
+        end
+      end
+    end
+
+    def wait_for_child(pid)
+      deadline = Process.clock_gettime(Process::CLOCK_MONOTONIC) + TIMEOUT
+
+      loop do
+        return if Process.waitpid(pid, Process::WNOHANG)
+        raise "child did not exit" if Process.clock_gettime(Process::CLOCK_MONOTONIC) >= deadline
+
+        sleep 0.01
+      end
+    end
+
+    def terminate_child(pid)
+      return unless pid
+
+      begin
+        return if Process.waitpid(pid, Process::WNOHANG)
+
+        Process.kill("KILL", pid)
+        Process.wait(pid)
+      rescue Errno::ECHILD, Errno::ESRCH
+      end
+    end
+  end
+end