Skip to content

warn if max_claims is greater than or equal to connection pool size #109

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
bKP451 wants to merge 7 commits into
base: main
Choose a base branch
from
Open

warn if max_claims is greater than or equal to connection pool size #109

Changes from 2 commits
Commits
Show all changes
7 commits
Select commit Hold shift + click to select a range
0101b39
wip
bKP451 May 28, 2026
8e3a182
verbose logging
bKP451 May 28, 2026
2ec85e1
revert explicit raise of thread error
bKP451 Jun 9, 2026
f472191
chore: revert unnecessary changes from PR (debug logs, thread_pool_si…
bKP451 Jun 9, 2026
bc6c96f
feat: warn at startup when max_claims >= DB connection pool size
bKP451 Jun 9, 2026
dea0914
fix: resolve RSpec/SubjectStub offense in connection pool config chec…
bKP451 Jun 9, 2026
bcc4212
document connection pool size constraint for max_claims
bKP451 Jun 9, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
34 changes: 32 additions & 2 deletions lib/delayed/worker.rb
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -88,31 +88,40 @@ def work_off(num = 100)

while total < num
start = clock_time
say "Attempting to reserve up to #{num - total} job(s)", 'debug'
jobs = reserve_jobs
say 'No jobs reserved; exiting work_off loop', 'debug' if jobs.empty?
break if jobs.empty?

total += jobs.length
pool = Concurrent::FixedThreadPool.new(jobs.length)
say "Reserved #{jobs.length} job(s); dispatching batch", 'debug'
pool = Concurrent::FixedThreadPool.new(thread_pool_size(jobs.length))
jobs.each do |job|
job_say job, 'Queued for thread execution', 'debug'
pool.post do
thread_started = false
self.class.lifecycle.run_callbacks(:thread, self) do
thread_started = true
success.increment if perform(job)
rescue DeserializationError => e
handle_unrecoverable_error(job, e)
rescue Exception => e # rubocop:disable Lint/RescueException
handle_erroring_job(job, e)
end
rescue Exception => e # rubocop:disable Lint/RescueException
say "Job thread crashed with #{e.class.name}: #{e.message}", 'error'
handle_thread_error(job, e, thread_started)
end
end

say 'Waiting for worker threads to finish', 'debug'
pool.shutdown
pool.wait_for_termination
say "Batch finished with #{success.value} successful job(s) out of #{total} attempted so far", 'debug'

break if stop? # leave if we're exiting

elapsed = clock_time - start
say format('Batch elapsed %.4f seconds', elapsed), 'debug'
interruptable_sleep(self.class.min_reserve_interval - elapsed)
end

Expand Down Expand Up @@ -215,6 +224,16 @@ def handle_unrecoverable_error(job, error)
failed(job)
end

def handle_thread_error(job, error, thread_started)
phase = thread_started ? 'after perform' : 'before perform'
job_say job, "thread crashed #{phase} with #{error.class.name}: #{error.message}", 'error'
return if thread_started

handle_erroring_job(job, error)
rescue Exception => inner_error # rubocop:disable Lint/RescueException
job_say job, "could not record pre-perform thread crash: #{inner_error.class.name}: #{inner_error.message}", 'error'
end

# The backend adapter may return either a list or a single job
# In some backends, this can be controlled with the `max_claims` config
# Either way, we map this to an array of job instances
Expand All @@ -235,6 +254,17 @@ def reload!
Rails.application.reloader.reload! if defined?(Rails.application.reloader) && Rails.application.reloader.check!
end

def thread_pool_size(job_count)
return job_count unless Delayed::Job.respond_to?(:connection_pool)

pool_size = Delayed::Job.connection_pool.size
return job_count unless pool_size

[job_count, [pool_size - 1, 1].max].min
rescue StandardError
job_count
end

@smudge smudge Jun 1, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

There may be a way to more proactively (e.g. during worker boot/initialization) establish if Delayed::Worker.max_claims > Delayed::Job.connection_pool.size rather than performing this thread_pool_size logic on every pickup loop. (My understanding is that Delayed::Job.connection_pool.size is informed by the pool size config in database.yml, and should not change once the app has loaded.)

The current pickup strategy is also intended to avoid picking up more work than the worker can immediately begin working off (to avoid holding unworked jobs in memory), so it may make sense to raise or warn up front (again, during boot / worker initialization) if a misconfiguration is detected.

@bKP451 bKP451 Jun 8, 2026

Copy link
Copy Markdown
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@smudge,

Yeah it is a great way to raise on warn upfront during app initializer ( config/initializers/delayed.rb ) for
Delayed::Worker.max_claims > Delayed::Job.connection_pool.size

On our project, our actual problem was max_claims being equal to pool_size. On investigation we found that main worker ( i.e server ) also needs to connect to DB to keep track of threads, locking, unlocking, polling etc. Therefor a worker thread will throw ActiveRecord::ConnectionTimeoutError if it cannot get DB connection after awaiting for a checkout_time ( i.e ActiveRecord::Base.connection_pool.checkout_timeout ). For long running jobs, DB connection won't get free for that worker thread and exception is raised

Job thread crashed with ActiveRecord::ConnectionTimeoutError: could not obtain a connection from the pool within 5.000 seconds (waited 5.001 seconds); all pooled connections were in use
Here 5.000 seconds is checkout_time

Problem illustration

   MAX_CLAIMS = 5, POOL_SIZE = 5


   main worker(server) housekeeping ─🔑          ← needs 1
   Job 1 ─🔑                        \
   Job 2 ─🔑                         \
   Job 3 ─🔑                          ├─ needs 5
   Job 4 ─🔑                         /
   Job 5 ─❌ (no key left!)         /

   6 requests  ▶  5 keys  ▶  somebody loses

We solved it by reducing concurrent worker threads to pool minus 1

Override MAX_CLAIMS at config/initializers/delayed.rb

# Reserve 1 DB connection for the worker's own housekeeping (polling + locking jobs).
db_pool_size = ActiveRecord::Base.connection_pool.size
Delayed::Worker.max_claims = [db_pool_size - 1, 1].max

WDYT ?

@smudge smudge Jun 8, 2026 •
edited
Loading

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Yes, that's a good callout -- we always include a buffer in our connection pool size (both for web and worker counts). We also sometimes need more than 1 connection per thread, but the same general rule applies there too -- basically, if your code generally needs N connections, and your max claims is M, you want N * (M + 1) connections. The most common case is N=1 though, and I think that would be easy enough to detect with >= (rather than the > I had originally proposed):

Delayed::Worker.max_claims >= Delayed::Job.connection_pool.size


def clock_time
Process.clock_gettime(Process::CLOCK_MONOTONIC)
end
Expand Down