Merge pull request rails#43282 from theojulienne/guard-against-instrumentation-exceptions

Guard against subscriber exceptions in ActiveSupport::Notifications

Author: rafaelfranca

activerecord/test/cases/adapter_test.rb

@@ -213,9 +213,10 @@ def test_numeric_value_out_of_ranges_are_translated_to_specific_exception
     def test_exceptions_from_notifications_are_not_translated
       original_error = StandardError.new("This StandardError shouldn't get translated")
       subscriber = ActiveSupport::Notifications.subscribe("sql.active_record") { raise original_error }
-      actual_error = assert_raises(StandardError) do
+      wrapped_error = assert_raises(ActiveSupport::Notifications::InstrumentationSubscriberError) do
         @connection.execute("SELECT * FROM posts")
       end
+      actual_error = wrapped_error.exceptions.first
 
       assert_equal original_error, actual_error
 

activerecord/test/cases/query_cache_test.rb

@@ -489,11 +489,13 @@ def test_query_cache_does_not_allow_sql_key_mutation
       payload[:sql].downcase!
     end
 
-    assert_raises FrozenError do
+    error = assert_raises ActiveSupport::Notifications::InstrumentationSubscriberError do
       ActiveRecord::Base.cache do
         assert_queries(1) { Task.find(1); Task.find(1) }
       end
     end
+
+    assert error.exceptions.first.is_a?(FrozenError)
   ensure
     ActiveSupport::Notifications.unsubscribe subscriber
   end

activesupport/lib/active_support/notifications/fanout.rb

@@ -7,6 +7,16 @@
 
 module ActiveSupport
   module Notifications
+    class InstrumentationSubscriberError < RuntimeError
+      attr_reader :exceptions
+
+      def initialize(exceptions)
+        @exceptions = exceptions
+        exception_class_names = exceptions.map { |e| e.class.name }
+        super "Exception(s) occurred within instrumentation subscribers: #{exception_class_names.join(', ')}"
+      end
+    end
+
     # This is a default queue implementation that ships with Notifications.
     # It just pushes events to all registered log subscribers.
     #
@@ -59,19 +69,32 @@ def unsubscribe(subscriber_or_name)
       end
 
       def start(name, id, payload)
-        listeners_for(name).each { |s| s.start(name, id, payload) }
+        iterate_guarding_exceptions(listeners_for(name)) { |s| s.start(name, id, payload) }
       end
 
       def finish(name, id, payload, listeners = listeners_for(name))
-        listeners.each { |s| s.finish(name, id, payload) }
+        iterate_guarding_exceptions(listeners) { |s| s.finish(name, id, payload) }
       end
 
       def publish(name, *args)
-        listeners_for(name).each { |s| s.publish(name, *args) }
+        iterate_guarding_exceptions(listeners_for(name)) { |s| s.publish(name, *args) }
       end
 
       def publish_event(event)
-        listeners_for(event.name).each { |s| s.publish_event(event) }
+        iterate_guarding_exceptions(listeners_for(event.name)) { |s| s.publish_event(event) }
+      end
+
+      def iterate_guarding_exceptions(listeners)
+        exceptions = nil
+
+        listeners.each do |s|
+          yield s
+        rescue Exception => e
+          exceptions ||= []
+          exceptions << e
+        end
+      ensure
+        raise InstrumentationSubscriberError.new(exceptions) unless exceptions.nil?
       end
 
       def listeners_for(name)

activesupport/test/notifications/evented_notification_test.rb

@@ -5,6 +5,9 @@
 module ActiveSupport
   module Notifications
     class EventedTest < ActiveSupport::TestCase
+      # we expect all exception types to be handled, so test with the most basic type
+      class BadListenerException < Exception; end
+
       class Listener
         attr_reader :events
 
@@ -27,6 +30,24 @@ def call(name, start, finish, id, payload)
         end
       end
 
+      class BadStartListener < Listener
+        def start(name, id, payload)
+          raise BadListenerException
+        end
+
+        def finish(name, id, payload)
+        end
+      end
+
+      class BadFinishListener < Listener
+        def start(name, id, payload)
+        end
+
+        def finish(name, id, payload)
+          raise BadListenerException
+        end
+      end
+
       def test_evented_listener
         notifier = Fanout.new
         listener = Listener.new
@@ -71,6 +92,54 @@ def test_listen_to_everything
         ], listener.events
       end
 
+      def test_listen_start_exception_consistency
+        notifier = Fanout.new
+        listener = Listener.new
+        notifier.subscribe nil, BadStartListener.new
+        notifier.subscribe nil, listener
+
+        assert_raises InstrumentationSubscriberError do
+          notifier.start  "hello", 1, {}
+        end
+        assert_raises InstrumentationSubscriberError do
+          notifier.start  "world", 1, {}
+        end
+        notifier.finish  "world", 1, {}
+        notifier.finish  "hello", 1, {}
+
+        assert_equal 4, listener.events.length
+        assert_equal [
+          [:start,  "hello", 1, {}],
+          [:start,  "world", 1, {}],
+          [:finish,  "world", 1, {}],
+          [:finish,  "hello", 1, {}],
+        ], listener.events
+      end
+
+      def test_listen_finish_exception_consistency
+        notifier = Fanout.new
+        listener = Listener.new
+        notifier.subscribe nil, BadFinishListener.new
+        notifier.subscribe nil, listener
+
+        notifier.start  "hello", 1, {}
+        notifier.start  "world", 1, {}
+        assert_raises InstrumentationSubscriberError do
+          notifier.finish  "world", 1, {}
+        end
+        assert_raises InstrumentationSubscriberError do
+          notifier.finish  "hello", 1, {}
+        end
+
+        assert_equal 4, listener.events.length
+        assert_equal [
+          [:start,  "hello", 1, {}],
+          [:start,  "world", 1, {}],
+          [:finish,  "world", 1, {}],
+          [:finish,  "hello", 1, {}],
+        ], listener.events
+      end
+
       def test_evented_listener_priority
         notifier = Fanout.new
         listener = ListenerWithTimedSupport.new