AsyncDataloader: ClosedQueueError persists in 2.6.5 and can hang requests indefinitely (follow-up to #5654)

[vytautaskubilius]

We hit the ClosedQueueError from #5654 on 2.6.5, which includes the #5656 fix - so it looks like the fix doesn't cover at least one code path. In our case the error isn't harmless: losing the ensure-block push means the run's accounting never balances, and the whole request hangs indefinitely.

Versions:

• graphql 2.6.5
• async 2.42.0
• Rails 8.1.3 with config.active_support.isolation_level = :fiber
• Ruby 3.4, unicorn (single-threaded workers)

A fan-out-heavy query spawns ~40 concurrent source/job tasks for us, which starves the ActiveRecord connection pool and produces 5s ActiveRecord::ConnectionTimeoutErrors inside source fetches. A task that gets woken from one of those waits after its round's queues have already been closed then fails the run.finished_tasks.push($! || task) in the ensure of spawn_source_task:

Async::Queue::ClosedError: Cannot enqueue items to a closed queue!
  async-2.42.0/lib/async/queue.rb:65 in 'Async::Queue#push'
  graphql-2.6.5/lib/graphql/dataloader/async_dataloader.rb:351 in 'block (2 levels) in GraphQL::Dataloader::AsyncDataloader#spawn_source_task'
  async-2.42.0/lib/async/task.rb:224 in 'block in Async::Task#run'

After that the run never completes - the root task waits forever and the reactor sits idle. We watched it for over two minutes with periodic backtrace dumps of the main thread, all identical:

async-2.42.0/lib/async/scheduler.rb:456:in 'IO::Event::Selector::EPoll#select'
...
async-2.42.0/lib/kernel/sync.rb:30:in 'Kernel#Sync'
graphql-2.6.5/lib/graphql/dataloader/async_dataloader.rb:197:in 'GraphQL::Dataloader::AsyncDataloader#run'
graphql-2.6.5/lib/graphql/execution/interpreter.rb:86:in 'block in GraphQL::Execution::Interpreter.run_all'

The race is timing-dependent: the exact same query against the same data sometimes completes in ~5s with the connection timeouts collected as field errors, and sometimes wedges as above. In production (unicorn with a 30s worker timeout) the wedged requests get killed silently, which is what we were originally debugging.

I don't have a minimal script, but the reproduction shape is: any query whose source fan-out exceeds the AR pool size produces late-finishing tasks via checkout timeouts, and each late task rolls the dice on this race - it triggers within a couple of attempts on our data. Happy to run patches or provide more diagnostics against our reproduction.

Pinning back to 2.6.3 avoids the problem entirely (verified A/B on the same environment).

[rmosolgo]

Hey, thanks for the detailed report and sorry about the ongoing trouble with this :S I'm going to try working up a replication, so I can better understand how this error is being mis-handled. I'll follow up here when I've got it "working" (failing).

[rmosolgo]

👋 just a quick check in with an unsucessful replication attempt. In the script below, I tried to replicate the error:

• AR 8.1.3, Async 2.42.0, GraphQL-Ruby 2.6.5
• Isolation level :fiber
• Shorter checkout_timeout (1), simulate slow queries with connection.execute("SELECT pg_sleep(1.2)")
• It raises ConnectionTimeoutError, but the error is propagated to the ExampleSchema.execute(...) call; it doesn't block the Sync { ... } call.

At first, it didn't fail, so I tried making it deeper, with a second layer of dataloader calls, but it still didn't fail.

Here's the script:

Try to replicate hanging `Sync`

require "bundler/inline"

gemfile(true) do
  gem "graphql", "2.6.5" # or `git checkout v2.6.5` and use `path: "./"` for easier debugging
  gem "activerecord", require: "active_record"
  gem "pg"
  gem "async", "2.42.0"
end

ActiveSupport::IsolatedExecutionState.isolation_level = :fiber

ActiveRecord::Base.establish_connection(
  host: "localhost",
  adapter: "postgresql",
  username: "postgres",
  database: "async_dataloader_test",
  checkout_timeout: 1,
)

ActiveRecord::Schema.define(force: true) do
  create_table :things, force: true do |t|
    t.string :name
  end
end

class Thing < ActiveRecord::Base
end

Thing.destroy_all
(1..40).each { |i| Thing.create!(id: i, name: "Thing ##{i}") }
ActiveRecord::Base.logger = Logger.new(STDOUT)

class ExampleSchema < GraphQL::Schema
  class ThingSource < GraphQL::Dataloader::Source
    def initialize(_id)
    end
    def fetch(keys)
      keys.map!(&:to_i)
      ::Thing.connection.execute("SELECT pg_sleep(1.2);")
      things = ::Thing.where(id: keys)
      keys.map { |k| things.find { |t| t.id == k } }
    end
  end
  class Thing < GraphQL::Schema::Object
    field :name, String

    field :things, [Thing]

    def things
      id = object.id
      new_ids = [
        (id * 10) + 1,
        (id * 10) + 2,
        (id * 10) + 3,
        (id * 10) + 4,
      ]
      reqs = new_ids.map { |id| dataloader.with(ThingSource, id).request(id) }
      reqs.map(&:load)
    end
  end

  class Query < GraphQL::Schema::Object
    field :thing, Thing, resolve_static: true do
      argument :id, ID
    end

    def thing(id:)
      dataload(ThingSource, id, id)
    end
  end

  query(Query)
  use GraphQL::Dataloader::AsyncDataloader
end


query_str = "{#{(1..3).map { |i| "\n  t#{i}: thing(id: #{i}) { name things { name } }"}.join}}"
3.times do
  begin
    pp ExampleSchema.execute(query_str).to_h
  rescue StandardError => err
    puts "Rescued: #{err.class}, #{err.message}"
    puts err.backtrace
  end
end

puts "Finished successfully"

Could you share an example query that exhibits this problem? (Changing the field and argument names would be fine -- I'm curious about the structure. Maybe I can imitate it more closely in the replication.)

---

Another update, I was able to get a ClosedQueueError by putting the pg_sleep in a field instead of a Source#fetch:

    def name
      ::Thing.connection.execute("SELECT pg_sleep(1.2);")
      object.name
    end

But this error was handled as expected -- raised to the top level, not causing a hung Sync ... :S

[rmosolgo]

I took a few minutes to scan the error handling flow in AsyncDataloader. I realized that, if it was working how I expected, then we should never get warnings from Async about unhandled exceptions in Tasks. But we do get those warnings.

I reworked the error handling flow somewhat here: #5672 There's some chance that the change will fix the problem for you (at a minimum, it checks for .closed? on queues...). If you can replicate the error locally, please give that patch a try.

[vytautaskubilius]

Thanks for the quick turnaround! I tested the patch today. Since we can only reproduce this against production data, the setup was a Rails console in a pod running our production image, with the installed gem's lib/ overwritten from three refs, replaying the failing query in a fresh console per run:

• v2.6.5: reproduces - ClosedQueueError warnings from Async, then the request hangs indefinitely.
• master at the PR merge base (8b9f621): same as v2.6.5, so nothing else on master since the tag changes the behaviour.
• AsyncDataloader: Rework task waiting #5672 head (da3d99e): the ClosedQueueError warnings are gone, but the request still hangs indefinitely. The main thread ends up parked in the reactor with nothing left to run:

/usr/local/bundle/gems/async-2.42.0/lib/async/scheduler.rb:456:in 'IO::Event::Selector::EPoll#select'
/usr/local/bundle/gems/async-2.42.0/lib/async/scheduler.rb:456:in 'Async::Scheduler#run_once!'
...
/usr/local/bundle/gems/async-2.42.0/lib/kernel/sync.rb:30:in 'Kernel#Sync'
.../lib/graphql/dataloader/async_dataloader.rb:208:in 'GraphQL::Dataloader::AsyncDataloader#run'

(Line 208 is the patch's Sync do call site - in stock 2.6.5 it's line 197 - so the patched file was definitely the one loaded.)

I suspect the closed? guard in mark_finished turns the previous raise into a silent drop: a task that finishes after its round's close_queues still loses its completion, so @finished_count never catches up to @started_count and the next wait_for_queues blocks on @finished_all_tasks.wait forever. The warnings look like a symptom of the lost completion rather than the cause of the hang.

For what it's worth on the trigger: this query fans out to ~40 concurrent source tasks against a small ActiveRecord pool (4 connections), so a wave of them spends ~5s waiting for a connection and then raises ActiveRecord::ConnectionTimeoutError. The tasks that finish late after that pile-up seem to be the ones that lose their completions.

You asked about the query structure - here's a skeleton of the failing document (fragments inlined, scalar fields collapsed):

query AppletEditPage($id: String!) {
  applet(id: $id) {
    # ~10 scalars
    channel { # ~12 scalars
      live_channel { ... }
    }
    channels { # list: same ~12 scalars per item
      live_channel { ... }
    }
    live_applet { ... }
    tags { ... }
    normalized_applet {
      # ~40 scalars
      channels { # list
        live_channel { ... }       # via one fragment
        live_channels { ... }      # via another fragment (list)
      }
      background_images { ... }
      value_propositions { ... }
      permissions { live_channel { ... } }
      configurations { live_configurations { ... } }
      ingredients { service { ... } }
      stored_fields { ... }
      enablement_rules { ... }
    }
    # These three branches are structurally identical - each step of the
    # applet loads its definition, that definition's channel with its
    # live_channels, plus ingredients and field definitions:
    applet_trigger {
      trigger {
        channel { live_channels { ... } }
        ingredients { ... }
        trigger_fields { ... }
      }
    }
    applet_queries {  # list
      query {
        channel { live_channels { ... } }
        ingredients { ... }
        query_fields { ... }
      }
    }
    applet_actions {  # list
      action {
        channel { live_channels { ... } }
        ingredients { ... }
        action_fields { ... }
      }
    }
  }
}

Nearly every object/list field above resolves through a Dataloader source doing an ActiveRecord lookup. Two things the skeleton doesn't show: the original document is five fragments composed by different frontend components, so the same channel -> live_channels subtree appears six-plus times through different paths (lots of overlapping source batches rather than one clean tree), and the three step branches multiply by the applet's step count - that's what pushes this particular data shape to ~40 concurrent source tasks while most applets stay under the pool limit.

Happy to test further patches - the repro is reliable on our side.

[rmosolgo]

Thanks for giving that branch a try and for sharing all these details. I think you're right that the real problem must not be the ClosedQueueErrors themselves...

I'm trying to think through the kind of deadlock you described. wait_for_queues waits for a promise called @finished_all_tasks:

graphql-ruby/lib/graphql/dataloader/async_dataloader.rb

Line 99 in 8b9f621

@finished_all_tasks.wait

That promise finishes on two conditions. In the success case, all tasks complete:

graphql-ruby/lib/graphql/dataloader/async_dataloader.rb

Lines 127 to 129 in 8b9f621

	if @finished_count == @started_count
	@finished_all_tasks.resolve(true)
	end

In the error case, an error is pushed into the finished tasks queue, then @finished_all_tasks passes it along to the waiting task:

graphql-ruby/lib/graphql/dataloader/async_dataloader.rb

Lines 123 to 124 in 8b9f621

	if t_or_err.is_a?(StandardError)
	@finished_all_tasks.reject(t_or_err)

That is the condition that I expect to unblock wait_for_queues when a #fetch call hits a ConnectionTimeoutError.

Obviously something is getting stuck ... but I don't see how it can be this. Do you see something I'm missing?

Anyways, I'm going to try a couple of things next:

• Extend my attempt at replicating this bug to include a greater variety of lists and reused, cached values from Dataloader, as you described in the comment above. Maybe some other codepath is not handling an error correctly...?
• Rethink the queue design. The thing that always makes this weird is that, by design, tasks don't complete with Dataloader; they yield instead. But I'm going to try again to research this and see if I can spot the flaw.

[vytautaskubilius]

I managed to capture the whole thing. I instrumented Run#new_queues / close_queues / wait_for_queues plus every push into the two queues (with a generation label per new_queues call), and had a watchdog thread dump the async task tree with backtraces mid-hang, then replayed until a run wedged. Two corrections to my earlier guess, both straight from the trace:

• @finished_all_tasks doesn't fail to resolve - it resolves too early.
• The dataloader run actually completes all its work afterwards; what hangs forever is the reactor.

Here's the fatal window, a run_sources round that spawned 5 source tasks (gen8 = the 8th new_queues generation of this request, task ids are object ids, line numbers are from 2.6.5):

[+0.8377s] new_queues gen8 via run_sources
[+0.8378s] push gen8.started task=13a10 status=initialized   (spawn_source_task, line 355)
[+0.8378s] push gen8.started task=13a18 status=initialized
[+0.8378s] push gen8.started task=13a20 status=initialized
[+0.8378s] push gen8.started task=13a28 status=initialized
[+0.8378s] push gen8.started task=13a30 status=initialized
[+0.8378s] wait_for_queues ENTER gen8 started_count=0 finished_count=0
[+0.8426s] push gen8.finished task=13a10   (ensure push, line 351)
[+0.8427s] push gen8.finished task=13a18
[+0.8428s] push gen8.finished task=13a20
[+0.8428s] wait_for_queues EXIT gen8                          <- resolved early
[+0.8428s] close_queues gen8 started_count=3 finished_count=2
[+0.8428s] close gen8.finished pending_size=1                 <- 13a20's completion, never consumed
[+0.8430s] push gen8.finished [QUEUE ALREADY CLOSED] task=13a28  <- the ClosedQueueError
[+0.8436s] close gen8.started pending_size=1                  <- 13a30, never popped

My read of the sequence:

1. All five tasks are pushed into started_tasks before wait_for_queues lets the consumers start.
2. The two counters advance in two independently scheduled consumer tasks, and @finished_count == @started_count is checked on every finish. It held transiently at 2 == 2 - two tasks counted as started, two completions consumed - while the remaining started_tasks pushes were still unconsumed, so @finished_all_tasks resolved with work still in both queues.
3. run_sources re-checked its loop condition, found nothing pending any more (the tasks that did run had drained the shared pending_sources array), exited, and close_queues cancelled both consumers.
4. Task 13a28 was mid-flight and finished 0.2ms after the close, pushing into the closed queue - that's the ClosedQueueError. It's only a symptom of the early close, which is why the closed? guard in AsyncDataloader: Rework task waiting #5672 silences the warnings without changing the hang.
5. Task 13a30 was never popped, so task.run was never called on it. It stays initialized forever.

The request then carries on normally for a bit - the sources were all actually fetched, the remaining rounds complete, and the dataloader's root task finishes. But the orphaned task keeps the reactor alive. This is the task tree while the process hangs, identical at 15s and 45s:

#<Async::Reactor:0x13a88 1 children (running)>
  #<Async::Task:0x13a90 (completed)>          <- Sync's task
    #<Async::Task:0x12ba8 (completed)>        <- the dataloader root task
      #<Async::Task:0x13a30 (initialized)>    <- spawned, never run, unfinishable

Nothing can ever resume 13a30 (its queue is closed and its consumer cancelled), Sync's scheduler loop won't return while it exists, so the request hangs until our process supervisor kills the worker at 30s.

One more data point: this particular run had no fetch errors at all - zero ConnectionTimeoutErrors - so the race looks intrinsic to the counting scheme, and errors or slow fetches just widen the window. (On your error-flow question specifically: fetch errors don't reach the finished_tasks reject path anyway - Source#run_pending_keys rescues them into @results and they re-raise later in the requesting fiber via result_for. But per the above, that turned out not to be what's hanging us.)

Happy to re-run this instrumentation against whatever you try next.

[rmosolgo]

Wow, that's so helpful -- thanks for your in-depth research here, @vytautaskubilius!

This rings a bell. I had trouble getting this shared counter to work right in the first place, which is why I made the @started_tasks queue actually start the task and added the @finished_first_pass promise. Obviously that wasn't a good enough solution. I've been reading (again...) about concurrent job queues and what I'm going to try is improving detection of "has the work all been scheduled?" (by passing the array of jobs to the Run, so it can check if there are any; same thing with pending_sources) and "Are there any outstanding jobs?" by keeping a list of running jobs instead of the two counts.

[rmosolgo]

👋 I continued building on #5672 :

• Use a single bus for task running status (new/paused/resumed/finished/error), which makes it easy to keep a plain Ruby Array of running tasks, removing the need for @started_count and @finished_count.
• Distinguish between "no running tasks" and "all work finished" -- these are different conditions but were mixed in the previous implementations. Update callers to wait on the proper condition.
• Unify the logic between the jobs queues and sources queues. There's still some divergence, but they're handled inline instead of using duplicated methods.

The result is a lot clearer control flow.

I didn't get a replication of this issue, but given the raciness of it, I'm not optimistic about getting it. But this new implementation should address the basic problem. Want to try it again?

(And honestly, I think I'm going to remove the changes to error handling because user code is already wrapped with a rescue. The warnings from Async were helpful in raising a flag that something was actually broken. 👀 )