Fix exception guards on multiple subscriber types

Previously in https://github.com/rails/rails/pull/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 https://github.com/rails/rails/pull/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.

activesupport/lib/active_support/notifications/fanout.rb

@@ -17,26 +17,30 @@ module ActiveSupport
     end
 
     module FanoutIteration # :nodoc:
-      def iterate_guarding_exceptions(listeners)
-        exceptions = nil
+      private
+        def iterate_guarding_exceptions(collection)
+          exceptions = nil
 
-        listeners.each do |s|
-          yield s
-        rescue Exception => e
-          exceptions ||= []
-          exceptions << e
-        end
-
-        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 @@ module ActiveSupport
       #     handle.finish
       #   end
       class Handle
+        include FanoutIteration
+
         def initialize(notifier, name, id, payload) # :nodoc:
           @name = name
           @id = id
@@ -236,7 +242,7 @@ module ActiveSupport
           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 @@ module ActiveSupport
           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,6 +126,9 @@ module ActiveSupport
         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, {}
@@ -133,11 +136,13 @@ module ActiveSupport
         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