AsyncDataloader: don't reuse another dataloader's run when nested

lib/graphql/dataloader/async_dataloader.rb

@@ -73,7 +73,7 @@ def initialize(dataloader, total_fiber_limit, jobs_fiber_limit)
 
         attr_accessor :trace, :root_task
 
-        attr_reader :jobs, :lazies_at_depth, :jobs_fiber_limit,  :snoozed_jobs_condition, :snoozed_sources_condition
+        attr_reader :dataloader, :jobs, :lazies_at_depth, :jobs_fiber_limit,  :snoozed_jobs_condition, :snoozed_sources_condition
 
         def jobs_bandwidth?
           running_count < @jobs_fiber_limit
@@ -193,11 +193,19 @@ def lazy_at_depth(depth, lazy)
       end
 
       def active_run
-        @pending_run || Async::Task.current?&.graphql_async_dataloader_run || raise(GraphQL::Error, "No available Run to append to, GraphQL-Ruby bug")
+        @pending_run || current_task_run || raise(GraphQL::Error, "No available Run to append to, GraphQL-Ruby bug")
+      end
+
+      # The current task's run, but only if it belongs to this dataloader. A different
+      # dataloader may be running inside one of our tasks (or vice versa), e.g. a query
+      # executed from a resolver or a subscription trigger; its run must not be reused.
+      def current_task_run
+        run = Async::Task.current?&.graphql_async_dataloader_run
+        run if run&.dataloader.equal?(self)
       end
 
       def run_isolated
-        previous_run = Async::Task.current?&.graphql_async_dataloader_run
+        previous_run = current_task_run
         prev_pending_keys = {}
         # Clear pending loads but keep already-cached records
         # in case they are useful to the given block.
@@ -233,7 +241,7 @@ def run_isolated
 
       def run(trace_query_lazy: nil)
         trace = Fiber[:__graphql_current_multiplex]&.current_trace
-        run = @pending_run || Async::Task.current?&.graphql_async_dataloader_run || raise(GraphQL::Error, "No available Run, GraphQL-Ruby internal bug")
+        run = @pending_run || current_task_run || raise(GraphQL::Error, "No available Run, GraphQL-Ruby internal bug")
         @pending_run = nil
         run.trace = trace
         first_pass = true

spec/graphql/dataloader/async_dataloader_spec.rb

@@ -466,6 +466,32 @@ def author(name:)
       end
     end
 
+    describe "when another dataloader runs inside a job" do
+      class EchoSource < GraphQL::Dataloader::Source
+        def fetch(keys)
+          keys.map { |k| "v#{k}" }
+        end
+      end
+
+      # A query executed from inside a resolver (or a subscription trigger) gets its own
+      # dataloader, but shares the outer dataloader's task tree. `run_isolated` used to
+      # mistake the outer task's run for its own and clear `@pending_run`, so the inner
+      # `run` took over the outer run's queues and the outer run never finished.
+      it "doesn't take over the outer dataloader's run" do
+        outer = GraphQL::Dataloader::AsyncDataloader.new
+        inner_result = nil
+        outer.append_job do
+          inner = GraphQL::Dataloader::AsyncDataloader.new
+          inner.run_isolated { inner.with(EchoSource).load(1) }
+          inner.append_job { inner_result = inner.with(EchoSource).load(2) }
+          inner.run
+        end
+
+        Timeout.timeout(5) { outer.run }
+        assert_equal "v2", inner_result
+      end
+    end
+
     describe "stress test" do
       RNG = Random.new(20260724)
       MAX_ITERATIONS = 2000