diff --git a/README.md b/README.md index 7181aa2e..a7038d54 100644 --- a/README.md +++ b/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. diff --git a/bin/que b/bin/que index 307076ac..386c8e79 100755 --- a/bin/que +++ b/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, diff --git a/lib/que.rb b/lib/que.rb index 7c1bee80..4e4b8ce5 100644 --- a/lib/que.rb +++ b/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" diff --git a/lib/que/adapters/active_record_with_lock.rb b/lib/que/adapters/active_record_with_lock.rb index a7bfaedf..f4a7c54d 100644 --- a/lib/que/adapters/active_record_with_lock.rb +++ b/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( diff --git a/lib/que/centralized_locker.rb b/lib/que/centralized_locker.rb new file mode 100644 index 00000000..5087faa1 --- /dev/null +++ b/lib/que/centralized_locker.rb @@ -0,0 +1,269 @@ +# frozen_string_literal: true + +require "prometheus/client" +require_relative "leaky_bucket" +require_relative "locker" + +module Que + # The CentralizedLocker runs a single background thread per process that polls + # Postgres for jobs, locks them, and pushes them into a shared JobBuffer for worker + # threads (CentralizedWorker) to consume. + # + # This backs `architecture: :centralized` on WorkerGroup - see the README for context + # on why it exists (fixing lock contention that scales with worker count) and how to + # opt into it. The original per-worker-thread Locker remains the default + # (`architecture: :legacy`) and is unaffected by this class. + # + # Centralising locking behind one Locker per process (rather than one per worker + # thread) means a process with N worker threads makes one batched lock query per poll + # cycle, instead of N independent single-job lock queries competing with each other. + # + # For performance, the locker keeps track of the last job ID it locked per queue, and + # uses that job's ID to begin the next attempt to acquire jobs. There is a point where + # the performance of lock acquisition goes off a cliff, and using the previous job's ID + # delays the point at which that occurs. + # + # For more information, see the 'Predicate Specificity' chapter of: + # https://brandur.org/postgres-queues + class CentralizedLocker + RUN_AT_CURSOR_RESET = Locker::RUN_AT_CURSOR_RESET + + DEFAULT_WAKE_INTERVAL = 5.0 # seconds + + METRICS = [ + # This is the only metric unique to CentralizedLocker - everything else below + # reuses Locker's existing Prometheus constants (rather than declaring duplicates + # with the same metric names), since only one architecture is ever active in a + # given process and operators shouldn't need separate dashboards depending on + # which is selected. + JobsLockedTotal = Prometheus::Client::Counter.new( + :que_locker_jobs_locked_total, + docstring: "Counter of jobs locked and pushed into the job buffer", + labels: %i[queue primary_queue worked_queue], + ), + ].freeze + + def initialize( + queue:, + cursor_expiry:, + job_buffer:, + window: nil, + budget: nil, + secondary_queues: [], + run_at_cursor: false, + wake_interval: DEFAULT_WAKE_INTERVAL + ) + @queue = queue + @cursor_expiry = cursor_expiry + @run_at_cursor = run_at_cursor + @job_buffer = job_buffer + @wake_interval = wake_interval + @queue_cursors = {} + @run_at_lower_bounds = {} + @queue_expires_at = {} + @secondary_queues = secondary_queues + @consolidated_queues = Array.wrap(queue).concat(secondary_queues) + + # Create a bucket that has 100% capacity, so even when we don't apply a limit we + # have a valid bucket that we can use everywhere + @leaky_bucket = LeakyBucket.new(window: window || 1.0, budget: budget || 1.0) + + @stop = false + @last_poll_result = nil + + # Initialise the cursor maps + handle_expired_cursors! + end + + attr_reader :thread + + # Starts the background thread that polls for jobs. Returns self. + def start + @thread = Thread.new { poll_loop } + self + end + + # Ask the poll loop to stop after its current cycle. Does not block - use + # wait_until_stopped to wait for the thread to actually exit (and for buffered jobs to + # be unlocked). + def stop! + @stop = true + end + + def wait_until_stopped + @thread&.join + end + + def alive? + !!@thread&.alive? + end + + def healthy? + @last_poll_result != :postgres_error + end + + private + + def poll_loop + until @stop + poll + break if @stop + + # If the buffer's still got room after that poll, we didn't find enough work to + # fill it (the queue's quiet, or every attempt is failing) - back off for a full + # wake_interval rather than busy-looping against Postgres. Only wait specifically + # for a freed-up slot when we filled the buffer completely, since then there's + # reason to think more work is waiting right behind it. + if @job_buffer.available_capacity.positive? + @job_buffer.wait_for_stop(@wake_interval) + else + @job_buffer.wait_for_space(@wake_interval) + end + end + ensure + # Drain whatever's left in the buffer (jobs we locked but no worker had claimed yet) + # so we don't leave advisory locks held on jobs nobody is going to work. + unlock_jobs(@job_buffer.clear) + end + + # Runs a single poll cycle: for each consolidated queue (primary, then secondaries), in + # order, lock as many due jobs as there's free space in the job buffer for, and push + # them in. Updates @last_poll_result (used by #healthy?) as a side effect, so that + # calling #poll directly (e.g. in tests) reflects health correctly too. + def poll + handle_expired_cursors! + + @consolidated_queues.each do |queue| + space = @job_buffer.available_capacity + break if space <= 0 + + jobs = lock_jobs_for(queue, space) + next if jobs.empty? + + # jobs is ordered by (priority, run_at, job_id) - see the ORDER BY in + # lock_jobs_batch - so the last entry is the furthest we've progressed through + # the queue in that order. + last = jobs.last + @queue_cursors[queue] = last[:job_id] + @run_at_lower_bounds[queue] = last[:run_at] if @run_at_cursor + + rejected = @job_buffer.push(*jobs) + unlock_jobs(rejected) if rejected.any? + end + + @last_poll_result = :ok + rescue PG::Error, Adapters::UnavailableConnection + @last_poll_result = :postgres_error + end + + # Locks up to `limit` jobs from `queue`. If the current cursor turns up nothing, retry + # once from the start of the queue - it's not necessarily the case that there aren't + # jobs in the queue, we may have been excluding candidates due to the cursor that are + # now due to be worked. We leave cursors for other queues alone; if a later poll of + # this queue starts finding jobs again, handle_expired_cursors! will eventually reset + # it based on the expiry rather than this immediate retry. + def lock_jobs_for(queue, limit) + cursor = @queue_cursors.fetch(queue, 0) + run_at_lower_bound = @run_at_lower_bounds.fetch(queue, RUN_AT_CURSOR_RESET) + + jobs = lock_jobs_in(queue, cursor, run_at_lower_bound, limit) + + if jobs.empty? && cursor != 0 + reset_cursor_for!(queue) + jobs = lock_jobs_in(queue, 0, RUN_AT_CURSOR_RESET, limit) + end + + jobs + end + + def lock_jobs_in(queue, cursor, run_at_lower_bound, limit) + observe(nil, Locker::ThrottleSecondsTotal, worked_queue: queue) { @leaky_bucket.refill } + jobs = @leaky_bucket.observe { execute_lock_jobs_in(queue, cursor, run_at_lower_bound, limit) } + + verify_existence(queue, jobs) + end + + def execute_lock_jobs_in(queue, cursor, run_at_lower_bound, limit) + strategy = cursor.zero? ? "full" : "cursor" + # Que.execute JSON-encodes Array params (see Adapters::Base#execute), which isn't + # valid input for a native Postgres array - build the `{1,2,3}` literal ourselves. + held_ids = "{#{@job_buffer.held_ids.join(',')}}" + + observe(Locker::AcquireTotal, Locker::AcquireSecondsTotal, + worked_queue: queue, + strategy: strategy) do + Que.execute(:lock_jobs_batch, [queue, cursor, run_at_lower_bound, limit, held_ids]) + end + end + + # Check that jobs we just locked haven't just been worked and destroyed by another + # locker (it's possible to lock a job that's just been destroyed because pg locks + # don't obey MVCC). This can happen with jobs that are already being worked when we + # begin our lock query: the job row exists but is locked at the point that we + # materialise our job rows for use in the recursive query. At some point after that, + # but before we attempt to take our lock, the original worker destroys the job row and + # unlocks the advisory lock. We then attempt to lock the ID, and succeed, despite the + # job having already been worked. + # + # Any jobs that didn't survive this check are unlocked immediately and excluded from + # what gets pushed into the buffer, so we never hand a worker a job that's already + # been completed. + def verify_existence(queue, jobs) + return jobs if jobs.empty? + + # Que.execute JSON-encodes Array params (see Adapters::Base#execute), which isn't + # valid input for a native Postgres array - build the `{1,2,3}` literal ourselves so + # it's passed through as a plain string and `$1::bigint[]` can cast it directly. + id_array_literal = "{#{jobs.map { |job| job[:job_id] }.join(',')}}" + + surviving_ids = observe(Locker::ExistsTotal, Locker::ExistsSecondsTotal, worked_queue: queue) do + Que.execute(:check_jobs_batch, [id_array_literal]).map { |row| row[:job_id] } + end.to_set + + destroyed, survived = jobs.partition { |job| !surviving_ids.include?(job[:job_id]) } + unlock_jobs(destroyed) if destroyed.any? + + JobsLockedTotal.increment(by: survived.size, + labels: { worked_queue: queue, queue: @queue, + primary_queue: @queue }) + + survived + end + + def unlock_jobs(jobs) + jobs.each do |job| + observe(Locker::UnlockTotal, Locker::UnlockSecondsTotal, worked_queue: job[:queue]) do + Que.execute(:unlock_job, [job[:job_id]]) + end + end + end + + def handle_expired_cursors! + @consolidated_queues.each do |queue| + queue_cursor_expires_at = @queue_expires_at.fetch(queue, monotonic_now) + reset_cursor_for!(queue) if queue_cursor_expires_at < monotonic_now + end + end + + def reset_cursor_for!(queue) + @queue_cursors[queue] = 0 + @run_at_lower_bounds[queue] = RUN_AT_CURSOR_RESET + @queue_expires_at[queue] = monotonic_now + @cursor_expiry + end + + def observe(metric, metric_duration, labels = {}) + now = monotonic_now + yield + ensure + metric&.increment(labels: labels.merge(queue: @queue, primary_queue: @queue)) + metric_duration&.increment( + by: monotonic_now - now, + labels: labels.merge(queue: @queue, primary_queue: @queue), + ) + end + + def monotonic_now + Process.clock_gettime(Process::CLOCK_MONOTONIC) + end + end +end diff --git a/lib/que/centralized_worker.rb b/lib/que/centralized_worker.rb new file mode 100644 index 00000000..28df1a72 --- /dev/null +++ b/lib/que/centralized_worker.rb @@ -0,0 +1,234 @@ +# frozen_string_literal: true + +require "pg" +require "benchmark" + +require_relative "worker" +require_relative "centralized_locker" +require_relative "job_timeout_error" + +module Que + # CentralizedWorker pulls already-locked jobs from a shared JobBuffer (filled by a + # single CentralizedLocker per process) rather than locking its own jobs directly. + # + # This backs `architecture: :centralized` on WorkerGroup - see the README for context + # on why it exists and how to opt into it. The original per-worker-thread Worker + # remains the default (`architecture: :legacy`) and is unaffected by this class. + # + # Deliberately reuses Worker's Prometheus metric constants and LongRunningMetricTracer + # rather than declaring its own copies - only one architecture is ever active in a + # given process, and operators shouldn't need separate dashboards depending on which is + # selected. + class CentralizedWorker + DEFAULT_QUEUE = Worker::DEFAULT_QUEUE + DEFAULT_WAKE_INTERVAL = Worker::DEFAULT_WAKE_INTERVAL + + def initialize( + job_buffer:, + queue: DEFAULT_QUEUE, + wake_interval: DEFAULT_WAKE_INTERVAL + ) + @queue = queue + @wake_interval = wake_interval + @job_buffer = job_buffer + @tracer = Worker::LongRunningMetricTracer.new(self) + @stop = false # instruct worker to stop + @stopped = false # mark worker as having stopped + @current_running_job = nil + @last_work_result = nil + end + + attr_reader :metrics + + def work_loop + return if @stop + + @tracer.trace(Worker::RunningSecondsTotal, queue: @queue, primary_queue: @queue) do + loop do + job = @tracer.trace(Worker::SleepingSecondsTotal, queue: @queue, primary_queue: @queue) do + @job_buffer.shift + end + + case work(job) + when :postgres_error + Que.logger&.info(event: "que.postgres_error", wake_interval: @wake_interval) + @tracer.trace(Worker::SleepingSecondsTotal, queue: @queue, primary_queue: @queue) do + sleep(@wake_interval) + end + when :job_not_found + # The buffer only yields false when it's been stopped (or in the brief window + # during shutdown before our own @stop flag is set) - avoid hot-looping while + # we wait for that flag to catch up. + sleep(0.01) unless @stop + when :job_worked + nil # immediately try to pull another job + end + + break if @stop + end + end + ensure + @stopped = true + end + + # Executes a single already-locked job (or does nothing but record :job_not_found if + # job is falsy, which happens when the JobBuffer has no job and has been stopped). + def work(job) + return (@last_work_result = :job_not_found) unless job + + Que.adapter.checkout do + log_keys = { + priority: job["priority"], + # TODO alerting / monitoring pre addition of secondary queues + # assume that a worker would not work a job from another queue. + # With the addition of secondary queues this is no longer true. + # To support moving to the new `primary_queue` field without + # disrupting existing tooling we "lie" and say the queue worked + # is the primary queue. Longer term alerting / monitoring should move + # to look at `primary_queue` and `worked_queue` instead. + queue: @queue, + primary_queue: @queue, + worked_queue: job["queue"], + handler: job["job_class"], + job_class: actual_job_class_name(job["job_class"], job["args"]), + job_error_count: job["error_count"], + que_job_id: job["job_id"], + } + + labels = { + job_class: actual_job_class_name(job["job_class"], job["args"]), + priority: job["priority"], + queue: @queue, + primary_queue: @queue, + worked_queue: job["queue"], + } + + begin + klass = class_for(job[:job_class]) + + log_keys.merge!( + (klass.log_context_proc&.call(job) || {}), + ) + + Que.logger&.info( + log_keys.merge( + event: "que_job.job_begin", + msg: "Job acquired, beginning work", + latency: job["latency"], + ), + ) + + # Note the time spent waiting in the queue before being processed, and update + # the jobs worked count here so that latency_seconds_total / worked_total + # doesn't suffer from skew. + Worker::JobLatencySecondsTotal.increment(by: job[:latency], labels: labels) + Worker::JobWorkedTotal.increment(labels: labels) + + duration = Benchmark.measure do + # TODO: _run -> run_and_destroy(*job[:args]) + @tracer.trace(Worker::JobWorkedSecondsTotal, labels) do + klass.new(job).tap do |job_instance| + @current_running_job = job_instance + begin + job_instance._run + ensure + @current_running_job = nil + end + end + end + end.real + + Que.logger&.info( + log_keys.merge( + event: "que_job.job_worked", + msg: "Successfully worked job", + duration: duration, + ), + ) + rescue StandardError, NotImplementedError, JobTimeoutError => e + Worker::JobErrorTotal.increment(labels: labels) + Que.logger&.error( + log_keys.merge( + event: "que_job.job_error", + msg: "Job failed with error", + error: e.inspect, + ), + ) + + # For compatibility with que-failure, we need to allow failure handlers to be + # defined on the job class. + if klass.respond_to?(:handle_job_failure) + klass.handle_job_failure(e, job) + else + handle_job_failure(e, job) + end + end + end + + @last_work_result = :job_worked + rescue PG::Error, Adapters::UnavailableConnection => _e + # In the event that our Postgres connection is bad, we don't want that error to halt + # the work loop. Instead, we should let the work loop sleep and retry. + @last_work_result = :postgres_error + ensure + unlock(job) + end + + def stop! + @stop = true + @current_running_job&.stop! + end + + def stopped? + @stopped + end + + def healthy? + @last_work_result != :postgres_error + end + + def collect_metrics + @tracer.collect + end + + private + + def unlock(job) + return unless job + + Que.execute(:unlock_job, [job[:job_id]]) + rescue PG::Error, Adapters::UnavailableConnection + # If we can't unlock (e.g. the connection just died), the advisory lock will be + # released anyway once this backend's connection is torn down/recycled. + ensure + # Tell the Locker we're done with this job regardless of whether the unlock query + # itself succeeded - otherwise it stays in the JobBuffer's held-ids set forever, + # permanently excluding it from every future lock query on this process. + @job_buffer.release(job[:job_id]) if job + end + + # Set the error and retry with back-off + def handle_job_failure(error, job) + count = job[:error_count].to_i + 1 + + Que.execute( + :set_error, [ + count, + (count**4) + 3, # exponentially back off when retrying failures + "#{error.message}\n#{error.backtrace.join("\n")}", + *job.values_at(*Job::JOB_INSTANCE_FIELDS), + ] + ) + end + + def class_for(string) + Que.constantize(string) + end + + def actual_job_class_name(class_name, args) + return args.first["job_class"] if class_name.include?("ActiveJob::QueueAdapters") + + class_name + end + end +end diff --git a/lib/que/job_buffer.rb b/lib/que/job_buffer.rb new file mode 100644 index 00000000..91dfa999 --- /dev/null +++ b/lib/que/job_buffer.rb @@ -0,0 +1,151 @@ +# frozen_string_literal: true + +module Que + # A bounded, thread-safe queue of already-locked jobs, shared between a single process's + # Locker (the producer, which polls Postgres and pushes locked jobs in) and its pool of + # Workers (the consumers, which block on #shift until a job is available). + # + # Centralising job acquisition behind this buffer means only the Locker ever queries + # Postgres to lock jobs - worker threads never do, so a process with N worker threads + # makes one lock query per poll cycle instead of N. + class JobBuffer + def initialize(maximum_size:) + @maximum_size = maximum_size + @array = [] + + # Every job_id currently locked by this process and not yet released - both jobs + # still sitting in @array, and ones already shifted out to a worker that's actively + # processing them. The Locker excludes these from every lock query (see Locker + # #poll), because pg_try_advisory_lock is reentrant *within the same session*: if we + # didn't exclude them, a poll could re-lock (and re-dispatch) a job we already hold, + # since our own session is always able to "acquire" a lock it already has. + @held_ids = Set.new + + @stopping = false + @mutex = Mutex.new + @job_available = ConditionVariable.new + @space_available = ConditionVariable.new + end + + attr_reader :maximum_size + + # Push newly-locked jobs onto the buffer, sorted so that higher-priority (and then + # earlier-run_at) jobs are shifted first. Returns any jobs that didn't fit, so the + # caller (the Locker) can unlock them again rather than leave them buffered forever. + def push(*jobs) + sync do + return jobs if @stopping + + space = _available_capacity + accepted = jobs.first(space) + rejected = jobs.drop(accepted.size) + + unless accepted.empty? + @array.concat(accepted) + @array.sort_by! { |job| [job[:priority], job[:run_at], job[:job_id]] } + accepted.each { |job| @held_ids << job[:job_id] } + @job_available.broadcast + end + + rejected + end + end + + # Blocks until a job is available or the buffer is stopped (in which case it returns + # false). + def shift + sync do + loop do + return false if @stopping + + unless @array.empty? + job = @array.shift + @space_available.broadcast + return job + end + + @job_available.wait(@mutex) + end + end + end + + # Blocks (up to `timeout` seconds) until there's free capacity in the buffer, or the + # buffer is stopped. Used by the Locker to back off when its last poll filled the + # buffer completely, while waking promptly whenever a worker frees up a slot. + def wait_for_space(timeout) + sync do + return if @stopping || _available_capacity.positive? + + @space_available.wait(@mutex, timeout) + end + end + + # Blocks up to `timeout` seconds, or until the buffer is stopped - unlike + # #wait_for_space, this always waits the full timeout even when space is already + # available. Used by the Locker when its last poll *didn't* fill the buffer (e.g. the + # queue's empty, or every attempt is failing), so it backs off for a full cycle + # instead of busy-looping against Postgres as fast as the CPU allows. + def wait_for_stop(timeout) + sync do + return if @stopping + + @space_available.wait(@mutex, timeout) + end + end + + # Called by a Worker once it's done with a job (whether it succeeded, failed, or the + # unlock itself errored) - marks the job_id as no longer held, so a future poll is free + # to consider it again. + def release(job_id) + sync { @held_ids.delete(job_id) } + end + + # A snapshot of every job_id currently held (buffered or checked out to a worker), + # for the Locker to exclude from its next lock query. + def held_ids + sync { @held_ids.to_a } + end + + def available_capacity + sync { _available_capacity } + end + + def size + sync { @array.size } + end + + def stop + sync do + @stopping = true + @job_available.broadcast + @space_available.broadcast + end + end + + def stopping? + sync { @stopping } + end + + # Drains and returns all buffered (but not yet claimed) jobs, so the caller can unlock + # them. Used during graceful shutdown, so we never leave an advisory lock held on a job + # that's sitting unclaimed in the buffer. + def clear + sync do + jobs = @array + @array = [] + jobs.each { |job| @held_ids.delete(job[:job_id]) } + jobs + end + end + + private + + def _available_capacity + maximum_size - @array.size + end + + def sync(&block) + @mutex.synchronize(&block) + end + end +end diff --git a/lib/que/middleware/worker_health_check.rb b/lib/que/middleware/worker_health_check.rb index d4636db6..e1fe4544 100644 --- a/lib/que/middleware/worker_health_check.rb +++ b/lib/que/middleware/worker_health_check.rb @@ -2,25 +2,42 @@ module Que module Middleware - # Rack application that reflects the health of all workers in the worker - # group. Returns 503 if all workers have encountered a postgres error on - # their last work cycle, 200 otherwise. + # Rack application that reflects the health of the worker group's Locker and its + # workers. Returns 503 if the process's Locker has died or is stuck in a persistent + # postgres_error state (in which case no worker in this process will ever find new + # work, regardless of how healthy the workers themselves look), or if every worker + # itself has encountered a postgres error on its last work cycle. Returns 200 + # otherwise. # - # A single worker in postgres_error state may be transient; only when all - # workers are unhealthy is the pod restarted with fresh database connections. + # A single worker in postgres_error state may be transient; only when the Locker is + # unhealthy, or all workers are unhealthy, is the pod restarted with fresh database + # connections. class WorkerHealthCheck def initialize(worker_group) @worker_group = worker_group end def call(_env) - workers = @worker_group.workers - if workers.empty? || workers.any?(&:healthy?) + if healthy? [200, { "Content-Type" => "text/plain" }, ["healthy"]] else [503, { "Content-Type" => "text/plain" }, ["unhealthy"]] end end + + private + + def healthy? + return false if locker_unhealthy? + + workers = @worker_group.workers + workers.empty? || workers.any?(&:healthy?) + end + + def locker_unhealthy? + locker = @worker_group.locker + locker && !locker.healthy? + end end end end diff --git a/lib/que/sql.rb b/lib/que/sql.rb index 5dc82df2..69181583 100644 --- a/lib/que/sql.rb +++ b/lib/que/sql.rb @@ -96,6 +96,74 @@ module Que AND job_id = $4::bigint ), + # Identical to lock_job, except that it locks up to $4 jobs in a single round trip + # instead of exactly one. Used by the (single, per-process) Locker to fill its + # JobBuffer, rather than by each worker locking its own single job. + # + # As with lock_job, the top-level SELECT's LIMIT is what bounds the recursion - Postgres + # pulls rows from the recursive term lazily, so it stops iterating as soon as either + # $4 jobs have been locked or there are no more candidates left in the queue. + # + # $5 is an exclusion list of job_ids the Locker already holds (buffered, or checked + # out to a worker). This is essential for correctness, not just an optimisation: + # pg_try_advisory_lock succeeds again if *the same session* already holds the lock, so + # without this exclusion, a poll that resets its job_id cursor back to the start of the + # queue (see Locker#lock_jobs_for) would re-lock - and re-dispatch - a job this same + # Locker is already working, causing it to run twice. + lock_jobs_batch: %{ + WITH RECURSIVE jobs AS ( + SELECT (j).*, pg_try_advisory_lock((j).job_id) AS locked + FROM ( + SELECT j + FROM que_jobs AS j + WHERE queue = $1::text + AND job_id > $2 + AND NOT job_id = ANY($5::bigint[]) + AND run_at >= COALESCE($3::timestamptz, '-infinity'::timestamptz) + AND run_at <= now() + AND retryable = true + ORDER BY priority, run_at, job_id + LIMIT 1 + ) AS t1 + UNION ALL ( + SELECT (j).*, pg_try_advisory_lock((j).job_id) AS locked + FROM ( + SELECT ( + SELECT j + FROM que_jobs AS j + WHERE queue = $1::text + AND NOT job_id = ANY($5::bigint[]) + AND run_at >= COALESCE($3::timestamptz, '-infinity'::timestamptz) + AND run_at <= now() + AND retryable = true + AND (priority, run_at, job_id) > (jobs.priority, jobs.run_at, jobs.job_id) + ORDER BY priority, run_at, job_id + LIMIT 1 + ) AS j + FROM jobs + WHERE jobs.job_id IS NOT NULL + LIMIT 1 + ) AS t1 + ) + ) + SELECT queue, priority, run_at, job_id, job_class, retryable, args, error_count, + extract(epoch from (now() - run_at)) as latency + FROM jobs + WHERE locked + LIMIT $4::bigint + }, + + # Batched counterpart to the existence re-check done after taking a job's advisory + # lock (pg locks don't obey MVCC, so a job can be locked despite having just been + # destroyed by the worker that was already working it - see Locker for details). + # job_id is globally unique, so it alone is sufficient to identify surviving jobs. + check_jobs_batch: %( + SELECT job_id + FROM que_jobs + WHERE retryable = true + AND job_id = ANY($1::bigint[]) + ), + set_error: %{ UPDATE que_jobs SET error_count = $1::integer, @@ -169,6 +237,15 @@ module Que SELECT pg_advisory_unlock($1) }, + # Note: job_id is a strict `>` cursor bound here (unlike lock_job's `>=`). This query + # is shared by both Adapters::ActiveRecordWithLock#lock_job_with_lock_database + # (single job at a time) and #lock_jobs_with_lock_database (used by + # architecture: :centralized to lock a batch across several calls within one poll). + # The batched caller advances its cursor to a job_id it may have *just* locked in a + # previous call within the same poll, without necessarily having released it yet - an + # inclusive `>=` would let this query re-select that same row, and since + # pg_try_advisory_lock succeeds again for a lock the same session already holds, + # that job would be locked (and dispatched) a second time. find_job_to_lock: %{ SELECT queue, @@ -184,7 +261,7 @@ module Que WHERE queue = $1::text AND run_at <= now() AND retryable = true - AND job_id >= $2 + AND job_id > $2 AND run_at >= COALESCE($3::timestamptz, '-infinity'::timestamptz) ORDER BY priority, run_at, job_id FOR UPDATE SKIP LOCKED diff --git a/lib/que/worker_group.rb b/lib/que/worker_group.rb index 328f0eff..0c0ddf07 100644 --- a/lib/que/worker_group.rb +++ b/lib/que/worker_group.rb @@ -1,10 +1,25 @@ # frozen_string_literal: true require_relative "worker" +require_relative "centralized_worker" +require_relative "locker" +require_relative "centralized_locker" +require_relative "job_buffer" module Que class WorkerGroup DEFAULT_STOP_TIMEOUT = 5 # seconds + + # :legacy - each Worker owns its own Locker and independently polls Postgres (the + # architecture Que has always used). Safe, proven default. + # :centralized - a single CentralizedLocker per process, batching lock queries into a + # shared JobBuffer that CentralizedWorker threads pull from. Reduces the number of + # concurrent lock queries hitting Postgres from one-per-worker-thread to + # one-per-process, at the cost of being newer and less battle-tested. See README + # for guidance on rolling this out. + ARCHITECTURES = %i[legacy centralized].freeze + DEFAULT_ARCHITECTURE = :legacy + METRICS = [ ActiveWorkersCount = Prometheus::Client::Gauge.new( :que_worker_group_active_workers_count, @@ -15,40 +30,135 @@ class WorkerGroup :que_worker_group_expected_workers_count, docstring: "Number of configured workers", ), + + LockerAliveGauge = Prometheus::Client::Gauge.new( + :que_worker_group_locker_alive, + docstring: "Whether this process's single CentralizedLocker thread is alive " \ + "(1) or dead (0). Only meaningful with architecture: :centralized - " \ + "the CentralizedLocker is what polls Postgres for jobs there, so if " \ + "it's dead, no worker in this process will ever find new work. " \ + "Always 0 under architecture: :legacy, where there is no single " \ + "locker thread.", + ), ].freeze - def self.start(count, **kwargs) + def self.start( + count, + architecture: DEFAULT_ARCHITECTURE, + queue: Worker::DEFAULT_QUEUE, + wake_interval: Worker::DEFAULT_WAKE_INTERVAL, + lock_cursor_expiry: wake_interval, + lock_window: nil, + lock_budget: nil, + secondary_queues: [], + run_at_cursor: false + ) + unless ARCHITECTURES.include?(architecture) + raise ArgumentError, + "unknown architecture: #{architecture.inspect} (expected one of #{ARCHITECTURES.inspect})" + end + Que.logger.info( msg: "Starting workers", event: "que.worker.start", worker_count: count, + architecture: architecture, ) - workers = Array.new(count) { Worker.new(**kwargs) } + kwargs = { + queue: queue, + wake_interval: wake_interval, + lock_cursor_expiry: lock_cursor_expiry, + lock_window: lock_window, + lock_budget: lock_budget, + secondary_queues: secondary_queues, + run_at_cursor: run_at_cursor, + } + + case architecture + when :legacy then start_legacy(count, **kwargs) + when :centralized then start_centralized(count, **kwargs) + end + end + + def self.start_legacy( + count, queue:, wake_interval:, lock_cursor_expiry:, lock_window:, lock_budget:, secondary_queues:, + run_at_cursor: + ) + workers = + Array.new(count) do + Worker.new( + queue: queue, + wake_interval: wake_interval, + lock_cursor_expiry: lock_cursor_expiry, + lock_window: lock_window, + lock_budget: lock_budget, + secondary_queues: secondary_queues, + run_at_cursor: run_at_cursor, + ) + end + + new(workers, start_worker_threads(workers), nil, nil) + end + private_class_method :start_legacy + + def self.start_centralized( + count, queue:, wake_interval:, lock_cursor_expiry:, lock_window:, lock_budget:, secondary_queues:, + run_at_cursor: + ) + job_buffer = JobBuffer.new(maximum_size: count) + + locker = + CentralizedLocker.new( + queue: queue, + cursor_expiry: lock_cursor_expiry, + window: lock_window, + budget: lock_budget, + secondary_queues: secondary_queues, + run_at_cursor: run_at_cursor, + wake_interval: wake_interval, + job_buffer: job_buffer, + ).start + + workers = + Array.new(count) do + CentralizedWorker.new(queue: queue, wake_interval: wake_interval, job_buffer: job_buffer) + end + + new(workers, start_worker_threads(workers), locker, job_buffer) + end + private_class_method :start_centralized + + # We want to ensure that our control flow for Que threads and our metric endpoints + # are prioritised above CPU intensive work of jobs. If people use the metrics + # endpoint as a liveness check then its essential we don't timeout when we do busy + # work. + # + # We can set our worker thread priority to be a value less than our parent thread to + # ensure this is the case. Different Ruby runtimes have different priority systems, + # but -10 should do the trick for all of them. + def self.start_worker_threads(workers) worker_threads = workers.map { |worker| Thread.new { worker.work_loop } } - # We want to ensure that our control flow for Que threads and our metric endpoints - # are prioritised above CPU intensive work of jobs. If people use the metrics - # endpoint as a liveness check then its essential we don't timeout when we do busy - # work. - # - # We can set our worker thread priority to be a value less than our parent thread to - # ensure this is the case. Different Ruby runtimes have different priority systems, - # but -10 should do the trick for all of them. worker_threads.each_with_index do |thread, index| thread.priority = Thread.current.priority - 10 thread.name = "worker-thread-#{index}" end - new(workers, worker_threads) + worker_threads end + private_class_method :start_worker_threads - def initialize(workers, worker_threads) + # locker and job_buffer are only present under architecture: :centralized - nil under + # :legacy, where each worker owns its own locker instead. + def initialize(workers, worker_threads, locker, job_buffer) @workers = workers @worker_threads = worker_threads + @locker = locker + @job_buffer = job_buffer end - attr_reader :workers + attr_reader :workers, :locker def collect_metrics # Threads that are no longer running will have either a status of `nil` (terminated @@ -62,11 +172,25 @@ def collect_metrics ActiveWorkersCount.set(active_threads.count) ExpectedWorkersCount.set(workers.count) + LockerAliveGauge.set(@locker&.alive? ? 1 : 0) end def stop(timeout = DEFAULT_STOP_TIMEOUT) Que.logger.info(msg: "Asking workers to finish", event: "que.worker.finish_wait") + # Under architecture: :centralized, stop polling for new jobs, wake up (and stop) + # any workers blocked waiting on the buffer, and wait for the locker to + # drain+unlock anything it had already buffered but that no worker had claimed + # yet. We do all of this before tearing down worker threads below, so we never + # hand a freshly-locked job to a worker that's already being asked to stop. Under + # :legacy there's no shared locker/buffer to stop - each worker's own locker just + # stops when the worker itself does, below. + if @locker + @locker.stop! + @job_buffer.stop + @locker.wait_until_stopped + end + @workers.each(&:stop!) # Asynchronously stop all workers, sending a join method call with a timeout. This diff --git a/spec/integration/integration_spec.rb b/spec/integration/integration_spec.rb index f1106d4e..9edde370 100644 --- a/spec/integration/integration_spec.rb +++ b/spec/integration/integration_spec.rb @@ -4,14 +4,14 @@ require "que/worker" # required to prevent autoload races # rubocop:disable RSpec/DescribeClass -RSpec.describe "multiple workers" do +RSpec.shared_examples "multiple workers" do |architecture| context "with one worker and many jobs" do it "works each job exactly once" do 10.times.each { |i| FakeJob.enqueue(i) } expect(QueJob.count).to eq(10) - with_workers(1) { wait_for_jobs_to_be_worked } + with_workers(1, architecture: architecture) { wait_for_jobs_to_be_worked } expect(QueJob.count).to eq(0) expect(FakeJob.log.count).to eq(10) @@ -26,7 +26,7 @@ expect(QueJob.count).to eq(2) - with_workers(1) { wait_for_jobs_to_be_worked(timeout: 1) } + with_workers(1, architecture: architecture) { wait_for_jobs_to_be_worked(timeout: 1) } expect(QueJob.count).to eq(1) expect(FakeJob.log.count).to eq(1) @@ -40,7 +40,7 @@ expect(QueJob.count).to eq(2) - with_workers(1, secondary_queues: ["non-default"]) do + with_workers(1, architecture: architecture, secondary_queues: ["non-default"]) do wait_for_jobs_to_be_worked(timeout: 1) end @@ -55,7 +55,7 @@ expect(QueJob.count).to eq(3) - with_workers(1, secondary_queues: ["non-default"]) do + with_workers(1, architecture: architecture, secondary_queues: ["non-default"]) do wait_for_jobs_to_be_worked(timeout: 1) end @@ -71,7 +71,7 @@ expect(QueJob.count).to eq(1) - with_workers(5) { wait_for_jobs_to_be_worked } + with_workers(5, architecture: architecture) { wait_for_jobs_to_be_worked } expect(QueJob.count).to eq(0) expect(FakeJob.log.count).to eq(1) @@ -95,7 +95,7 @@ expect(QueJob.count).to eq(3) - with_workers(5) { wait_for_jobs_to_be_worked } + with_workers(5, architecture: architecture) { wait_for_jobs_to_be_worked } expect(QueJob.count).to eq(0) expect(User.count).to eq(3) @@ -110,7 +110,7 @@ # Sleep to let the worker pick-up the SleepJob, then stop the worker with an # aggressive timeout. This should cause JobTimeout to be raised in the worker # thread. - with_workers(1, stop_timeout: 0.01) { sleep 0.1 } + with_workers(1, architecture: architecture, stop_timeout: 0.01) { sleep 0.1 } sleep_job = QueJob.last @@ -125,7 +125,7 @@ # Sleep 0.1s to let the worker pick-up the SleepJob, then stop the worker with a # a long enough timeout to let an iteration of sleep complete. - with_workers(1, stop_timeout: 0.3) { sleep 0.1 } + with_workers(1, architecture: architecture, stop_timeout: 0.3) { sleep 0.1 } expect(QueJob.count).to eq(0) expect(InterruptibleSleepJob.log.count).to eq(1) @@ -133,4 +133,17 @@ end end end + +# Run the exact same behavioural suite against both locking architectures (see +# Que::WorkerGroup) - the two implementations are meant to be indistinguishable from the +# outside, and this is our regression coverage for that. +RSpec.describe "multiple workers, across architectures" do + context "with architecture: legacy" do + it_behaves_like "multiple workers", :legacy + end + + context "with architecture: centralized" do + it_behaves_like "multiple workers", :centralized + end +end # rubocop:enable RSpec/DescribeClass diff --git a/spec/lib/que/centralized_locker_spec.rb b/spec/lib/que/centralized_locker_spec.rb new file mode 100644 index 00000000..3486b007 --- /dev/null +++ b/spec/lib/que/centralized_locker_spec.rb @@ -0,0 +1,247 @@ +# frozen_string_literal: true + +require "spec_helper" + +RSpec.describe Que::CentralizedLocker do + subject(:locker) do + described_class.new( + queue: queue, + cursor_expiry: cursor_expiry, + job_buffer: job_buffer, + ) + end + + let(:queue) { "default" } + let(:cursor_expiry) { 0 } + let(:job_buffer) { Que::JobBuffer.new(maximum_size: buffer_size) } + let(:buffer_size) { 10 } + + # Avoid leaking held advisory locks across examples for jobs left buffered but never + # worked by a test. + after do + job_buffer.clear.each { |job| Que.execute(:unlock_job, [job[:job_id]]) } + end + + describe "#poll (via #start)" do + before { allow(Que).to receive(:execute).and_call_original } + + # Runs a single poll cycle directly (rather than starting the background thread), + # so we can assert deterministically on what got locked and pushed. + def poll + locker.send(:poll) + end + + # Simulates a worker successfully finishing the job, so its slot in the buffer frees + # up and its advisory lock is released, exactly as Worker#unlock would (including + # releasing it from the buffer's held-ids set). + def work(job) + job_hash = job_buffer.shift + expect(job_hash[:job_id]).to eql(job[:job_id]) + Que.execute(:unlock_job, [job_hash[:job_id]]) + job_buffer.release(job_hash[:job_id]) + + # Destroy the job to simulate the behaviour of the queue, and allow our lock query + # to discover new jobs. + QueJob.find(job[:job_id]).destroy! + end + + # held_ids is computed from the buffer's actual current state, since it changes as + # jobs are locked/pushed/worked over the course of a test. + def expect_to_lock_with(cursor:, run_at_lower_bound: nil, limit: buffer_size) + held_ids = "{#{job_buffer.held_ids.join(',')}}" + expect(Que).to receive(:execute). + with(:lock_jobs_batch, [queue, cursor, run_at_lower_bound, limit, held_ids]) + end + + context "with no jobs to lock" do + it "scans entire table and leaves the buffer empty" do + expect(Que).to receive(:execute).with(:lock_jobs_batch, [queue, 0, nil, buffer_size, "{}"]) + + poll + + expect(job_buffer.size).to eq(0) + end + end + + context "with just one job to lock" do + let!(:job_1) { FakeJob.enqueue(1, queue: queue, priority: 1).attrs } + let(:cursor_expiry) { 60 } + + # Pretend time isn't moving, as we don't want to test cursor expiry here + before { allow(Process).to receive(:clock_gettime).and_return(0) } + + # We want our workers to start from the front of the queue immediately after finding + # no jobs are available to work. + it "will use a cursor until no jobs are found" do + expect_to_lock_with(cursor: 0) + poll + work(job_1) + + expect_to_lock_with(cursor: job_1[:job_id]) + poll + + expect_to_lock_with(cursor: 0) + poll + end + end + + context "with jobs to lock" do + let!(:job_1) { FakeJob.enqueue(1, queue: queue, priority: 1).attrs } + let!(:job_2) { FakeJob.enqueue(2, queue: queue, priority: 2).attrs } + let!(:job_3) { FakeJob.enqueue(3, queue: queue, priority: 3).attrs } + + it "locks all due jobs (up to the buffer's capacity) in a single poll" do + expect_to_lock_with(cursor: 0) + poll + + expect(job_buffer.size).to eq(3) + end + + context "when the buffer has less capacity than there are jobs" do + let(:buffer_size) { 2 } + + it "locks only as many jobs as fit" do + expect_to_lock_with(cursor: 0, limit: 2) + poll + + expect(job_buffer.size).to eq(2) + end + end + + # rubocop:disable RSpec/SubjectStub + # rubocop:disable RSpec/InstanceVariable + context "on subsequent polls" do + context "with non-zero cursor expiry" do + let(:cursor_expiry) { 5 } + let(:buffer_size) { 1 } + + before do + # we need this to avoid flakiness during resetting the cursor. + # Cursors are reset in the beginning when the locker class object is created. + # It is reset in handle_expired_cursors! method. Sometimes the execution is fast enough that + # the condition to reset is not met because the Process.clock_gettime remains same(monotonic_now method). + locker.instance_variable_get(:@queue_expires_at)[queue] = + Process.clock_gettime(Process::CLOCK_MONOTONIC) + cursor_expiry + allow(locker).to receive(:monotonic_now) { @epoch } + end + + # This test simulates repeated polling. We're trying to prove that the locker + # will use the previous job's ID as a cursor until the expiry has elapsed, + # after which we'll reset. + it "continues lock from previous job id, until cursor expires" do + @epoch = Process.clock_gettime(Process::CLOCK_MONOTONIC) + expect_to_lock_with(cursor: 0, limit: 1) + poll + work(job_1) + + @epoch += 2 + expect_to_lock_with(cursor: job_1[:job_id], limit: 1) + poll + work(job_2) + + @epoch += cursor_expiry # our cursor should now expire + expect_to_lock_with(cursor: 0, limit: 1) + poll + work(job_3) + end + end + end + # rubocop:enable RSpec/SubjectStub + # rubocop:enable RSpec/InstanceVariable + end + + context "with run_at_cursor enabled" do + subject(:locker) do + described_class.new( + queue: queue, cursor_expiry: 60, job_buffer: job_buffer, run_at_cursor: true, + ) + end + + let(:buffer_size) { 1 } + let!(:job_1) { FakeJob.enqueue(1, queue: queue, priority: 1).attrs } + + before do + allow(Process).to receive(:clock_gettime).and_return(0) + locker.instance_variable_get(:@queue_expires_at)[queue] = 60 + end + + it "advances the run_at cursor to the last locked job's run_at after locking" do + expect_to_lock_with(cursor: 0, run_at_lower_bound: nil, limit: 1) + poll + work(job_1) + + expect_to_lock_with(cursor: job_1[:job_id], run_at_lower_bound: job_1[:run_at], limit: 1) + poll + end + + it "resets both cursors when the expiry elapses" do + expect_to_lock_with(cursor: 0, run_at_lower_bound: nil, limit: 1) + poll + work(job_1) + + allow(Process).to receive(:clock_gettime).and_return(61) + + expect_to_lock_with(cursor: 0, run_at_lower_bound: nil, limit: 1) + poll + end + end + + context "with run_at_cursor disabled (default)" do + let(:buffer_size) { 1 } + let!(:job_1) { FakeJob.enqueue(1, queue: queue, priority: 1).attrs } + let(:cursor_expiry) { 60 } + + before do + allow(Process).to receive(:clock_gettime).and_return(0) + locker.instance_variable_get(:@queue_expires_at)[queue] = 60 + end + + it "always passes nil as the run_at lower bound regardless of jobs worked" do + expect_to_lock_with(cursor: 0, run_at_lower_bound: nil, limit: 1) + poll + work(job_1) + + expect_to_lock_with(cursor: job_1[:job_id], run_at_lower_bound: nil, limit: 1) + poll + end + end + end + + describe "existence re-check (pg locks don't obey MVCC)" do + let!(:job_1) { FakeJob.enqueue(1, queue: queue, priority: 1).attrs } + + it "unlocks and excludes jobs that no longer exist by the time the lock is taken" do + # Simulate the race: the row was destroyed between being scanned and being locked, + # by having the existence re-check come back empty for a job the lock query returned. + allow(Que).to receive(:execute).and_call_original + allow(Que).to receive(:execute). + with(:check_jobs_batch, ["{#{job_1[:job_id]}}"]). + and_return([]) + + expect(Que).to receive(:execute).with(:unlock_job, [job_1[:job_id]]).and_call_original + + locker.send(:poll) + + expect(job_buffer.size).to eq(0) + end + end + + describe "#healthy?" do + it "is true before any poll has happened" do + expect(locker).to be_healthy + end + + it "is true after a poll finds nothing" do + locker.send(:poll) + expect(locker).to be_healthy + end + + it "is false after a postgres error" do + allow(Que).to receive(:execute). + with(:lock_jobs_batch, [queue, 0, nil, buffer_size, "{}"]). + and_raise(PG::Error) + locker.send(:poll) + expect(locker).to_not be_healthy + end + end +end diff --git a/spec/lib/que/centralized_worker_spec.rb b/spec/lib/que/centralized_worker_spec.rb new file mode 100644 index 00000000..d0d68e23 --- /dev/null +++ b/spec/lib/que/centralized_worker_spec.rb @@ -0,0 +1,341 @@ +# frozen_string_literal: true + +require "spec_helper" + +RSpec.describe Que::CentralizedWorker do + subject(:worker) { described_class.new(job_buffer: job_buffer) } + + let(:job_buffer) { Que::JobBuffer.new(maximum_size: 10) } + + # Worker no longer locks jobs itself (that's the Locker's job - see locker_spec.rb), so + # to exercise Worker#work with a realistic, already-locked job hash (including + # DB-computed fields like `latency`), lock whatever's next in the queue directly via the + # same SQL a Locker would use, then hand it to the worker under test. + def work_next_job(queue: "default") + locked_job = Que.execute(:lock_job, [queue, 0, nil]).first + worker.work(locked_job) + end + + describe ".work" do + subject(:work) { work_next_job } + + it "returns job_not_found if there's no job to work" do + expect(worker.work(nil)).to eq(:job_not_found) + end + + context "when there's a job to work" do + let!(:job) { FakeJob.enqueue(1) } + + it "works the job" do + expect(work).to eq(:job_worked) + expect(FakeJob.log).to eq([1]) + expect(QueJob.count).to eq(0) + end + + it "logs the work without custom log context" do + expect(Que.logger).to receive(:info). + with(hash_including( + event: "que_job.job_begin", + handler: "FakeJob", + job_class: "FakeJob", + job_error_count: 0, + msg: "Job acquired, beginning work", + priority: 100, + queue: "default", + primary_queue: "default", + que_job_id: job.attrs["job_id"], + latency: an_instance_of(BigDecimal), + )) + + expect(Que.logger).to receive(:info). + with(hash_including( + duration: kind_of(Float), + event: "que_job.job_worked", + handler: "FakeJob", + job_class: "FakeJob", + job_error_count: 0, + msg: "Successfully worked job", + priority: 100, + queue: "default", + primary_queue: "default", + que_job_id: job.attrs["job_id"], + )) + work + end + + context "with custom log context" do + let!(:job) do + klass = Class.new(FakeJob) do + custom_log_context ->(attrs) { + { + custom_log_1: attrs[:args][0], + custom_log_2: "test-log", + } + } + + @log = [] + end + + stub_const("FakeJobWithCustomLogs", klass) + + FakeJobWithCustomLogs.enqueue(1) + end + + it "logs the work with custom log context" do + expect(Que.logger).to receive(:info). + with(hash_including( + event: "que_job.job_begin", + handler: "FakeJobWithCustomLogs", + job_class: "FakeJobWithCustomLogs", + job_error_count: 0, + msg: "Job acquired, beginning work", + priority: 100, + queue: "default", + primary_queue: "default", + que_job_id: job.attrs["job_id"], + latency: an_instance_of(BigDecimal), + custom_log_1: 1, + custom_log_2: "test-log", + )) + + expect(Que.logger).to receive(:info). + with(hash_including( + duration: kind_of(Float), + event: "que_job.job_worked", + handler: "FakeJobWithCustomLogs", + job_class: "FakeJobWithCustomLogs", + job_error_count: 0, + msg: "Successfully worked job", + priority: 100, + queue: "default", + primary_queue: "default", + que_job_id: job.attrs["job_id"], + custom_log_1: 1, + custom_log_2: "test-log", + )) + work + end + end + end + + context "when a job raises an exception" do + it "rescues it" do + ExceptionalJob.enqueue(1) + + expect(work).to eq(:job_worked) + end + + it "logs the work" do + klass = Class.new(ExceptionalJob) do + custom_log_context ->(attrs) { + { + first_arg: attrs[:args][0], + } + } + + @log = [] + end + + stub_const("ExceptionalJobWithCustomLogging", klass) + + ExceptionalJobWithCustomLogging.enqueue(1) + + expect(Que.logger).to receive(:info). + with(hash_including( + event: "que_job.job_begin", + handler: "ExceptionalJobWithCustomLogging", + job_class: "ExceptionalJobWithCustomLogging", + msg: "Job acquired, beginning work", + first_arg: 1, + )) + + expect(Que.logger).to receive(:error). + with(hash_including( + event: "que_job.job_error", + handler: "ExceptionalJobWithCustomLogging", + job_class: "ExceptionalJobWithCustomLogging", + msg: "Job failed with error", + error: "#", + first_arg: 1, + )) + + work + end + + context "and the job has a failure handler" do + let(:job_class) { ExceptionalJob::WithFailureHandler } + + it "calls it" do + job_class.enqueue(1) + + expect(work).to eq(:job_worked) + + expect(job_class.log.count).to eq(2) + expect(job_class.log.first).to eq([:run, 1]) + + method, error, job = job_class.log.second + + expect(method).to eq(:handle_job_failure) + expect(error.class).to eq(ExceptionalJob::Error) + expect(job).to be_a(Hash) + end + end + + context "when the job doesn't have a failure handler" do + let(:job_class) { ExceptionalJob } + + it "calls the default one" do + job_class.enqueue("foo") + + expect(work).to eq(:job_worked) + + job = QueJob.first + + expect(job.error_count).to eq(1) + expect(job.run_at).to be > postgres_now + expect(job.last_error).to match("bad argument foo") + end + end + + context "when the job is no longer defined" do + it "marks it as failed" do + job_options = { + queue: "default", + priority: 1, + run_at: nil, + job_class: "JobNotFound", + retryable: true, + } + + Que.adapter.execute(:insert_job, [*job_options.values_at(*Que::Job::JOB_OPTIONS), []]) + + expect(work).to eq(:job_worked) + + job = QueJob.first + + expect(job.error_count).to eq(1) + expect(job.run_at).to be > postgres_now + expect(job.last_error).to match("uninitialized constant JobNotFound") + end + end + + context "when postgres raises an exception while processing a job" do + it "rescues it and returns an error" do + ExceptionalJob.enqueue(1) + + allow(Que).to receive(:execute).and_call_original + expect(Que).to receive(:execute).with(:set_error, anything).and_raise(PG::Error) + + expect(work).to eq(:postgres_error) + end + end + + context "when postgres raises a bad connection error while processing a job" do + before do + allow(Que).to receive(:execute).and_call_original + allow(Que).to receive(:execute).with(:set_error, anything).and_raise(PG::ConnectionBad) + + # Ensure we don't have any currently leased connections, since in a thread + # using with_connection this would never be the case (but in specs it + # sometimes is). + pool.disconnect! + end + + let(:pool) { ActiveRecord::Base.connection_pool } + + it "rescues it and returns an error" do + ExceptionalJob.enqueue(1) + + expect(work).to eq(:postgres_error) + end + + it "removes the connection from the connection pool" do + # Unlike the sibling example above, we need a job to already be enqueued (and + # locked) before `work` is called at all, since Worker no longer locks its own + # jobs - so the pool's connection count won't be 0 at this point the way it was + # when Worker's own lock_job call was what first touched the database. + ExceptionalJob.enqueue(1) + + expect { work }.to_not(change { pool.connections.count }) + end + end + + context "when we time out checking out a new connection" do + it "rescues it and returns an error" do + ExceptionalJob.enqueue(1) + + allow(Que).to receive(:execute).and_call_original + expect(Que). + to receive(:execute).with(:set_error, anything). + and_raise(ActiveRecord::ConnectionTimeoutError) + + expect(work).to eq(:postgres_error) + end + end + + context "when we can't connect to postgres" do + it "rescues it and returns an error" do + ExceptionalJob.enqueue(1) + + allow(Que).to receive(:execute).and_call_original + expect(Que). + to receive(:execute).with(:set_error, anything). + and_raise(ActiveRecord::ConnectionNotEstablished) + + expect(work).to eq(:postgres_error) + end + end + end + end + + describe "#healthy?" do + it "is true before the worker has done any work" do + expect(worker).to be_healthy + end + + it "is true when no job is found" do + work_next_job + expect(worker).to be_healthy + end + + it "is true after successfully working a job" do + FakeJob.enqueue(1) + work_next_job + expect(worker).to be_healthy + end + + it "is false after a postgres error" do + ExceptionalJob.enqueue(1) + + # A PG::Error raised while destroying a *successful* job would be caught by the + # job's own inner rescue (PG::Error is a StandardError) and routed to the ordinary + # failure/retry path rather than surfacing as :postgres_error. To trigger a genuine + # :postgres_error, the failure needs to happen in the error-handling path itself + # (set_error), which nothing catches. + allow(Que).to receive(:execute).and_call_original + expect(Que).to receive(:execute).with(:set_error, anything).and_raise(PG::Error) + + work_next_job + expect(worker).to_not be_healthy + end + + it "recovers once work succeeds after a postgres error" do + job = ExceptionalJob.enqueue(1) + + allow(Que).to receive(:execute).and_call_original + expect(Que).to receive(:execute).with(:set_error, anything).and_raise(PG::Error) + + work_next_job + expect(worker).to_not be_healthy + + # The set_error update never completed (it raised), so the job row is still sat in + # the queue, still due, and unlocked (Worker#work always unlocks in its ensure + # block). Remove it directly so the next poll finds nothing, simulating the + # underlying issue having cleared. + QueJob.find(job.attrs["job_id"]).destroy! + + work_next_job + expect(worker).to be_healthy + end + end +end diff --git a/spec/lib/que/job_buffer_spec.rb b/spec/lib/que/job_buffer_spec.rb new file mode 100644 index 00000000..a8c7f33d --- /dev/null +++ b/spec/lib/que/job_buffer_spec.rb @@ -0,0 +1,145 @@ +# frozen_string_literal: true + +require "spec_helper" + +RSpec.describe Que::JobBuffer do + subject(:buffer) { described_class.new(maximum_size: maximum_size) } + + let(:maximum_size) { 2 } + + def job(job_id, priority: 1, run_at: Time.now) + { queue: "default", priority: priority, run_at: run_at, job_id: job_id } + end + + describe "#push" do + it "accepts jobs up to the maximum size" do + rejected = buffer.push(job(1), job(2)) + expect(rejected).to eq([]) + expect(buffer.size).to eq(2) + end + + it "rejects jobs that don't fit, returning them to the caller" do + accepted_job = job(1) + rejected_job = job(2) + + buffer.push(job(0), accepted_job) + rejected = buffer.push(rejected_job) + + expect(rejected).to eq([rejected_job]) + expect(buffer.size).to eq(2) + end + + it "orders jobs by priority, then run_at, then job_id" do + low = job(3, priority: 10) + high = job(1, priority: 1) + + buffer.push(low) + buffer.push(high) + + expect(buffer.shift).to eq(high) + expect(buffer.shift).to eq(low) + end + + it "returns jobs unchanged once the buffer is stopping" do + buffer.stop + jobs = [job(1), job(2)] + expect(buffer.push(*jobs)).to eq(jobs) + expect(buffer.size).to eq(0) + end + end + + describe "#shift" do + it "returns a pushed job immediately" do + pushed_job = job(1) + buffer.push(pushed_job) + expect(buffer.shift).to eq(pushed_job) + end + + it "blocks until a job is pushed" do + result = nil + thread = Thread.new { result = buffer.shift } + + sleep 0.05 + expect(result).to be_nil + + pushed_job = job(1) + buffer.push(pushed_job) + thread.join(1) + + expect(result).to eq(pushed_job) + end + + it "returns false for all waiters once stopped" do + threads = Array.new(3) { Thread.new { buffer.shift } } + sleep 0.05 + + buffer.stop + + results = threads.map { |t| t.join(1).value } + expect(results).to all(eq(false)) + end + end + + describe "#available_capacity" do + it "reflects free slots" do + expect(buffer.available_capacity).to eq(2) + buffer.push(job(1)) + expect(buffer.available_capacity).to eq(1) + buffer.shift + expect(buffer.available_capacity).to eq(2) + end + end + + describe "#wait_for_space" do + it "returns immediately if space is already available" do + expect { buffer.wait_for_space(5) }.to take_less_than(1) + end + + it "blocks until a slot frees up" do + buffer.push(job(1), job(2)) + + woke = false + thread = Thread.new do + buffer.wait_for_space(5) + woke = true + end + + sleep 0.05 + expect(woke).to eq(false) + + buffer.shift + + thread.join(1) + expect(woke).to eq(true) + end + + it "returns once the buffer is stopped" do + buffer.push(job(1), job(2)) + + thread = Thread.new { buffer.wait_for_space(5) } + sleep 0.05 + + buffer.stop + expect(thread.join(1)).to_not be_nil + end + end + + describe "#clear" do + it "drains and returns all buffered jobs" do + buffer.push(job(1), job(2)) + jobs = buffer.clear + + expect(jobs.map { |j| j[:job_id] }).to contain_exactly(1, 2) + expect(buffer.size).to eq(0) + end + end + + RSpec::Matchers.define :take_less_than do |seconds| + match do |actual| + start = Process.clock_gettime(Process::CLOCK_MONOTONIC) + actual.call + Process.clock_gettime(Process::CLOCK_MONOTONIC) - start < seconds + end + supports_block_expectations + end +end diff --git a/spec/lib/que/middleware/worker_health_check_spec.rb b/spec/lib/que/middleware/worker_health_check_spec.rb index 656041f0..6a8b3d1b 100644 --- a/spec/lib/que/middleware/worker_health_check_spec.rb +++ b/spec/lib/que/middleware/worker_health_check_spec.rb @@ -5,10 +5,11 @@ RSpec.describe Que::Middleware::WorkerHealthCheck do subject(:health_check) { described_class.new(worker_group) } - let(:worker_group) { instance_double(Que::WorkerGroup, workers: workers) } + let(:worker_group) { instance_double(Que::WorkerGroup, workers: workers, locker: locker) } + let(:locker) { instance_double(Que::CentralizedLocker, healthy?: true) } describe "#call" do - context "when all workers are healthy" do + context "when the locker and all workers are healthy" do let(:workers) do [ instance_double(Que::Worker, healthy?: true), @@ -60,5 +61,21 @@ expect(status).to eq(200) end end + + context "when the locker is unhealthy, even if all workers are healthy" do + let(:locker) { instance_double(Que::CentralizedLocker, healthy?: false) } + let(:workers) do + [ + instance_double(Que::Worker, healthy?: true), + instance_double(Que::Worker, healthy?: true), + ] + end + + it "returns 503 with an unhealthy body" do + status, _, body = health_check.call({}) + expect(status).to eq(503) + expect(body).to eq(["unhealthy"]) + end + end end end diff --git a/spec/spec_helper.rb b/spec/spec_helper.rb index daa4e224..9dca3738 100644 --- a/spec/spec_helper.rb +++ b/spec/spec_helper.rb @@ -60,6 +60,7 @@ FakeJob.log = [] ExceptionalJob.log = [] ExceptionalJob::WithFailureHandler.log = [] + InterruptibleSleepJob.log = [] # In normal runtime we'll new-up metrics with consistent labels, but in tests we'll # create workers with all sorts of configurations. Ensure we clear the registry to @@ -77,9 +78,11 @@ end end -def with_workers(num, stop_timeout: 5, secondary_queues: [], &block) +def with_workers(num, stop_timeout: 5, secondary_queues: [], architecture: Que::WorkerGroup::DEFAULT_ARCHITECTURE, + &block) Que::WorkerGroup.start( num, + architecture: architecture, wake_interval: 0.01, secondary_queues: secondary_queues, ).tap(&block).stop(stop_timeout)