Queue pending Dataloader sources

[ydah]

This changes GraphQL::Dataloader to keep a queue of pending sources instead of repeatedly scanning every source in @source_cache.

Previously, run_pending_steps and spawn_source_fiber walked the full source cache to find pending sources. That makes scheduling overhead grow with the total number of cached source instances, even when only a few sources are pending. Now sources enqueue themselves when they become pending, and the run loop drains that queue.

This also updates GraphQL::Dataloader::AsyncDataloader to use the same pending source queue and adds coverage for avoiding full-cache scans and preserving pending sources across run_isolated.

Benchmark

This benchmark isolates GraphQL::Dataloader scheduling overhead by using a no-op fetch. Real DB/API-backed workloads will see smaller end-to-end gains, but this shows the internal source lookup cost that this PR removes.

Ruby: ruby 4.0.0 (2025-12-25 revision 553f1675f3) +PRISM

sources	cycles	master	this PR	speedup	master pending? checks	this PR pending? checks
100	2,000	0.0667s	0.0057s	11.6x	901,200	4,000
1,000	2,000	0.6099s	0.0059s	102.7x	9,003,000	4,000
5,000	1,000	1.3737s	0.0048s	286.5x	20,510,500	2,000
10,000	500	1.3360s	0.0058s	230.9x	20,145,250	1,000

Benchmark script

# frozen_string_literal: true

$LOAD_PATH.unshift(File.expand_path("lib", ARGV.fetch(0, ".")))

require "graphql"

class BenchSource < GraphQL::Dataloader::Source
  @pending_checks = 0

  class << self
    attr_accessor :pending_checks
  end

  def self.reset!
    self.pending_checks = 0
  end

  def initialize(batch_key)
    @batch_key = batch_key
  end

  def fetch(keys)
    keys
  end

  def pending?
    self.class.pending_checks += 1
    super
  end
end

def median(values)
  values.sort.fetch(values.length / 2)
end

def measure(source_count, cycles)
  dataloader = GraphQL::Dataloader.new
  source_count.times { |idx| dataloader.with(BenchSource, idx) }

  dataloader.append_job do
    cycles.times do |idx|
      dataloader.with(BenchSource, idx % source_count).load(idx)
    end
  end

  BenchSource.reset!
  GC.start
  started_at = Process.clock_gettime(Process::CLOCK_MONOTONIC)
  dataloader.run
  elapsed = Process.clock_gettime(Process::CLOCK_MONOTONIC) - started_at

  [elapsed, BenchSource.pending_checks]
end

puts "ruby=#{RUBY_DESCRIPTION}"
puts "graphql=#{GraphQL::VERSION}"
puts "sources,cycles,seconds,pending_checks"

[[100, 2_000], [1_000, 2_000], [5_000, 1_000], [10_000, 500]].each do |source_count, cycles|
  times = []
  checks = nil

  3.times do
    elapsed, checks = measure(source_count, cycles)
    times << elapsed
  end

  puts [
    source_count,
    cycles,
    format("%.6f", median(times)),
    checks,
  ].join(",")
end

[rmosolgo]

Thanks so much for this great speed-up! It makes a lot of sense: previously, each Source knew whether it was waiting for data, but the parent Dataloader didn't know: instead, it would search each source, looking for something pending. Now, the Dataloader has its own list of pending sources, so it doesn't have to search the whole cache. Awesome 🤘

I have one question about the implementation but this looks good as-is. Edit: oops, I see that manual concurrency tests are failing. I can take a look at those now 👀

[rmosolgo]

It looks like this has is being used as a set. Why not use Ruby's Set instead of a Hash?

[rmosolgo]

I pushed a change so that drain_pending_sources would be called before the while loop that spins up Fibers for calling Sources. That way, those fibers could share pending_sources queue -- allowing them to run in parallel. It looked good in local tests, now we'll see CI 👀 Let me know what you think of that change.

[rmosolgo]

I identified a couple other issues after pushing that change:

1. The previous implementation would pick up newly-added pending sources during def spawn_source_fiber. I added a drain_pending_sources call to that method to keep that behavior. It's good for when one source calls another source; you can resolve those in the same loop.

2. This new implementation changes the order that sources are resolved in. Previously, they'd be ordered according to how they were added to the @sources_cache hash. Now, they're ordered according to their position in the @pending_sources array. The same sources get resolved during run loops, but possibly in different order. This doesn't matter -- I don't think GraphQL-Ruby made any guarantees about this, and I don't think it changes an results -- but it shows up in Perfetto traces because packets appear in different orders. I rebuilt the snapshots after this change.

[ydah]

Thank you very much for your prompt review and for taking the time to suggest the necessary changes.
I sincerely apologize, but I accidentally squashed the changes you had committed. I have added you as a co-author to the commit messages for both commits.

I believe the issue has now been resolved. I would greatly appreciate it if you could kindly review the changes once again at your convenience.