Fix exception guards on multiple subscriber types

Previously in rails#43282, which shipped
with Rails 7.0, we added the guarantee that if an exception occurred
within an ActiveSupport::Notificaitons subscriber that the remaining
subscribers would still be run.

This was broken in rails#44469, where the
different types of subscribers were broken into groups by type for
performance. Although we would guard exceptions and allways run all (or
none) of the same group, that didn't work cross-group with different
types of subscriber.

Author: jhawthorn

activesupport/lib/active_support/notifications/fanout.rb

@@ -17,26 +17,30 @@ def initialize(exceptions)
     end
 
     module FanoutIteration # :nodoc:
-      def iterate_guarding_exceptions(listeners)
-        exceptions = nil
-
-        listeners.each do |s|
-          yield s
-        rescue Exception => e
-          exceptions ||= []
-          exceptions << e
-        end
+      private
+        def iterate_guarding_exceptions(collection)
+          exceptions = nil
 
-        if exceptions
-          if exceptions.size == 1
-            raise exceptions.first
-          else
-            raise InstrumentationSubscriberError.new(exceptions), cause: exceptions.first
+          collection.each do |s|
+            yield s
+          rescue Exception => e
+            exceptions ||= []
+            exceptions << e
           end
-        end
 
-        listeners
-      end
+          if exceptions
+            exceptions = exceptions.flat_map do |exception|
+              exception.is_a?(InstrumentationSubscriberError) ? exception.exceptions : [exception]
+            end
+            if exceptions.size == 1
+              raise exceptions.first
+            else
+              raise InstrumentationSubscriberError.new(exceptions), cause: exceptions.first
+            end
+          end
+
+          collection
+        end
     end
 
     # This is a default queue implementation that ships with Notifications.
@@ -222,6 +226,8 @@ def groups_for(name) # :nodoc:
       #     handle.finish
       #   end
       class Handle
+        include FanoutIteration
+
         def initialize(notifier, name, id, payload) # :nodoc:
           @name = name
           @id = id
@@ -236,7 +242,7 @@ def start
           ensure_state! :initialized
           @state = :started
 
-          @groups.each do |group|
+          iterate_guarding_exceptions(@groups) do |group|
             group.start(@name, @id, @payload)
           end
         end
@@ -249,7 +255,7 @@ def finish_with_values(name, id, payload) # :nodoc:
           ensure_state! :started
           @state = :finished
 
-          @groups.each do |group|
+          iterate_guarding_exceptions(@groups) do |group|
             group.finish(name, id, payload)
           end
         end

activesupport/test/notifications/evented_notification_test.rb

@@ -126,18 +126,23 @@ def test_listen_finish_multiple_exception_consistency
         listener = Listener.new
         notifier.subscribe nil, BadFinishListener.new
         notifier.subscribe nil, BadFinishListener.new
+        notifier.subscribe(nil) { |*args| raise "foo" }
+        notifier.subscribe(nil) { |obj| raise "foo" }
+        notifier.subscribe(nil, monotonic: true) { |obj| raise "foo" }
         notifier.subscribe nil, listener
 
         notifier.start  "hello", 1, {}
         notifier.start  "world", 1, {}
         error = assert_raises InstrumentationSubscriberError do
           notifier.finish  "world", 1, {}
         end
+        assert_equal 5, error.exceptions.count
         assert_instance_of BadListenerException, error.cause
 
         error = assert_raises InstrumentationSubscriberError do
           notifier.finish  "hello", 1, {}
         end
+        assert_equal 5, error.exceptions.count
         assert_instance_of BadListenerException, error.cause
 
         assert_equal 4, listener.events.length