Subscribing to notifications while inside the said instrumented section.

The issue is that on the exit from Instrumenter#instrument section,
an Evented listener will run into an error because its thread local
(Thread.current[:_timestack]) has not been set up by the #start
method (this obviously happens because the Evented listeners didn't
exist at the time, since no subscribtion to that section was made yet).

Note: support for subscribing to instrumented sections, while being
inside those instrumented sections, might be removed in the future.

Maybe fixes #21873.

activesupport/lib/active_support/notifications/fanout.rb

@@ -42,8 +42,8 @@ def start(name, id, payload)
         listeners_for(name).each { |s| s.start(name, id, payload) }
       end
 
-      def finish(name, id, payload)
-        listeners_for(name).each { |s| s.finish(name, id, payload) }
+      def finish(name, id, payload, listeners = listeners_for(name))
+        listeners.each { |s| s.finish(name, id, payload) }
       end
 
       def publish(name, *args)

activesupport/lib/active_support/notifications/instrumenter.rb

@@ -15,14 +15,15 @@ def initialize(notifier)
       # and publish it. Notice that events get sent even if an error occurs
       # in the passed-in block.
       def instrument(name, payload={})
-        start name, payload
+        # some of the listeners might have state
+        listeners_state = start name, payload
         begin
           yield payload
         rescue Exception => e
           payload[:exception] = [e.class.name, e.message]
           raise e
         ensure
-          finish name, payload
+          finish_with_state listeners_state, name, payload
         end
       end
 
@@ -36,6 +37,10 @@ def finish(name, payload)
         @notifier.finish name, @id, payload
       end
 
+      def finish_with_state(listeners_state, name, payload)
+        @notifier.finish name, @id, payload, listeners_state
+      end
+
       private
 
       def unique_id

activesupport/test/notifications_test.rb

@@ -42,6 +42,21 @@ def test_subscribed
       ActiveSupport::Notifications.instrument(name)
       assert_equal expected, events
     end
+
+    def test_subsribing_to_instrumentation_while_inside_it
+      # the repro requires that there are no evented subscribers for the "foo" event,
+      # so we have to duplicate some of the setup code
+      old_notifier = ActiveSupport::Notifications.notifier
+      ActiveSupport::Notifications.notifier = ActiveSupport::Notifications::Fanout.new
+
+      ActiveSupport::Notifications.subscribe('foo', TestSubscriber.new)
+
+      ActiveSupport::Notifications.instrument('foo') do
+        ActiveSupport::Notifications.subscribe('foo') {}
+      end
+    ensure
+      ActiveSupport::Notifications.notifier = old_notifier
+    end
   end
 
   class UnsubscribeTest < TestCase