From 0a48975c1565c64a965872ce2bd9ef38452e05b0 Mon Sep 17 00:00:00 2001 From: JP Camara Date: Thu, 30 Jul 2026 09:28:15 -0400 Subject: [PATCH] Add claim cursors: graveyard-resistant polling on PostgreSQL Long-running queries pin the xmin horizon, vacuum cannot reap claimed ready executions, and the poll query scans from the low end of the polling index -- exactly where claimed rows go to die -- so every poll burns CPU proportional to how long the horizon has been pinned. Each process now remembers one lexicographic (priority, id) position per queue key: fast polls make a single row-constructor index seek past the dead-tuple graveyard across all priorities, an empty seek clears the position and suppresses claim queries entirely, and cursor-free discovery seeds and heals the position. The cursor is not a lower bound for live work (a lower-id claim can roll back after we advance past it, sequences are not commit-ordered, and an empty SKIP LOCKED seek can mean rows were merely locked elsewhere), so discovery is fundamental to finding work rather than defensive healing. Discovery itself is bounded by a floor held in process memory. The frequent pass (claim_cursors_discovery_interval, one second by default) claims only ids above the floor recorded before the last full pass, so when the planner seeks its cost tracks recent throughput rather than graveyard size (a sequential-scan plan caps at one heap pass), while still finding all newly enqueued work. Only the full pass (claim_cursors_full_discovery_interval, ten seconds by default, zero disables the split) pays the whole scan, healing rollback overshoot and late commits within one full interval. The floor is process memory -- the highest id this process observed before its last full pass -- so nothing is read from catalogs, sequences, or settings; a cached or rewound sequence merely delays a row until the next full pass, like a late commit, and a fresh process runs full passes until it has observed the stream. A floored pass never asks the database to order by priority: it reads a bounded window of region rows in id order over a (queue_name, id) index (typically a range seek from the floor; a degenerate-statistics plan caps at one heap pass), orders the window in memory, and locks its picks by bare id equality with no ORDER BY, giving no plan an ordering reason to walk an index through the graveyard. A region larger than the window keeps priority order only within it until the window slides or the next full pass reaches past it: an explicit, bounded slack. Claim ordering within a priority changes from enqueue order (job_id) to readiness order (id), matching the cursor's index position; the new polling indexes are added alongside the old pair so a generated schema matches a migrated one, and the README includes the concurrent migration. Row-constructor seeks and the motivating pathology are PostgreSQL-specific, so other adapters keep the classic path; config.solid_queue.claim_cursors = false disables everything. Pinned-horizon bench (200k dead tuples): the frequent discovery pass drops from 49ms and 200k buffer hits to 9.8ms and 1k, fast polls touch single-digit buffers, and a 200-poll loop touches 95.7x fewer buffers than classic polling. Co-Authored-By: Claude Fable 5 --- Gemfile.lock | 2 +- README.md | 121 +++- app/models/solid_queue/queue_selector.rb | 15 +- app/models/solid_queue/ready_execution.rb | 161 +++++- .../ready_execution/claim_cursors.rb | 152 +++++ .../install/templates/db/queue_schema.rb | 3 + lib/solid_queue.rb | 3 + lib/solid_queue/version.rb | 2 +- test/dummy/config/environments/test.rb | 7 + test/dummy/db/queue_schema.rb | 3 + .../solid_queue/claim_cursors_drain_test.rb | 139 +++++ test/models/solid_queue/claim_cursors_test.rb | 545 ++++++++++++++++++ test/test_helper.rb | 3 + 13 files changed, 1137 insertions(+), 19 deletions(-) create mode 100644 app/models/solid_queue/ready_execution/claim_cursors.rb create mode 100644 test/models/solid_queue/claim_cursors_drain_test.rb create mode 100644 test/models/solid_queue/claim_cursors_test.rb diff --git a/Gemfile.lock b/Gemfile.lock index 20ce27795..7ef8b8519 100644 --- a/Gemfile.lock +++ b/Gemfile.lock @@ -1,7 +1,7 @@ PATH remote: . specs: - solid_queue (1.5.0) + solid_queue (1.5.1) activejob (>= 7.1) activerecord (>= 7.1) concurrent-ruby (>= 1.3.1) diff --git a/README.md b/README.md index 700dec3d2..171d8daa4 100644 --- a/README.md +++ b/README.md @@ -24,6 +24,7 @@ Solid Queue can be used with SQL databases such as MySQL, PostgreSQL, or SQLite, - [Database configuration](#database-configuration) - [Other configuration settings](#other-configuration-settings) - [Validating the configuration](#validating-the-configuration) +- [Claim cursors on PostgreSQL](#claim-cursors-on-postgresql) - [Lifecycle hooks](#lifecycle-hooks) - [Errors when enqueuing](#errors-when-enqueuing) - [Concurrency controls](#concurrency-controls) @@ -305,24 +306,26 @@ We recommend not mixing queue order with priorities but either choosing one or t ### Queues specification and performance -To keep polling performant and ensure a covering index is always used, Solid Queue only does two types of polling queries: +To keep polling performant and ensure an order-supporting index is always used, Solid Queue's classic claim path only does two types of polling queries: ```sql -- No filtering by queue -SELECT job_id +SELECT id, job_id, priority FROM solid_queue_ready_executions -ORDER BY priority ASC, job_id ASC +ORDER BY priority ASC, id ASC LIMIT ? FOR UPDATE SKIP LOCKED; -- Filtering by a single queue -SELECT job_id +SELECT id, job_id, priority FROM solid_queue_ready_executions WHERE queue_name = ? -ORDER BY priority ASC, job_id ASC +ORDER BY priority ASC, id ASC LIMIT ? FOR UPDATE SKIP LOCKED; ``` +(On PostgreSQL, [claim cursors](#claim-cursors-on-postgresql) add a few more shapes -- a cursor-guided seek, an id-ordered window read, and a keyed lock -- each covered by the polling indexes.) + The first one (no filtering by queue) is used when you specify ```yml queues: * @@ -431,6 +434,114 @@ Both commands validate the configuration for the current Rails environment. On s `bin/jobs check` accepts the same options as `bin/jobs start` (e.g. `--config_file`, `--recurring_schedule_file`, `--skip-recurring`). The rake task honors the same environment variables Solid Queue already uses: `SOLID_QUEUE_CONFIG`, `SOLID_QUEUE_RECURRING_SCHEDULE`, and `SOLID_QUEUE_SKIP_RECURRING`. To validate a specific environment's configuration, set `RAILS_ENV`, for example `RAILS_ENV=production bin/jobs check`. +## Claim cursors on PostgreSQL + +On PostgreSQL, long-running queries and transactions pin the xmin horizon: vacuum can't +remove dead tuples from `solid_queue_ready_executions`, and because the claim query scans +from the low end of the polling index -- exactly where claimed rows go to die -- every poll +re-scans the accumulated graveyard, burning CPU proportionally to how long the horizon has +been pinned. + +Claim cursors make polls graveyard-resistant: each process remembers one `(priority, id)` +position per queue. Fast polls use that position for a single lexicographic index seek past +dead tuples, across all priorities. When a seek comes up empty, the position is cleared and +polls skip the claim query until discovery is due. + +The cursor is not a lower bound for live work. A competing claim of a lower id can roll +back after this process advances past it, PostgreSQL sequences are not commit-ordered, +so a row can allocate a lower id and commit late, and an empty seek under `SKIP LOCKED` +can mean the remaining rows were merely locked by a peer at that instant. Discovery +queries that don't use the cursor are therefore a fundamental part of finding work, not +merely defensive healing, and they come in two forms. + +A **floored** pass runs on the short cadence +(`config.solid_queue.claim_cursors_discovery_interval`, 1 second by default). Its floor is +held in process memory: the highest ready execution id this process had observed before the +last unbounded pass. Rows arriving since then have greater ids, at every priority, so +`id > floor` finds strictly-higher-priority arrivals without rescanning the graveyard buried +underneath them. The pass covers the region below the cursor and the fast seek covers the +region above it, in the same poll, so claims keep their `(priority, id)` order. Because a +floored pass only sees part of the index, it never advances the cursor -- it may only seed +one that an empty seek cleared. + +An **unbounded** pass runs on the long cadence +(`config.solid_queue.claim_cursors_full_discovery_interval`, 10 seconds by default). It +scans the graveyard, records the next floor, and is the only pass that can reach a row +sitting at or below the floor -- which happens only when a competing claim rolled back or a +commit landed after the floor was read. It is also the only pass allowed to move the cursor +anywhere. + +So the graveyard scan is paid once per full interval rather than once per second. A floored +pass never asks the database to order by priority: it reads a bounded window of the +region's rows in `id` order -- typically a range seek starting at the floor; if degenerate +all-dead statistics steer the planner to a sequential scan instead, the cost caps at one +pass over the table's pages, still far below the classic per-tuple scan -- orders the +window in memory, and locks its picks by bare id equality with no `ORDER BY`, a shape that +gives no plan an ordering reason to walk an index through the graveyard. When a region outgrows the +window (500 rows), claims keep `(priority, id)` order only within the window until it +slides or the next full pass reaches past it: an explicit, bounded slack. Its cost is bounded by rows enqueued since the last full pass, independent of +graveyard size, and a floored pass over an empty region costs a few buffer reads. Still +validate with `EXPLAIN (ANALYZE, BUFFERS)` against your own version and data. The tradeoff +of the split is that the two rare below-floor cases wait for the next unbounded pass -- at +most the full interval (a per-process jitter only ever schedules it earlier) plus the short +cadence it piggybacks on, instead of the short interval alone. Raise the interval to spend +less CPU while a horizon is pinned, lower it to heal faster. Setting it to `0` makes every +discovery pass unbounded. + +The floor is process memory: the highest ready execution id this process observed before +its last full pass. Nothing is read from catalogs, sequences, or settings. It leans on ids +being handed out in increasing order (the default); a cached (`CACHE > 1`), rewound +(`setval`), or backwards-restored sequence, or explicit id inserts, merely delay the +affected row until the next unbounded pass -- the same healing bound as a late commit. A +freshly booted process runs unbounded passes until its first pass has observed the stream. +Polling is graveyard-resistant, not dead-tuple-immune. + +This changes claim ordering within a priority from enqueue order (`job_id`) to the order +jobs became ready (`id`): a scheduled job coming due now queues behind already-waiting jobs +of the same priority instead of jumping ahead of them. Claims are ordered this way on every +path, including when cursors are disabled, so the polling indexes need to lead with `id`. +Existing installations add them alongside the current ones: + +```ruby +class AddClaimCursorPollIndexes < ActiveRecord::Migration[7.1] + disable_ddl_transaction! + + def change + add_index :solid_queue_ready_executions, [ :priority, :id ], + name: "index_solid_queue_poll_all_by_id", algorithm: :concurrently + add_index :solid_queue_ready_executions, [ :queue_name, :priority, :id ], + name: "index_solid_queue_poll_by_queue_and_id", algorithm: :concurrently + add_index :solid_queue_ready_executions, [ :queue_name, :id ], + name: "index_solid_queue_poll_by_queue_floored", algorithm: :concurrently + end +end +``` + +Run this before deploying the upgrade: until the new indexes exist, claims fall back to +sorting instead of an index scan. Keeping the old `job_id` indexes means either version of +the gem has an index matching its ordering, so the upgrade and the migration can ship +independently and a rollback needs no database change. New installs generate both index +generations for the same reason, so a fresh schema matches a migrated one; the `job_id` +pair will be dropped from the generated schema once the old ordering is retired. Note +`CREATE INDEX CONCURRENTLY` waits for transactions older than itself, so a pinned horizon +can delay it considerably. + +The old indexes are then unused, costing write throughput and space on the busiest table. +Dropping them is optional cleanup, safe to defer, and worth confirming first: + +```sql +SELECT indexrelname, idx_scan FROM pg_stat_user_indexes +WHERE relname = 'solid_queue_ready_executions'; +``` + +Once `idx_scan` stops advancing on the `job_id` indexes and every process runs the new +version, `remove_index ..., algorithm: :concurrently` retires them. + +Cursors are enabled by default on PostgreSQL (`config.solid_queue.claim_cursors = false` +to disable) and inactive on other databases, where the PostgreSQL vacuum pathology doesn't +apply. Disabling cursors keeps the readiness ordering, so it is a kill switch for the +cursor queries, not for the index requirement. + ## Lifecycle hooks In Solid queue, you can hook into two different points in the supervisor's life: diff --git a/app/models/solid_queue/queue_selector.rb b/app/models/solid_queue/queue_selector.rb index d0d61dd14..4dff650e0 100644 --- a/app/models/solid_queue/queue_selector.rb +++ b/app/models/solid_queue/queue_selector.rb @@ -9,15 +9,22 @@ def initialize(queue_list, relation) @relation = relation end - def scoped_relations + # Keyed by queue name, or :all for the all-queues relation -- a symbol so a + # queue literally named "*" cannot collide with it -- so claim cursors stay + # attached to their queue as the eligible set changes between polls + def relations_by_queue case - when all? then [ relation.all ] - when none? then [ relation.none ] + when all? then { all: relation.all } + when none? then {} else - queue_names.map { |queue_name| relation.queued_as(queue_name) } + queue_names.index_with { |queue_name| relation.queued_as(queue_name) } end end + def scoped_relations + relations_by_queue.values + end + private def all? include_all_queues? && paused_queues.empty? diff --git a/app/models/solid_queue/ready_execution.rb b/app/models/solid_queue/ready_execution.rb index ea14c193a..3f35eb2ff 100644 --- a/app/models/solid_queue/ready_execution.rb +++ b/app/models/solid_queue/ready_execution.rb @@ -4,12 +4,18 @@ module SolidQueue class ReadyExecution < Execution scope :queued_as, ->(queue_name) { where(queue_name: queue_name) } + # Within a priority, claims follow the order rows became ready (id) rather + # than enqueue order (job_id), matching the claim cursor's index position. + scope :ordered, -> { order(priority: :asc, id: :asc) } + assumes_attributes_from_job + FLOORED_WINDOW = 500 + class << self def claim(queue_list, limit, process_id) - QueueSelector.new(queue_list, self).scoped_relations.flat_map do |queue_relation| - select_and_lock(queue_relation, process_id, limit).tap do |locked| + QueueSelector.new(queue_list, self).relations_by_queue.flat_map do |key, queue_relation| + select_and_lock(key, queue_relation, process_id, limit).tap do |locked| limit -= locked.size end end @@ -19,19 +25,155 @@ def aggregated_count_across(queue_list) QueueSelector.new(queue_list, self).scoped_relations.map(&:count).sum end + # Cursors, floors and discovery deadlines are positions in one table's id + # space, so a process polling several databases keeps a registry per + # datastore. The pool is the datastore's identity: config names can + # collide (every raw hash or URL config resolves to "primary"), pools + # cannot. + def claim_cursors + claim_cursors_registry.compute_if_absent(connection_pool) { ClaimCursors.new } + end + private - def select_and_lock(queue_relation, process_id, limit) + # Keyed to the current pid so forked processes always cold-start: + # inherited cursors, floors, and deadlines belong to the parent + def claim_cursors_registry + if @claim_cursors_pid != ::Process.pid + @claim_cursors_pid = ::Process.pid + @claim_cursors_registry = Concurrent::Map.new + end + + @claim_cursors_registry + end + + def select_and_lock(key, queue_relation, process_id, limit) return [] if limit <= 0 - transaction do - candidates = select_candidates(queue_relation, limit) - lock_candidates(candidates, process_id) + unless claim_cursors_enabled? + return claim_classically(queue_relation, process_id, limit) + end + + discovery, floor, position = claim_cursors.state(key) + + case discovery + when :full then claim_discovering(key, queue_relation, process_id, limit) + when :floored then claim_discovering_above(key, floor, position, queue_relation, process_id, limit) + else + position ? claim_along_cursor(key, position, queue_relation, process_id, limit) : [] + end + end + + def claim_classically(queue_relation, process_id, limit) + _candidates, claimed = claim_candidates(queue_relation, process_id, limit) + claimed + end + + # Cursor-free claim in full (priority, id) order. The one pass that reaches + # rows a floored pass cannot see, so the one that may move the cursor + # anywhere. Its floor is the highest id observed before the pass, so + # every row that arrives after it is above it. + def claim_discovering(key, queue_relation, process_id, limit) + floor = claim_cursors.observed_max_id + candidates, claimed = claim_candidates(queue_relation, process_id, limit) + + claim_cursors.observe(candidates.map(&:id).max) + claim_cursors.record_full_discovery(key, position_of(candidates.last), floor) + + claimed + end + + # Discovery bounded below by the floor: the highest id observed before + # the last full pass, which every row arriving since then exceeds. The + # database is never asked to order this pass: the window of region rows + # is fetched in id order, the (priority, id) ordering and the pick + # happen in memory, and the picked rows are locked by bare id equality + # with no ORDER BY -- so no plan has an ordering reason to walk an index + # through the graveyard. A stale window row just misses its lock. One + # transaction for both claim halves, so a failure above the cursor + # cannot strand committed below-cursor claims; the deadline is recorded + # after it returns, and only once the region is drained, or rows below + # the cursor would wait on lower-priority work above it. A region larger + # than the window trades strict priority order for boundedness inside + # the window until it drains or the next full pass, a documented slack. + def claim_discovering_above(key, floor, position, queue_relation, process_id, limit) + drained = seed = nil + + claimed = transaction(requires_new: true) do + # The window is read, not consumed: observing it would raise the + # floor over rows merely beyond this pick, demoting them to + # full-pass healing. The floor follows claimed candidates only. + window = queue_relation.where("id > ?", floor).order(:id).limit(FLOORED_WINDOW) + .select(:id, :job_id, :priority).to_a + region = position ? window.select { |row| (position_of(row) <=> position) <= 0 } : window + picked = region.sort_by { |row| position_of(row) }.first(limit) + + below_claimed = lock_picked(picked, queue_relation, process_id) + drained = window.size < FLOORED_WINDOW && picked.size < limit + seed = position_of(picked.last) unless position + + if position && below_claimed.size < limit + below_claimed + claim_along_cursor(key, position, queue_relation, process_id, limit - below_claimed.size) + else + below_claimed + end end + + claim_cursors.seed(key, seed) if seed + claim_cursors.record_floored_discovery(key) if drained + + claimed end - def select_candidates(queue_relation, limit) + # Bare id equality and no ORDER BY: SKIP LOCKED may still lose rows to + # peers, and the claim order is the in-memory (priority, id) pick order + def lock_picked(picked, queue_relation, process_id) + return [] if picked.empty? + + relocked = queue_relation.where(id: picked.map(&:id)) + .non_blocking_lock.select(:id, :job_id, :priority).to_a.index_by(&:id) + candidates = picked.filter_map { |row| relocked[row.id] } + + Array(lock_candidates(candidates, process_id)) + end + + # One lexicographic index seek across every priority in the queue key. + def claim_along_cursor(key, position, queue_relation, process_id, limit) + candidates, claimed = claim_candidates( + queue_relation.where("(priority, id) > (?, ?)", *position), process_id, limit + ) + + if candidates.any? + claim_cursors.observe(candidates.map(&:id).max) + claim_cursors.advance(key, position_of(candidates.last)) + else + # Under SKIP LOCKED, empty can also mean every remaining row was + # locked elsewhere; either way the next discovery re-checks + claim_cursors.clear(key, position) + end + + claimed + end + + def claim_candidates(relation, process_id, limit) + candidates = nil + + claimed = transaction do + candidates = select_candidates(relation, limit) + Array(lock_candidates(candidates, process_id)) + end + + [ candidates, claimed ] + end + + # Row-constructor index seeks and the motivating dead-tuple pathology + # are PostgreSQL-specific. + def claim_cursors_enabled? + SolidQueue.claim_cursors && connection_db_config.adapter.match?(/postg/i) + end + + def select_candidates(relation, limit) # Force query execution here with #to_a to avoid unintended FOR UPDATE query executions - queue_relation.ordered.limit(limit).non_blocking_lock.select(:id, :job_id).to_a + relation.ordered.limit(limit).non_blocking_lock.select(:id, :job_id, :priority).to_a end def lock_candidates(executions, process_id) @@ -43,6 +185,9 @@ def lock_candidates(executions, process_id) end end + def position_of(execution) + [ execution.priority, execution.id ] if execution + end def discard_jobs(job_ids) Job.release_all_concurrency_locks Job.where(id: job_ids) diff --git a/app/models/solid_queue/ready_execution/claim_cursors.rb b/app/models/solid_queue/ready_execution/claim_cursors.rb new file mode 100644 index 000000000..87caade0c --- /dev/null +++ b/app/models/solid_queue/ready_execution/claim_cursors.rb @@ -0,0 +1,152 @@ +# frozen_string_literal: true + +module SolidQueue + class ReadyExecution + # Process-local claim cursors: for each queue key, the last (priority, id) + # position this process observed. Cursor-guided queries can descend the + # polling index past dead tuples instead of scanning them. + # + # A cursor is not a lower bound for live work. A competing claim below it + # can roll back (its own transaction or an enclosing one), a row can + # receive a lower sequence id but commit after the cursor advances, and an + # empty SKIP LOCKED seek can mean the remaining rows were locked elsewhere + # rather than gone. Discovery is therefore fundamental to finding work, not + # merely defensive healing, and it comes in two forms. A floored pass runs + # on the short cadence, bounded below by the id watermark the last full + # pass recorded, which every row allocated since then exceeds. The + # unbounded full pass runs on the long cadence; it is the only pass that + # reaches rows at or below that watermark, and so the only one allowed to + # move a cursor. + class ClaimCursors + # Deadlines land up to this fraction BELOW the interval: worker processes + # that booted together don't scan the graveyard in lockstep, and the + # interval stays an upper bound on how long healing waits + FULL_DISCOVERY_JITTER = 0.25 + + def initialize(clock: -> { ::Process.clock_gettime(::Process::CLOCK_MONOTONIC) }) + @clock = clock + @mutex = Mutex.new + @positions = {} + @floors = {} + @observed_max_id = nil + @next_discovery_at = {} + @next_full_discovery_at = {} + end + + # The highest ready execution id this process has seen anywhere. Ids are + # allocated in increasing order, so it is a floor above which all newer + # rows land; anything that breaks that (sequence caching, setval) only + # delays a row until the next full pass, like a late commit. + def observe(id) + @mutex.synchronize do + @observed_max_id = id if id && (@observed_max_id.nil? || id > @observed_max_id) + end + end + + def observed_max_id + @mutex.synchronize { @observed_max_id } + end + + def state(key) + @mutex.synchronize { [ discovery_without_lock(key), @floors[key], @positions[key]&.dup ] } + end + + def position(key) + @mutex.synchronize { @positions[key]&.dup } + end + + def floor(key) + @mutex.synchronize { @floors[key] } + end + + def advance(key, position) + @mutex.synchronize do + current = @positions[key] + @positions[key] = position.dup if current.nil? || (current <=> position).negative? + end + end + + def clear(key, expected_position) + @mutex.synchronize do + @positions.delete(key) if @positions[key] == expected_position + end + end + + def discovery_due?(key) + @mutex.synchronize { due_without_lock?(@next_discovery_at, key) } + end + + def full_discovery_due?(key) + @mutex.synchronize { due_without_lock?(@next_full_discovery_at, key) } + end + + def record_full_discovery(key, position, floor) + @mutex.synchronize do + @positions[key] = position&.dup + @floors[key] = floor + @next_discovery_at[key] = monotonic_time + SolidQueue.claim_cursors_discovery_interval + @next_full_discovery_at[key] = monotonic_time + jittered_full_discovery_interval + end + end + + # A floored pass only sees the part of the index above its floor, so it may + # seed a missing position but must never advance or clear one: the rows its + # floor hid would end up below the cursor, out of reach of every query but + # the next full pass. + def seed(key, position) + @mutex.synchronize do + @positions[key] = position.dup if @positions[key].nil? + end + end + + def record_floored_discovery(key) + @mutex.synchronize do + @next_discovery_at[key] = monotonic_time + SolidQueue.claim_cursors_discovery_interval + end + end + + def expire_discovery!(key) + @mutex.synchronize { @next_discovery_at[key] = monotonic_time } + end + + def expire_full_discovery!(key) + @mutex.synchronize do + @next_discovery_at[key] = monotonic_time + @next_full_discovery_at[key] = monotonic_time + end + end + + def reset! + @mutex.synchronize do + @positions.clear + @floors.clear + @observed_max_id = nil + @next_discovery_at.clear + @next_full_discovery_at.clear + end + end + + private + def discovery_without_lock(key) + return nil unless due_without_lock?(@next_discovery_at, key) + return :full if @floors[key].nil? || due_without_lock?(@next_full_discovery_at, key) + + :floored + end + + def due_without_lock?(deadlines, key) + deadline = deadlines[key] + deadline.nil? || monotonic_time >= deadline + end + + def jittered_full_discovery_interval + interval = SolidQueue.claim_cursors_full_discovery_interval + interval - rand * FULL_DISCOVERY_JITTER * interval + end + + def monotonic_time + @clock.call + end + end + end +end diff --git a/lib/generators/solid_queue/install/templates/db/queue_schema.rb b/lib/generators/solid_queue/install/templates/db/queue_schema.rb index 85194b6a8..20ffd22fd 100644 --- a/lib/generators/solid_queue/install/templates/db/queue_schema.rb +++ b/lib/generators/solid_queue/install/templates/db/queue_schema.rb @@ -71,7 +71,10 @@ t.datetime "created_at", null: false t.index [ "job_id" ], name: "index_solid_queue_ready_executions_on_job_id", unique: true t.index [ "priority", "job_id" ], name: "index_solid_queue_poll_all" + t.index [ "priority", "id" ], name: "index_solid_queue_poll_all_by_id" t.index [ "queue_name", "priority", "job_id" ], name: "index_solid_queue_poll_by_queue" + t.index [ "queue_name", "priority", "id" ], name: "index_solid_queue_poll_by_queue_and_id" + t.index [ "queue_name", "id" ], name: "index_solid_queue_poll_by_queue_floored" end create_table "solid_queue_recurring_executions", force: :cascade do |t| diff --git a/lib/solid_queue.rb b/lib/solid_queue.rb index 654a6be63..02f4a3476 100644 --- a/lib/solid_queue.rb +++ b/lib/solid_queue.rb @@ -26,6 +26,9 @@ module SolidQueue mattr_accessor :app_executor, :on_thread_error, :connects_to mattr_accessor :use_skip_locked, default: true + mattr_accessor :claim_cursors, default: true + mattr_accessor :claim_cursors_discovery_interval, default: 1.second + mattr_accessor :claim_cursors_full_discovery_interval, default: 10.seconds mattr_accessor :process_heartbeat_interval, default: 60.seconds mattr_accessor :process_alive_threshold, default: 5.minutes diff --git a/lib/solid_queue/version.rb b/lib/solid_queue/version.rb index 967d4cc48..bb2ee0ed7 100644 --- a/lib/solid_queue/version.rb +++ b/lib/solid_queue/version.rb @@ -1,5 +1,5 @@ module SolidQueue - VERSION = "1.5.0" + VERSION = "1.5.1" def self.next_major_version Gem::Version.new(VERSION).segments.first + 1 diff --git a/test/dummy/config/environments/test.rb b/test/dummy/config/environments/test.rb index 4a07d140a..10a4c80fe 100644 --- a/test/dummy/config/environments/test.rb +++ b/test/dummy/config/environments/test.rb @@ -60,6 +60,13 @@ config.solid_queue.shutdown_timeout = 2.seconds + # Keep job pickup snappy for latency-sensitive integration tests, and keep the + # unbounded pass frequent enough that anything only it can heal heals inside a + # test's patience, while still leaving most passes floored. Cursor cadence + # behavior is pinned deterministically in ClaimCursorsTest + config.solid_queue.claim_cursors_discovery_interval = 0.05.seconds + config.solid_queue.claim_cursors_full_discovery_interval = 1.second + config.log_formatter = proc do |severity, timestamp, progname, msg| ts = timestamp.getlocal.strftime("%H:%M:%S.%3N") "#{ts} #{msg}\n" diff --git a/test/dummy/db/queue_schema.rb b/test/dummy/db/queue_schema.rb index 697c2e928..76c96b91f 100644 --- a/test/dummy/db/queue_schema.rb +++ b/test/dummy/db/queue_schema.rb @@ -83,7 +83,10 @@ t.datetime "created_at", null: false t.index ["job_id"], name: "index_solid_queue_ready_executions_on_job_id", unique: true t.index ["priority", "job_id"], name: "index_solid_queue_poll_all" + t.index ["priority", "id"], name: "index_solid_queue_poll_all_by_id" t.index ["queue_name", "priority", "job_id"], name: "index_solid_queue_poll_by_queue" + t.index ["queue_name", "priority", "id"], name: "index_solid_queue_poll_by_queue_and_id" + t.index ["queue_name", "id"], name: "index_solid_queue_poll_by_queue_floored" end create_table "solid_queue_recurring_executions", charset: "utf8mb4", collation: "utf8mb4_0900_ai_ci", force: :cascade do |t| diff --git a/test/models/solid_queue/claim_cursors_drain_test.rb b/test/models/solid_queue/claim_cursors_drain_test.rb new file mode 100644 index 000000000..2f286613e --- /dev/null +++ b/test/models/solid_queue/claim_cursors_drain_test.rb @@ -0,0 +1,139 @@ +require "test_helper" + +# Property test: random interleavings of enqueues, claims, rollbacks, deadline +# expiries, kill-switch flips, and below-floor arrivals must always drain to +# exactly-once delivery. Reproduce a failure with DRAIN_SEED= and raise +# the workload with DRAIN_ITERATIONS/DRAIN_OPS. +class SolidQueue::ClaimCursorsDrainTest < ActiveSupport::TestCase + PROCESS_ID = 42 + QUEUES = %w[ drain_one drain_two ].freeze + + setup do + skip "Claim cursors require PostgreSQL" unless SolidQueue::Record.connection_db_config.adapter.match?(/postg/i) + SolidQueue::ReadyExecution.claim_cursors.reset! + @original_discovery_interval = SolidQueue.claim_cursors_discovery_interval + @original_full_discovery_interval = SolidQueue.claim_cursors_full_discovery_interval + SolidQueue.claim_cursors_discovery_interval = 10.minutes + SolidQueue.claim_cursors_full_discovery_interval = 10.minutes + end + + teardown do + SolidQueue::ReadyExecution.claim_cursors.reset! + SolidQueue.claim_cursors = true + SolidQueue.claim_cursors_discovery_interval = @original_discovery_interval if @original_discovery_interval + SolidQueue.claim_cursors_full_discovery_interval = @original_full_discovery_interval if @original_full_discovery_interval + end + + test "random interleavings always drain to exactly-once delivery" do + iterations = Integer(ENV.fetch("DRAIN_ITERATIONS", "25")) + ops_per_iteration = Integer(ENV.fetch("DRAIN_OPS", "40")) + + iterations.times do |iteration| + seed = ENV["DRAIN_SEED"]&.to_i || Random.new_seed + rng = Random.new(seed) + context = "iteration #{iteration}, DRAIN_SEED=#{seed}" + + enqueued, planted, claimed_log = run_random_ops(rng, ops_per_iteration) + + surface_drained = drain(full: false) + + all_claimed = claimed_ids(claimed_log, surface_drained) + missing_surface = (enqueued - planted) - all_claimed + assert_empty missing_surface, + "jobs above the floor not drained by floored+fast passes alone (#{context}): #{missing_surface.inspect}" + + all_claimed += claimed_ids([ drain(full: true) ]) + missing = enqueued - all_claimed + assert_empty missing, "jobs stranded after full passes (#{context}): #{missing.inspect}" + + duplicates = all_claimed.tally.select { |_, count| count > 1 } + assert_empty duplicates, "jobs claimed more than once (#{context}): #{duplicates.inspect}" + + assert_equal 0, SolidQueue::ReadyExecution.count, "ready executions left behind (#{context})" + + SolidQueue::ClaimedExecution.delete_all + SolidQueue::Job.delete_all + SolidQueue::ReadyExecution.claim_cursors.reset! + SolidQueue.claim_cursors = true + end + end + + private + def run_random_ops(rng, count) + enqueued = [] + planted = [] + claimed_log = [] + low_id_pool = reserve_low_ids(rng) + + count.times do + case rng.rand(100) + when 0...35 # enqueue a small batch at random priority and queue + rng.rand(1..4).times do + job = AddToBufferJob.set(queue: QUEUES.sample(random: rng), priority: rng.rand(0..3)).perform_later("op") + enqueued << job.provider_job_id + end + when 35...65 # claim from a random scope + claimed_log << claim(rng.rand(1..4), queues: [ QUEUES.sample(random: rng), "*" ].sample(random: rng)) + when 65...75 # a claim that rolls back: rows must become claimable again, + # though possibly only by the full pass, the designed healing bound + SolidQueue::Record.transaction do + rolled_back = claim(rng.rand(1..3), queues: "*") + planted.concat(claimed_ids([ rolled_back ])) + raise ActiveRecord::Rollback + end + when 75...85 # deadline expiries in random combinations + key = [ :all, *QUEUES ].sample(random: rng) + rng.rand(2).zero? ? cursors.expire_discovery!(key) : cursors.expire_full_discovery!(key) + when 85...93 # a row arrives below the floor, like a late commit + if (slot = low_id_pool.pop) + SolidQueue::ReadyExecution.insert_all!([ { id: slot[:ready_id], job_id: slot[:job_id], queue_name: slot[:queue_name], priority: 0, created_at: Time.current } ]) + planted << slot[:job_id] + enqueued << slot[:job_id] + end + else # kill switch flip: one classic claim, then back on + SolidQueue.claim_cursors = false + claimed_log << claim(rng.rand(1..3), queues: "*") + SolidQueue.claim_cursors = true + end + end + + [ enqueued, planted, claimed_log ] + end + + # Jobs whose ready rows are deleted up front, freeing ids below every + # future floor for the late-commit op to reuse + def reserve_low_ids(rng) + Array.new(3) do + job = AddToBufferJob.set(queue: QUEUES.sample(random: rng)).perform_later("reserved") + execution = SolidQueue::ReadyExecution.find_by!(job_id: job.provider_job_id) + slot = { ready_id: execution.id, job_id: job.provider_job_id, queue_name: execution.queue_name } + execution.delete + slot + end + end + + # full: false drains with floored+fast passes only -- everything above the + # floor must be reachable without an unbounded scan + def drain(full:) + claimed = [] + 50.times do + [ :all, *QUEUES ].each { |key| full ? cursors.expire_full_discovery!(key) : cursors.expire_discovery!(key) } + batch = claim(10, queues: "*") + claimed << batch + break if batch.empty? && SolidQueue::ReadyExecution.count.zero? + end + claimed + end + + def claimed_ids(*logs) + logs.flatten.map { |execution| execution.try(:job_id) }.compact + end + + def claim(limit, queues:) + SolidQueue::ReadyExecution.claim(queues, limit, PROCESS_ID) + end + + def cursors + SolidQueue::ReadyExecution.claim_cursors + end +end diff --git a/test/models/solid_queue/claim_cursors_test.rb b/test/models/solid_queue/claim_cursors_test.rb new file mode 100644 index 000000000..061cdd93f --- /dev/null +++ b/test/models/solid_queue/claim_cursors_test.rb @@ -0,0 +1,545 @@ +require "test_helper" + +class SolidQueue::ClaimCursorsTest < ActiveSupport::TestCase + PROCESS_ID = 42 + + setup do + skip "Claim cursors require PostgreSQL" unless SolidQueue::Record.connection_db_config.adapter.match?(/postg/i) + SolidQueue::ReadyExecution.claim_cursors.reset! + # The dummy app shortens the interval for integration latency; these tests + # control discovery explicitly via expire_discovery! + @original_discovery_interval = SolidQueue.claim_cursors_discovery_interval + @original_full_discovery_interval = SolidQueue.claim_cursors_full_discovery_interval + SolidQueue.claim_cursors_discovery_interval = 10.minutes + SolidQueue.claim_cursors_full_discovery_interval = 10.minutes + end + + teardown do + SolidQueue::ReadyExecution.claim_cursors.reset! + SolidQueue.claim_cursors = true + SolidQueue.claim_cursors_discovery_interval = @original_discovery_interval if @original_discovery_interval + SolidQueue.claim_cursors_full_discovery_interval = @original_full_discovery_interval if @original_full_discovery_interval + end + + test "advance is monotonic per key" do + cursors.advance(:all, [ 0, 10 ]) + cursors.advance(:all, [ 0, 5 ]) + assert_equal [ 0, 10 ], cursors.position(:all) + + cursors.advance(:all, [ 1, 1 ]) + assert_equal [ 1, 1 ], cursors.position(:all) + end + + test "clear only removes the position it observed" do + cursors.advance(:all, [ 0, 10 ]) + + cursors.clear(:all, [ 0, 5 ]) # stale observation from a slower thread + assert_equal [ 0, 10 ], cursors.position(:all) + + cursors.clear(:all, [ 0, 10 ]) + assert_nil cursors.position(:all) + end + + test "reset clears the observed floor source" do + cursors.observe(42) + assert_equal 42, cursors.observed_max_id + + cursors.reset! + assert_nil cursors.observed_max_id + end + + test "observations are shared across queue keys" do + AddToBufferJob.set(queue: :first).perform_later("a") + claim(1, queues: "first") # observes an id + + AddToBufferJob.set(queue: :second).perform_later("b") + cursors.expire_full_discovery!("second") + claim(1, queues: "second") + + # Ids share one table-wide namespace, so the second queue's full pass + # records a floor from the first queue's observation stream + assert_not_nil cursors.floor("second") + end + + test "a row inserted with an anomalous low id heals on the next full pass" do + seed_cursor_and_floor + + job = SolidQueue::Job.create!(queue_name: "default", class_name: "AddToBufferJob", arguments: "{}", priority: 0) + SolidQueue::ReadyExecution.where(job_id: job.id).delete_all + low_id = cursors.floor(:all) - 1 + SolidQueue::ReadyExecution.insert_all!([ { id: low_id, job_id: job.id, queue_name: "default", priority: 0, created_at: Time.current } ]) + + cursors.expire_discovery!(:all) + assert_empty claim(1) # hidden below the floor, like a setval rewind would be + + cursors.expire_full_discovery!(:all) + assert_equal 1, claim(1).size + end + + test "cursors are tracked independently per queue" do + AddToBufferJob.set(queue: :first).perform_later("a") + AddToBufferJob.set(queue: :second).perform_later("b") + + claim(1, queues: "first") + first_position = cursors.position("first") + assert_not_nil first_position + assert_nil cursors.position("second") + assert_equal 1, SolidQueue::ReadyExecution.count + + claim(1, queues: "second") + assert_not_nil cursors.position("second") + assert_equal first_position, cursors.position("first") + end + + test "an empty discovery suppresses claim queries until the next one is due" do + assert_empty claim(1) # discovery against an empty queue records a nil position + + queries = capture_candidate_queries do + 3.times { assert_empty claim(1) } + end + assert_empty queries + end + + test "discovery seeds a single lexicographic position" do + 2.times { |i| AddToBufferJob.perform_later(i) } + executions = SolidQueue::ReadyExecution.ordered.to_a + + assert_equal 1, claim(1).size + assert_equal position_of(executions.first), cursors.position(:all) + + assert_equal 1, claim(1).size + assert_equal position_of(executions.second), cursors.position(:all) + end + + test "fast path claims across priorities with one cursor query" do + AddToBufferJob.set(priority: 1).perform_later("seed") + claim(1) + + AddToBufferJob.set(priority: 1).perform_later("high") + AddToBufferJob.set(priority: 5).perform_later("low") + + queries = capture_candidate_queries do + claimed_jobs = claim(2).sort_by(&:id).map { |execution| SolidQueue::Job.find(execution.job_id) } + assert_equal [ 1, 5 ], claimed_jobs.map(&:priority) + end + + assert_equal 1, queries.size + assert_includes queries.sole, "(priority, id) > (" + end + + test "fast path immediately finds new work at the current priority" do + AddToBufferJob.set(priority: 3).perform_later("seed") + claim(1) + + AddToBufferJob.set(priority: 3).perform_later("next") + + assert_equal 1, claim(1).size + assert_equal 0, SolidQueue::ReadyExecution.count + end + + test "empty seek clears the position and skips cursor queries until discovery is due" do + AddToBufferJob.perform_later("seed") + claim(1) + + queries = capture_candidate_queries { assert_empty claim(1) } + assert_equal 1, queries.size + assert_nil cursors.position(:all) + assert_not cursors.discovery_due?(:all) + + queries = capture_candidate_queries do + 3.times { assert_empty claim(1) } + end + assert_empty queries + end + + test "higher priority arrivals are picked up once discovery is due" do + AddToBufferJob.set(priority: 5).perform_later("seed") + claim(1) + + AddToBufferJob.set(priority: 1).perform_later("higher") + + assert_empty claim(1) + assert_nil cursors.position(:all) + + cursors.expire_discovery!(:all) + claimed = claim(1) + + assert_equal 1, claimed.size + assert_equal 1, SolidQueue::Job.find(claimed.sole.job_id).priority + end + + test "discovery heals an overshot position" do + AddToBufferJob.perform_later("seed") + claim(1) + + AddToBufferJob.perform_later("stranded") + execution = SolidQueue::ReadyExecution.sole + cursors.advance(:all, [ execution.priority, execution.id + 1000 ]) + + assert_empty claim(1) + assert_nil cursors.position(:all) + + cursors.expire_discovery!(:all) + claimed = claim(1) + + assert_equal 1, claimed.size + assert_equal "stranded", SolidQueue::Job.find(claimed.sole.job_id).arguments.dig("arguments").first + end + + test "within a priority, claims follow readiness order, not enqueue order" do + scheduled = AddToBufferJob.set(wait: 5.minutes).perform_later("scheduled first") + immediate = AddToBufferJob.perform_later("enqueued second") + + travel_to(6.minutes.from_now) { SolidQueue::ScheduledExecution.dispatch_next_batch(10) } + + claimed_jobs = claim(2).sort_by(&:id).map { |execution| SolidQueue::Job.find(execution.job_id) } + assert_equal [ immediate.job_id, scheduled.job_id ], claimed_jobs.map(&:active_job_id) + end + + test "discovery is unbounded until a pass has observed a floor" do + AddToBufferJob.perform_later("seed") + + queries = capture_candidate_queries { assert_equal 1, claim(1).size } + + assert_equal 1, queries.size + assert_not_includes queries.sole, "id > " + assert_nil cursors.floor(:all) # nothing observed before the first pass + + AddToBufferJob.perform_later("next") + cursors.expire_full_discovery!(:all) + claim(1) + + assert_not_nil cursors.floor(:all) + end + + test "a floored pass sweeps below the cursor and seeks above it in one poll" do + seed_cursor_and_floor + + AddToBufferJob.perform_later("next") + cursors.expire_discovery!(:all) + + selects = nil + queries = capture_candidate_queries do + selects = capture_ready_selects { assert_equal 1, claim(1).size } + end + + # The arrival sits above the cursor, so the empty in-memory pick issues no + # below-cursor lock at all; the window read and the above-cursor seek remain + assert_equal 1, queries.size + assert_includes queries.sole, "(priority, id) > (" + assert selects.any? { |sql| sql.include?("id > ") && sql.exclude?("FOR UPDATE") } + end + + test "a floored pass finds a higher priority arrival without scanning below the watermark" do + seed_cursor_and_floor(priority: 5) + position = cursors.position(:all) + + AddToBufferJob.set(priority: 1).perform_later("higher") + cursors.expire_discovery!(:all) + + selects = nil + queries = capture_candidate_queries do + selects = capture_ready_selects do + claimed = claim(1) + assert_equal 1, SolidQueue::Job.find(claimed.sole.job_id).priority + end + end + + assert_equal 1, queries.size + assert_match(/"id" (=|IN)/, queries.sole) + + # The window is ordered by id alone and the lock query carries no ORDER BY, + # so no plan has an ordering reason to walk an index past the graveyard + window = selects.find { |sql| sql.include?("id > ") && sql.exclude?("FOR UPDATE") } + assert_includes window, %(ORDER BY "solid_queue_ready_executions"."id" ASC) + assert_no_match(/ORDER BY.*priority/im, window) + assert_no_match(/ORDER BY/i, queries.sole) + + # A pass that only sees part of the index must not move the cursor: the rows + # its floor hid would fall below it and out of every other query + assert_equal position, cursors.position(:all) + end + + test "successive floored passes reach every arrival below the cursor" do + AddToBufferJob.set(priority: 9).perform_later("seed") + claim(1) + + # The higher priority arrival takes the higher id, so a watermark that + # followed the claims would jump over the one left behind + lower = AddToBufferJob.set(priority: 5).perform_later("lower priority, lower id") + higher = AddToBufferJob.set(priority: 1).perform_later("higher priority, higher id") + + cursors.expire_discovery!(:all) + assert_equal higher.job_id, claimed_active_job_ids(1).sole + + cursors.expire_discovery!(:all) + assert_equal lower.job_id, claimed_active_job_ids(1).sole + end + + test "a limit-filling sweep keeps discovery due until the region below the cursor drains" do + AddToBufferJob.set(priority: 9).perform_later("seed") + claim(1) + + AddToBufferJob.set(priority: 1).perform_later("highest") + AddToBufferJob.set(priority: 5).perform_later("middle") + AddToBufferJob.set(priority: 10).perform_later("lowest") + + cursors.expire_discovery!(:all) + first = claim(1) + assert_equal 1, SolidQueue::Job.find(first.sole.job_id).priority + + # The sweep filled its limit, so without any deadline expiring the next + # poll must continue below the cursor, not claim the lower-priority row + # sitting above it + second = claim(1) + assert_equal 5, SolidQueue::Job.find(second.sole.job_id).priority + + third = claim(1) + assert_equal 10, SolidQueue::Job.find(third.sole.job_id).priority + end + + test "a region larger than the window keeps priority order only within it until it slides" do + seed_cursor_and_floor(priority: 9) + + now = Time.current + filler_ids = SolidQueue::Job.insert_all!( + Array.new(SolidQueue::ReadyExecution::FLOORED_WINDOW) { { queue_name: "default", class_name: "AddToBufferJob", arguments: "{}", priority: 10, created_at: now, updated_at: now } }, + returning: [ :id ] + ).rows.flatten + high_id = SolidQueue::Job.insert_all!( + [ { queue_name: "default", class_name: "AddToBufferJob", arguments: "{}", priority: 1, created_at: now, updated_at: now } ], + returning: [ :id ] + ).rows.flatten.sole + SolidQueue::ReadyExecution.insert_all!( + filler_ids.map { |id| { job_id: id, queue_name: "default", priority: 10, created_at: now } } + + [ { job_id: high_id, queue_name: "default", priority: 1, created_at: now } ] + ) + + cursors.expire_discovery!(:all) + + # The high-priority row's id lies beyond the window, so a lower-priority + # claim precedes it: the documented slack of a region larger than the window + slack = claim(1) + assert_equal 10, SolidQueue::Job.find(slack.sole.job_id).priority + + # Consuming a row slides the window over it, and priority order resumes + healed = claim(1) + assert_equal 1, SolidQueue::Job.find(healed.sole.job_id).priority + end + + test "a floored pass with no cursor spans every priority and seeds one" do + seed_cursor_and_floor(priority: 5) + assert_empty claim(1) + assert_nil cursors.position(:all) + + AddToBufferJob.set(priority: 1).perform_later("higher") + cursors.expire_discovery!(:all) + + queries = capture_candidate_queries { assert_equal 1, claim(1).size } + + assert_equal 1, queries.size + assert_match(/"id" (=|IN)/, queries.sole) + assert_not_includes queries.sole, "(priority, id) <= (" + assert_not_nil cursors.position(:all) + end + + test "a row below both the cursor and the watermark waits for the unbounded pass" do + AddToBufferJob.set(priority: 9).perform_later("seed") + claim(1) + seed_position = cursors.position(:all) + + AddToBufferJob.set(priority: 1).perform_later("stranded") + stranded = SolidQueue::ReadyExecution.sole + + # What a rolled back claim or a commit landing after the watermark was read + # leaves behind: a row below the cursor that the floor also hides + cursors.record_full_discovery(:all, seed_position, stranded.id) + + cursors.expire_discovery!(:all) + assert_empty claim(1) + + cursors.expire_full_discovery!(:all) + assert_equal 1, claim(1).size + end + + test "the unbounded pass runs on its own cadence" do + AddToBufferJob.perform_later("seed") + claim(1) + + assert_not cursors.discovery_due?(:all) + assert_not cursors.full_discovery_due?(:all) + + cursors.expire_discovery!(:all) + assert cursors.discovery_due?(:all) + assert_not cursors.full_discovery_due?(:all) + + AddToBufferJob.perform_later("next") + cursors.expire_full_discovery!(:all) + queries = capture_candidate_queries { assert_equal 1, claim(1).size } + + assert_equal 1, queries.size + assert_not_includes queries.sole, "id > " + assert_not_includes queries.sole, "(priority, id)" + end + + test "a zero full discovery interval keeps every discovery pass unbounded" do + SolidQueue.claim_cursors_full_discovery_interval = 0 + AddToBufferJob.perform_later("seed") + claim(1) + + AddToBufferJob.perform_later("next") + cursors.expire_discovery!(:all) + + queries = capture_candidate_queries { assert_equal 1, claim(1).size } + + assert_equal 1, queries.size + assert_not_includes queries.sole, "id > " + end + + test "jitter only schedules the unbounded pass earlier, never later" do + now = 0.0 + clocked = SolidQueue::ReadyExecution::ClaimCursors.new(clock: -> { now }) + + 3.times do + clocked.record_full_discovery(:all, [ 0, 1 ], 1) + + now += 0.7 * SolidQueue.claim_cursors_full_discovery_interval + assert_not clocked.full_discovery_due?(:all) + + now += 0.3 * SolidQueue.claim_cursors_full_discovery_interval + assert clocked.full_discovery_due?(:all) + end + end + + test "cursor state is scoped to the datastore" do + cursors.advance(:all, [ 0, 10 ]) + + # The pool is the datastore's identity: two datastores never share one + SolidQueue::ReadyExecution.stubs(:connection_pool).returns(Object.new) + other_cursors = SolidQueue::ReadyExecution.claim_cursors + + assert_nil other_cursors.position(:all) + other_cursors.advance(:all, [ 0, 99 ]) + + SolidQueue::ReadyExecution.unstub(:connection_pool) + assert_equal [ 0, 10 ], cursors.position(:all) + ensure + other_cursors&.reset! + end + + test "disabling claim cursors uses the classic path without changing cursor state" do + SolidQueue.claim_cursors = false + AddToBufferJob.perform_later("classic") + + queries = capture_candidate_queries { assert_equal 1, claim(1).size } + + assert_equal 1, queries.size + assert_not_includes queries.sole, "(priority, id) > (" + assert_nil cursors.position(:all) + assert cursors.discovery_due?(:all) + end + + private + def claim(limit, queues: "*") + SolidQueue::ReadyExecution.claim(queues, limit, PROCESS_ID) + end + + def cursors + SolidQueue::ReadyExecution.claim_cursors + end + + def position_of(execution) + [ execution.priority, execution.id ] + end + + def seed_cursor_and_floor(priority: 0) + AddToBufferJob.set(priority: priority).perform_later("floor seed") + claim(1) # observes the seed id + AddToBufferJob.set(priority: priority).perform_later("cursor seed") + cursors.expire_full_discovery!(:all) + claim(1) # a full pass that records the observed floor and the position + end + + def claimed_active_job_ids(limit) + claim(limit).map { |execution| SolidQueue::Job.find(execution.job_id).active_job_id } + end + + def capture_ready_selects + queries = [] + subscriber = ActiveSupport::Notifications.subscribe("sql.active_record") do |event| + sql = event.payload[:sql] + queries << sql if sql.start_with?("SELECT") && sql.include?("solid_queue_ready_executions") + end + + yield + queries + ensure + ActiveSupport::Notifications.unsubscribe(subscriber) if subscriber + end + + def capture_candidate_queries + queries = [] + subscriber = ActiveSupport::Notifications.subscribe("sql.active_record") do |event| + sql = event.payload[:sql] + queries << sql if sql.include?("solid_queue_ready_executions") && sql.include?("FOR UPDATE SKIP LOCKED") + end + + yield + queries + ensure + ActiveSupport::Notifications.unsubscribe(subscriber) if subscriber + end +end + +# Uses a second database connection, so rows must be committed and visible +# across connections +class SolidQueue::ClaimCursorsContentionTest < ActiveSupport::TestCase + self.use_transactional_tests = false + + PROCESS_ID = 42 + + setup do + skip "Claim cursors require PostgreSQL" unless SolidQueue::Record.connection_db_config.adapter.match?(/postg/i) + SolidQueue::ReadyExecution.claim_cursors.reset! + @original_discovery_interval = SolidQueue.claim_cursors_discovery_interval + SolidQueue.claim_cursors_discovery_interval = 10.minutes + end + + teardown do + SolidQueue::ReadyExecution.claim_cursors.reset! + SolidQueue.claim_cursors_discovery_interval = @original_discovery_interval if @original_discovery_interval + end + + test "rows skipped as locked elsewhere are rediscovered after the peer rolls back" do + AddToBufferJob.perform_later("seed") + claim(1) # discovery seeds the cursor + + AddToBufferJob.perform_later("contended") + execution = SolidQueue::ReadyExecution.sole + + peer = SolidQueue::Record.connection_pool.checkout + peer.execute("BEGIN") + peer.execute("SELECT id FROM solid_queue_ready_executions WHERE id = #{execution.id} FOR UPDATE") + + assert_empty claim(1) # the only row is locked, so the seek looks empty + assert_nil SolidQueue::ReadyExecution.claim_cursors.position(:all) + + peer.execute("ROLLBACK") + + SolidQueue::ReadyExecution.claim_cursors.expire_discovery!(:all) + claimed = claim(1) + + assert_equal 1, claimed.size + assert_equal "contended", SolidQueue::Job.find(claimed.sole.job_id).arguments.dig("arguments").first + ensure + if peer + peer.execute("ROLLBACK") rescue nil + SolidQueue::Record.connection_pool.checkin(peer) + end + end + + private + def claim(limit) + SolidQueue::ReadyExecution.claim("*", limit, PROCESS_ID) + end +end diff --git a/test/test_helper.rb b/test/test_helper.rb index 99d00ba27..c86e27a08 100644 --- a/test/test_helper.rb +++ b/test/test_helper.rb @@ -46,6 +46,9 @@ def destroy_records end setup do + # Tests model fresh processes: claim cursors are process-local state that + # must not outlive the records they were derived from + SolidQueue::ReadyExecution.claim_cursors.reset! @_on_thread_error = SolidQueue.on_thread_error SolidQueue.on_thread_error = silent_on_thread_error_for(ExpectedTestError, @_on_thread_error) ActiveJob::QueueAdapters::SolidQueueAdapter.stopping = false