AsyncDataloader: avoid ClosedError race condition with straggler fibers

[drhops]

Problem

AsyncDataloader can leak Async::Queue::ClosedError into resolver code, and lose the task's own error report when:

1. A job fiber raises a StandardError, so run_queue's check_error! re-raises and its ensure runs run.close_queues, and
2. At that moment, sibling tasks are still mid-flight, parked in non-Dataloader waits (e.g. sleep, socket waits).

Introduced with the queue-based rewrite in #5479 (2.6.4+). Related prior reports: #5654 (#5656) and #5671 (#5672, #5679). All of those fixes are in 2.6.8, but this race survives them, which matches the #5671 reporter's observation that the warnings went away while a hang remained.

Production stacktrace

Seen intermittently in production (Puma, Ruby 3.4, graphql 2.6.8, async 2.36.0, config.active_support.isolation_level = :fiber, fiber_limit: 16, sources doing Redis + Postgres I/O):

Async::Queue::ClosedError: Cannot enqueue items to a closed queue!
  async-2.36.0/lib/async/queue.rb:65:in 'Async::Queue#push'
  graphql-2.6.8/lib/graphql/dataloader/async_dataloader.rb:39:in 'AsyncDataloader#yield'
  graphql-2.6.8/lib/graphql/dataloader/source.rb:105:in 'Source#sync'
  graphql-2.6.8/lib/graphql/dataloader/source.rb:84:in 'Source#load_all'
  ... resolver frames ...
=> rescued at async_dataloader.rb:339, where push([:task_error, err]) raises
  ClosedError again => task dies as an unhandled exception, error report lost

Minimal repro

Single-threaded, no GraphQL execution needed. A job error while siblings sit in plain sleeps is the whole trigger:

dataloader = GraphQL::Dataloader::AsyncDataloader.new

3.times do |j|
  dataloader.append_job do
    sleep(rand(0.003))                              # non-Dataloader IO
    dataloader.with(SlowSource, rand(0.002)).load(j) # -> yield -> ClosedError
  end
end

dataloader.append_job do
  sleep(rand(0.003))
  raise "boom"
end

dataloader.run # should raise "boom"; instead stragglers intermittently
               # hit ClosedError and their task_error reports are lost

With randomized delays this reproduces within a few dozen iterations on master, against both async 2.36.0 and async 2.44.1. A single straggler job (SlowSource) reproduces the issue ~1% of runs, while three stragglers increases repro rate to ~5%.

Root cause

Closing the channel and stopping the fibers that report to it are meant to be one atomic act, but the scheduler can interleave another fiber between them.

The stragglers are accounted for and sit in @running_tasks, so bookkeeping is accurate. On the error path the round is deliberately abandoned with straggler fibers still parked: check_error! raises, the ensure closes the tasks_channel, and root_task.cancel is about to stop everyone. But teardown itself yields control: cancelling a non-current task transfers into it (IO::Event::Selector#raise pushes the calling fiber onto the ready list, then fiber.raises into the target), and each transfer gives the reactor a chance to resume whatever else became ready.

So a parked straggler's wakeup (sleep timer, IO completion, etc) has three possible timings:

• Before the error is processed => it wakes while the channel is open, yields, parks on the condition; later cancelled cleanly. Fine.
• After cancellation reaches it => Stop is raised at its suspension point; it unwinds without pushing. Fine.
• Inside the teardown window => after close_queues, before its own cancel. The reactor resumes it and it runs one normal slice: its dataloader.yield pushes [:paused_task, ...] into the closed channel and ClosedError leaks into resolver code (async_dataloader.rb:39); its rescue then pushes [:task_error, ClosedError] into the same closed channel (:339) → a second ClosedError, an unhandled task exception, and a lost error report.

The window is microseconds wide, which is why the failure is intermittent and timing-dependent, and why every additional mid-flight fiber is another independent chance to land in it.

Fix

Route all four task-message pushes through a new Run#push_task_message, which tolerates a closed channel:

• :task_error payloads are recorded directly on the Run (@task_error ||= err) so an error report is never lost to a closed channel.
• yield's two push sites do push_task_message(...) || task.stop: a straggler that finds the channel closed stops itself, exactly what root_task.cancel was about to do a moment later, instead of leaking ClosedError into resolver code.
• :finished_task on a closed channel is discarded: the round's accounting is already finished, and the task completes cleanly.

Tests

New regression test in spec/graphql/dataloader/async_dataloader_spec.rb ("doesn't leak Async::Queue::ClosedError into straggler fibers or lose the original error"): seeded randomized straggler/error timing over up to 500 dataloader runs, asserting that no ClosedError reaches job code and that dataloader.run always re-raises the injected job error. It fails on master within ~10 iterations and passes with this change (~2s). Existing dataloader specs (including the #5671 stress test) still pass.