Should you get an async queue closed error after upgrading from 2.25.0 to 2.26.0?

[bkuhlmann]

Hello. 👋

I've been using the following code to build release notes by shelling to the system with Async 2.25.0 and earlier:

barrier = Async::Barrier.new

Sync do
  tasks = projects_data.all.select(&:publishable?).sort_by(&:name).map do |project|
    root = BUILD_DIR.join("feeds/projects").mkpath

    barrier.async do
      CACHE_DIR.join("projects", project.name).change_dir do
        system "milestoner build --format feed --max 10 --tail tag " \
               "--output #{root} --basename #{project.name}"
      end
    end
  end

  tasks.each(&:wait)
ensure
  barrier.stop
end

All this does is run the Milestoner CLI to build release notes for each project in the cache directory (nothing fancy). I never had a problem with this until upgrading to 2.26.0 and higher. Starting with 2.26.0, I started getting the following stack dumps for each failed task:

Example Stack Dump

3.07s     warn: Async::Task [oid=0x4dc8] [ec=0x4dd0] [pid=19909] [2025-08-31 14:28:20 -0600]
              | Task may have ended with unhandled exception.
              |   Async::Queue::ClosedError: Cannot enqueue items to a closed queue.
              |   → $USER/.cache/frum/versions/3.4.5/lib/ruby/gems/3.4.0/gems/async-2.28.0/lib/async/queue.rb:76 in 'Async::Queue#enqueue'
              |     $USER/.cache/frum/versions/3.4.5/lib/ruby/gems/3.4.0/gems/async-2.28.0/lib/async/queue.rb:125 in 'Async::Queue#signal'
              |     $USER/.cache/frum/versions/3.4.5/lib/ruby/gems/3.4.0/gems/async-2.28.0/lib/async/barrier.rb:53 in 'block in Async::Barrier#async'
              |     $USER/.cache/frum/versions/3.4.5/lib/ruby/gems/3.4.0/gems/async-2.28.0/lib/async/task.rb:183 in 'block in Async::Task#run'
              |     $USER/.cache/frum/versions/3.4.5/lib/ruby/gems/3.4.0/gems/async-2.28.0/lib/async/task.rb:427 in 'block in Async::Task#schedule'
              |   Caused by Async::Stop: Task was stopped
              |   → $USER/.cache/frum/versions/3.4.5/lib/ruby/gems/3.4.0/gems/async-2.28.0/lib/async/scheduler.rb:414 in 'IO::Event::Selector::KQueue#process_wait'
              |     $USER/.cache/frum/versions/3.4.5/lib/ruby/gems/3.4.0/gems/async-2.28.0/lib/async/scheduler.rb:414 in 'Async::Scheduler#process_wait'
              |     lib/site/build.rb:543 in 'Kernel#system'
              |     lib/site/build.rb:543 in 'block (4 levels) in <main>'
              |     $USER/.cache/frum/versions/3.4.5/lib/ruby/gems/3.4.0/gems/refinements-13.4.0/lib/refinements/pathname.rb:27 in 'block in change_dir'
              |     $USER/.cache/frum/versions/3.4.5/lib/ruby/gems/3.4.0/gems/refinements-13.4.0/lib/refinements/pathname.rb:27 in 'Dir.chdir'
              |     $USER/.cache/frum/versions/3.4.5/lib/ruby/gems/3.4.0/gems/refinements-13.4.0/lib/refinements/pathname.rb:27 in 'change_dir'
              |     lib/site/build.rb:542 in 'block (3 levels) in <main>'
              |     $USER/.cache/frum/versions/3.4.5/lib/ruby/gems/3.4.0/gems/async-2.28.0/lib/async/barrier.rb:51 in 'block in Async::Barrier#async'
              |     $USER/.cache/frum/versions/3.4.5/lib/ruby/gems/3.4.0/gems/async-2.28.0/lib/async/task.rb:183 in 'block in Async::Task#run'
              |     $USER/.cache/frum/versions/3.4.5/lib/ruby/gems/3.4.0/gems/async-2.28.0/lib/async/task.rb:427 in 'block in Async::Task#schedule'
              |   Caused by Async::Stop::Cause: Stopping task!
              |   → $USER/.cache/frum/versions/3.4.5/lib/ruby/gems/3.4.0/gems/async-2.28.0/lib/async/task.rb:256 in 'Async::Task#stop'
              |     $USER/.cache/frum/versions/3.4.5/lib/ruby/gems/3.4.0/gems/async-2.28.0/lib/async/barrier.rb:93 in 'block in Async::Barrier#stop'
              |     $USER/.cache/frum/versions/3.4.5/lib/ruby/gems/3.4.0/gems/async-2.28.0/lib/async/list.rb:298 in 'Async::List::Iterator#each'
              |     $USER/.cache/frum/versions/3.4.5/lib/ruby/gems/3.4.0/gems/async-2.28.0/lib/async/list.rb:311 in 'Async::List::Iterator.each'
              |     $USER/.cache/frum/versions/3.4.5/lib/ruby/gems/3.4.0/gems/async-2.28.0/lib/async/list.rb:180 in 'Async::List#each'
              |     $USER/.cache/frum/versions/3.4.5/lib/ruby/gems/3.4.0/gems/async-2.28.0/lib/async/barrier.rb:92 in 'Async::Barrier#stop'
              |     lib/site/build.rb:552 in 'block in <main>'
              |     $USER/.cache/frum/versions/3.4.5/lib/ruby/gems/3.4.0/gems/async-2.28.0/lib/async/task.rb:183 in 'block in Async::Task#run'
              |     $USER/.cache/frum/versions/3.4.5/lib/ruby/gems/3.4.0/gems/async-2.28.0/lib/async/task.rb:427 in 'block in Async::Task#schedule'

If I add a rescue Async::Queue::ClosedError to the Sync block, I'll still get the stack dumps but the code will complete and produce the desired output.

Based on the Release Notes, I see there were Barrier improvements but nothing that says a breaking change was introduced. Am I missing something or doing something wrong? I wouldn't expect a minor upgrade to break like this. If I need to log a proper issue for this, let me know, but figured I'd ask first.

[ioquatix]

At first glance that does seem a bit odd. I'll review it in more detail.

Okay, here is the fix: #414

Also, in your code better to replace tasks.each(&:wait) with barrier.wait.

[bkuhlmann]

Thanks!

Not sure why I didn't see this reply sooner but have upgraded to Async 2.29.1 and the issue is resolved. I switched to barrier.wait as well. The new error indicating that the barrier has been stopped is helpful as well.

I didn't realize, until now, that you can't define the barrier once and then reuse elsewhere. This means, in order to upgrade to 2.29.1, I had to define a new instance of my barrier for the specific Sync block (as shown above) in order to not get the stopped barrier. After that, everything worked as before.

[ioquatix]

I was actually a little disappointed that I had to make that change. But the push towards (thread) safer primitives has meant that barrier should be one-shot - because there is no easy way to model creating tasks then stopping, and then creating more tasks, stopping again, etc => race conditions. So, I decided to clean up the semantics, and basically you experienced the result of that, so sorry about that.

[bkuhlmann]

Yeah, I was thinking about that too because I wanted a way to define the barrier once which is interesting because I got away with it (without realizing it was a mistake) until 2.26.0. 😅 Good lesson to learn, though.