Que Documentation

Que uses a priority-based worker pool. You can manage capacity at different priority levels using --worker-priorities or --worker-count.

• --worker-priorities [LIST]: Assigns specific priorities to workers. For example, --worker-priorities=10,30,50,any,any,any reserves one worker for priority < 10, one for < 30, one for < 50, and three for any priority.
• --worker-count [COUNT]: A shorthand for creating N workers at the any priority level.
• Combining both: If both are provided, --worker-count will pad or trim the list with any workers. For example, --worker-priorities=20,30,40 --worker-count=6 results in 20,30,40,any,any,any.

Que attempts to block worker processes from exiting until jobs complete to avoid data corruption during transactions. However, if jobs are long-running, a normal exit might hang.

If you need to force a shutdown, use SIGKILL. Unlike a normal exit (which uses Thread#kill), SIGKILL causes the Ruby process to end immediately. When the connection to PostgreSQL is lost, the database will roll back any open transactions and unlock the job, allowing it to be retried by another worker. This is the approach used by platforms like Heroku (SIGTERM, then SIGKILL after a timeout).

There are two primary ways to approach testing Que jobs:

1. Synchronous execution: Set Que::Job.run_synchronously = true. This causes JobClass.enqueue to execute the job's logic immediately in the same process, as if you called JobClass.run(*args) directly.
2. Asynchronous assertion: Leave run_synchronously disabled and assert on the state of the job records stored in the database after enqueuing.

Que requires database schema updates for certain releases. Use Que.migrate!(version: X) to upgrade or downgrade. To remove Que entirely, migrate to version 0.

In an ActiveRecord migration:

class UpdateQue < ActiveRecord::Migration[5.0]
  def self.up
    Que.migrate!(version: 3)
  end

  def self.down
    Que.migrate!(version: 2)
  end
end

Manually in a console:

Que.migrate!(version: 3)
Que.db_version # => 3

Que supports multiple queues within a single job table. This is useful for scenarios where different codebases share the same queue.

Running workers for a specific queue: Use the --queue-name flag or the -q flag. You can specify multiple queues.

Enqueuing jobs to a specific queue:

1. At enqueue time: Pass job_options: { queue: 'name' } to the enqueue method.
2. In the Job class: Set self.queue = 'name' within the job class definition.

# Run workers for specific queues
que --queue-name credit_cards
# OR
que -q default -q credit_cards

# Enqueue with specific queue
ProcessCreditCard.enqueue(current_user.id, job_options: { queue: 'credit_cards' })

# Set default queue for a class
class ProcessCreditCard < Que::Job
  self.queue = 'credit_cards'
end

Because the que_jobs table is high-churn, dead tuples can accumulate and slow down job acquisition. While PostgreSQL's autovacuum usually handles this, you may need to run a manual VACUUM if your database is under heavy load or has large tables that delay autovacuum processes.

Example of a recurring manual vacuum job using Sequel:

class ManualVacuumJob < CronJob
  self.priority = 1 # Highest priority to keep the table healthy
  INTERVAL = 300

  def run(args)
    DB.run "VACUUM VERBOSE ANALYZE que_jobs"
  end
end

class ManualVacuumJob < CronJob
  self.priority = 1 # set this to the highest priority since it keeps the table healthy for other jobs
  INTERVAL = 300

  def run(args)
    DB.run "VACUUM VERBOSE ANALYZE que_jobs"
  end
end

Long-running jobs or hanging network requests can block workers and prevent graceful shutdowns. Always use timeouts on operations prone to hanging, such as HTTP requests.

class ScrapeStuff < Que::Job
  def run(url_to_scrape)
    # Use the timeout feature of your HTTP library
    result = YourHTTPLibrary.get(url_to_scrape, timeout: 5)

    ActiveRecord::Base.transaction do
      # Insert result...
      destroy
    end
  end
end

When using Sequel, assign the Sequel database instance directly to Que.connection.

If you are using Sequel's migrator, ensure you require 'que' and set the connection within your migration blocks to ensure the schema updates correctly.

DB = Sequel.connect(ENV['DATABASE_URL'])
Que.connection = DB

# In migrations:
require 'que'
Sequel.migration do
  up do
    Que.connection = self
    Que.migrate!(version: 7)
  end
  down do
    Que.connection = self
    Que.migrate!(version: 0)
  end
end

To use Que with ActiveRecord in a non-Rails application, establish your connection and then tell Que to use the ActiveRecord connection pool.

ActiveRecord::Base.establish_connection(ENV['DATABASE_URL'])

require 'que'
Que.connection = ActiveRecord