Centralized job locking into one Locker+JobBuffer per process

README.md

@@ -113,6 +113,65 @@ To determine what happens when a job is queued, you can set Que's mode. There ar
 **If you're using ActiveRecord to dump your database's schema, [set your schema_format to :sql](http://guides.rubyonrails.org/migrations.html#types-of-schema-dumps) so that Que's table structure is managed correctly.** (You can use schema_format as :ruby if you want but keep in mind this is highly advised against, as some parts of Que will not work.)
 
 
+## Running workers (`bin/que`)
+
+This fork ships a `bin/que` executable for running a pool of worker threads in a
+dedicated process:
+
+```shell
+$ bin/que ./config/environment.rb
+```
+
+Run `bin/que -h` for the full list of options. The most commonly used ones:
+
+  * `-c, --worker-count [COUNT]` - number of worker threads to run (default: 1)
+  * `-q, --queue-name [NAME]` - primary queue to work jobs from (default: `default`)
+  * `--secondary-queues a,b` - additional queues to fall back to once the primary queue is empty
+  * `-i, --wake-interval [SECONDS]` - how often to poll Postgres when there's no work available (default: 5.0)
+  * `-p, --metrics-port [PORT]` - serve Prometheus metrics and a health-check endpoint (`/`) on this port
+  * `--timeout [SECONDS]` - how long to wait for in-flight jobs to finish on SIGINT/SIGTERM before forcing them to stop
+
+### Job-locking architecture (`-a`/`--architecture`)
+
+Que supports two strategies for how worker threads acquire (lock) jobs from Postgres,
+selected with `-a`/`--architecture` on the command line, or the `QUE_ARCHITECTURE`
+environment variable:
+
+  * **`legacy`** (the default) - each worker thread independently polls Postgres and
+    locks its own jobs. This is the architecture Que has always used, and remains the
+    default because it's the most battle-tested.
+  * **`centralized`** - a single Locker thread per process polls Postgres and locks jobs
+    in batches, handing them out to worker threads through an in-memory buffer. This
+    exists to fix a scaling problem with `legacy`: with N worker threads, `legacy` runs
+    N independent lock queries against `que_jobs` on every poll cycle, and contention
+    between them gets worse as you add more workers. `centralized` collapses that down
+    to one lock query per process, regardless of worker count.
+
+```shell
+# Explicitly select an architecture on the command line:
+$ bin/que --architecture=centralized ./config/environment.rb
+
+# Or via the environment (useful for toggling per-deployment without a code change):
+$ QUE_ARCHITECTURE=centralized bin/que ./config/environment.rb
+```
+
+The two architectures are meant to be indistinguishable from the outside - same job
+semantics ("works each job exactly once"), same Prometheus metrics, same health-check
+behavior (`Que::Middleware::WorkerHealthCheck`) - so it's safe to flip between them with
+just a redeploy/restart, no migration or code change required. If you're seeing lock
+contention or CPU usage that scales with worker count, try `centralized` on a
+low-traffic queue or environment first, watch it for a while, and roll it out further
+from there before making it the default in your deployment configuration - `legacy` has
+years of production usage behind it, `centralized` doesn't yet.
+
+If you're constructing a `Que::WorkerGroup` directly (rather than via `bin/que`), the
+same option is available as `architecture:`:
+
+```ruby
+Que::WorkerGroup.start(4, architecture: :centralized)
+```
+
+
 ## Related Projects
 
   * [que-web](https://github.com/statianzo/que-web) is a Sinatra-based UI for inspecting your job queue.

bin/que

@@ -51,6 +51,10 @@ OptionParser.new do |opts|
     options.run_at_cursor = true
   end
 
+  opts.on("-a", "--architecture [ARCHITECTURE]", String, "Job-locking architecture to use: 'legacy' (default; one Locker per worker thread) or 'centralized' (one Locker per process, batching lock queries into a shared buffer - reduces concurrent lock queries against Postgres at higher worker counts, but is newer and less battle-tested; see README)") do |architecture|
+    options.architecture = architecture
+  end
+
   opts.on("-c", "--worker-count [COUNT]", Integer, "Start this many threaded workers") do |worker_count|
     options.worker_count = worker_count
   end
@@ -108,6 +112,12 @@ run_at_cursor    = options.run_at_cursor || false
 worker_count     = options.worker_count  || 1
 timeout          = options.timeout
 secondary_queues = options.secondary_queues || []
+architecture     = (options.architecture || ENV["QUE_ARCHITECTURE"] || Que::WorkerGroup::DEFAULT_ARCHITECTURE).to_sym
+
+unless Que::WorkerGroup::ARCHITECTURES.include?(architecture)
+  $stdout.puts "Bad architecture: '#{architecture}' (expected one of: #{Que::WorkerGroup::ARCHITECTURES.join(', ')})"
+  exit 1
+end
 
 Que.logger ||= Logger.new($stdout)
 
@@ -127,6 +137,7 @@ end
 
 worker_group = Que::WorkerGroup.start(
   worker_count,
+  architecture: architecture,
   queue: queue_name,
   wake_interval: wake_interval,
   lock_cursor_expiry: cursor_expiry,

lib/que.rb

@@ -5,12 +5,15 @@
 require_relative "que/adapters/base"
 require_relative "que/job"
 require_relative "que/job_timeout_error"
+require_relative "que/job_buffer"
 require_relative "que/leaky_bucket"
 require_relative "que/locker"
+require_relative "que/centralized_locker"
 require_relative "que/migrations"
 require_relative "que/sql"
 require_relative "que/version"
 require_relative "que/worker"
+require_relative "que/centralized_worker"
 require_relative "que/worker_group"
 require_relative "que/middleware/worker_collector"
 require_relative "que/middleware/queue_collector"

lib/que/adapters/active_record_with_lock.rb

@@ -38,6 +38,9 @@ def execute(command, params = [])
         when :lock_job
           queue, cursor, run_at_lower_bound = params
           lock_job_with_lock_database(queue, cursor, run_at_lower_bound)
+        when :lock_jobs_batch
+          queue, cursor, run_at_lower_bound, limit, held_ids = params
+          lock_jobs_with_lock_database(queue, cursor, run_at_lower_bound, limit, held_ids)
         when :unlock_job
           job_id = params[0]
           unlock_job(job_id)
@@ -65,6 +68,61 @@ def lock_job_with_lock_database(queue, cursor, run_at_lower_bound = Que::Locker:
         end
       end
 
+      # Batched counterpart to lock_job_with_lock_database: keeps looping through
+      # candidates (skipping any we fail to advisory-lock, same as the single-job
+      # version) until we've locked `limit` jobs or there are no more candidates left.
+      #
+      # We deliberately do NOT delegate to the generic recursive-CTE `lock_jobs_batch` SQL
+      # here (unlike the default adapter) - that query takes its advisory locks on
+      # whatever connection runs it, which for this adapter must be the dedicated
+      # lock_connection_pool, not the job_connection_pool used for find_job_to_lock. Row
+      # skipping is instead handled by `FOR UPDATE SKIP LOCKED` in find_job_to_lock, one
+      # candidate row at a time, exactly as the single-job version does.
+      #
+      # held_ids (a "{1,2,3}" Postgres array literal, same format the Locker builds for
+      # the generic SQL path) lists job_ids the Locker already holds. We must skip these
+      # without even attempting pg_try_advisory_lock: since that lock is reentrant within
+      # a session, re-locking a job we already hold would trivially "succeed" again and
+      # get double-dispatched to another worker.
+      def lock_jobs_with_lock_database(queue, cursor, run_at_lower_bound, limit, held_ids)
+        held_id_set = parse_id_array_literal(held_ids)
+        locked_jobs = []
+
+        while locked_jobs.size < limit
+          job_locked = false
+
+          candidate =
+            observe(duration_metric: FindJobSecondsTotal, labels: { queue: queue }) do
+              Que.transaction do
+                job_to_lock = Que.execute(:find_job_to_lock, [queue, cursor, run_at_lower_bound])
+                break job_to_lock if job_to_lock.empty?
+
+                cursor = job_to_lock.first["job_id"]
+
+                unless held_id_set.include?(cursor)
+                  job_locked = pg_try_advisory_lock?(cursor)
+                  observe(count_metric: FindJobHitTotal, labels: { queue: queue, job_hit: job_locked })
+                end
+
+                job_to_lock
+              end
+            end
+
+          # No more candidates left in the table for this queue - stop looking.
+          break if candidate.empty?
+
+          # Whether or not we won the lock, the cursor has advanced past this candidate,
+          # so the next iteration looks further down the queue.
+          locked_jobs.concat(candidate) if job_locked
+        end
+
+        locked_jobs
+      end
+
+      def parse_id_array_literal(literal)
+        literal.to_s.delete_prefix("{").delete_suffix("}").split(",").reject(&:empty?).to_set(&:to_i)
+      end
+
       def pg_try_advisory_lock?(job_id)
         checkout_lock_database_connection do |conn|
           conn.execute(