Support for async streaming in Solid Queue for AI response handling

[darkamenosa]

Hi folks!

I've been using Solid Queue in my Rails app and loving it so far. However, I've run into an interesting use case that I'd love to
discuss and potentially see supported.

I'm building an application that heavily relies on streaming responses from AI services (think ChatGPT, Claude, etc.). These AI responses can be quite lengthy and come in as continuous streams that need to be processed and forwarded to users in real-time.

Currently, from what I understand reading through the documentation and codebase, Solid Queue's thread-based model might have some limitations when it comes to handling these long-running streaming operations efficiently. The synchronous nature of job processing means that each streaming AI response essentially blocks a worker thread for the entire duration of the stream, which could be several seconds or even minutes for longer responses.

My current workaround involves spinning up multiple separate queue servers to increase concurrency and handle more simultaneous streams, but this feels like throwing hardware at what might be better solved architecturally.

I'm wondering if there are plans (or if it would even make sense) to add async/streaming support to Solid Queue? Something that could handle these long-running streaming operations without blocking worker threads would be amazing. Maybe using Ruby's Fiber scheduler or integrating with async gems?

Would love to hear your thoughts on:

1. Whether this is something you've considered for Solid Queue's roadmap
2. If there are better patterns I should be using for this use case
3. Whether this aligns with Solid Queue's design philosophy

Thanks for all the great work on this gem! 🙏

[rosa]

Hey @darkamenosa, thanks for opening this issue and for your kind words 🙏 I hadn't thought about this at all, TBH, until I read https://paolino.me/async-ruby-is-the-future/. That means it's not in the roadmap. I'm not quite sure yet! My first instinct is to rely on a different Active Job backend for this, like https://github.com/socketry/async-job-adapter-active_job or https://github.com/tuwukee/jiggler, in the same way you'd replace Puma with Falcon if you wanted to rely on fibers instead of threads.

This might change in the future, of course, depending on the direction Rails goes.

[darkamenosa]

Hi @rosa,

Sorry for the late reply. I missed this message.

Yes. I got the same thought after read that post. I contacted the guy, he suggested using a combination of:

• https://github.com/socketry/async-job-adapter-active_job : this for AI stream or IO tasks
• And Solid Queue: For CPU heavy tasks

For example:

# Base job for LLM operations
class LLMJob < ApplicationJob
  self.queue_adapter = :async_job
end

# All LLM jobs inherit from this
class ChatResponseJob < LLMJob
  def perform(conversation_id, message)
    # Runs with async-job adapter
    response = RubyLLM.chat.ask(message)
    # ...
  end
end

# CPU-intensive jobs use default adapter (e.g., Sidekiq)
class ImageProcessingJob < ApplicationJob
  def perform(image_id)
    # Runs with your default adapter
    # ...
  end
end

I like the idea. But it seems I can't make it work. Solid queue worked as expect, however that async_job adapter didn't work as expected. I'm still figure it out.

I will keep update to let you know about the progress.

[film42]

@darkamenosa if you're concerned about losing progress if a job needs to restart, you can look into capturing each emitted chunk and pass that back in as input to the LLM. Something like:

contents = <<~EOF
#{prompt}

I'm sorry! I cut you off before you could finish.
Please continue writing your response from exactly where you left off! 

Here is what you wrote out before I stopped you:
#{prior_output}"

Might work, might not!

[rosa]

Hey! Thanks for the thoughtful discussion. To sum up, a combination of different job backends is for now the recommended solution: use Solid Queue for discrete job processing, and long-running streaming I/O operations are better served by fibre-based async job backends.

Going to close this, thanks again for chiming in 🙏🏻

[crmne]

FWIW, I ended up working on this and opened a first implementation in #728.

It adds async worker execution mode, so Solid Queue can handle cooperative I/O-bound jobs without switching to a different backend.