Building a task queue, Part 2

Friday, November 11, 2022

This post is part 2 of a series where I build a task queue system. Other parts:

Next, I'll beef up my queue library with some extra features.

Bulk dispatch

First, I'll add the ability to dispatch jobs in bulk. This way rather than calling Queuable.dispatch in a loop n times and issuing n database queries, the user can make a single call which issues a single query. This part is pretty easy:

# Add a `dispatch_many` method to our Queueable class
module Gator
  class Queueable
    # ...
    def self.dispatch_many(args, **options)
      jobs = do |job_args|
          id: Gator::Models::Job.generate_job_id,
          args: job_args.to_json,
      Gator::Models::Job.multi_insert(jobs) "Enqueued #{args.size} #{} jobs"
# Usage:
get "/queue/:count?" do
  count = Integer(params[:count] || 1)
  args = { |i| ["Nellie #{i}", "Buster #{i}"] }
  "Queued #{count} jobs"


Next up, supporting a delay before dispatching a job. Say, we want to send a welcome email to a user two hours after they signed up. This should look like:

def signup
  # Create user...
  SendWelcomeEmail.dispatch(, wait: 2 * 60 * 60)

Or just sending an email at a specific date/time (like a payment reminder on a certain date):

SendPaymentReminder.dispatch(, at: reminder_date)

Since we planned ahead for this with the next_execution_at column, this is also straightforward:

def self.dispatch(*args, wait: nil, at: nil)
  job =
    name:, args:, 
    next_execution_at: wait ? ( + wait) : (at ? at : nil),

def self.dispatch_many(args, wait: nil, at: nil)
  next_execution_at = wait ? ( + wait) : (at ? at : nil)

  jobs = do |job_args|
      id: Gator::Models::Job.generate_job_id,
      args: job_args.to_json,
# Usage:
delay = count % 15
Jobs::DoSomeStuff.dispatch("I was dispatched #{delay} minutes ago — at #{}", wait: delay * 60)
INFO [2022-11-01 23:40:43] HIIII I was dispatched 3 minutes ago — at 2022-11-01 23:37:40 +0100 and 
INFO [2022-11-01 23:40:43] Processed job id=job_0426d65614121577f4d3aba1 result=succeeded args=["I was dispatched 3 minutes ago — at 2022-11-01 23:37:40 +0100"]

Of course, there's no guarantee the job will execute at exactly that time. Our workers only poll the database at intervals, so if the execution time reaches during a sleep, the job has to wait for the next poll. Also, a worker might be busy with another job that takes some time. Overall, most timing systems can only guarantee that the task won't be executed until after the timer is up, but not necessarily immediately.

Named queues

Not all jobs are equal. Some are business-critical, some take longer, and so on. In an app with a lot of background jobs, jobs are often assigned to different queues, which are typically named by priority (eg "critical", "high", "default", "low") or by function ("emails", "transactions"), or a mix of both.

By default, all jobs are pushed to the default queue, and the worker executes all jobs on all queues, but using multiple queues allows for greater flexibility and control over how jobs are processed:

  • You can spin up a new worker process and assign it to a queue that gets a lot of jobs, so they aren't kept waiting for long.
  • You could even have a separate server dedicated to a queue with jobs that use a lot of resources (CPU/memory), so they don't affect other jobs.
  • You can tell a worker what queues to prioritize, so jobs on that queue get processed sooner.
  • Workers can be independent, so a problem with one worker doesn't mean there's a backlog on all your jobs

Different job systems do this differently.

  • BullMQ and Que set the priority directly on a job. You can set this to any number, and jobs with a higher priority are executed before any other waiting jobs of a lower (or no) priority.
  • Resque does not support explicit priorities, but instead will check queues in the order you list them. So, for instance, if you specify high,default for a worker, the default queue will only be checked when the high queue is empty. One big weakness of this implicit ordering is queue starvation: in a single-threaded worker, if you keep enqueuing high jobs at a high rate, default jobs will never be processed. The only way to get around this is to start up a dedicated worker for each queue.
  • Sidekiq supports Resque-like implicit ordering as well as explicitly assigning priorities to queues (["critical", 7], ["default", 2]). In this case, this means that when picking a job from the queue, the worker is likely to pick a critical job 7 times out of 9, and a default job 2 times.

I'll be implementing the explicit queue priority approach, as I think it's the clearest and most predictable approach.

To implement this, we allow the user to specify the queue by defining a static queue property:

class SendStorageAlmostUsedUpEmail < Gator::Queueable
  queue_on :critical

We'll also support overriding it when enqueuing a job; you might have a job that should only be sent as high priority in certain scenarios:

case percent_storage_used
when 95
  SendStorageUsageEmail.dispatch(, 95, queue: :critical)
  SendStorageUsageEmail.dispatch(, 80)

(Neat trick: this allows us achieve Que's "set a priority on a job" approach—just use the queue as the priority.)

Right, let's go. First, we update the Queueable class so a job can specify its queue and override it at execution time.

def self.queue_on(queue)
  @queue = queue

class << self
  attr_reader :queue

def self.dispatch(*args, wait: nil, at: nil, queue: nil)
  job =
    name:, args:, 
    next_execution_at: wait ? ( + wait) : (at ? at : nil),
    queue: queue ? queue : (self.queue || 'default'),
  ) "Enqueued job id=#{} args=#{job.args} queue=#{job.queue}"

Now, over to the job server. Our CLI work script needs to change to accept a list of queues: do |parser|
  parser.on("-r", "--require FILE", "File to load at startup. Use this to boot your app.")

  parser.on("-q", "--queue QUEUE", Array,
    "A queue for this worker to check. Pass this flag multiple times to set multiple queues. " +
      "Separate a queue from its priority with a comma. Example: '-q critical,8 -q default,1'"
  ) do |q, priority|
    options[:queue] ||= []
    options[:queue] << [q, priority ? Integer(priority) : 1]
end.parse!(into: options)

require options.delete(:require) if options[:require]

require_relative '../worker'

# CLI option name is `queue`, 
# but Worker argument name is `queues`
options[:queues] = options.delete(:queue)

w =**options)

As for the worker itself, we need to adjust the worker to take an optional list of queues to check, along with their priorities. The queue list should be composed of [<queue_name>, <priority>] tuples. You can also just pass a queue name to default the priority to 1. For example, in the list [[:critical, 6], [:high, 4], [:default, 2], :low], :low has a priority of 1.

We'll also adjust our check_for_jobs query to respect the queues the worker is assigned to. There are three rules:

  • If no queues were specified, just pick the next job as usual.
  • If queues were specified, but they all have the same priority, only pick up jobs that are in any of the queues.
  • If queues were specified, and have different priorities, pick a job randomly using the queue priorities as weights.

So here we go (showing only the changed bits):

module Gator
  class Worker
    attr_reader :polling_interval, :logger, :worker_id, :queues

    def initialize(queues: [])
      @polling_interval = 5
      @logger =
      @worker_id = "wrk_" + SecureRandom.hex(8)
      # Normalize the list of queues
      @queues = (queues || []).map do |q|
        q.kind_of?(Array) ? q : [q, 1]
      # If all queues have the same priority,
      # we can precompute the query, to save time
      all_queues_have_same_priority = { _1[1] }.uniq.size == 1
      @queues_filter = if all_queues_have_same_priority
    def run "Worker #{worker_id} ready"
      queues.empty? ?"Watching all queues") :"Watching queues: #{ { |q, p| "#{q} (priority=#{p})" }.join(', ')}")

      loop do
        if (job = next_job)
          error = execute_job(job)
          cleanup(job, error)
          sleep polling_interval
    # ...

    def check_for_jobs
      query = Models::Job.where(state: "waiting", reserved_by: nil)
      query = query.where { (next_execution_at =~ nil) | (next_execution_at <= }
      if queues.empty? "Checking all queues"
        queues_to_check = queues_filter
        query = query.where(queue: queues_to_check) "Checking queues #{queues_to_check}"
      logger.debug query.sql

    def queues_filter
      return @queues_filter if @queues_filter

      # Weighted random sampling formula from Efraimidis and Spirakis,
      # thanks to
      queue_to_check, _priority = queues.max_by(1) { |(_name, priority)| rand ** (1.0/priority) }.first

    def execute_job(job)
      Object.const_get(*job.args) "Processed job id=#{} result=succeeded queue=#{job.queue}"
    rescue => e "Processed job id=#{} result=failed queue=#{job.queue}"
    # ...

I've added a bunch of logging statements so we can see what's going on at this stage.

Now, here's what happens when we start the worker without specifying any queues (bin/work.rb).

When we specify queues with the same priority (bin/work.rb -q batch,2 -q default,2):

And when we specify different priorities (bin/work.rb -q critical, 3 -q default):


Just to be sure our weighted random algorithm works, let's let it run for a bit. I've turned off the SQL logging, so all that shows is the queue being checked.

We're specifying a priority of 3 for critical and 1 for default. This means jobs processed from these queues should be in the ratio 3:1. How did we do?

Our screenshot has 20 lines of "Checking..." logs, and 14 of them are for critical, while 6 are for default. So we have a ratio of 14:6, which is 2.3:1. Not exact, but good enough. (In fact, if the worker had checked critical just once more instead of default, we'd have had 15:5, which is exactly 3:1 😄.)

Anyway, this is a small sample size. As the number of jobs grows larger, the ratio will tend towards the expected value. (We could also write a test for it, but I've intentionally ignored testing for now.)

Okay, finally, let's watch it in action. I'm creating two jobs, a critical one and a regular-ass one.

class Jobs::CriticalJob < Gator::Queueable
  queue_on :critical

  def handle

class Jobs::RegularAssJob < Gator::Queueable
  def handle

Let's do some dispatching (while the worker is stopped), just to see it in action:

3.times { Jobs::RegularAssJob.dispatch("arg") }
Jobs::RegularAssJob.dispatch_many([["arg"]] * 4, queue: :critical)

(By the way, I've switched from making a route for every thing I need to running things in pry. pry -r ./app/boot.rb)

That's a total of 3 jobs to the default queue, and 5 to the critical queue. Now, if we start the worker: we can see it processes the 5 critical jobs before it even gets to the default.

Note: it isn't always like this! There's every chance the weighted sample gives us the lower priority first, or even twice or three times in a row. As mentioned earlier, the effects are more obvious with a large sample size.

In fact, here's my test with a larger (albeit still small) sample size. I've modified the worker to loop only 200 times and log to stdout:

def run
- loop do
+ 200.times do
    if (job = next_job)
      # ...
 ruby .\gator\bin\work.rb -q critical,3 -q default `
  | Select-String -Pattern '["critical"]' -SimpleMatch `
  | Measure-Object -Line

Lines Words Characters Property
----- ----- ---------- --------

(It's PowerShell, so sorry if the syntax looks strange. 🤣)

critical was checked times, so that's a ratio of 143:57, which is 2.51:1.


Final feature for the day is automatic retries. Implementation:

  • We'll support different retry policies.
  • The user can specify the retry policy per job class (or write a custom policy).
  • After a job is executed, if it falls, we'll use the specified retry policy to calculate the next time we should try again.

Here's what the API will look like:

# Retry every 5 minutes until success, or at most 5 times
retry_with interval: 5 * 60, max_retries: 5

# Retry with exponential backoff
retry_with interval: :exponential

# Put retries on a different queue
retry_with interval: :exponential, queue: :retries

# Custom retry logic
retry_with(queue: :retries) do |exception, retry_count|
  # Don't retry if the record has been deleted
  next false if exception === RecordNotFound
  next :exponential if exception === ExternalServiceDown
  # Retry after 5 minutes, then 10, then 25
  retry_intervals = [5 * 60, 10 * 60, 25 * 60]

In summary, your choices:

  • Specify an interval, either as a number or :exponential for exponential backoff, along with optional max_retries and queue parameters.
  • Pass a block that will be called after every failure. The block can return the next interval (or :exponential), or false to stop retrying.

As usual, we update the Queueable class:

def self.retry_with(
  interval: nil, max_retries: 10, queue: nil, &block
  @retry_strategy = { interval:, max_retries:, queue:, block: }

  class << self
    attr_reader :queue, :retry_strategy

In the worker, the main change is in our cleanup method:

    def cleanup(job, error = nil)
      job.reserved_by = nil
      job.attempts += 1
      job.last_executed_at =
      job_class = Object.const_get(
      if error
        job.state = "failed"
        job.error_details = error
        set_retry_details(job_class.retry_strategy, job, error) if job_class.retry_strategy
        job.state = "succeeded"
    def set_retry_details(retry_strategy, job, error)
      retry_strategy => { interval:, queue:, max_retries:, block: } # Ruby 3 "destructuring" 😍

      retry_count = job.attempts - 1
      if max_retries && retry_count >= max_retries
        job.state = "dead"

      if block
        decision =, retry_count)
        if decision == false
          job.state = "dead"

        interval = decision

      interval = (30 + (retry_count) ** 5) if interval == :exponential
      job.next_execution_at = job.last_executed_at + interval

      job.queue = queue if queue

Let's talk about the exponential option there. Exponential backoff is a common strategy used to retry a task at progressively longer intervals. A good reason for this might be to reduce the load on the resource being accessed or give the problem time to be resolved.

It's called "exponential" because it's typically in the shape of an exponential curve, y = a + bᶜ—y increases slowly at first, but then faster and faster as b (or c) increases. In the real world, though, no one is checking your maths. The important thing is that the intervals increase at a reasonable rate. In fact, the Resque backoffs plugin uses a fixed list of intervals 1m, 10m, 1h, 3h, 6h, rather than a specific formula. Sidekiq's formula is approximately next_interval = n⁴ + 15, where n is the current retry count. So the first retry happens after 15s, the second happens 16s after the first, the third happens 31s after the second, and so on.

I think Sidekiq's intervals are too close together, at least in the early stages; 3 retries within a minute is overkill. I went with interval = n^5 + 30, because I like the spacing of the intervals. Let's check the first 15 retry intervals:

def y(x) = 30 + (x ** 5)

ONE_HOUR = 60 * 60

def print_duration(v)
  return "#{v} seconds" if v < 60
  return "#{v/60.0} minutes" if v < ONE_HOUR
  return "#{v/ONE_HOUR} hours, #{(v % ONE_HOUR)/60.0} minutes" if v < ONE_DAY
  "#{v/ONE_DAY} days, #{(v % ONE_DAY)/ONE_HOUR} hours, #{(v % ONE_HOUR)/60.0} minutes"

(0...15).map { print_duration y(_1) }
 "30 seconds",                                                 
 "31 seconds",                                                 
 "1.0333333333333334 minutes",                                 
 "4.55 minutes",                                               
 "17.566666666666666 minutes",                                 
 "52.583333333333336 minutes",                                 
 "2 hours, 10.1 minutes",                                      
 "4 hours, 40.61666666666667 minutes",                         
 "9 hours, 6.633333333333334 minutes",
 "16 hours, 24.65 minutes",
 "1 days, 3 hours, 47.166666666666664 minutes",
 "1 days, 20 hours, 44.68333333333333 minutes",
 "2 days, 21 hours, 7.7 minutes",
 "4 days, 7 hours, 8.716666666666667 minutes",
 "6 days, 5 hours, 24.233333333333334 minutes"

Finally, we have to update the query in our check_for_jobs method to also look for failed jobs whose retry time is due.

query = Models::Job.where(state: "waiting").
  where { (next_execution_at =~ nil) | (next_execution_at <= }.
  or(state: "failed", next_execution_at: (
  where(reserved_by: nil)

And finally, demo time! With a two-minute interval and a custom queue:

class Jobs::WithRetryInterval < Gator::Queueable
  queue_on :interval
  retry_with interval: 2 * 60, queue: :retries

  def handle(*)
    logger.warn "Going to fail..."
    raise "Always fails"


Retries every 2 minutes and keeps going, because we didn't set a max_retries. ✅

With exponential retry and a limit:

class Jobs::WithExponentialRetry < Gator::Queueable
  queue_on :exponential
  retry_with interval: :exponential, max_retries: 4

  def handle(*)
    logger.warn "Going to fail..."
    raise "Always fails"


You can see the time intervals between retries matches up with our earlier calculations.✅

With a block:

class Jobs::WithCustomRetryLogic < Gator::Queueable
  queue_on :custom
  retry_with do |exception, retry_count|
    [10, 20, 30][retry_count] || false

  def handle(*)
    logger.warn "Going to fail..."
    raise "Always fails"


Uses our custom retry intervals (10s, 20s, 30s), then stops. ✅

Phew, that's a lot! See you in Part 3! Code on GitHub.

Hey👋. I write about interesting software engineering challenges. Want to get updated when I publish new posts? Just visit

(Confession: I built Tentacle.✋ It helps you keep a clean inbox by combining your favourite blogs into one weekly newsletter.)

Powered By Swish