Bound the claim cursor discovery pass with a sequence watermark

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)
@@ -431,6 +432,100 @@ 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
+the `solid_queue_ready_executions` sequence high-water mark, read just before the last
+unbounded pass: every row allocated since then has a greater id, 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, and
+everything in between is bounded by how much work has been enqueued since. The tradeoff is
+that those two rare 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's guarantee leans on ids being handed out in order, so it is only used when the
+id sequence's cache is `1` (PostgreSQL's default) on PostgreSQL 10 or newer, where the
+catalog exposes the cache setting. Anything else -- an older server, `CACHE` above 1, a
+table without its own sequence -- simply records no floor, and every discovery pass runs
+unbounded. 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[8.0]
+  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
+  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,21 @@ def initialize(queue_list, relation)
       @relation = relation
     end
 
-    def scoped_relations
+    # Keyed by queue name, or "*" for the all-queues relation, 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 { "*" => 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,16 @@ 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
 
     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 +23,168 @@ 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)
+        def claim_cursors_registry
+          @claim_cursors_registry ||= Concurrent::Map.new
+        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
 
-        def select_candidates(queue_relation, limit)
+        # 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 read before the claim's snapshot, so every row
+        # allocated after it is above it.
+        def claim_discovering(key, queue_relation, process_id, limit)
+          floor = id_watermark
+          candidates, claimed = claim_candidates(queue_relation, process_id, limit)
+
+          claim_cursors.record_full_discovery(key, position_of(candidates.last), floor)
+
+          claimed
+        end
+
+        # Discovery bounded below by the last full pass's watermark, which every
+        # row allocated since then exceeds. A strictly-higher-priority arrival is
+        # still found, without rescanning the graveyard buried underneath it.
+        def claim_discovering_above(key, floor, position, queue_relation, process_id, limit)
+          floored_relation = queue_relation.where("id > ?", floor)
+
+          if position
+            claim_below_cursor(key, floored_relation, position, queue_relation, process_id, limit)
+          else
+            claim_seeding_cursor(key, floored_relation, process_id, limit)
+          end
+        end
+
+        # The cursor owns everything above itself and the fast path claims it in
+        # the same poll: disjoint regions keep claims in (priority, id) order, and
+        # leave the cursor to advance only over a range nothing was hidden from.
+        def claim_below_cursor(key, floored_relation, position, queue_relation, process_id, limit)
+          _candidates, claimed = claim_candidates_without_sorts(
+            floored_relation.where("(priority, id) <= (?, ?)", *position), process_id, limit
+          )
+
+          claim_cursors.record_floored_discovery(key)
+
+          return claimed if claimed.size >= limit
+
+          claimed + claim_along_cursor(key, position, queue_relation, process_id, limit - claimed.size)
+        end
+
+        # With no cursor to bound it, a floored pass spans every priority, and the
+        # prefix it claims is a position nothing live was hidden below.
+        def claim_seeding_cursor(key, floored_relation, process_id, limit)
+          candidates, claimed = claim_candidates_without_sorts(floored_relation, process_id, limit)
+
+          claim_cursors.record_floored_discovery(key, position_of(candidates.last))
+
+          claimed
+        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.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
+
+        # Over an all-dead table every plan estimates near zero cost, and the tie
+        # can land on a legacy job_id polling index plus a sort that re-reads the
+        # whole graveyard a floored query exists to skip. Sorts add nothing the
+        # polling indexes don't already provide, so claim with them off -- inside
+        # a savepoint, restored before it releases and reverted by PostgreSQL if
+        # it rolls back, so the settings can never escape a caller's transaction.
+        def claim_candidates_without_sorts(relation, process_id, limit)
+          transaction(requires_new: true) do
+            sort_settings = connection.select_rows(
+              "SELECT name, setting FROM pg_settings WHERE name IN ('enable_sort', 'enable_incremental_sort')"
+            )
+            sort_settings.each { |name, _| connection.execute("SET LOCAL #{name} = OFF") }
+
+            claim_candidates(relation, process_id, limit).tap do
+              sort_settings.each { |name, setting| connection.execute("SET LOCAL #{name} = #{connection.quote(setting)}") }
+            end
+          end
+        end
+
+        # Every id the sequence hands out from now on exceeds this, at every
+        # priority, so a pass that reads it before its own snapshot bounds the
+        # passes that follow. That stops holding if the sequence caches blocks of
+        # ids per backend, and the catalog exposing the cache setting arrived in
+        # PostgreSQL 10 -- caching, a missing sequence, and an old server all
+        # degrade the same way: no floor, every pass unbounded.
+        def id_watermark
+          return unless connection.database_version >= 10_00_00
+          return unless id_sequence_cache == 1
+
+          connection.select_value("SELECT CASE WHEN is_called THEN last_value END FROM #{connection.quote_table_name(sequence_name)}")
+        end
+
+        def id_sequence_cache
+          connection.select_value(
+            sanitize_sql_array([ "SELECT seqcache FROM pg_sequence WHERE seqrelid = to_regclass(?)", sequence_name ])
+          )
+        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 +196,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)