Async::Job and DB pool management

[agalloch]

Hello @ioquatix and thank you for building this set of wonderful machinery! ✨

I've got a question about using Async::Job in a standard Rails app which does both IO waiting and state tracking in the DB. I've built a prototype async job processing server into an existing app, adding ActiveJob, async-job-adapter-active_job and async-job-processor-redis to the mix so ActiveJob takes care of enqueuing the jobs to Redis and a server script dequeues and executes them. I've run different kinds of simulated enqueue/process loads on that app.

What I've found is that under certain conditions the job executor's Fibers get to exhaust the DB pool. This is expected since to my knowledge ActiveRecord is not built for async usage, and so I've started looking for a built-in way to control backpressure. Other than Async::Idler's max load parameter, it was to no avail.

What is the recommended/built-in way to keep consumer Fiber spawning under control? Of course, I could implement my custom Async::Job::Processor::Redis::Server with some sort of semaphore to achieve that but this seems expensive in the long run. Maybe I'm doing something wrong?

Cheers!

---

Test setup

Rails 7.2.2.2
Ruby 3.4.2
macOS Tahoe
Apple M3

[trevorturk]

I'm curious if you've set the Rails isolation to fiber? https://blog.saeloun.com/2022/02/23/rails-fiber-safe-connection-pools/ in case it might help here...

[ioquatix]

What I've found is that under certain conditions the job executor's Fibers get to exhaust the DB pool. This is expected since to my knowledge ActiveRecord is not built for async usage, and so I've started looking for a built-in way to control backpressure. Other than Async::Idler's max load parameter, it was to no avail.

Do you mean you are getting timeouts trying to use the database connection?

[agalloch]

Thank you for the quick response, @ioquatix!

Yup, I got the

ActiveRecord::ConnectionTimeoutError - could not obtain a connection from the pool within 5.000 seconds (waited 5.002 seconds); all pooled connections were in use

in the Rails log. I've tested (on dev machine) with pool size of 5, 20, 40 and 100 connections.

As for setting the isolation_level = :fiber - no, haven't tried that yet. I'm not sure how would that affect the rest of the application which runs vanilla Rails (you know, ActiveRecord, Postgres, Passenger, Sidekiq) and is fairly busy with concurrent workloads.

[ioquatix]

You need to use fiber isolation level, can you try it in your dev environment?

[agalloch]

I've just tried config.active_support.isolation_level = :fiber and I don't see any difference regarding the DB pool exhaustion (i.e. it still times out as before).

Also, if I'm not mistaken, it seems the isolation level is set to :fiber implicitly anyways here: https://github.com/socketry/async-job-adapter-active_job/blob/e64d00a4ce804a589d9e56ae7b8ab29e66334ff1/lib/async/job/adapter/active_job/railtie.rb#L73

Btw, this is my test async worker:

def perform(id)
  ActiveRecord::Base.connection_pool.with_connection do |connection|
    _, updated_at =
      connection.
        execute("select id, updated_at from foos where id = #{id} limit 1").
        values.
        flatten

    sleep 5

    connection.execute "update foos set updated_at = '#{5.minutes.since updated_at}' where id = #{id}"
  end
end

[trevorturk]

I think that with_connection looks right https://blog.saeloun.com/2024/07/01/rails-adds-activerecord-base-with-connection but sleeping in the middle might be the issue? IIRC there's a Task.sleep or something that's fiber safe, but I think it could still be an issue to sleep in the with_connection block?

[agalloch]

Hmm, according to the async gem's documentation Async::Task#sleep is deprecated. Should I still use it?

[trevorturk]

I'm not sure... I think the first question is... can you avoid sleep in the middle of the with_connection block? That new system was designed to check out and back in as quickly as possible to avoid blocking. (In the past, it would hold the connection for an entire request once anything Active Record was triggered, so I thiiiink what you have might be an anti-pattern, instead, consider two with_connection blocks?)

[ioquatix]

You should not use Task#sleep any more.

You will only be able to run as many concurrent queries as you have in your pool. Sleeping while executing a query isn't ideal, best to only hold a connection while doing the query.

Here are some notes: https://gist.github.com/ioquatix/f419bf18d7af8ba5cda81fdadee458f6

You should confirm connection_pool.active_connection? is nil outside of the with_connection block.

Consider experimenting with config.active_record.permanent_connection_checkout = :disallowed to find connection leaks.

[agalloch]

Of course, I'm not running sleep in with_connection blocks - it's just the simplest example I can share publicly. The problem

ActiveRecord::ConnectionTimeoutError - could not obtain a connection from the pool within 5.000 seconds (waited 5.002 seconds); all pooled connections were in use

is present in real production code too, it's just not that easy to demonstrate. Also, I don't think I'm leaking connections because if I wrap example code in a semaphore:

semaphore = Async::Semaphore.new

semaphore.acquire do
  ActiveRecord::Base.connection_pool.with_connection do |connection|
    connection.execute(<SOME SQL>)
    sleep 5
    connection.execute(<SOME SQL>)
  end
end

the problem goes away.

Let's take a step back and focus on the original question - is there a built-in way to control backpressure?

[ioquatix]

ActiveRecord::Base.connection_pool.with_connection is already applying back-pressure with a timeout. The back pressure is the timeout, using a semaphore just waits indefinitely (serialises the job executions).

If your goal is to run more concurrent jobs at once, you need to increase the number of available connections or hold them for a shorter period of time, e.g. use with_connection around each operation:

ActiveRecord::Base.connection_pool.with_connection do |connection|
  connection.execute(<SOME SQL>)
end

sleep 5

ActiveRecord::Base.connection_pool.with_connection do |connection|
  connection.execute(<SOME SQL>)
end

This allows better sharing of connections. Rails 8 does this by default IIRC.