Claim cursors: graveyard-resistant polling on PostgreSQL

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)

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:

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?

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)