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

[drhops]

Problem

An AsyncDataloader run hangs forever when a second dataloader is used inside one of its jobs and that second dataloader calls run_isolated before its first run.

In our case, a mutation triggered a subscription from an after_commit hook, and the subscription document is executed inline (Schema.execute => its own dataloader) from within the mutation's job fiber.

Introduced with the queue-based rewrite in #5479 (2.6.4+). Not fixed by #5672/#5679, and independent of #5702 (that race is collateral of the same nesting in our case: on master this repro also logs Async::Queue::ClosedError from async_dataloader.rb:341, because the inner run closes the outer run's channel).

Minimal repro

class EchoSource < GraphQL::Dataloader::Source
  def fetch(keys) = keys.map { |k| "v#{k}" }
end

outer = GraphQL::Dataloader::AsyncDataloader.new
outer.append_job do
  inner = GraphQL::Dataloader::AsyncDataloader.new
  inner.run_isolated { inner.with(EchoSource).load(1) }   # e.g. argument coercion
  inner.append_job { inner.with(EchoSource).load(2) }
  inner.run
end
outer.run  # never returns

Production symptom (Puma, graphql 2.6.8, async 2.36.0): every mutation that triggered an inline subscription sat idle after ~0.2s of work until Rack::Timeout (10s) fired, with the root task parked in Run#wait_for_activity, and no job tasks alive.

Root cause

run_isolated decides whether it is nested by looking at the current task's run:

previous_run = Async::Task.current?&.graphql_async_dataloader_run
...
ensure
  if previous_run
    Async::Task.current.graphql_async_dataloader_run = previous_run
    @pending_run = nil   # "clear the one created in #run"
  end

Inside the outer dataloader's job fiber, that run belongs to the outer dataloader, but run_isolated treats it as its own nesting case and clears the inner dataloader's @pending_run. The inner dataloader's next run (and active_run/append_job) then falls back the same way, @pending_run || Async::Task.current?&.graphql_async_dataloader_run, and picks up the outer run. It calls new_queues/run_queue on it, replacing the outer run's tasks_channel/@activity, and closes them when it finishes. The outer root task is still blocked on the old @activity condition, which nothing signals anymore, so the reactor sits in select with no timers or IO. Meanwhile the outer job's :finished_task push lands on a closed channel (ClosedError).

Fix

Only treat the current task's run as ours if it belongs to this dataloader:

def current_task_run
  run = Async::Task.current?&.graphql_async_dataloader_run
  run if run&.dataloader.equal?(self)
end

Used by active_run, run_isolated (previous_run) and run. Run already holds @dataloader, but this exposes it. Same-dataloader nesting (run_isolated from a resolver of the same query) is unchanged; a foreign run is now simply not a candidate, so the inner dataloader keeps its own @pending_run and runs its own queues.

Tests

spec/graphql/dataloader/async_dataloader_spec.rb: "when another dataloader runs inside a job, it doesn't take over the outer dataloader's run".